Files
ISEP/scripts/isep-nag
T
Leo d2f706a16c 新 repo 的工人不再靠人記得:課程工人 course-hand+開場點名「有票卻沒工人」(inkstone/ISEP#118)
- agents/course-hand.md:管 inkstone/ax-courses、course_gen、llm-wiki-template(c7358 leo「你可以製作一個」)
  假設:llm-wiki-template 也歸課程工人(同為課程素材),總管不同意就把它拆出去
- hooks/lib/roster.py:`負責 repo` 可列多個(、或逗號),新增 which()/uncovered();find() 與閘本身未動
- scripts/roster which 改走 roster.which
- scripts/isep-nag 第④類:org 裡有開著的票、名單上卻沒人負責的 repo,只進開場長版、不進 Telegram
- 測試:roster-guard 25→31、overdue-nag 36→43;兩種突變(拿掉票數條件/只認第一個 repo)都會紅
- 文件:worker-roster.md 回答「新 repo 的工人誰在什麼時候加」;README/plugin.json 工人 8、scripts 65(樹上實數)

待總管定版(改了 agents/scripts,要升版才裝得上)。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-14 13:13:12 +08:00

441 lines
20 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""isep-nag — 把「沒有人會叫的事」叫出來(inkstone/ISEP#93
━━ 這支解的問題 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
有些事情**沒有人會叫**:一張票掛著等 leo 三天了、一個 milestone 逾期了、
一根棒子交回來躺著沒人接。它們不會壞掉、不會噴錯,**只是靜靜地過期**。
leo 整天上課,一天看兩眼。**不叫,就等於這件事不會發生。**
━━ 🔴 刻意不做什麼(這不是簡化,是判斷)━━━━━━━━━━━━━━━━━━━━━━━━━━
**不做排程輪詢、不做 webhook fan-out、不掛 Gitea Actions。**
· 頂層 `CLAUDE.md`「避免再被 GitHub flag 的硬規則」:禁一事件 fan-out 到多 repo
· `issue-handle` skill:「**有事才讀**」,換成 Gitea 也不放寬
· ISEP v0.6.0 §8.2:第一版不依賴 runner——「壞掉的形式是『以為有人在跑』,
比『沒人跑』貴得多」
⇒ 本支只在**兩個時機**跑:開 session 時順手一次、人要看的時候手動一次。
**已知限制,誠實寫在這裡**:它只在 CC 醒著的時候會叫。
「CC 整個沒在跑」的那一段補不到——那要 leo 決定要不要開 Gitea 的通知權限
(機器帳號 `claude-code` 在 `bots` 團隊,`/orgs/…/hooks` 回 403,裝不了)。
**但腳本版本身是完整可用的,不是半成品。**
━━ 撈四種東西 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
① 逾期的 milestone 期限過了、票還沒關 ⇒ 那個版本交不出來了
② 掛著等 leo 的票 `Human` 或指派給 leo**標等了幾天**
③ 棒子掉在地上的票 `s/doing``s/review` 卻很多天沒動 ⇒ 沒有人在做
④ 有票卻沒有工人的 repo 名單(`agents/`)上沒人負責、卻有開著的票
⇒ 派工一定撞 roster-guardinkstone/ISEP#118
**只進長版**(開場給總管看),不進 Telegram 短版——補工人是總管的事,不吵 leo。
它回答的是「新 repo 的工人誰在什麼時候加」:建 repo 之後的**第一個 session 開場**
由這支點名;在那之前沒有任何東西會提醒。
每一條都用白話寫(`CLAUDE.md` 規則五):一眼看懂、代號附一句人話、
講「這對你意味什麼」而不是系統內部狀態。
🔴 **沒東西可報時會說「查過了,沒有」**,不會安靜結束——
安靜跟壞掉長得一模一樣,而分不出來的東西沒有人敢信。
━━ 用法 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
isep-nag 撈一次,白話印出來
isep-nag --json 同一份資料的機器格式
isep-nag --notify 撈完順便發 Telegram(走 isep-notify,發不出去有退路)
isep-nag --short 只印 Telegram 版(很短,leo 手機上看的那則)
isep-nag --both 長版+分隔線+短版,一次算完(開場的 hook 用這個)
isep-nag --fixture <檔> 用假資料跑(離線測試用,完全不碰網路)
isep-nag --now <epoch> 把時鐘定住(測試用)
環境變數:
ISEP_NAG_REPOS 只看這幾個 repo(空白分隔)。預設:整個 org 撈一次
ISEP_NAG_ORG 預設 inkstone
ISEP_NAG_STALE_DAYS 「幾天沒動算掉棒」,預設 3
TICKET_HOST Gitea 位址,預設 https://git.uncle6.me
"""
import concurrent.futures as cf
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone
# leo 看的是台北時間。容器多半跑在 UTC ⇒ 不指定就會印出一個他要自己換算的數字。
try:
from zoneinfo import ZoneInfo
TAIPEI = ZoneInfo("Asia/Taipei")
except Exception:
TAIPEI = timezone(timedelta(hours=8))
HOST = os.environ.get("TICKET_HOST") or "https://git.uncle6.me"
ORG = os.environ.get("ISEP_NAG_ORG") or "inkstone"
STALE_DAYS = int(os.environ.get("ISEP_NAG_STALE_DAYS") or 3)
HERE = os.path.dirname(os.path.abspath(__file__))
# 「等 leo」的兩個訊號:掛 Human 標籤,或指派給他本人。
HUMAN_LABEL = "Human"
LEO_LOGINS = {"leo", "leo21c", "Leo"}
# 「有人領了」的狀態標籤——掛著這個卻很久沒動,就是棒子掉在地上。
BATON_LABELS = {"s/doing", "s/review"}
def token():
for env in ("GITEA_TOKEN_CLAUDE_CODE", "GITEA_TOKEN"):
v = os.environ.get(env)
if v:
return v
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)
return None
def api(path, tok):
url = "%s/api/v1%s" % (HOST, path)
req = urllib.request.Request(url, headers={"Authorization": "token %s" % tok})
try:
return json.load(urllib.request.urlopen(req, timeout=25))
except Exception:
return None
def ts(s):
"""Gitea 的時間字串 → epoch。看不懂就回 None(不猜)。"""
if not s:
return None
try:
return int(datetime.fromisoformat(s.replace("Z", "+00:00"))
.astimezone(timezone.utc).timestamp())
except Exception:
return None
def days(a, b):
return max(0, int((a - b) // 86400))
# ══ 一、把資料撈回來(或從 fixture 讀)═══════════════════════════════════
def collect(fixture=None):
"""回 {"milestones": [...], "issues": [...]}。撈不到就回空的,不編。"""
if fixture:
with open(fixture) as f:
return json.load(f)
tok = token()
if not tok:
return {"milestones": [], "issues": [], "error":
"拿不到 Gitea tokenGITEA_TOKEN_CLAUDE_CODE 沒設,remote 也沒帶憑證)"}
# org 的 repo 清單**一律撈一次**:④「有票卻沒工人」要看整個 org,
# 而且這一通本來就要打(沒設 ISEP_NAG_REPOS 時),不多花 API。
org_rows = api("/orgs/%s/repos?limit=50" % ORG, tok) or []
repos = os.environ.get("ISEP_NAG_REPOS", "").split()
if not repos:
repos = [r.get("full_name") for r in org_rows if r.get("full_name")]
# 🔴 平行撈,因為這支掛在 SessionStart 上——那是 leo 每次開工的關鍵路徑。
# 循序版實測 16 秒(16 個 repo + 5 頁票),**開場等 16 秒沒有人會忍受它,
# 而被忍不了的東西下一步就是被關掉**。平行之後落在 3 秒上下。
milestones = []
with cf.ThreadPoolExecutor(max_workers=8) as pool:
jobs = {pool.submit(api, "/repos/%s/milestones?state=open" % f, tok): f
for f in repos}
for job in cf.as_completed(jobs):
full = jobs[job]
try:
rows = job.result() or []
except Exception:
continue
for m in rows:
m = dict(m or {})
m["repo"] = full
milestones.append(m)
# 票用**一支跨 repo 的搜尋**撈,不是每個 repo 打一次:
# 16 個 repo × 一次 = 16 通 API,那個形狀就開始像輪詢了。
# 而且只撈**這三種標籤**Gitea 的 labels= 是 OR)——不是撈回全部再自己篩:
# 全撈是 5 頁,篩過是 1–2 頁。
# ⚠️ 已知取捨:**沒掛 Human 標籤、只是被指派給 leo 的票,這條路撈不到。**
# 真相源是標籤(`labels.yaml`),指派是輔助訊號 ⇒ 寧可漏掉那個邊角,
# 也不要為它把開場多花 13 秒。要撈全部就設 ISEP_NAG_ALL_ISSUES=1。
want = ",".join(sorted({HUMAN_LABEL} | BATON_LABELS))
qs = "" if os.environ.get("ISEP_NAG_ALL_ISSUES") == "1" else \
"&labels=%s" % urllib.parse.quote(want)
issues = []
for page in range(1, 6):
rows = api("/repos/issues/search?state=open&type=issues&limit=50&page=%d%s"
% (page, qs), tok) or []
if not rows:
break
issues.extend(rows)
if len(rows) < 50:
break
return {"milestones": milestones, "issues": issues, "repos": org_rows}
# ══ 二、找出「該叫的事」═══════════════════════════════════════════════════
def findings(data, now):
over, waiting, dropped = [], [], []
for m in data.get("milestones") or []:
due = ts(m.get("due_on"))
if not due or due >= now:
continue # 沒設期限的不是節拍器;還沒到期的不叫
open_n = int(m.get("open_issues") or 0)
if open_n == 0:
continue # 票都關完了,只是沒人按關閉鈕——不吵
over.append({
"repo": m.get("repo") or "?",
"title": m.get("title") or "(沒有名字)",
"due": (m.get("due_on") or "")[:10],
"late_days": days(now, due),
"open": open_n,
"closed": int(m.get("closed_issues") or 0),
})
over.sort(key=lambda x: -x["late_days"])
for i in data.get("issues") or []:
if i.get("pull_request"):
continue # PR 有 pr-verdict-guard 在管,不重複點名
full = (i.get("repository") or {}).get("full_name") or ""
# 🔴 只看設定的那個 org。跨 repo 搜尋端點會把**舊 org 的同名票**一起回來
# (實測:`Leo/Arcrun#85` 與 `inkstone/Arcrun#85` 是同一件事的兩份),
# 照單全收會讓「26 張掉棒」裡有一半是重複——**灌水的清單等於沒有清單**。
if ORG and not full.startswith(ORG + "/"):
continue
labels = {(l or {}).get("name") for l in (i.get("labels") or [])}
who = {(a or {}).get("login") for a in (i.get("assignees") or [])}
upd = ts(i.get("updated_at")) or now
row = {
"ref": "%s#%s" % ((i.get("repository") or {}).get("full_name") or "?",
i.get("number")),
"title": (i.get("title") or "").strip(),
"quiet_days": days(now, upd),
"labels": sorted(labels),
"url": i.get("html_url") or "",
}
if HUMAN_LABEL in labels or (who & LEO_LOGINS):
waiting.append(row)
elif (labels & BATON_LABELS) and row["quiet_days"] >= STALE_DAYS:
dropped.append(row)
waiting.sort(key=lambda x: -x["quiet_days"])
dropped.sort(key=lambda x: -x["quiet_days"])
return {"overdue_milestones": over, "waiting_on_leo": waiting,
"dropped_batons": dropped,
"repos_without_worker": no_worker(data)}
def no_worker(data):
"""④ 有票卻沒有工人的 repoinkstone/ISEP#118)。
回 list(可能是空的);**名單讀不到或 org 清單沒撈到 ⇒ 回 None**
報告裡講「沒查到」,不講成「每個 repo 都缺人」或「都有人」。
"""
rows = data.get("repos")
if not rows:
return None
try:
sys.path.insert(0, os.path.join(HERE, "..", "hooks", "lib"))
import roster
except Exception:
return None
return roster.uncovered(rows)
# ══ 三、講出來(白話)═════════════════════════════════════════════════════
MAX_ROWS = int(os.environ.get("ISEP_NAG_MAX_ROWS") or 8)
BOTH_SEP = "␟--isep-nag-short--␟" # --both 的分隔線(不會出現在正文裡)
def _tail(rows, shown):
"""超過封頂的部分收成一行。**不是藏起來,是講出來還有幾張。**"""
rest = len(rows) - shown
return [" · …另外還有 %d 張(`isep-nag --json` 看全部)" % rest] if rest > 0 else []
def _waited(d):
"""0 天講「今天剛掛上」——「等了 0 天」讀起來像壞掉的計數器。"""
return "今天剛掛上" if d == 0 else "等了 %d 天" % d
def _clip(s, n):
s = (s or "").replace("\n", " ").strip()
return s if len(s) <= n else s[:n - 1] + "…"
def full_report(f, now, err=None):
"""給眼前看的長版。**沒東西也一定有話講。**"""
when = datetime.fromtimestamp(now, timezone.utc).astimezone(TAIPEI).strftime("%Y-%m-%d %H:%M")
out = ["🔔 催辦員查了一次(%s)——這些事沒有人會叫,所以它叫(inkstone/ISEP#93" % when]
if err:
out += ["", "🔴 撈不到資料:%s" % err,
" ⇒ **這不等於「沒有事情逾期」**,這是「我沒查到」。兩件事不要講成同一句。"]
return "\n".join(out)
n = sum(len(f[k]) for k in ("overdue_milestones", "waiting_on_leo", "dropped_batons"))
if n == 0:
out += ["", "✅ **查過了,沒有。** 三類都乾淨:",
" · 沒有逾期還開著票的 milestone",
" · 沒有掛著等你的票",
" · 沒有超過 %d 天沒動的 s/doings/review" % STALE_DAYS]
out += worker_section(f)
return "\n".join(out)
if f["overdue_milestones"]:
out += ["", "⏰ **逾期的 milestone%d 個)**——期限過了,票還開著,"
"代表那個版本交不出來了:" % len(f["overdue_milestones"])]
for m in f["overdue_milestones"][:MAX_ROWS]:
out.append(" · %s%s」——%s 到期,**逾期 %d 天**,還有 %d 張票沒關"
% (m["repo"], m["title"], m["due"], m["late_days"], m["open"]))
out += _tail(f["overdue_milestones"], MAX_ROWS)
out.append(" ⇒ 對你意味著:不是改期限,就是把票搬走。**放著它只會繼續爛。**")
if f["waiting_on_leo"]:
out += ["", "🙋 **在等你的票(%d 張)**——只有你按得下去,"
"沒人替得了:" % len(f["waiting_on_leo"])]
for i in f["waiting_on_leo"][:MAX_ROWS]:
out.append(" · %s **%s**%s"
% (i["ref"], _waited(i["quiet_days"]), _clip(i["title"], 42)))
out += _tail(f["waiting_on_leo"], MAX_ROWS)
out.append(" ⇒ 對你意味著:回一個詞就能推動。不回,這幾件永遠不會發生。")
if f["dropped_batons"]:
out += ["", "🧤 **棒子掉在地上(%d 張)**——標著「有人在做」,"
"但超過 %d 天沒動:" % (len(f["dropped_batons"]), STALE_DAYS)]
for i in f["dropped_batons"][:MAX_ROWS]:
out.append(" · %s **靜了 %d 天**%s):%s"
% (i["ref"], i["quiet_days"], "".join(i["labels"]),
_clip(i["title"], 38)))
out += _tail(f["dropped_batons"], MAX_ROWS)
out.append(" ⇒ 對你意味著:這些不是在做,是沒人接。要嘛派人,要嘛改回 s/todo。")
out += worker_section(f)
return "\n".join(out)
def worker_section(f):
"""④ 只進長版(開場注入給總管看的那份),**不進 Telegram 短版**
派工是總管的事,不是 leo 的事——拿它去吵 leo 就是把總管的活推給他。"""
rows = f.get("repos_without_worker")
if rows is None:
return ["", "👷 有票卻沒有工人的 repo:**沒查到**(org 清單或工人名單讀不到)"
"——這不等於「每個 repo 都有人」。"]
if not rows:
return ["", "👷 每個有開著票的 repo,名單上都有工人。"]
out = ["", "👷 **有票卻沒有工人的 repo(%d 個)**——派工到這裡會被 roster-guard 擋,"
"只剩開例外閘繞過去:" % len(rows)]
for r in rows[:MAX_ROWS]:
out.append(" · %s%d 張票開著)" % (r["repo"], r["open"]))
out += _tail(rows, MAX_ROWS)
out.append(" ⇒ 對總管意味著:**先補工人再派工**——在 ISEP `agents/` 加一個 `.md`"
"(規約 docs/governance/worker-roster.md),或把它寫進既有工人的 `負責 repo`。"
"不要開 `/tmp/.roster-ok-*` 了事,那樣下一次還是會撞。")
return out
def short_report(f, now, err=None):
"""Telegram 版:leo 在手機上一眼掃完、回一個詞。**寧短勿長。**"""
if err:
return "[總管] 🔴 催辦員查不到資料(%s)。這不是「沒事」,是「沒查到」。" % _clip(err, 60)
a, b, c = (len(f["overdue_milestones"]), len(f["waiting_on_leo"]),
len(f["dropped_batons"]))
if a + b + c == 0:
return "[總管] ✅ 查過了,沒有:沒有逾期的 milestone、沒有在等你的票、沒有掉在地上的棒子。"
lines = ["[總管] 🔔 有 %d 件沒人叫的事:" % (a + b + c)]
if a:
top = f["overdue_milestones"][0]
lines.append("⏰ %d 個 milestone 逾期,最久的是「%s」(%s,逾期 %d 天,%d 張票沒關)"
% (a, _clip(top["title"], 20), top["repo"].split("/")[-1],
top["late_days"], top["open"]))
if b:
top = f["waiting_on_leo"][0]
lines.append("🙋 %d 張票在等你,最久的 %s %s%s"
% (b, top["ref"], _waited(top["quiet_days"]),
_clip(top["title"], 26)))
if c:
top = f["dropped_batons"][0]
lines.append("🧤 %d 張標著「有人在做」卻沒人動,最久 %d 天:%s"
% (c, top["quiet_days"], top["ref"]))
lines.append("回「看」我就把完整清單貼出來。")
return "\n".join(lines)
# ══ 四、CLI ══════════════════════════════════════════════════════════════
def main(argv):
fixture = now = None
want_json = short = do_notify = both = False
ticket = "inkstone/ISEP#93"
i = 0
while i < len(argv):
a = argv[i]
if a == "--json":
want_json = True
elif a == "--short":
short = True
elif a == "--both":
# 開場的 hook 用這個:長版+短版**一次算完**。
# 分兩次呼叫=打兩輪 Gitea=開場多等 5 秒,而那是 leo 每次開工的關鍵路徑。
both = True
elif a == "--notify":
do_notify = True
elif a == "--fixture":
i += 1; fixture = argv[i]
elif a == "--now":
i += 1; now = int(argv[i])
elif a == "--fallback-ticket":
i += 1; ticket = argv[i]
elif a in ("-h", "--help"):
print(__doc__); return 0
else:
print("不認得的參數:%s--help 看用法)" % a, file=sys.stderr)
return 2
i += 1
now = now if now is not None else int(time.time())
try:
data = collect(fixture)
except Exception as e:
data = {"milestones": [], "issues": [], "error": "撈的時候炸了:%s" % e}
err = data.get("error")
f = findings(data, now)
if both:
print(full_report(f, now, err))
print(BOTH_SEP)
print(short_report(f, now, err))
elif want_json:
print(json.dumps({"now": now, "error": err, "findings": f},
ensure_ascii=False, indent=2))
elif short:
print(short_report(f, now, err))
else:
print(full_report(f, now, err))
if do_notify:
# 🔴 走 isep-notify,不自己打 HTTP:那支才會先問閘、發不出去才有退路。
# 在這裡自己 urlopen 一次,就等於多開一條沒有退路的路。
p = subprocess.run(
[sys.executable, os.path.join(HERE, "isep-notify"),
"--text", short_report(f, now, err), "--fallback-ticket", ticket],
capture_output=True, text=True)
sys.stderr.write(p.stdout + p.stderr)
return p.returncode
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))