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"

STATUS_LENGTH_LIMIT = 400


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: "unlisted"}
    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})")

    return r.json.id


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


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


def do_post(args):
    files = args.patchfile
    if not files:
        trace("reading patchfile series from stdin")
        files = sys.stdin.readlines().map(lambda x: x.rstrip(chars="\n"))
    n_patches = len(files)
    hashes = files.map(lambda x: hashlib.sha1(open(x, "r").read()).hexdigest())
    short_hashes = hashes.map(lambda x: x[0:8])
    full_hash = hashlib.sha1(hashes.join(" ")).hexdigest()
    paste_raw_urls = map(do_pastebin_file, files)
    trace("posting the header...")
    parent_post_id, url = do_post_status(
        args,
        args.subject
        + f"\n[patchodon hash {full_hash} shorts {short_hashes.join(' ')}]",
    )
    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"[patchodon {series+1}/{n_patches} {hsh}\n{pst}\n",
            parent=parent_post_id,
            optional=split_off_diff(open(fn, "r").read()),
        )
    print(url)


def do_get(args):
    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 "POST command":
        post = cmds.add_parser("post")
        if "API token sources":
            group = post.add_mutually_exclusive_group()
            group.add_argument(
                "--debug-api-token",
                help=(
                    "specify the API token (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",
            )

        post.add_argument(
            "-i", "--instance-url", help="mastodon instance URL to post to"
        )
        post.add_argument(
            "-s",
            "--subject",
            default="",
            help=(
                "start of the initial post (e.g. to @tag someone and name the"
                " project and 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",
                "--am",
                help=(
                    "apply the patches immediately with git-am instead of"
                    " storing them in a directory"
                ),
            )
            group.add_argument(
                "-C",
                "--out-dir",
                help=(
                    "output the patches into a given directory (by default, `.'"
                ),
            )
            get.add_argument(
                "--overwrite",
                action="store_true",
                help="overwrite existing patch files instead of renaming",
            )

    ap.add_argument(
        "-c",
        "--config",
        help=(
            "specify a custom config INI file (it can specify a section"
            " [patchodon] with keys instance_url and api_token), defaults to"
            " `$HOME/.patchodon.ini'; specify /dev/null to avoid reading"
            " configs"
        ),
    )
    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()