chktime/chktime.py

#!/usr/bin/env python3
"""chktime - Claude Code のタスク(セッション)ごとのやりとりの時刻を一覧表示するローカルWebアプリ.

~/.claude/projects/<project>/<sessionId>.jsonl を解析し、
  * プロジェクト一覧
  * セッション(タスク)一覧 … 開始/終了時刻・所要時間・ターン数
  * ターン(ユーザー発言→Claude応答)ごとの時刻・初回応答までの待ち・処理時間・空き時間
を JSON API + ブラウザUI で提供する。標準ライブラリのみで動作する。

  python chktime.py            # http://127.0.0.1:8787 を開く
  python chktime.py --port 9000 --no-browser
"""

from __future__ import annotations

import argparse
import csv
import errno
import io
import json
import os
import re
import threading
import webbrowser
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse

APP_DIR = Path(__file__).resolve().parent
WEB_DIR = APP_DIR / "web"

TS_RE = re.compile(rb'"timestamp":"([^"]+)"')
SYSTEM_REMINDER_RE = re.compile(r"<system-reminder>.*?</system-reminder>", re.S)
TAG_RE = re.compile(r"<([a-z-]+)>(.*?)</\1>", re.S)

PROMPT_LIMIT = 4000
SUMMARY_LIMIT = 200

# ---------------------------------------------------------------- 解析ロジック


def parse_ts(value):
    """ISO8601(UTC) 文字列 -> epoch 秒(float)。失敗したら None。"""
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None


def iso(epoch):
    if epoch is None:
        return None
    return datetime.fromtimestamp(epoch, timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def clean_text(text):
    """system-reminder などノイズを落とした本文を返す。"""
    text = SYSTEM_REMINDER_RE.sub("", text or "")
    return text.strip()


def classify_user(obj):
    """user レコードを人間の発言として解釈する。対象外なら None。

    戻り値: (kind, text)
      kind: prompt / command / interrupt / resume
    """
    if obj.get("isMeta") or obj.get("isSidechain"):
        return None
    origin = obj.get("origin")
    if isinstance(origin, dict) and origin.get("kind") not in (None, "human"):
        return None

    content = (obj.get("message") or {}).get("content")
    extras = []
    if isinstance(content, str):
        text = content
    elif isinstance(content, list):
        parts = []
        for block in content:
            if not isinstance(block, dict):
                continue
            btype = block.get("type")
            if btype == "tool_result":
                return None  # ツール実行結果 = 人間の発言ではない
            if btype == "text":
                parts.append(block.get("text") or "")
            elif btype in ("image", "document"):
                extras.append(btype)
        text = "\n".join(parts)
    else:
        return None

    text = clean_text(text)

    if text.startswith("[Request interrupted"):
        return ("interrupt", text)
    if "<local-command-stdout>" in text:
        return None
    if text.startswith("Caveat: The messages below were generated by the user while running"):
        return None
    if "<command-name>" in text:
        tags = dict(TAG_RE.findall(text))
        name = (tags.get("command-name") or "").strip()
        args = (tags.get("command-args") or "").strip()
        if not name:
            return None
        return ("command", (name + " " + args).strip())
    if text.startswith("This session is being continued from a previous conversation"):
        return ("resume", "(前の会話からの継続 / コンパクト後の再開)")

    if extras:
        text = (text + " " if text else "") + "[" + "+".join(extras) + "]"
    if not text:
        return None
    return ("prompt", text)


def new_turn(index, kind, text, ts):
    return {
        "i": index,
        "kind": kind,
        "prompt": text[:PROMPT_LIMIT],
        "truncated": len(text) > PROMPT_LIMIT,
        "ts": ts,
        "firstReplyTs": None,
        "endTs": ts,
        "assistant": 0,
        "toolCalls": 0,
        "tools": {},
        "sidechain": 0,
        "interrupted": False,
        "models": [],
    }


def parse_file(path: Path):
    """1セッションの jsonl を解析して dict を返す。"""
    turns = []
    titles = {}
    meta = {}
    counts = {"user": 0, "assistant": 0, "toolResult": 0, "sidechain": 0}
    start = end = None
    pending_prompt = None  # turn 開始待ち (queue-operation より後の user 本体を採用)

    with path.open("rb") as fh:
        for raw in fh:
            raw = raw.strip()
            if not raw:
                continue

            # tool_result 行は巨大なので JSON パースせず時刻だけ拾う
            if b'"tool_result"' in raw and b'"type":"user"' in raw:
                match = TS_RE.search(raw)
                ts = parse_ts(match.group(1).decode("utf-8")) if match else None
                counts["toolResult"] += 1
                if ts is not None:
                    start = ts if start is None else min(start, ts)
                    end = ts if end is None else max(end, ts)
                    if turns:
                        turns[-1]["endTs"] = max(turns[-1]["endTs"] or ts, ts)
                continue

            try:
                obj = json.loads(raw)
            except (ValueError, UnicodeDecodeError):
                continue

            rtype = obj.get("type")

            if rtype == "custom-title":
                titles["custom"] = obj.get("customTitle")
                continue
            if rtype == "ai-title":
                titles["ai"] = obj.get("aiTitle")
                continue
            if rtype == "last-prompt":
                titles["last"] = obj.get("lastPrompt")
                continue

            ts = parse_ts(obj.get("timestamp"))
            if ts is not None:
                start = ts if start is None else min(start, ts)
                end = ts if end is None else max(end, ts)

            for key in ("cwd", "gitBranch", "version", "slug", "sessionId"):
                if not meta.get(key) and obj.get(key):
                    meta[key] = obj[key]

            if rtype == "user":
                counts["user"] += 1
                parsed = classify_user(obj)
                if not parsed:
                    if obj.get("isSidechain"):
                        counts["sidechain"] += 1
                        if turns and ts is not None:
                            turns[-1]["sidechain"] += 1
                            turns[-1]["endTs"] = max(turns[-1]["endTs"] or ts, ts)
                    continue
                kind, text = parsed
                if kind == "interrupt":
                    if turns:
                        turns[-1]["interrupted"] = True
                        if ts is not None:
                            turns[-1]["endTs"] = max(turns[-1]["endTs"] or ts, ts)
                    continue
                pending_prompt = None
                turns.append(new_turn(len(turns) + 1, kind, text, ts))
                continue

            if rtype == "assistant":
                if obj.get("isSidechain"):
                    counts["sidechain"] += 1
                    if turns and ts is not None:
                        turns[-1]["sidechain"] += 1
                        turns[-1]["endTs"] = max(turns[-1]["endTs"] or ts, ts)
                    continue
                counts["assistant"] += 1
                if turns:
                    turn = turns[-1]
                    turn["assistant"] += 1
                    if ts is not None:
                        if turn["firstReplyTs"] is None:
                            turn["firstReplyTs"] = ts
                        turn["endTs"] = max(turn["endTs"] or ts, ts)
                    model = (obj.get("message") or {}).get("model")
                    if model and model not in turn["models"]:
                        turn["models"].append(model)
                    content = (obj.get("message") or {}).get("content")
                    if isinstance(content, list):
                        for block in content:
                            if isinstance(block, dict) and block.get("type") == "tool_use":
                                name = block.get("name") or "?"
                                turn["tools"][name] = turn["tools"].get(name, 0) + 1
                                turn["toolCalls"] += 1
                continue

            # その他 (system / attachment / mode / queue-operation) はセッション範囲にのみ反映。
            # 再開時に書かれる事があり、ターンの所要時間を伸ばしてしまうため無視する。

    _ = pending_prompt  # 予約

    # ターンごとの派生値
    prev_end = None
    active = 0.0
    tool_total = 0
    for turn in turns:
        ts = turn["ts"]
        turn["waitMs"] = ms(turn["firstReplyTs"], ts)
        turn["durMs"] = ms(turn["endTs"], ts)
        turn["idleMs"] = ms(ts, prev_end)
        turn["tsIso"] = iso(ts)
        turn["endIso"] = iso(turn["endTs"])
        turn["summary"] = one_line(turn["prompt"], SUMMARY_LIMIT)
        turn["toolList"] = sorted(turn["tools"].items(), key=lambda kv: (-kv[1], kv[0]))
        tool_total += turn["toolCalls"]
        if turn["durMs"]:
            active += turn["durMs"]
        prev_end = turn["endTs"] if turn["endTs"] is not None else prev_end

    first_prompt = next((t["prompt"] for t in turns if t["kind"] == "prompt"), "")
    title = (
        titles.get("custom")
        or titles.get("ai")
        or one_line(first_prompt, 120)
        or titles.get("last")
        or meta.get("slug")
        or path.stem
    )

    stat = path.stat()
    return {
        "id": path.stem,
        "file": str(path),
        "title": title,
        "titleSource": ("custom" if titles.get("custom") else "ai" if titles.get("ai") else "prompt"),
        "cwd": meta.get("cwd"),
        "gitBranch": meta.get("gitBranch"),
        "version": meta.get("version"),
        "slug": meta.get("slug"),
        "start": iso(start),
        "end": iso(end),
        "startEpoch": start,
        "endEpoch": end,
        "wallMs": ms(end, start),
        "activeMs": int(active) if active else 0,
        "turnCount": sum(1 for t in turns if t["kind"] in ("prompt", "command")),
        "userMsgs": counts["user"],
        "assistantMsgs": counts["assistant"],
        "toolResults": counts["toolResult"],
        "sidechainMsgs": counts["sidechain"],
        "toolCalls": tool_total,
        "bytes": stat.st_size,
        "mtime": stat.st_mtime,
        "turns": turns,
    }


def ms(later, earlier):
    if later is None or earlier is None:
        return None
    return int(round((later - earlier) * 1000))


def one_line(text, limit):
    text = " ".join((text or "").split())
    return text[:limit] + ("…" if len(text) > limit else "")


def summary_of(session):
    out = {k: v for k, v in session.items() if k != "turns"}
    turns = session["turns"]
    out["firstPrompt"] = one_line(next((t["prompt"] for t in turns if t["kind"] == "prompt"), ""), SUMMARY_LIMIT)
    out["lastPrompt"] = one_line(turns[-1]["prompt"] if turns else "", SUMMARY_LIMIT)
    models = []
    for turn in turns:
        for model in turn["models"]:
            if model not in models:
                models.append(model)
    out["models"] = models
    return out


# ---------------------------------------------------------------- ストア


class Store:
    """jsonl の解析結果を (mtime, size) でキャッシュする。"""

    def __init__(self, root: Path):
        self.root = root
        self._cache = {}
        self._lock = threading.Lock()

    def projects(self):
        items = []
        if not self.root.is_dir():
            return items
        for entry in sorted(self.root.iterdir()):
            if not entry.is_dir():
                continue
            files = list(entry.glob("*.jsonl"))
            if not files:
                continue
            stats = [f.stat() for f in files]
            newest = max(files, key=lambda f: f.stat().st_mtime)
            items.append(
                {
                    "id": entry.name,
                    "path": sniff_cwd(newest) or decode_project_dir(entry.name),
                    "sessions": len(files),
                    "bytes": sum(s.st_size for s in stats),
                    "lastActivity": iso(max(s.st_mtime for s in stats)),
                    "lastEpoch": max(s.st_mtime for s in stats),
                }
            )
        items.sort(key=lambda p: p["lastEpoch"], reverse=True)
        return items

    def session_files(self, project_id):
        directory = self.root / project_id
        if not directory.is_dir() or os.path.sep in project_id or project_id in ("..", "."):
            raise FileNotFoundError(project_id)
        return sorted(directory.glob("*.jsonl"))

    def session(self, path: Path):
        stat = path.stat()
        key = str(path)
        with self._lock:
            hit = self._cache.get(key)
            if hit and hit[0] == (stat.st_mtime, stat.st_size):
                return hit[1]
        parsed = parse_file(path)
        with self._lock:
            self._cache[key] = ((stat.st_mtime, stat.st_size), parsed)
        return parsed

    def sessions(self, project_id):
        out = [summary_of(self.session(p)) for p in self.session_files(project_id)]
        out.sort(key=lambda s: s["startEpoch"] or 0, reverse=True)
        return out

    def one(self, project_id, session_id):
        for path in self.session_files(project_id):
            if path.stem == session_id:
                return self.session(path)
        raise FileNotFoundError(session_id)

    def clear(self):
        with self._lock:
            self._cache.clear()


CWD_RE = re.compile(rb'"cwd":"((?:[^"\\]|\\.)*)"')


def sniff_cwd(path: Path, max_lines=200):
    """jsonl の先頭から cwd を拾う (プロジェクトの実パス表示用)。"""
    try:
        with path.open("rb") as fh:
            for _, raw in zip(range(max_lines), fh):
                match = CWD_RE.search(raw)
                if match:
                    return json.loads(b'"' + match.group(1) + b'"')
    except OSError:
        pass
    return None


def decode_project_dir(name):
    """'C--debug-ImgChkSystem' -> 'C:\\debug\\ImgChkSystem' (推定)。"""
    if len(name) > 3 and name[1:3] == "--" and name[0].isalpha():
        return name[0].upper() + ":\\" + name[3:].replace("-", "\\")
    return name.replace("-", "/")


# ---------------------------------------------------------------- HTTP


def fmt_dur(msec):
    if msec is None:
        return ""
    seconds = msec / 1000.0
    if seconds < 60:
        return f"{seconds:.1f}s"
    minutes, seconds = divmod(int(round(seconds)), 60)
    if minutes < 60:
        return f"{minutes}m{seconds:02d}s"
    hours, minutes = divmod(minutes, 60)
    return f"{hours}h{minutes:02d}m"


def sessions_csv(sessions):
    buf = io.StringIO(newline="")
    writer = csv.writer(buf)
    writer.writerow(["session_id", "title", "start", "end", "wall", "active", "turns", "assistant_msgs", "tool_calls", "branch"])
    for s in sessions:
        writer.writerow(
            [s["id"], s["title"], s["start"] or "", s["end"] or "", fmt_dur(s["wallMs"]),
             fmt_dur(s["activeMs"]), s["turnCount"], s["assistantMsgs"], s["toolCalls"], s.get("gitBranch") or ""]
        )
    return buf.getvalue()


def turns_csv(session):
    buf = io.StringIO(newline="")
    writer = csv.writer(buf)
    writer.writerow(["#", "kind", "time", "end", "first_reply", "duration", "idle_before", "assistant_msgs", "tool_calls", "prompt"])
    for t in session["turns"]:
        writer.writerow(
            [t["i"], t["kind"], t["tsIso"] or "", t["endIso"] or "", fmt_dur(t["waitMs"]), fmt_dur(t["durMs"]),
             fmt_dur(t["idleMs"]), t["assistant"], t["toolCalls"], one_line(t["prompt"], 500)]
        )
    return buf.getvalue()


class Server(ThreadingHTTPServer):
    # Windows では SO_REUSEADDR が「使用中のポートにも bind できる」意味になり、
    # 二重起動が検出できなくなるため無効にする。
    allow_reuse_address = False
    daemon_threads = True


class Handler(BaseHTTPRequestHandler):
    server_version = "chktime"
    store: Store = None  # set on the server class

    def log_message(self, fmt, *args):  # 静かに
        pass

    def do_GET(self):
        parsed = urlparse(self.path)
        route = parsed.path
        query = parse_qs(parsed.query)
        try:
            if route.startswith("/api/"):
                return self.api(route, query)
            return self.static(route)
        except FileNotFoundError as exc:
            self.send_json({"error": f"not found: {exc}"}, 404)
        except Exception as exc:  # noqa: BLE001 - UI にエラーを返す
            self.send_json({"error": f"{type(exc).__name__}: {exc}"}, 500)

    def api(self, route, query):
        store = self.server.store
        if route == "/api/projects":
            return self.send_json({"root": str(store.root), "projects": store.projects()})
        if route == "/api/sessions":
            project = query.get("project", [""])[0]
            sessions = store.sessions(project)
            if query.get("format", [""])[0] == "csv":
                return self.send_text(sessions_csv(sessions), "text/csv", f"{project}-sessions.csv")
            return self.send_json({"project": project, "sessions": sessions})
        if route == "/api/session":
            project = query.get("project", [""])[0]
            session_id = query.get("id", [""])[0]
            session = store.one(project, session_id)
            if query.get("format", [""])[0] == "csv":
                return self.send_text(turns_csv(session), "text/csv", f"{session_id}-turns.csv")
            return self.send_json(session)
        if route == "/api/reload":
            store.clear()
            return self.send_json({"ok": True})
        raise FileNotFoundError(route)

    def static(self, route):
        name = "index.html" if route in ("/", "") else route.lstrip("/")
        target = (WEB_DIR / name).resolve()
        if not str(target).startswith(str(WEB_DIR)) or not target.is_file():
            raise FileNotFoundError(name)
        ctype = {
            ".html": "text/html; charset=utf-8",
            ".js": "text/javascript; charset=utf-8",
            ".css": "text/css; charset=utf-8",
        }.get(target.suffix, "application/octet-stream")
        body = target.read_bytes()
        self.send_response(200)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def send_json(self, payload, status=200):
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def send_text(self, text, ctype, filename):
        body = ("\ufeff" + text).encode("utf-8")  # Excel 対応の BOM
        self.send_response(200)
        self.send_header("Content-Type", f"{ctype}; charset=utf-8")
        self.send_header("Content-Disposition", f'attachment; filename="{filename}"')
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


def main():
    parser = argparse.ArgumentParser(description="Claude Code のタスク別やりとり時刻ビューア")
    parser.add_argument("--port", type=int, default=8787)
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--dir", default=str(Path.home() / ".claude" / "projects"), help="projects ディレクトリ")
    parser.add_argument("--no-browser", action="store_true")
    args = parser.parse_args()

    root = Path(args.dir).expanduser()
    if not root.is_dir():
        raise SystemExit(f"見つかりません: {root}")

    url = f"http://{args.host}:{args.port}/"
    try:
        server = Server((args.host, args.port), Handler)
    except OSError as exc:
        if exc.errno not in (errno.EADDRINUSE, getattr(errno, "WSAEADDRINUSE", 10048)):
            raise
        # 既に起動済み: ブラウザだけ開いて終了する (ショートカットの二重起動対策)
        print(f"すでに起動しています: {url}")
        if not args.no_browser:
            webbrowser.open(url)
        return
    server.store = Store(root)
    print(f"chktime: {root}")
    print(f"  -> {url}  (Ctrl+C で終了)")
    if not args.no_browser:
        threading.Timer(0.4, lambda: webbrowser.open(url)).start()
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nbye")


if __name__ == "__main__":
    main()