Merge remote-tracking branch 'origin/main' into fix/cloud-wiring-isep90
# Conflicts: # docs/TESTING.md
This commit is contained in:
claude-code
Executable
+201
@@ -0,0 +1,201 @@
|
||||
#!/usr/bin/env python3
|
||||
"""hooks/lib/countdown.py —— 算「這件事已經花了多久／還剩多久」的那一行。
|
||||
|
||||
這是 helper，不是閘（inkstone/ISEP#40 S7：lib/ 底下的東西不算一支手寫的閘）。
|
||||
兩個呼叫者共用它，所以時間的算法只有一份：
|
||||
· UserPromptSubmit（`countdown-guard.sh` 注入模式）—— 把這一行送到模型眼前
|
||||
· Stop（同一支的查核模式）—— 擋下來時要把同一行原樣印出來給它抄
|
||||
|
||||
── 基準從哪一刻算起（inkstone/ISEP#63 要我自己判斷並說明理由）──────────
|
||||
leo 的原話是「**隨時看自己拖了多久**」。「拖」指向的是**任務**，不是 session，
|
||||
但「任務」在機器上沒有起點——沒有任何欄位記著「我什麼時候接手這條線」。
|
||||
可機械取得、又真的對應到一個交付的起點，只有一個：**這段對話的第一則訊息**。
|
||||
|
||||
而那正好就是總管的交貨單位：CLAUDE.md 規則三點七「**一段對話 ＝ 一個 release**」。
|
||||
⇒ 已過時間 ＝ 這個 release 已經燒掉多久。**這個數字不需要任何人維護，也不會說謊。**
|
||||
|
||||
resume 的 session 取「transcript 第一列的時間」而不是「hook 第一次被叫到的時間」：
|
||||
接關不是重新開始，昨天燒掉的兩小時仍然算在這個交付上。兩者都拿得到時取較早的。
|
||||
|
||||
── 第二個數字：今天的收工線 ──────────────────────────────────────────
|
||||
leo 2026-08-28 07:40：「今天我整天上課……**下午四點左右**希望已經都完成了」。
|
||||
⇒ 台北 16:00。過了 16:00 之後**不立刻換算到明天**——那會讓超時消失，
|
||||
而超時正是他要看的東西。過線後四小時內顯示「已超過 X」，
|
||||
台北 20:00 之後才滾到隔天（那時已經是下一個工作日的倒數了）。
|
||||
|
||||
── 第三個數字：主線 milestone 的期限（可選）────────────────────────────
|
||||
從快取檔讀，**這支從不打網路**：UserPromptSubmit 走在每一則訊息的關鍵路徑上，
|
||||
在那裡打網路 ＝ 每一句話都先等一次 HTTP。快取由 SessionStart 那支負責更新
|
||||
（`scripts/countdown-milestone-refresh.sh`），拿不到就整段不顯示，不編數字。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
TAIPEI = timezone(timedelta(hours=8))
|
||||
MARKER = "⏱"
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
"""現在。測試用 ISEP_COUNTDOWN_NOW（epoch 秒）把時鐘定住。"""
|
||||
override = os.environ.get("ISEP_COUNTDOWN_NOW", "").strip()
|
||||
if override:
|
||||
try:
|
||||
return datetime.fromtimestamp(float(override), tz=timezone.utc)
|
||||
except Exception:
|
||||
pass
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _parse_ts(raw) -> "datetime | None":
|
||||
if not isinstance(raw, str) or not raw.strip():
|
||||
return None
|
||||
s = raw.strip().replace("Z", "+00:00")
|
||||
try:
|
||||
dt = datetime.fromisoformat(s)
|
||||
except Exception:
|
||||
return None
|
||||
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def transcript_start(path: str) -> "datetime | None":
|
||||
"""transcript 第一列帶時間戳的那一列 ＝ 這段對話真正的起點。"""
|
||||
if not path or not os.path.exists(path):
|
||||
return None
|
||||
try:
|
||||
with open(path) as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
row = json.loads(line)
|
||||
except Exception:
|
||||
continue
|
||||
dt = _parse_ts(row.get("timestamp"))
|
||||
if dt:
|
||||
return dt
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def state_dir() -> str:
|
||||
d = os.environ.get("ISEP_COUNTDOWN_STATE_DIR", "").strip()
|
||||
if not d:
|
||||
d = os.path.join(os.path.expanduser("~"), ".claude", "isep-countdown")
|
||||
try:
|
||||
os.makedirs(d, exist_ok=True)
|
||||
except Exception:
|
||||
pass
|
||||
return d
|
||||
|
||||
|
||||
def session_start(session_id: str, transcript_path: str, now: datetime) -> datetime:
|
||||
"""起點 ＝ transcript 第一列與「第一次看到這個 session」兩者取較早。
|
||||
|
||||
第一次看到就落一個檔；transcript 沒有時間戳（或讀不到）時它就是唯一的憑據。
|
||||
"""
|
||||
cands = []
|
||||
ts = transcript_start(transcript_path)
|
||||
if ts:
|
||||
cands.append(ts)
|
||||
|
||||
sid = "".join(c for c in (session_id or "") if c.isalnum() or c in "-_")[:80]
|
||||
if sid:
|
||||
p = os.path.join(state_dir(), sid + ".start")
|
||||
seen = None
|
||||
try:
|
||||
with open(p) as f:
|
||||
seen = datetime.fromtimestamp(float(f.read().strip()), tz=timezone.utc)
|
||||
except Exception:
|
||||
seen = None
|
||||
if seen is None:
|
||||
seen = min(cands) if cands else now
|
||||
try:
|
||||
with open(p, "w") as f:
|
||||
f.write("%d" % seen.timestamp())
|
||||
except Exception:
|
||||
pass
|
||||
cands.append(seen)
|
||||
|
||||
return min(cands) if cands else now
|
||||
|
||||
|
||||
def deadline(now: datetime) -> datetime:
|
||||
"""今天的收工線（台北 HH:MM）。過線後四小時內不滾，讓超時看得見。"""
|
||||
hhmm = os.environ.get("ISEP_COUNTDOWN_DEADLINE", "16:00").strip() or "16:00"
|
||||
try:
|
||||
hh, mm = (int(x) for x in hhmm.split(":", 1))
|
||||
except Exception:
|
||||
hh, mm = 16, 0
|
||||
local = now.astimezone(TAIPEI)
|
||||
d = local.replace(hour=hh, minute=mm, second=0, microsecond=0)
|
||||
if local >= d + timedelta(hours=4):
|
||||
d += timedelta(days=1)
|
||||
return d
|
||||
|
||||
|
||||
def milestone() -> "tuple[str, datetime] | None":
|
||||
"""主線 milestone 的期限。快取檔一行：`YYYY-MM-DD|<名稱>`。沒有就沒有。"""
|
||||
raw = os.environ.get("ISEP_MILESTONE_DUE", "").strip()
|
||||
name = os.environ.get("ISEP_MILESTONE_NAME", "").strip()
|
||||
if not raw:
|
||||
p = os.path.join(state_dir(), "milestone-due")
|
||||
try:
|
||||
with open(p) as f:
|
||||
raw = f.read().strip()
|
||||
except Exception:
|
||||
return None
|
||||
if "|" in raw:
|
||||
raw, name = raw.split("|", 1)
|
||||
raw, name = raw.strip(), name.strip()
|
||||
if not raw:
|
||||
return None
|
||||
dt = _parse_ts(raw) or _parse_ts(raw + "T23:59:59+08:00")
|
||||
if not dt:
|
||||
return None
|
||||
return (name or "主線", dt)
|
||||
|
||||
|
||||
def human(delta: timedelta) -> str:
|
||||
mins = int(abs(delta).total_seconds()) // 60
|
||||
h, m = divmod(mins, 60)
|
||||
if h and m:
|
||||
return "%d 小時 %d 分" % (h, m)
|
||||
if h:
|
||||
return "%d 小時" % h
|
||||
return "%d 分" % m
|
||||
|
||||
|
||||
def line(session_id: str, transcript_path: str) -> str:
|
||||
now = _now()
|
||||
start = session_start(session_id, transcript_path, now)
|
||||
dl = deadline(now)
|
||||
|
||||
parts = ["%s 已過 %s" % (MARKER, human(now - start))]
|
||||
left = dl - now
|
||||
if left.total_seconds() >= 0:
|
||||
parts.append("距收工線（台北 %s）剩 %s" % (dl.strftime("%H:%M"), human(left)))
|
||||
else:
|
||||
parts.append("🔴 已超過收工線（台北 %s）%s" % (dl.strftime("%H:%M"), human(left)))
|
||||
|
||||
ms = milestone()
|
||||
if ms:
|
||||
name, due = ms
|
||||
gap = due - now
|
||||
if gap.total_seconds() >= 0:
|
||||
parts.append("主線 %s 剩 %s" % (name, human(gap)))
|
||||
else:
|
||||
parts.append("🔴 主線 %s 已逾期 %s" % (name, human(gap)))
|
||||
|
||||
return "｜".join(parts)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
payload = json.load(sys.stdin)
|
||||
except Exception:
|
||||
payload = {}
|
||||
print(line(payload.get("session_id") or "", payload.get("transcript_path") or ""))
|
||||
Reference in New Issue
Block a user