258 lines
11 KiB
Python
258 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""pr-verdict — 給 PR 一個結論，一個動作把該做的全做完（inkstone/ISEP#81）。
|
||
|
||
leo 的問題（票上原文）：「我交出去的 PR 沒有人去看它⋯⋯票看起來是『已交付』，
|
||
但東西沒有進 main，也就沒有進版本，**leo 手上永遠不會出現它**。」
|
||
|
||
「有結論」只有三種，缺一不可：
|
||
|
||
pr-verdict merge <owner/repo#N> 併 ＋ **當場刪掉那條 branch**
|
||
pr-verdict reject <owner/repo#N> --ticket <票> -F <檔> 退回：理由寫**票上** ＋ 關 PR
|
||
pr-verdict changes <owner/repo#N> --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 <owner/repo#N> [--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} <owner/repo#N> --ticket <owner/repo#N> -F <檔> [--dry-run]")
|
||
owner, repo, num = parse_ref(argv[0])
|
||
ticket = opt(argv, "--ticket")
|
||
if not ticket:
|
||
die("🚫 缺 --ticket <owner/repo#N>：結論要寫在**票**上。\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()
|