patchodon/patchodon.py

#!/usr/bin/env python3

import sys, os
import argparse
import requests
import hashlib
import time

# NOTES: html2text: html2text
# the replies are listed by context, should be link-listed to avoid issues,
# should specify next hash to provide some kind of a filter
# visibility public+unlisted, all unlisted, all private, all direct

VERSION = "0.1.0"

DPASTE_URL = "https://dpaste.com"  # TODO any good way to parametrize this?

STATUS_LENGTH_LIMIT = 400  # TODO obtain from instance


def trace(x):
    sys.stderr.write(sys.argv[0] + ": " + x + "\n")


def api_token(args):
    if args.debug_api_token:
        return args.debug_api_token
    if args.env_api_token:
        return os.environ["PATCHODON_API_TOKEN"]
    raise ("API token not specified")


def do_post_status(args, body, parent=None, optional=None):
    if len(body) > STATUS_LENGTH_LIMIT:
        raise ("required status body too long")

    if not args.instance_url:
        raise ("mastodon instance not specified")

    token = api_token(args)

    headers = {"Authorization": f"Bearer {token}"}
    st = body + (
        "\n" + optional[0 : (STATUS_LENGTH_LIMIT - len(body) - 1)]
        if optional
        else ""
    )
    data = {"status": st, "visibility": "direct"}  # TODO parametrize direct
    if parent:
        data["in_reply_to_id"] = parent

    r = requests.post(
        args.instance_url + "/api/v1/statuses", data=data, headers=headers
    )

    if r.status_code != 200:
        raise ("mastodon status posting failed ({r.status_code})")

    rj = r.json()
    return (rj["id"], rj["url"])


def do_pastebin_file(file):
    # DPASTE API USE RULES:
    # - user-agent must be set properly
    # - 1 second between requests
    trace(f"sending `{file}' to dpaste...")
    r = requests.post(
        DPASTE_URL + "/api/v2/",
        data={
            "content": open(file, "r").read(),
            "syntax": "diff",
            "title": os.path.basename(file),
            "expiry_days": 1,  # TODO remove after testing
        },
        headers={"User-agent": f"patchodon v{VERSION}"},
    )
    time.sleep(1.1)
    if r.status_code != 201:
        raise (f"dpaste POST failed for `{file}'")
    return r.headers["location"] + ".txt"


def split_off_diff(s):
    return s.split("\ndiff --git ")[0]


def mapl(f, xs):
    return list(map(f, xs))


def mayline(s):
    if s:
        return s + "\n"
    else:
        return ""


def do_post(args):
    files = args.patchfile
    if not files:
        trace("reading patchfile series from stdin")
        files = mapl(lambda x: x.rstrip(chars="\n"), sys.stdin.readlines())
    n_patches = len(files)
    hashes = mapl(
        lambda x: hashlib.sha1(open(x, "r").read().encode()).hexdigest(), files
    )
    short_hashes = mapl(lambda x: x[0:8], hashes)
    full_hash = hashlib.sha1(" ".join(hashes).encode()).hexdigest()
    paste_raw_urls = mapl(do_pastebin_file, files)
    trace("posting the header...")
    parent_post_id, url = do_post_status(
        args,
        f"{mayline(args.recipient)}{mayline(args.subject)}"
        f"[patchodon hash {full_hash} / {' '.join(short_hashes)}]",
    )
    for fn, pst, hsh, series in zip(
        files, paste_raw_urls, hashes, range(n_patches)
    ):
        trace(f"posting patch {series+1}/{n_patches}...")
        parent_post_id, _ = do_post_status(
            args,
            f"{mayline(args.recipient)}"
            f"[patchodon {series+1}/{n_patches} {hsh}]\n"
            f"{pst}\n",
            parent=parent_post_id,
            optional=split_off_diff(open(fn, "r").read()),
        )
    print(url)


def do_get(args):
    # search & resolve the status ID on the instance (if configured)
    # get context, all replies from the same author as the original status ID
    # cycle the same for all hashes
    # verify the full hash
    # pass to git-am OR throw to a directory
    pass


def main():
    ap = argparse.ArgumentParser(
        prog=sys.argv[0],
        epilog="patchodon.py version " + VERSION + " is a free software.",
        description="Publicly send and receive git patch series via Mastodon.",
    )

    cmds = ap.add_subparsers(required=True, dest="command")

    if "API token sources":
        group = cmds.add_mutually_exclusive_group()
        group.add_argument(
            "--debug-api-token",
            help=(
                "specify the API token on command line (not very secure,"
                " good for debugging only)"
            ),
        )
        group.add_argument(
            "-e",
            "--env-api-token",
            action="store_true",
            help="get the API token from environment PATCHODON_API_TOKEN",
        )

    cmds.add_argument(
        "-i",
        "--instance-url",
        help=(
            "mastodon instance URL to use, such as `https://mastodon.example/'"
        ),
    )

    if "POST command":
        post = cmds.add_parser("post")
        post.add_argument(
            "-r",
            "--recipient",
            default=None,
            help=(
                "user tag to prepend to all posted statuses (required esp. for"
                " direct sending of statuses)"
            ),
        )
        post.add_argument(
            "-s",
            "--subject",
            default=None,
            help=(
                "opening text of the initial post, ideally used to specify the"
                " target project and patch topic"
            ),
        )
        post.add_argument(
            "patchfile",
            nargs="*",
            help=(
                "filenames of the patch series; taken from stdin if none are"
                " specified (useful for piping the output of git-format-patch"
                " into patchodon)"
            ),
        )

    if "GET command":
        get = cmds.add_parser("get")
        get.add_argument(
            "patch_url",
            help=(
                "root URL of the status where the patch was posted (the status"
                " should contain the patch hash)"
            ),
        )
        if "output possibilities":
            group = get.add_mutually_exclusive_group()
            group.add_argument(
                "-a",
                "--run-git-am",
                action="store_true",
                help=(
                    "apply the patches immediately with git-am instead of"
                    " storing them in a directory"
                ),
            )
            group.add_argument(
                "-C",
                "--out-prefix",
                default="./patchodon-",
                help=(
                    "write the numbered patchfiles to files with a given prefix"
                    " (the default `./patchodon-' will produce files like"
                    " `./patchodon-0001.patch')"
                ),
            )
            get.add_argument(
                "--overwrite",
                action="store_true",
                help="overwrite existing patch files instead of failing",
            )

    ap.add_argument(
        "-c",
        "--config",
        default=os.environ["HOME"] + "/.patchodon.ini",
        help=(
            "specify a custom config INI file that may specify a section"
            " [patchodon] with keys instance_url and api_token; defaults to"
            " `$HOME/.patchodon.ini', specify `/dev/null' to avoid config"
            " loading"
        ),
    )
    args = ap.parse_args()

    if args.command == "post":
        do_post(args)
    elif args.command == "get":
        do_get(args)
    else:
        raise ("fatal: args borked")


if __name__ == "__main__":
    main()