ISEP/hooks/lib/countdown.py

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