yoshikazuandClaude Opus 5
0205404797
薄いバーで開始〜終了の全体、濃い部分で応答が動いていた区間を重ねて描く。 放置していた時間や、セッションを並行して動かしていた時間が見える。 - /api/timeline: セッションごとに全体の範囲と応答中の区間 (spans) を返す (隙間 1 秒以内のターンは結合、期間指定 days= / 全期間、CSV も可) - ガントタブ: 1日/3日/7日/14日/30日/全期間の切替、1〜3日では時刻目盛りに 自動で切替、土日の帯・日境界線・現在時刻の点線、横断時はプロジェクト別の色 - 行のホバーで内訳、クリックでそのタスクのやりとりを開く - ラベル列とバーの行を22pxで揃え、時間軸ヘッダは縦スクロール中も固定 - ペイン幅に応じてラベル幅を詰め、横スクロールを出さない Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
827 lines
31 KiB
Python
827 lines
31 KiB
Python
#!/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, timedelta, 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 all_sessions(self):
|
|
"""全プロジェクトのタスクを1つのリストにまとめる (開始が新しい順)。"""
|
|
out = []
|
|
for project in self.projects():
|
|
for path in self.session_files(project["id"]):
|
|
summary = summary_of(self.session(path))
|
|
summary["project"] = project["id"]
|
|
summary["projectPath"] = summary.get("cwd") or project["path"]
|
|
out.append(summary)
|
|
out.sort(key=lambda s: s["startEpoch"] or 0, reverse=True)
|
|
return out
|
|
|
|
def daily_stats(self, project_id=None, days=30):
|
|
"""ターンをローカル日付で集計する。project_id=None なら全プロジェクト。
|
|
|
|
days>0 なら「今日から遡って days 日」、それ以外は全期間。
|
|
活動のない日も 0 の行として埋めるので、そのまま横軸に使える。
|
|
"""
|
|
paths = {p["id"]: p["path"] for p in self.projects()}
|
|
if project_id:
|
|
if project_id not in paths:
|
|
raise FileNotFoundError(project_id)
|
|
targets = [project_id]
|
|
else:
|
|
targets = list(paths)
|
|
|
|
today = datetime.now().date()
|
|
cutoff = today - timedelta(days=days - 1) if days and days > 0 else None
|
|
|
|
per_day = {}
|
|
per_project = {}
|
|
seen = set()
|
|
for pid in targets:
|
|
for path in self.session_files(pid):
|
|
session = self.session(path)
|
|
for turn in session["turns"]:
|
|
if turn["ts"] is None:
|
|
continue
|
|
day = datetime.fromtimestamp(turn["ts"]).date()
|
|
if cutoff and day < cutoff:
|
|
continue
|
|
dur = turn["durMs"] or 0
|
|
counted = 1 if turn["kind"] in ("prompt", "command") else 0
|
|
bucket = per_day.setdefault(day, {"activeMs": 0, "turns": 0, "sessions": set(), "byProject": {}})
|
|
bucket["activeMs"] += dur
|
|
bucket["turns"] += counted
|
|
bucket["sessions"].add(session["id"])
|
|
bucket["byProject"][pid] = bucket["byProject"].get(pid, 0) + dur
|
|
total = per_project.setdefault(pid, {"activeMs": 0, "turns": 0, "sessions": set()})
|
|
total["activeMs"] += dur
|
|
total["turns"] += counted
|
|
total["sessions"].add(session["id"])
|
|
seen.add(session["id"])
|
|
|
|
if per_day:
|
|
start = cutoff or min(per_day)
|
|
end = max(max(per_day), today)
|
|
elif cutoff:
|
|
start, end = cutoff, today # データなしでも指定範囲は 0 で埋める
|
|
else:
|
|
start = end = None
|
|
|
|
rows = []
|
|
cur = start
|
|
while cur and cur <= end:
|
|
bucket = per_day.get(cur)
|
|
rows.append(
|
|
{
|
|
"date": cur.isoformat(),
|
|
"dow": cur.weekday(),
|
|
"activeMs": bucket["activeMs"] if bucket else 0,
|
|
"turns": bucket["turns"] if bucket else 0,
|
|
"sessions": len(bucket["sessions"]) if bucket else 0,
|
|
"byProject": bucket["byProject"] if bucket else {},
|
|
}
|
|
)
|
|
cur += timedelta(days=1)
|
|
|
|
projects = sorted(
|
|
(
|
|
{
|
|
"id": pid,
|
|
"path": paths.get(pid, pid),
|
|
"activeMs": v["activeMs"],
|
|
"turns": v["turns"],
|
|
"sessions": len(v["sessions"]),
|
|
}
|
|
for pid, v in per_project.items()
|
|
),
|
|
key=lambda p: -p["activeMs"],
|
|
)
|
|
busiest = max(rows, key=lambda r: r["activeMs"], default=None)
|
|
return {
|
|
"scope": project_id or "",
|
|
"rangeDays": days if days and days > 0 else 0,
|
|
"days": rows,
|
|
"projects": projects,
|
|
"totals": {
|
|
"activeMs": sum(r["activeMs"] for r in rows),
|
|
"turns": sum(r["turns"] for r in rows),
|
|
"sessions": len(seen),
|
|
"activeDays": sum(1 for r in rows if r["activeMs"] or r["turns"]),
|
|
"busiestDate": busiest["date"] if busiest and busiest["activeMs"] else None,
|
|
"busiestMs": busiest["activeMs"] if busiest else 0,
|
|
},
|
|
}
|
|
|
|
def timeline(self, project_id=None, days=7):
|
|
"""ガント表示用。セッションを1本のバーとして、内側に応答中の区間を返す。
|
|
|
|
spans は各ターンの [発言時刻, 応答完了] を、隙間 1 秒以内なら繋げたもの。
|
|
"""
|
|
paths = {p["id"]: p["path"] for p in self.projects()}
|
|
if project_id:
|
|
if project_id not in paths:
|
|
raise FileNotFoundError(project_id)
|
|
targets = [project_id]
|
|
else:
|
|
targets = list(paths)
|
|
|
|
now = datetime.now().timestamp()
|
|
t0 = None
|
|
if days and days > 0:
|
|
midnight = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
|
|
t0 = (midnight - timedelta(days=days - 1)).timestamp()
|
|
|
|
rows = []
|
|
per_project = {}
|
|
for pid in targets:
|
|
for path in self.session_files(pid):
|
|
session = self.session(path)
|
|
start, end = session["startEpoch"], session["endEpoch"]
|
|
if start is None:
|
|
continue
|
|
if t0 is not None and (end if end is not None else start) < t0:
|
|
continue
|
|
spans = []
|
|
active = 0
|
|
for turn in session["turns"]:
|
|
a = turn["ts"]
|
|
if a is None:
|
|
continue
|
|
b = turn["endTs"] if turn["endTs"] is not None else a
|
|
if t0 is not None and b < t0:
|
|
continue
|
|
active += turn["durMs"] or 0
|
|
if spans and a - spans[-1][1] <= 1:
|
|
spans[-1][1] = max(spans[-1][1], b)
|
|
else:
|
|
spans.append([a, b])
|
|
rows.append(
|
|
{
|
|
"id": session["id"],
|
|
"project": pid,
|
|
"projectPath": session.get("cwd") or paths.get(pid, pid),
|
|
"title": session["title"],
|
|
"start": session["start"],
|
|
"end": session["end"],
|
|
"startEpoch": round(start, 1),
|
|
"endEpoch": round(end if end is not None else start, 1),
|
|
"wallMs": session["wallMs"],
|
|
"activeMs": active,
|
|
"turnCount": session["turnCount"],
|
|
"toolCalls": session["toolCalls"],
|
|
"spans": [[round(a, 1), round(b, 1)] for a, b in spans],
|
|
}
|
|
)
|
|
total = per_project.setdefault(pid, {"activeMs": 0, "sessions": 0})
|
|
total["activeMs"] += active
|
|
total["sessions"] += 1
|
|
|
|
rows.sort(key=lambda s: s["startEpoch"])
|
|
if t0 is None:
|
|
t0 = min((s["startEpoch"] for s in rows), default=now)
|
|
t1 = max([now] + [s["endEpoch"] for s in rows])
|
|
projects = sorted(
|
|
(
|
|
{"id": pid, "path": paths.get(pid, pid), "activeMs": v["activeMs"], "sessions": v["sessions"]}
|
|
for pid, v in per_project.items()
|
|
),
|
|
key=lambda p: -p["activeMs"],
|
|
)
|
|
return {
|
|
"scope": project_id or "",
|
|
"rangeDays": days if days and days > 0 else 0,
|
|
"from": round(t0, 1),
|
|
"to": round(t1, 1),
|
|
"now": round(now, 1),
|
|
"sessions": rows,
|
|
"projects": projects,
|
|
}
|
|
|
|
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 parse_days(value):
|
|
"""?days= の解釈。'all' / 0 / 不正値は全期間 (0) とみなす。"""
|
|
try:
|
|
days = int(value)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
return max(0, min(days, 3650))
|
|
|
|
|
|
def sessions_csv(sessions, with_project=False):
|
|
buf = io.StringIO(newline="")
|
|
writer = csv.writer(buf)
|
|
head = ["session_id", "title", "start", "end", "wall", "active", "turns", "assistant_msgs", "tool_calls", "branch"]
|
|
writer.writerow((["project"] if with_project else []) + head)
|
|
for s in sessions:
|
|
writer.writerow(
|
|
([s.get("projectPath") or s.get("project") or ""] if with_project else [])
|
|
+ [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 daily_csv(data):
|
|
buf = io.StringIO(newline="")
|
|
writer = csv.writer(buf)
|
|
projects = data["projects"]
|
|
writer.writerow(["date", "dow", "active", "active_ms", "turns", "sessions"] + [p["path"] for p in projects])
|
|
labels = "月火水木金土日"
|
|
for row in data["days"]:
|
|
writer.writerow(
|
|
[row["date"], labels[row["dow"]], fmt_dur(row["activeMs"]), row["activeMs"], row["turns"], row["sessions"]]
|
|
+ [fmt_dur(row["byProject"].get(p["id"])) for p in projects]
|
|
)
|
|
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()
|
|
|
|
|
|
def timeline_csv(data):
|
|
buf = io.StringIO(newline="")
|
|
writer = csv.writer(buf)
|
|
writer.writerow(["project", "session_id", "title", "start", "end", "wall", "active", "turns", "tool_calls"])
|
|
for s in data["sessions"]:
|
|
writer.writerow(
|
|
[s["projectPath"], s["id"], s["title"], s["start"] or "", s["end"] or "",
|
|
fmt_dur(s["wallMs"]), fmt_dur(s["activeMs"]), s["turnCount"], s["toolCalls"]]
|
|
)
|
|
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/all":
|
|
sessions = store.all_sessions()
|
|
if query.get("format", [""])[0] == "csv":
|
|
return self.send_text(sessions_csv(sessions, True), "text/csv", "all-sessions.csv")
|
|
return self.send_json({"project": "", "sessions": sessions})
|
|
if route == "/api/daily":
|
|
project = query.get("project", [""])[0]
|
|
data = store.daily_stats(project or None, parse_days(query.get("days", ["30"])[0]))
|
|
if query.get("format", [""])[0] == "csv":
|
|
return self.send_text(daily_csv(data), "text/csv", f"{project or 'all'}-daily.csv")
|
|
return self.send_json(data)
|
|
if route == "/api/timeline":
|
|
project = query.get("project", [""])[0]
|
|
data = store.timeline(project or None, parse_days(query.get("days", ["7"])[0]))
|
|
if query.get("format", [""])[0] == "csv":
|
|
return self.send_text(timeline_csv(data), "text/csv", f"{project or 'all'}-timeline.csv")
|
|
return self.send_json(data)
|
|
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()
|