ISEP/scripts/mainline

#!/usr/bin/env python3
"""mainline — 「現在的主線是哪一個」這個問題的**唯一指令**（inkstone/ISEP#82）。

leo 2026-08-27：「**14 個里程碑同時亮著卻不知道該看哪個**」。
當天實查 `GET /repos/inkstone/*/milestones?state=open` 回 14 個，其中「Mira 現代化」
這個名字同時活在 5 個 repo，另有 5 個已經逾期。
⇒ SOP 說的「**那個** active milestone」在現場根本沒有指涉對象——
  要注入不知道注哪一個，要擋跳線也不知道拿哪一個當基準。

這支不解「milestone 太多」（跨 repo 同名是既有做法，不動它），
它解的是「**這一刻我在做哪一條**」沒有答案：

    mainline                      現在的主線是哪一個（＝ show，唯一答案，離線）
    mainline list                 有哪些 open milestone 可以選（打網路）
    mainline set <owner/repo#id>  把某一個 milestone 標成主線（打網路，寫進 state）
    mainline clear                現在沒有主線（例：一條線收掉了，下一條還沒開始）
    mainline refresh [--members]  更新進度／期限（SessionStart 跑一次，不輪詢）
    mainline adopt <owner/repo#N> 「補收」：把一張票掛進主線（打網路）
    mainline has <owner/repo#N>   這張票在不在主線上（exit 0 ＝在，1 ＝不在）

🔴 **`<owner/repo#id>` 的 id 是 milestone 的 id，不是 issue 的號碼。**
   兩者在 Gitea 是兩套獨立編號，`mainline list` 印出來的就是可以直接貼的那一個。

── 答案為什麼唯一 ─────────────────────────────────────────────────
主線住在一個檔裡（`~/.claude/isep-countdown/mainline.json`），那個檔**放得下一條**。
（inkstone/ISEP#130 起它有一份隨 repo 走的鏡像：`InkStoneCo/system-dev/mainline.json`——
 雲端沒有家目錄那份，讀的就是它；`set`／`adopt`／`clear` 兩份一起寫，`refresh` 只寫家目錄。）
沒有清單、沒有優先序、沒有「前三名」——`set` 覆蓋掉的那條就不再是主線。
⇒ 問一次只會拿到 0 或 1 個答案，不會拿到 14 個。

── 主線是一個名字，錨在一個具體的 milestone 上 ────────────────────
Gitea 的 milestone 不能跨 repo，而同一條線常常同時開在 5 個 repo。所以
**錨**（期限／進度／目標宣告都讀它）是 `owner/repo#id`，
**成員判定**看的是「票所屬 milestone 的標題與主線標題相同」（不分 repo）。
兩件事都成立：同名跨 repo 照舊能用，而「現在在做哪一條」仍然只有一個答案。
"""
import importlib.machinery
import importlib.util
import os
import sys
import urllib.parse
from datetime import datetime, timezone

HERE = os.path.dirname(os.path.abspath(__file__))


def _load(name, path):
    """把一個檔載進來當模組。`scripts/ticket` 沒有 `.py` 副檔名，所以要明著給 loader
    （`spec_from_file_location` 只靠副檔名認不出它，會回 None）。"""
    loader = importlib.machinery.SourceFileLoader(name, path)
    spec = importlib.util.spec_from_file_location(name, path, loader=loader)
    mod = importlib.util.module_from_spec(spec)
    loader.exec_module(mod)
    return mod


# 憑證怎麼拿、API 怎麼打、票號怎麼解析——**全部沿用 `scripts/ticket` 那一份**。
# 🔴 刻意不自己再寫一份 token()：金鑰的取得只准有一條路（頂層 CLAUDE.md 金鑰鐵律
#   ＋ 2026-07-29 t145 的實害——同一把金鑰兩種寫法並存，必然漂移）。
ML = _load("isep_mainline", os.path.join(HERE, "..", "hooks", "lib", "mainline.py"))
try:
    T = _load("isep_ticket", os.path.join(HERE, "ticket"))
except Exception:
    # 🔴 載不到就只是「打不了網路的那幾個動作不能用」，不是整支壞掉。
    #   ISEP#82 驗收第 4 條要的是「沒有主線也不能整組壞掉」，同一個道理：
    #   `show` 是離線的，它不該因為別的零件壞了而回答不出來。
    T = None


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


def need_net():
    if T is None:
        die("🔴 載不到 `scripts/ticket`（憑證與 API 都沿用它），這個動作打不了網路。\n"
            "   `scripts/mainline show` 仍然可以用——它是離線的。")


def parse_ms_ref(s):
    """`owner/repo#<milestone id>`。與票號同形，所以錯手貼票號會拿到 404，訊息要講清楚。"""
    import re
    m = re.match(r"^([\w.-]+)/([\w.-]+)#(\d+)$", (s or "").strip())
    if not m:
        die("🔴 寫法是 owner/repo#<milestone id>，你給的是：%s\n"
            "（先跑 `scripts/mainline list` 看有哪些，那裡印的就是可以直接貼的）" % s)
    return m.group(1), m.group(2), int(m.group(3))


def org_repos(owner):
    rows = T.api_soft("/orgs/%s/repos?limit=100" % owner) or []
    return [r["name"] for r in rows if isinstance(r, dict) and r.get("name")]


def issues_in_milestone(owner, repo, title):
    q = urllib.parse.urlencode({"state": "all", "type": "issues",
                                "milestones": title, "limit": 100})
    rows = T.api_soft("/repos/%s/%s/issues?%s" % (owner, repo, q)) or []
    out = []
    for it in rows:
        if not isinstance(it, dict) or it.get("pull_request"):
            continue
        ms = it.get("milestone") or {}
        # 🔴 再比對一次標題：`milestones=` 這個參數在不同 Gitea 版本上對「名稱 vs id」
        #   的解讀不完全一致，拿回來的東西要自己確認過才算數（不要相信查詢字串）。
        if ML.norm(ms.get("title")) != ML.norm(title):
            continue
        out.append("%s/%s#%d" % (owner, repo, it["number"]))
    return out


def collect_members(owner, title, anchor_repo):
    """跨 repo 收齊「掛在同名 milestone 底下」的票。"""
    refs, scanned = [], []
    repos = org_repos(owner)
    if anchor_repo not in repos:
        repos.append(anchor_repo)
    for r in repos:
        got = issues_in_milestone(owner, r, title)
        if got:
            scanned.append((r, len(got)))
            refs.extend(got)
    return sorted(set(refs)), scanned


def fetch_milestone(owner, repo, mid):
    d = T.api_soft("/repos/%s/%s/milestones/%d" % (owner, repo, mid))
    if not isinstance(d, dict) or not d.get("title"):
        die("🔴 讀不到 %s/%s#%d 這個 milestone。\n"
            "  · id 是 milestone 的 id，不是 issue 的號碼（`scripts/mainline list` 會印對的）\n"
            "  · 或是這台機器現在拿不到 Gitea（那就先別標，不要標一個猜的）" % (owner, repo, mid))
    return d


# ── show ─────────────────────────────────────────────────────────────────
def cmd_show(argv):
    """離線。**永遠回一個答案**，包括「現在沒有主線」（ISEP#82 驗收第 4 條）。"""
    ms = ML.load()
    if not ms:
        print("🎯 現在沒有主線。")
        print()
        print("  這不是故障——沒有人標過，或上一條收掉了。")
        print("  要標一條：`scripts/mainline list` 看有哪些 → `scripts/mainline set <owner/repo#id>`")
        print()
        print("📌 在標之前，每回合的 🎯 那一行也會照樣講「現在沒有主線」，不會編一條出來。")
        return
    c, total, pct = ML.progress(ms)
    print("🎯 現在的主線：%s「%s」" % (ML.ref_of(ms), ms.get("title")))
    desc = (ms.get("description") or "").strip()
    if desc:
        first = desc.splitlines()[0].strip()
        # 只印第一行、最多 150 字：這是**一眼掃**的答案，不是 milestone 的全文。
        # leo 一天看兩眼（白話鐵律），整段貼過來他就不會看。
        print("   目標：" + (first[:150] + "…" if len(first) > 150 else first))
    print("   期限：%s" % ML.due_phrase(ms))
    print("   進度：%d／%d 張已關（%d%%）" % (c, total, pct))
    mem = sorted(ML.members(ms))
    byrepo = {}
    for r in mem:
        byrepo[r.split("#")[0]] = byrepo.get(r.split("#")[0], 0) + 1
    print("   屬於主線的票：%d 張%s" % (
        len(mem), ("（" + "、".join("%s %d" % (k, v) for k, v in sorted(byrepo.items())) + "）") if byrepo else ""))
    if ms.get("set_at"):
        print("   標定於：%s" % ms["set_at"])
    if ms.get("refreshed_at"):
        print("   最後更新：%s" % ms["refreshed_at"])
    print()
    print("📌 這是唯一答案——主線住在一個檔裡，那個檔放得下一條：%s" % ML.path())


# ── list ─────────────────────────────────────────────────────────────────
def cmd_list(argv):
    need_net()
    owner = argv[0] if argv else T.ORG
    rows = []
    for r in org_repos(owner):
        for m in (T.api_soft("/repos/%s/%s/milestones?state=open" % (owner, r)) or []):
            if not isinstance(m, dict):
                continue
            rows.append((owner, r, m))
    if not rows:
        die("🔴 一個 open milestone 都沒讀到——多半是拿不到 Gitea，不是真的沒有。\n"
            "（讀不到 ≠ 不存在。先修連線，不要因此標一個猜的主線。）", 1)
    now = datetime.now(timezone.utc)
    cur = ML.load()
    print("open milestone 共 %d 個（%s）：\n" % (len(rows), owner))
    for o, r, m in sorted(rows, key=lambda x: (ML.due_of(x[2]) or datetime.max.replace(tzinfo=timezone.utc))):
        c, total, pct = ML.progress(m)
        mark = "  ← 現在的主線" if cur and ML.ref_of(cur) == "%s/%s#%s" % (o, r, m.get("id")) else ""
        print("  %-28s %-24s %3d%%  期限 %s%s" % (
            "%s/%s#%s" % (o, r, m.get("id")), (m.get("title") or "")[:24],
            pct, ML.due_phrase(m, now), mark))
    print("\n📌 貼左邊那一欄：`scripts/mainline set <owner/repo#id>`")


# ── set ──────────────────────────────────────────────────────────────────
def cmd_set(argv):
    need_net()
    if not argv:
        die("用法：mainline set <owner/repo#milestone_id>")
    owner, repo, mid = parse_ms_ref(argv[0])
    m = fetch_milestone(owner, repo, mid)
    title = m.get("title")
    refs, scanned = collect_members(owner, title, repo)
    now = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M")
    ok = ML.save({
        "ref": "%s/%s#%d" % (owner, repo, mid),
        "owner": owner, "repo": repo, "id": mid,
        "title": title,
        "description": m.get("description") or "",
        "due_on": m.get("due_on") or "",
        "open_issues": m.get("open_issues") or 0,
        "closed_issues": m.get("closed_issues") or 0,
        "members": refs,
        "set_at": now, "refreshed_at": now,
    }, mirror=True)
    if not ok:
        die("🔴 寫不進 %s——主線沒有被標上。" % ML.path())
    print("✅ 主線已標定：%s/%s#%d「%s」" % (owner, repo, mid, title))
    if scanned:
        print("   跨 repo 收到同名 milestone 的票 %d 張：%s"
              % (len(refs), "、".join("%s %d" % (r, n) for r, n in scanned)))
    print(mirror_line())
    print()
    cmd_show([])


def mirror_line():
    """repo 那份有沒有跟著寫（inkstone/ISEP#130）：寫了要人去 commit，沒寫要講為什麼。"""
    rp = ML.repo_path(for_write=True)
    if rp and os.path.isfile(rp):
        return ("   已同步寫進 repo 那份：%s\n"
                "   📌 這份要 commit／開 PR 併進 inkstone/InkStoneCo，雲端 clone 才拿得到" % rp)
    return ("   ⚠️ 這台找不到 InkStoneCo 的 system-dev/（CLAUDE_PROJECT_DIR 或 cwd 底下都沒有），"
            "repo 那份沒更新——雲端仍會讀到舊主線")


def cmd_clear(argv):
    had = ML.load()
    rp = ML.repo_path()
    ML.clear(mirror=True)
    print("✅ 已清掉主線%s。現在沒有主線——每回合的 🎯 那一行會照實說。"
          % ("（原本是 %s「%s」）" % (ML.ref_of(had), had.get("title")) if had else ""))
    if rp:
        print("   repo 那份也刪了：%s（要 commit 才會到雲端）" % rp)


# ── refresh ──────────────────────────────────────────────────────────────
def cmd_refresh(argv):
    """SessionStart 跑一次。**拿不到就原封不動**——寧可資料舊，不要把主線弄丟。"""
    ms = ML.load()
    if not ms:
        return                      # 沒有主線就沒有東西要更新，安靜結束
    d = T.api_soft("/repos/%s/%s/milestones/%s" % (ms["owner"], ms["repo"], ms["id"]))
    if not isinstance(d, dict) or not d.get("title"):
        return                      # 讀不到 ≠ 主線不存在
    ms.update({
        "title": d.get("title") or ms["title"],
        "description": d.get("description") or ms.get("description") or "",
        "due_on": d.get("due_on") or ms.get("due_on") or "",
        "open_issues": d.get("open_issues") or 0,
        "closed_issues": d.get("closed_issues") or 0,
        "refreshed_at": datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M"),
    })
    if "--members" in argv:
        refs, _ = collect_members(ms["owner"], ms["title"], ms["repo"])
        if refs:
            ms["members"] = refs
    ML.save(ms)


# ── adopt（補收）──────────────────────────────────────────────────────────
def cmd_adopt(argv):
    need_net()
    if not argv:
        die("用法：mainline adopt <owner/repo#N>（把一張票掛進主線＝「補收」）")
    ms = ML.load()
    if not ms:
        die("🔴 現在沒有主線，沒有東西可以補收進去。先 `scripts/mainline set …`。")
    owner, repo, num = T.parse_ref(argv[0])
    cands = [m for m in (T.api_soft("/repos/%s/%s/milestones?state=all" % (owner, repo)) or [])
             if isinstance(m, dict) and ML.norm(m.get("title")) == ML.norm(ms["title"])]
    if not cands:
        die("""🔴 `%s/%s` 底下沒有叫「%s」的 milestone，所以這張票掛不進去。

兩條路，**都要人決定，這支不會替你選**：
  · 這件事真的屬於主線 ⇒ 在 `%s/%s` 開一個同名 milestone（記得設真的期限，
    `milestone-due-guard.sh` 會擋沒期限的），再跑一次 adopt
  · 這件事其實是另一條線 ⇒ 那就是**跳線**，不是補收。直接重送派工
    （本閘同一張票只擋一次），並在回覆說清楚為什麼現在要岔開主線。""" % (
            owner, repo, ms["title"], owner, repo))
    mid = cands[0]["id"]
    T.api("/repos/%s/%s/issues/%d" % (owner, repo, num), {"milestone": mid}, method="PATCH")
    ref = "%s/%s#%d" % (owner, repo, num)
    ms["members"] = sorted(set(ms.get("members") or []) | {ref})
    ML.save(ms, mirror=True)
    print("✅ 補收：%s 已掛進「%s」（%s/%s#%d）" % (ref, ms["title"], owner, repo, mid))
    print(mirror_line())


# ── has ──────────────────────────────────────────────────────────────────
def cmd_has(argv):
    if not argv:
        die("用法：mainline has <owner/repo#N>")
    ms = ML.load()
    ref = argv[0].strip()
    if not ms:
        print("🎯 現在沒有主線 ⇒ 沒有「不屬於主線」這回事。")
        return
    v = ML.belongs(ref, ms)
    if v is None:
        owner, repo, num = T.parse_ref(ref)
        it = T.api_soft("/repos/%s/%s/issues/%d" % (owner, repo, num))
        title = ((it or {}).get("milestone") or {}).get("title") if isinstance(it, dict) else None
        v = ML.belongs(ref, ms, ticket_milestone_title=title or "")
        if not isinstance(it, dict):
            print("⚪ 讀不到 %s，判不出來（讀不到 ≠ 不屬於）。" % ref)
            sys.exit(0)
    if v:
        print("✅ %s 在主線「%s」上。" % (ref, ms["title"]))
        sys.exit(0)
    print("🚧 %s **不在**主線「%s」上。\n"
          "   補收 ⇒ `scripts/mainline adopt %s`；真的是插件事 ⇒ 那是跳線，說清楚再做。"
          % (ref, ms["title"], ref))
    sys.exit(1)


CMDS = {"show": cmd_show, "list": cmd_list, "set": cmd_set, "clear": cmd_clear,
        "refresh": cmd_refresh, "adopt": cmd_adopt, "has": cmd_has}

if __name__ == "__main__":
    # `mainline | head` 是很自然的用法，別讓它吐一段 BrokenPipe 的 traceback
    # ——那會讓人以為這支壞了。
    try:
        import signal
        signal.signal(signal.SIGPIPE, signal.SIG_DFL)
    except Exception:
        pass
    if len(sys.argv) < 2:
        cmd_show([])
        sys.exit(0)
    if sys.argv[1] in ("-h", "--help", "help"):
        print(__doc__)
        sys.exit(0)
    if sys.argv[1] not in CMDS:
        print(__doc__)
        sys.exit(2)
    CMDS[sys.argv[1]](sys.argv[2:])