#!/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 auth_headers(args): if not args.instance_url: raise ("mastodon instance not specified") token = api_token(args) return {"Authorization": f"Bearer {token}"} def do_post_status(args, body, parent=None, optional=None): if len(body) > STATUS_LENGTH_LIMIT: raise ("required status body too long") headers = auth_headers(args) 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 find_head_post(args): headers = auth_headers(args) r = requests.get( args.instance_id + "/api/v2/search", params={"resolve": "true", "limit": "10", "q": args.patch_url}, ) print(r.__dict__) if r.status_code != 200: raise ("status URL search failed!") sts = filter(lambda x: x["url"] == args.patch_url, r.json()["statuses"]) if len(sts) < 1: raise ("status URL not found") if len(sts) > 1: raise ("ambiguous status URL?") st = sts[0] return (st["id"], st["account"]["id"], st["content"]) def do_get(args): st_id, st_acct_id, st_content = find_head_post(args) # parse out the hash and subhashes # get context, all replies from the same author as the original status ID, subhashes must match # repeat for all subhashes # verify the full hash # pass as one blob to git-am OR throw to a directory 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()