ISEP/hooks/lib/release_chain.py

"""release_chain — ISEP 運送鏈「tag → release」那兩站的**收貨端事實**，一份定義三處用
（inkstone/ISEP#67）

誰在用：
    scripts/release-ship        送貨：把已推的 tag 建成 release（前置條件從這裡問）
    scripts/release-check       覆核：逐版問 Gitea「到了嗎」
    hooks/lib/beacon_report.py  開場：SessionStart 報「打了 tag 卻沒 release」

🔴 為什麼收成一份：09-05 那版 release-ship／release-check 各自抄了一套「問 Gitea」
   的程式碼；要再讓 SessionStart 也問一次，就是第三份。同一個判準散在三處，
   修一份不會帶動另外兩份——`leo21c-write-guard.sh` 那一族就是這樣漂掉的
   （system-dev/wiki/mistakes.md 2026-09-07）。

🔴 讀一律匿名（D20：讀不帶憑證＝自由、不計次）。inkstone/ISEP 是公開 repo，
   tags／releases／某個 tag 底下的檔案／兩個 tag 之間的 commit 都不需要 token。
   只有 401／403 才退回帶 token 再試一次；token 只有 POST（建 release）才是必要的。

判準是結構不是關鍵字：某一版「送到了沒」＝它的 tag 與 release **在不在** Gitea 上。

離線：每個入口各自把 fixture dict 傳進來（鍵：tags／releases／manifest／commits），
      不打網路。fixture 的形狀在 `Chain.__init__` 的註解。
"""
import base64
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.request

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
try:
    from gitea_token import gitea_token as _gitea_token   # 唯一 token 來源（ISEP#47）：env→.env，不碰 remote URL
except Exception:      # 旁邊沒有 gitea_token.py（部分部署）也要能載入——讀 tag／release 一律匿名，不靠 token
    _gitea_token = None

DEFAULT_HOST = "https://git.uncle6.me"
DEFAULT_REPO = "inkstone/ISEP"
MANIFEST_PATH = ".claude-plugin/plugin.json"

# 運送鏈的三站：鍵、人話、怎麼補（補法的指令由呼叫端決定要不要換成絕對路徑）
STATIONS = [
    ("dingban", "定版 commit",
     "在 main 上 commit 一顆 `定版 v{v}`（＝ plugin.json 也改成那一版）"),
    ("tag", "tag（Gitea）",
     "打並推 tag：`git tag v{v} && git push origin v{v}`"),
    ("release", "release（Gitea）",
     "建 release：`{ship} v{v}`（這一站就是最常斷的那站）"),
]


class ChainError(Exception):
    """問不到收貨端（網路／HTTP 非 404）。呼叫端自己決定要死還是要閉嘴。"""


def vtuple(v):
    nums = re.findall(r"\d+", v or "")
    return tuple(int(x) for x in nums[:3]) if nums else (0,)


def tagname(v):
    v = (v or "").strip()
    return v if v.startswith("v") else "v" + v


def token_from_env_or_remote(host, root=None):
    """建 release（POST）才用得到。來源＝env／.env（ISEP#47），**不再從 remote URL 抽**——
    那會逼 token 明文嵌在 URL 裡，任何印出 remote 的訊息都會洩一次。解析順序見
    hooks/lib/gitea_token.py（先 env、再 $CLAUDE_PROJECT_DIR/.env、最後 git 根的 .env）。
    `host`／`root` 參數保留給呼叫端相容，現已不參與解析。拿不到回 ""（呼叫者自己判空）。"""
    if _gitea_token is not None:
        return _gitea_token() or ""
    # 退路：gitea_token.py 不在旁邊——只讀 env（仍不碰 remote URL，不重現 #47 的洩漏源）
    for env in ("GITEA_TOKEN_CLAUDE_CODE", "GITEA_TOKEN"):
        if os.environ.get(env):
            return os.environ[env]
    return ""


class Chain:
    """一個 repo 的運送鏈收貨端。

    fixture（離線用）的形狀：
        {"tags": ["v0.24.0", ...],
         "releases": ["v0.23.0", ...],
         "manifest": {"v0.24.0": "0.24.0", ...},     # 那個 tag 底下 plugin.json 的 version
         "commits": {"v0.23.0...v0.24.0": ["定版 v0.24.0（…）", "..."],   # 兩 tag 之間的 commit 標題
                     "v0.24.0": ["…"]}}                                     # 沒有上一個 release 時
    """

    def __init__(self, host=None, repo=None, fixture=None, timeout=25, token=None):
        self.host = (host or DEFAULT_HOST).rstrip("/")
        self.repo = repo or DEFAULT_REPO
        self.fx = fixture
        self.timeout = timeout
        self.token = token

    # ── HTTP：匿名優先 ───────────────────────────────────────────────────
    def _get(self, path, params=""):
        url = "%s/api/v1/repos/%s/%s%s" % (self.host, self.repo, path, params)
        headers = {"Cache-Control": "no-cache"}   # 09-05 實撞：邊緣快取的舊值被當 live
        for attempt in (0, 1):
            if attempt == 1:
                if not self.token:
                    raise ChainError("Gitea 對匿名讀回 401／403，而這裡沒有 token：%s" % url)
                headers["Authorization"] = "token " + self.token
            req = urllib.request.Request(url, headers=headers)
            try:
                with urllib.request.urlopen(req, timeout=self.timeout) as r:
                    return json.load(r)
            except urllib.error.HTTPError as e:
                if e.code == 404:
                    return None
                if e.code in (401, 403) and attempt == 0:
                    continue
                raise ChainError("Gitea %s：%s %s" % (e.code, url, e.read().decode()[:200]))
            except Exception as e:
                raise ChainError("連不上 %s：%s" % (url, e))
        return None

    def _names(self, path, key):
        names, page = set(), 1
        while True:
            rows = self._get(path, "?limit=50&page=%d" % page) or []
            for r in rows:
                n = r.get(key)
                if n:
                    names.add(n)
            if len(rows) < 50:
                return names
            page += 1

    # ── 收貨端事實 ───────────────────────────────────────────────────────
    def tags(self):
        if self.fx is not None:
            return set(self.fx.get("tags") or [])
        return self._names("tags", "name")

    def releases(self):
        if self.fx is not None:
            return set(self.fx.get("releases") or [])
        return self._names("releases", "tag_name")

    def release(self, tag):
        """這個 tag 的 release（dict），沒有就 None。"""
        if self.fx is not None:
            return {"tag_name": tag, "html_url": "(fixture)"} if tag in self.releases() else None
        return self._get("releases/tags/%s" % tag)

    def manifest_version_at(self, tag):
        """那個 tag 的樹裡 plugin.json 宣稱的版本（None ＝ tag 不在或檔案不在）。

        🔴 這是 release-ship 前置①的對象：**要建 release 的那個 tag 的樹**，
        不是「現在 checkout 出來的工作樹」——工作樹可能是任何分支，跟那個 tag 無關。
        """
        if self.fx is not None:
            return (self.fx.get("manifest") or {}).get(tag)
        d = self._get("contents/%s" % MANIFEST_PATH, "?ref=%s" % tag)
        if not d:
            return None
        try:
            raw = base64.b64decode(d.get("content") or "").decode("utf-8")
            return (json.loads(raw) or {}).get("version")
        except Exception:
            return None

    def commits_between(self, base, head):
        """base（可 None）到 head 之間的 commit 標題，**不含 merge commit**，舊在前。"""
        key = ("%s...%s" % (base, head)) if base else head
        if self.fx is not None:
            return list((self.fx.get("commits") or {}).get(key) or [])
        if base:
            d = self._get("compare/%s...%s" % (base, head)) or {}
            rows = d.get("commits") or []
        else:
            rows = self._get("commits", "?sha=%s&limit=50&stat=false&verification=false&files=false" % head) or []
            rows = list(reversed(rows))
        out = []
        for c in rows:
            if len(c.get("parents") or []) > 1:
                continue
            msg = ((c.get("commit") or {}).get("message") or "").strip()
            if msg:
                out.append(msg.splitlines()[0])
        return out

    def tag_commit_message(self, tag):
        """tag 指到的那顆 commit 的完整訊息（取 `定版 vX（人話）` 用）。"""
        if self.fx is not None:
            for lines in (self.fx.get("commits") or {}).values():
                for line in lines:
                    if "定版" in line and tag.lstrip("v") in line:
                        return line
            return ""
        t = self._get("tags/%s" % tag)
        sha = ((t or {}).get("commit") or {}).get("sha")
        if not sha:
            return ""
        c = self._get("git/commits/%s" % sha) or {}
        return ((c.get("commit") or {}).get("message") or "")

    # ── 判準 ─────────────────────────────────────────────────────────────
    def latest_release_before(self, ver, releases=None):
        rels = releases if releases is not None else self.releases()
        older = [t for t in rels if vtuple(t) < vtuple(ver)]
        return max(older, key=vtuple) if older else None


def gaps(ver, tags, releases, dingban=True):
    """某一版缺哪幾站 → [(key, label, fix_template)]，按運送順序。空 list ＝ 都在。"""
    t = tagname(ver)
    present = {"dingban": dingban, "tag": t in tags, "release": t in releases}
    return [(k, label, fix) for k, label, fix in STATIONS if not present[k]]


def dingban_title(message, ver):
    """從 `定版 vX.Y.Z（一句人話）` 抓括號那句——leo 既有 release 的命名慣例。"""
    m = re.search(r"定版\s+v?%s[（(]([^）)]+)[）)]" % re.escape(ver.lstrip("v")), message or "")
    return m.group(1).strip() if m else None