#!/usr/bin/env python3 """Make GitHub and Forgejo match manifest.toml. python3.14 quine-repos.py # plan: report drift, change nothing python3.14 quine-repos.py --apply # carry out the plan python3.14 quine-repos.py --apply ncdu ava # only these repos Credentials: the Forgejo admin token in $FORGEJO_TOKEN or ~/.config/forgejo/token; GitHub admin calls go through the `gh` CLI's login; push mirrors use each bucket's mirror_token_file. Never deletes a repository. """ import argparse, json, os, re, subprocess, sys, time, urllib.error, urllib.request try: import tomllib except ImportError: sys.exit("needs Python 3.11+ (try python3.14)") HERE = os.path.dirname(os.path.abspath(__file__)) OK = (200, 201, 202, 204) # ── Forgejo ────────────────────────────────────────────────────────────────── class Forgejo: def __init__(self, url, token): self.api = url.rstrip("/") + "/api/v1" self.token = token def call(self, method, path, body=None, timeout=60): req = urllib.request.Request( self.api + path, method=method, data=None if body is None else json.dumps(body).encode(), headers={"Authorization": f"token {self.token}", "Content-Type": "application/json"}) for attempt in range(4): # the forge sometimes drops connections for a few seconds try: with urllib.request.urlopen(req, timeout=timeout) as r: status, raw = r.status, r.read() break except urllib.error.HTTPError as e: status, raw = e.code, e.read() break except (urllib.error.URLError, TimeoutError) as e: if attempt == 3 or method != "GET": # don't retry writes: they may have landed raise RuntimeError(f"Forgejo {method} {path}: {e}") from None time.sleep(5 * (attempt + 1)) try: data = json.loads(raw) if raw else None except ValueError: data = raw.decode(errors="replace")[:300] return status, data def must(self, method, path, body=None, timeout=60): status, data = self.call(method, path, body, timeout) if status not in OK: raise RuntimeError(f"Forgejo {method} {path}: HTTP {status}: {data}") return data def org_repos(self, org): """{name: repo} for an org, or None if the org doesn't exist.""" if self.call("GET", f"/orgs/{org}")[0] == 404: return None repos, page = {}, 1 while batch := self.must("GET", f"/orgs/{org}/repos?limit=50&page={page}"): repos.update({r["name"]: r for r in batch}) page += 1 return repos def push_mirrors(self, owner, name): return self.must("GET", f"/repos/{owner}/{name}/push_mirrors") or [] def sync_push_mirror(self, owner, name, url): """Trigger a push-mirror sync and wait for its result.""" def current(): return next(m for m in self.push_mirrors(owner, name) if same_url(m["remote_address"], url)) before = current().get("last_update") self.must("POST", f"/repos/{owner}/{name}/push_mirrors-sync") for _ in range(40): time.sleep(6) m = current() if m.get("last_error"): raise RuntimeError(f"push mirror sync failed: {m['last_error'][:500]}") if m.get("last_update") != before and not m["last_update"].startswith("0001"): return raise RuntimeError("push mirror sync gave no result after 4 minutes") # ── GitHub (via gh) ────────────────────────────────────────────────────────── def gh(*args, check=True): p = subprocess.run(["gh", "api", *args], capture_output=True, text=True) if check and p.returncode: raise RuntimeError(f"gh api {' '.join(args)}: {(p.stderr or p.stdout).strip()}") return p def gh_org_repos(org): """{name: repo} for an org, or None if the org doesn't exist.""" r = gh(f"orgs/{org}", check=False) if r.returncode: err = (r.stderr or r.stdout).strip() if "HTTP 404" in err: return None # A bad, expired or under-scoped token isn't a missing org. raise RuntimeError(f"GitHub: can't read org {org}: {err}") out = gh("--paginate", f"orgs/{org}/repos?per_page=100&type=all", "--jq", ".[] | {name, private, archived, fork, description}").stdout return {r["name"]: r for r in map(json.loads, out.splitlines())} def gh_token(): return subprocess.run(["gh", "auth", "token"], capture_output=True, text=True, check=True).stdout.strip() # ── helpers ────────────────────────────────────────────────────────────────── def same_url(a, b): # https://github.com/o/r, ssh://git@github.com/o/r and git@github.com:o/r # are the same repo (a push mirror may use SSH, with a deploy key). def norm(u): u = u.lower().rstrip("/").removesuffix(".git") u = u.split("://", 1)[-1].split("@", 1)[-1] return u.replace(":", "/", 1) if "/" not in u.split(":", 1)[0] and ":" in u else u return norm(a) == norm(b) def gh_url(org, name): return f"https://github.com/{org}/{name}.git" def read_token(path): path = os.path.expanduser(path) return open(path).read().strip() if os.path.exists(path) else None class Plan: def __init__(self, name): self.name, self.steps, self.notes = name, [], [] def step(self, desc, fn): self.steps.append((desc, fn)) def note(self, msg): self.notes.append(msg) # ── planning ───────────────────────────────────────────────────────────────── class World: """Current state of every bucket org on both hosts.""" def __init__(self, manifest, fj): self.buckets = manifest["buckets"] self.fj = fj self.gh = {b["github"]: gh_org_repos(b["github"]) for b in self.buckets.values()} self.fjo = {b["forgejo"]: fj.org_repos(b["forgejo"]) for b in self.buckets.values()} self.missing_fj_orgs = set() @staticmethod def _find(orgs, name): hits = [(org, repos[name]) for org, repos in orgs.items() if repos and name in repos] return hits def find_gh(self, name): return self._find(self.gh, name) def find_fj(self, name): return self._find(self.fjo, name) def plan_repo(name, spec, world, fj, cfg): p = Plan(name) bucket = world.buckets[spec["bucket"]] primary = spec.get("primary", "forgejo") upstream = spec.get("upstream") # Forgejo pull-mirrors this, then push-mirrors to GitHub on_github = spec.get("github", True) explicit_archived = spec.get("archived", bucket.get("archived")) # None = not stated archived = bool(explicit_archived) gh_org, fj_org = bucket["github"], bucket["forgejo"] want_url = gh_url(gh_org, name) # GitHub: location and archive state ------------------------------------ gh_repo = None if on_github: hits = world.find_gh(name) if len(hits) > 1: p.note(f"on GitHub in several bucket orgs: {', '.join(o for o, _ in hits)}; fix by hand") return p if not hits: p.note(f"not found on GitHub in any bucket org (create it there, or set github = false)") return p cur_org, gh_repo = hits[0] if cur_org != gh_org and world.gh[gh_org] is None: p.note(f"GitHub org {gh_org} doesn't exist; create it at https://github.com/organizations/plan") return p if gh_repo["archived"] and explicit_archived is None: # Archived on GitHub but not in the manifest: don't undo it, ask for the manifest to catch up. p.note(f"archived on GitHub but the manifest doesn't say; set archived = true to keep it, or false to unarchive") return p if gh_repo["archived"] and (cur_org != gh_org or not archived): # archived repos can't be changed much; unarchive first, re-archive at the end if wanted p.step(f"GitHub: unarchive {cur_org}/{name}", lambda o=cur_org: gh("-X", "PATCH", f"repos/{o}/{name}", "-F", "archived=false")) gh_repo = {**gh_repo, "archived": False} if cur_org != gh_org: p.step(f"GitHub: transfer {cur_org}/{name} → {gh_org}/{name}", lambda o=cur_org: (gh("-X", "POST", f"repos/{o}/{name}/transfer", "-f", f"new_owner={gh_org}"), wait_gh(gh_org, name))) # Forgejo: location, existence, kind ------------------------------------- hits = world.find_fj(name) if len(hits) > 1: p.note(f"on Forgejo in several bucket orgs: {', '.join(o for o, _ in hits)}; fix by hand") return p fj_repo = hits[0][1] if hits else None if world.fjo[fj_org] is None: world.missing_fj_orgs.add(fj_org) # created once, before any repo steps if fj_repo is None: if not on_github and not upstream: p.note("Forgejo-only repo is missing from Forgejo; nothing to import from") return p how = f"pull mirror of {upstream}" if upstream else \ f"from GitHub ({'pull mirror' if primary == 'github' else spec.get('import', 'full')})" p.step(f"Forgejo: import {fj_org}/{name} {how}", lambda: import_repo(fj, name, spec, primary, gh_org, fj_org, gh_repo, cfg)) fj_repo = {"mirror": primary == "github" or bool(upstream), "archived": False, "private": bool(gh_repo and gh_repo["private"]), "_new": True} else: cur = fj_repo["owner"]["login"] if fj_repo["archived"] and (cur.lower() != fj_org.lower() or not archived): p.step(f"Forgejo: unarchive {cur}/{name}", lambda c=cur: fj.must("PATCH", f"/repos/{c}/{name}", {"archived": False})) if cur.lower() != fj_org.lower(): p.step(f"Forgejo: transfer {cur}/{name} → {fj_org}/{name}", lambda c=cur: fj.must("POST", f"/repos/{c}/{name}/transfer", {"new_owner": fj_org})) if primary == "github" and not fj_repo["mirror"]: p.note("should be a pull mirror of GitHub but is a regular repo; recreate it by hand (not automated: needs a delete)") if upstream and not fj_repo["mirror"]: p.note(f"should be a pull mirror of {upstream} but is a regular repo; recreate it by hand (not automated: needs a delete)") if upstream and fj_repo["mirror"] and not same_url(fj_repo.get("original_url", ""), upstream): p.note(f"pull mirror points at {fj_repo.get('original_url')}; change it to {upstream} in Settings → Mirror") if primary == "forgejo" and not upstream and fj_repo["mirror"]: p.note("is a pull mirror but Forgejo should be primary; use Settings → Convert to regular repository") if primary == "github" and fj_repo["mirror"] and on_github and not same_url(fj_repo.get("original_url", ""), want_url): p.note(f"pull mirror points at {fj_repo.get('original_url')}; change it to {want_url} in Settings → Mirror") if on_github and fj_repo["private"] != gh_repo["private"]: p.step(f"Forgejo: make {'private' if gh_repo['private'] else 'public'} to match GitHub", lambda: fj.must("PATCH", f"/repos/{fj_org}/{name}", {"private": gh_repo["private"]})) # Actions ----------------------------------------------------------------- # Upstream mirrors carry upstream's own CI (.github/workflows), which Forgejo # would start on every sync and leave "waiting" forever for runners we don't # have. So their Actions are off unless the manifest says otherwise. want_actions = spec.get("actions", False if upstream else None) # None = leave it if want_actions is not None and (fj_repo.get("_new") or fj_repo.get("has_actions") != want_actions): p.step(f"Forgejo: turn Actions {'on' if want_actions else 'off'}", lambda: fj.must("PATCH", f"/repos/{fj_org}/{name}", {"has_actions": want_actions})) # Description notes ------------------------------------------------------- # Forgejo says where a fork came from; GitHub (when Forgejo is primary) points at Forgejo. src = spec.get("forked_from") fj_note = ("forked from", src if "://" in src else f"https://github.com/{src}") if src else None # GitHub is public: point at the forge's public view, not its tailnet address. gh_note = ("mirror of", f"{cfg['public_url'].rstrip('/')}/{fj_org}/{name}.git") if primary == "forgejo" else None # The text before the note should match on both hosts: the manifest's # `description` if it has one, else an empty side borrows the other's. desc = spec.get("description") gh_base = desc or (described(gh_repo.get("description"), None) if on_github else "") fj_base = desc or described(fj_repo.get("description"), None) if on_github and (want := described(gh_base or fj_base, gh_note)) != (gh_repo.get("description") or ""): p.step(f"GitHub: description → {want!r}", lambda want=want: gh("-X", "PATCH", f"repos/{gh_org}/{name}", "-f", f"description={want}")) if fj_repo.get("_new"): p.step("Forgejo: set description note", lambda: set_fj_description(fj, fj_org, name, fj_note, gh_base, desc)) elif (want := described(fj_base or gh_base, fj_note)) != (fj_repo.get("description") or ""): p.step(f"Forgejo: description → {want!r}", lambda: set_fj_description(fj, fj_org, name, fj_note, gh_base, desc)) # Push mirrors (Forgejo primary) ----------------------------------------- if on_github and primary == "forgejo": mirrors = [] if fj_repo.get("_new") or (fj_repo["mirror"] and not upstream) else \ fj.push_mirrors(fj_repo["owner"]["login"], name) keep = [m for m in mirrors if same_url(m["remote_address"], want_url)] stale = [m for m in mirrors if "github.com" in m["remote_address"] and m not in keep] if archived and keep and not gh_repo["archived"]: p.step("Forgejo: final push-mirror sync before archiving", lambda: fj.sync_push_mirror(fj_org, name, want_url)) for m in stale + (keep if archived else []): p.step(f"Forgejo: remove push mirror → {m['remote_address']}", lambda m=m: fj.must("DELETE", f"/repos/{fj_org}/{name}/push_mirrors/{m['remote_name']}")) if not archived and not keep and (upstream or not fj_repo["mirror"]): token_file = bucket.get("mirror_token_file") if not token_file or not read_token(token_file): p.note(f"no GitHub mirror token for bucket {spec['bucket']} ({token_file or 'mirror_token_file unset'}); can't add push mirror") else: p.step(f"Forgejo: add push mirror → {want_url} (sync on commit + every {cfg['interval']}) and sync", lambda: add_push_mirror(fj, fj_org, name, want_url, read_token(token_file), cfg["interval"], gh_org)) # Renovate ----------------------------------------------------------------- plan_renovate(p, name, spec, fj_org, fj_repo, archived, fj, cfg) # Archive state ----------------------------------------------------------- if on_github and archived and not gh_repo["archived"]: p.step(f"GitHub: archive {gh_org}/{name}", lambda: gh("-X", "PATCH", f"repos/{gh_org}/{name}", "-F", "archived=true")) if archived and not fj_repo["archived"]: p.step(f"Forgejo: archive {fj_org}/{name}", lambda: fj.must("PATCH", f"/repos/{fj_org}/{name}", {"archived": True})) return p # Every event that can mean "look again": pushes (a merged update, a ticked # rebase box), PRs and their comments, and issues (the Dependency Dashboard). RENOVATE_EVENTS = [ "push", "issues", "issue_assign", "issue_label", "issue_milestone", "issue_comment", "pull_request", "pull_request_assign", "pull_request_label", "pull_request_milestone", "pull_request_comment", "pull_request_review_approved", "pull_request_review_rejected", "pull_request_review_comment", "pull_request_sync", "pull_request_review_request", ] def renovate_wanted(spec, cfg): rcfg = cfg["renovate"] if "renovate" in spec: return spec["renovate"] return (spec["bucket"] in rcfg.get("buckets", []) and not spec.get("forked_from") and not spec.get("upstream") and spec.get("primary", "forgejo") == "forgejo") def plan_renovate(p, name, spec, fj_org, fj_repo, archived, fj, cfg): """Renovate's access and webhook, for the repos it covers. Mirrors and archived repos can't take its PRs. Turning it off is reported, not undone.""" rcfg = cfg["renovate"] if not rcfg: return user, url = rcfg["user"], rcfg["webhook"] new = fj_repo.get("_new") has_user = has_hook = False if not new: # Where it is now: a repo moving buckets is transferred by an earlier # step, and its collaborators and webhooks go with it. cur = fj_repo["owner"]["login"] perm = fj.call("GET", f"/repos/{cur}/{name}/collaborators/{user}/permission") has_user = perm[0] == 200 and perm[1].get("permission") in ("write", "admin", "owner") has_hook = any(h.get("config", {}).get("url") == url for h in fj.must("GET", f"/repos/{cur}/{name}/hooks") or []) if not renovate_wanted(spec, cfg): if has_user or has_hook: p.note(f"renovate is off for it, but {user} still has access or its webhook; remove them by hand if meant") return if archived or fj_repo.get("mirror"): return if not has_user: p.step(f"Forgejo: give {user} write access (Renovate)", lambda: fj.must("PUT", f"/repos/{fj_org}/{name}/collaborators/{user}", {"permission": "write"})) if not has_hook: p.step(f"Forgejo: add Renovate's webhook → {url}", lambda: fj.must("POST", f"/repos/{fj_org}/{name}/hooks", { "type": "forgejo", "active": True, "branch_filter": "*", "events": RENOVATE_EVENTS, "config": {"url": url, "content_type": "json", "http_method": "post"}})) # ── actions ────────────────────────────────────────────────────────────────── NOTE_RE = re.compile(r"\s*\((?:forked from|mirror of) https?://[^)\s]+\)|^(?:forked from|mirror of) https?://\S+$", re.I) def described(desc, note): """desc with any of our notes stripped, then `note` = (label, url) appended.""" base = NOTE_RE.sub("", desc or "").strip() if not note: return base label, url = note label = label.capitalize() return f"{base} ({label} {url})" if base else f"{label} {url}" def set_fj_description(fj, org, name, note, fallback="", base=None): desc = fj.must("GET", f"/repos/{org}/{name}").get("description") or "" if (want := described(base or described(desc, None) or fallback, note)) != desc: fj.must("PATCH", f"/repos/{org}/{name}", {"description": want}) def wait_gh(org, name): for _ in range(20): if not gh(f"repos/{org}/{name}", check=False).returncode: return time.sleep(3) raise RuntimeError(f"GitHub transfer to {org}/{name} didn't complete within a minute") def import_repo(fj, name, spec, primary, gh_org, fj_org, gh_repo, cfg): url = gh_url(gh_org, name) # A Forgejo-only upstream mirror has no GitHub copy to take privacy from. body = {"clone_addr": url, "repo_owner": fj_org, "repo_name": name, "private": bool(gh_repo and gh_repo["private"])} if spec.get("upstream"): body |= {"clone_addr": spec["upstream"], "service": "git", "mirror": True, "mirror_interval": cfg["interval"]} elif primary == "github": body |= {"service": "git", "mirror": True, "mirror_interval": cfg["interval"]} if gh_repo["private"]: # stored for future syncs, so use the bucket's narrow token token = read_token(cfg["bucket"].get("mirror_token_file", "")) if not token: raise RuntimeError("private pull mirror needs the bucket's mirror_token_file") body |= {"auth_username": "x-access-token", "auth_password": token} elif spec.get("import", "full") == "code": body |= {"service": "git", "mirror": False, "auth_username": "x-access-token", "auth_password": gh_token()} else: body |= {"service": "github", "mirror": False, "auth_token": gh_token(), "issues": True, "pull_requests": True, "releases": True, "labels": True, "milestones": True, "wiki": True} status, data = fj.call("POST", "/repos/migrate", body, timeout=1800) if status == 504: raise RuntimeError("import hit the proxy timeout; Forgejo leaves the repo stuck 'migrating'. " "Delete it in Forgejo, set import = \"code\" (or import via the web UI), and re-run") if status not in OK: raise RuntimeError(f"import failed: HTTP {status}: {data}") def check_github_not_ahead(fj, org, name, gh_org): """Refuse to mirror over GitHub commits that Forgejo doesn't have (pushed after the import).""" p = gh("--paginate", f"repos/{gh_org}/{name}/git/matching-refs/", "--jq", '.[] | select(.ref | startswith("refs/heads/") or startswith("refs/tags/")) | "\\(.ref) \\(.object.sha)"', check=False) if p.returncode: if "is empty" in p.stdout + p.stderr: # empty GitHub repo: nothing to lose return raise RuntimeError(f"couldn't list GitHub refs: {(p.stderr or p.stdout).strip()}") out = p.stdout missing = [ref for ref, sha in (line.split() for line in out.splitlines()) if fj.call("GET", f"/repos/{org}/{name}/git/commits/{sha}?stat=false")[0] == 404 and fj.call("GET", f"/repos/{org}/{name}/git/tags/{sha}")[0] == 404] if missing: raise RuntimeError( f"GitHub has commits Forgejo lacks on {', '.join(missing[:5])}; the first mirror sync would " f"overwrite them. Pull them into Forgejo first (fetch from GitHub, push to Forgejo), then re-run") def add_push_mirror(fj, org, name, url, token, interval, gh_org): check_github_not_ahead(fj, org, name, gh_org) fj.must("POST", f"/repos/{org}/{name}/push_mirrors", { "remote_address": url, "remote_username": "oauth2", "remote_password": token, "interval": interval, "sync_on_commit": True}) fj.sync_push_mirror(org, name, url) # ── main ───────────────────────────────────────────────────────────────────── def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("repos", nargs="*", help="limit to these repos") ap.add_argument("--apply", action="store_true", help="make the changes (default: plan only)") ap.add_argument("--manifest", default=os.path.join(HERE, "manifest.toml")) args = ap.parse_args() with open(args.manifest, "rb") as f: manifest = tomllib.load(f) repos = manifest["repos"] for name, spec in repos.items(): if spec.get("bucket") not in manifest["buckets"]: sys.exit(f"{name}: unknown bucket {spec.get('bucket')!r}") unknown = [r for r in args.repos if r not in repos] if unknown: sys.exit(f"not in manifest: {', '.join(unknown)}") token = os.environ.get("FORGEJO_TOKEN") or read_token("~/.config/forgejo/token") if not token: sys.exit("no Forgejo token ($FORGEJO_TOKEN or ~/.config/forgejo/token)") fj = Forgejo(manifest["forgejo_url"], token) world = World(manifest, fj) # Every repo in a bucket org must be in the manifest; refuse to touch anything otherwise. unaccounted = [] for host, orgs in (("GitHub", world.gh), ("Forgejo", world.fjo)): for org, found in orgs.items(): for extra in sorted(set(found or {}) - set(repos)): hint = " (archived)" if found[extra]["archived"] else "" unaccounted.append(f"{host} {org}/{extra}{hint}") if unaccounted: print("\n".join(["✗ REPOS NOT IN THE MANIFEST — add them to manifest.toml before running:", *(f" {u}" for u in unaccounted)]), file=sys.stderr) sys.exit(2) interval = manifest.get("mirror_interval", "8h0m0s") plans = [] for name in args.repos or repos: spec = repos[name] cfg = {"interval": interval, "bucket": manifest["buckets"][spec["bucket"]], "forgejo_url": manifest["forgejo_url"], "public_url": manifest["public_url"], "renovate": manifest.get("renovate")} plans.append(plan_repo(name, spec, world, fj, cfg)) changed = failed = 0 blocked_orgs = set() for org in sorted(world.missing_fj_orgs): desc = f"Forgejo: create org {org}" if not args.apply: print(f"+ {desc}") continue try: fj.must("POST", "/orgs", {"username": org, "visibility": "public"}) print(f"✓ {desc}") changed += 1 except Exception as e: print(f"✗ {desc}\n {e}") failed += 1 blocked_orgs.add(org) for p in plans: if blocked_orgs and repos[p.name]["bucket"] in { b for b, v in manifest["buckets"].items() if v["forgejo"] in blocked_orgs}: print(f"✗ {p.name}: skipped, its Forgejo org couldn't be created") continue if not p.steps and not p.notes: print(f"✓ {p.name}") continue print(f"• {p.name}") for msg in p.notes: print(f" ! {msg}") for desc, fn in p.steps: if not args.apply: print(f" + {desc}") continue try: fn() print(f" ✓ {desc}", flush=True) changed += 1 except Exception as e: # stop this repo, carry on with the rest print(f" ✗ {desc}\n {e}", flush=True) failed += 1 break for host, orgs in (("GitHub", world.gh), ("Forgejo", world.fjo)): for org, found in orgs.items(): if found is None: print(f"? {host} org {org} doesn't exist") pending = sum(len(p.steps) for p in plans) + len(world.missing_fj_orgs) if args.apply: print(f"\n{changed} change(s) made, {failed} failed.") else: print(f"\n{pending} change(s) planned. Run with --apply to make them.") sys.exit(1 if failed else 0) if __name__ == "__main__": main()