ISEP/scripts/pr-verdict

#!/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

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "hooks", "lib"))
from gitea_token import gitea_token as _gitea_token   # 唯一 token 來源（ISEP#47）：env→.env，不碰 remote URL


def die(msg, code=2):
    print(msg, file=sys.stderr)
    sys.exit(code)


def token():
    # 來源＝env／.env（ISEP#47），不碰 remote URL。解析順序見 hooks/lib/gitea_token.py。
    tok = _gitea_token()
    if tok:
        return tok
    host = HOST.split("//")[-1].rstrip("/")
    die(f"🔴 拿不到 {host} 的 token（沒有 GITEA_TOKEN_CLAUDE_CODE，.env 裡也沒有）")


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()