#!/usr/bin/env python3 """pr-verdict — 給 PR 一個結論，一個動作把該做的全做完（inkstone/ISEP#81）。 leo 的問題（票上原文）：「我交出去的 PR 沒有人去看它⋯⋯票看起來是『已交付』， 但東西沒有進 main，也就沒有進版本，**leo 手上永遠不會出現它**。」 「有結論」只有三種，缺一不可： pr-verdict merge 併 ＋ **當場刪掉那條 branch** pr-verdict reject --ticket <票> -F <檔> 退回：理由寫**票上** ＋ 關 PR pr-verdict changes --ticket <票> -F <檔> 要求修改：寫**票上** ＋ REQUEST_CHANGES pr-verdict list 撈一次：誰還沒有結論 🔴 為什麼要有這支、而不是叫人記得多做一步： 「merge 完要刪 branch」「退回理由要寫在票上」都是**第二個動作**， 而所有掉棒的形狀都長一樣——第一個動作做了，第二個沒有。 把兩件事綁成一個指令，就沒有「忘了做第二件」這個狀態可以存在。 🔴 `--dry-run` 會把要打的每一通 API 印出來但不送出。 加在任何子指令上都有效；**收工前想確認自己按下去會發生什麼，用它。** """ import json import os import re import subprocess import sys import urllib.error import urllib.parse import urllib.request HOST = os.environ.get("PR_VERDICT_HOST", "https://git.uncle6.me") DRY = False def die(msg, code=2): print(msg, file=sys.stderr) sys.exit(code) def token(): root = os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd() host = HOST.split("//")[-1].rstrip("/") try: out = subprocess.run(["git", "-C", root, "remote", "-v"], capture_output=True, text=True, timeout=20).stdout except Exception: out = "" for line in out.splitlines(): if host in line: m = re.search(r"//[^:/]+:([^@]+)@", line) if m: return m.group(1) for env in ("GITEA_TOKEN_CLAUDE_CODE", "GITEA_TOKEN"): if os.environ.get(env): return os.environ[env] die(f"🔴 拿不到 {host} 的 token（remote 沒帶憑證，也沒有 GITEA_TOKEN_CLAUDE_CODE）") def api(path, payload=None, method=None, soft=False): verb = method or ("POST" if payload is not None else "GET") if DRY and verb != "GET": print(f" [dry-run] {verb} {path}" + (f"\n {json.dumps(payload, ensure_ascii=False)}" if payload else "")) return {} url = path if path.startswith("http") else f"{HOST}/api/v1{path}" data = json.dumps(payload).encode() if payload is not None else None req = urllib.request.Request( url, data=data, method=verb, headers={"Authorization": f"token {token()}", "Content-Type": "application/json"}) try: raw = urllib.request.urlopen(req, timeout=40).read() return json.loads(raw) if raw.strip() else {} except urllib.error.HTTPError as e: if soft: return {"__status__": e.code} die(f"🔴 Gitea {e.code} on {verb} {path}：{e.read().decode()[:300]}") def parse_ref(s): m = re.match(r"^([\w.-]+)/([\w.-]+)#(\d+)$", s.strip()) if not m: die(f"🔴 寫法是 owner/repo#N，你給的是：{s}") return m.group(1), m.group(2), int(m.group(3)) def read_body(argv, what): if "-F" not in argv: die(f"🚫 缺 -F <檔>：{what}要**寫在票上**，不是寫在訊息裡。\n" f" （inkstone/ISEP#81：「退回理由寫在票上（寫票，不是寫訊息）」）") p = argv[argv.index("-F") + 1] try: with open(p) as f: return f.read().strip() except Exception as e: die(f"🔴 讀不到 {p}：{e}") def opt(argv, name, default=None): return argv[argv.index(name) + 1] if name in argv else default def say_on_ticket(ref, body): """把結論貼進**票**（不是貼進 PR 的討論串——PR 會隨自己沉掉，票才是那條線的家）。 內文第一行帶【總管】身份欄，對齊 reply-identity-guard 的規約 （leo 2026-08-27：「subagent 回覆時要表明身份」）。 """ owner, repo, num = parse_ref(ref) if DRY: print(f" [dry-run] POST /repos/{owner}/{repo}/issues/{num}/comments") print(f" {body[:120]}…") return api(f"/repos/{owner}/{repo}/issues/{num}/comments", {"body": body}) # ── merge ──────────────────────────────────────────────────────────────── def cmd_merge(argv): if not argv: die("用法：pr-verdict merge [--style merge|squash|rebase] [--dry-run]") owner, repo, num = parse_ref(argv[0]) style = opt(argv, "--style", "merge") pr = api(f"/repos/{owner}/{repo}/pulls/{num}") branch = ((pr.get("head") or {}).get("ref")) or "" same_repo = ((pr.get("head") or {}).get("repo") or {}).get("full_name") == f"{owner}/{repo}" if not pr.get("merged"): if pr.get("state") != "open": die(f"🔴 {owner}/{repo}#{num} 不是 open（state={pr.get('state')}）——沒得 merge。") api(f"/repos/{owner}/{repo}/pulls/{num}/merge", {"Do": style, "delete_branch_after_merge": True}) if not DRY: print(f"✅ {owner}/{repo}#{num} 已 merge（{style}）") else: print(f"ℹ️ {owner}/{repo}#{num} 本來就是 merged，只補刪分支") # 🔴 第二件事：**當場複驗分支真的不見了**。 # Gitea 的 delete_branch_after_merge 在某些設定下不會生效（保護分支、 # 跨 fork），而「我下了刪除的指令」不等於「它被刪掉了」—— # 本 repo 心法：交出去之前，我知不知道它能不能用。 if not branch: print("⚠️ 拿不到 head 分支名，無法複驗——請自己確認分支已刪。") return if not same_repo: print(f"ℹ️ head 在別的 repo（fork），本指令不跨 repo 刪分支：{branch}") return if DRY: print(f" [dry-run] DELETE /repos/{owner}/{repo}/branches/{branch}（若還在）") return if _branch_alive(owner, repo, branch): api(f"/repos/{owner}/{repo}/branches/{urllib.parse.quote(branch, safe='')}", method="DELETE", soft=True) if _branch_alive(owner, repo, branch): die(f"🔴 分支 {branch} 刪不掉（保護分支？權限？）——**結論只給了一半**，" f"pr-verdict-guard 會繼續點名它。") print(f"✅ 分支 {branch} 已刪除（複驗過：查不到了）") def _branch_alive(owner, repo, br): r = api(f"/repos/{owner}/{repo}/branches/{urllib.parse.quote(br, safe='')}", soft=True) return isinstance(r, dict) and "name" in r # ── reject / changes ───────────────────────────────────────────────────── def _verdict_with_reason(argv, kind): if not argv: die(f"用法：pr-verdict {kind} --ticket -F <檔> [--dry-run]") owner, repo, num = parse_ref(argv[0]) ticket = opt(argv, "--ticket") if not ticket: die("🚫 缺 --ticket ：結論要寫在**票**上。\n" " PR 的討論串會隨 PR 一起沉掉；票才是那條線的家。") body = read_body(argv, "退回理由" if kind == "reject" else "修改要求") pr = api(f"/repos/{owner}/{repo}/pulls/{num}") head = f"{owner}/{repo}#{num}" url = (pr.get("html_url") or "") title = "退回" if kind == "reject" else "要求修改" say_on_ticket(ticket, f"【總管】**PR {title}**：{head}　{url}\n\n{body}") if not DRY: print(f"✅ {title}理由已寫進票 {ticket}") if kind == "reject": api(f"/repos/{owner}/{repo}/issues/{num}", {"state": "closed"}, method="PATCH") if not DRY: print(f"✅ {head} 已關閉") br = ((pr.get("head") or {}).get("ref")) or "" same = ((pr.get("head") or {}).get("repo") or {}).get("full_name") == f"{owner}/{repo}" if br and same and not DRY and _branch_alive(owner, repo, br): api(f"/repos/{owner}/{repo}/branches/{urllib.parse.quote(br, safe='')}", method="DELETE", soft=True) print(f"✅ 分支 {br} 已刪除") else: # REQUEST_CHANGES：**讓「已經給過結論」變成 Gitea 查得到的事實**， # 這樣 pr-verdict-guard 就不會再把它當成被遺忘的 PR（＝擋等待）。 api(f"/repos/{owner}/{repo}/pulls/{num}/reviews", {"event": "REQUEST_CHANGES", "body": f"修改要求寫在 {ticket}"}) if not DRY: print(f"✅ {head} 已留下 REQUEST_CHANGES review（閘看得到，不會再點名它）") if DRY: print("\n[dry-run] 以上一通都沒有送出。") return print("\n📌 兩件事是同一個動作——不會只寫了理由而 PR 還開著" "（reject），也不會留了 review 而票上查不到理由（changes）。") # ── list ───────────────────────────────────────────────────────────────── def cmd_list(argv): orgs = [o.strip().lower() for o in os.environ.get("PR_VERDICT_ORGS", "inkstone").split(",") if o.strip()] rows = api("/repos/issues/search?type=pulls&state=open&limit=50") import time now = time.time() n = 0 for pr in rows: full = (pr.get("repository") or {}).get("full_name") or "" if "/" not in full or (orgs and full.split("/", 1)[0].lower() not in orgs): continue try: t = time.mktime(time.strptime(pr["created_at"][:19], "%Y-%m-%dT%H:%M:%S")) age = int((now - t) / 86400) except Exception: age = -1 who = [a["login"] for a in (pr.get("assignees") or [])] mark = "🕒" if who else "❌" print(f"{mark} {full}#{pr['number']:<5} 開了 {age:>3} 天 " f"{'指派＝' + ','.join(who) if who else '沒有人被指派'} {pr['title'][:34]}") n += 1 print(f"\n共 {n} 個 open PR（org：{'/'.join(orgs) or '全部'}）。" f"　❌＝沒有人在管，pr-verdict-guard 會點名它") def main(): global DRY argv = sys.argv[1:] if "--dry-run" in argv: DRY = True argv.remove("--dry-run") if not argv: print(__doc__) sys.exit(0) cmd, rest = argv[0], argv[1:] if cmd == "merge": cmd_merge(rest) elif cmd == "reject": _verdict_with_reason(rest, "reject") elif cmd == "changes": _verdict_with_reason(rest, "changes") elif cmd == "list": cmd_list(rest) else: print(__doc__) die(f"🔴 沒有這個子指令：{cmd}") if __name__ == "__main__": main()