#!/usr/bin/env python3 """Статистика по транскриптам сессий Claude Code для ретро обвязки. Только чтение. Вывод: таблица по сессиям + агрегаты + примеры блокировок.""" import json, sys, glob, os, re, collections, datetime d = sys.argv[1] files = sorted(glob.glob(os.path.join(d, "*.jsonl"))) rows = [] agg = collections.Counter() blocked_examples = collections.defaultdict(list) denied_examples = [] system_subtypes = collections.Counter() hook_names = collections.Counter() skills = collections.Counter() agent_models = collections.Counter() agent_types = collections.Counter() first_prompts = {} PAT_BLOCK = re.compile(r"BLOCKED[^\n]{0,160}") PAT_DENY = re.compile(r"(Auto-Mode Bypass|permission denied|user denied|denied by|The user doesn't want|rejected this tool)", re.I) def text_of(content): if isinstance(content, str): return content out = [] if isinstance(content, list): for b in content: if isinstance(b, dict): if b.get("type") == "text": out.append(b.get("text", "")) elif b.get("type") == "tool_result": out.append(text_of(b.get("content", ""))) return "\n".join(out) for f in files: sid = os.path.basename(f)[:8] c = collections.Counter() title = "" pm = "" t0 = t1 = None usage_in = usage_out = cache_r = cache_c = 0 human_prompts = 0 first_prompt = "" for line in open(f, encoding="utf-8", errors="replace"): try: j = json.loads(line) except Exception: continue t = j.get("type") if t == "ai-title" and not title: title = j.get("aiTitle", "") if t == "permission-mode": pm = j.get("permissionMode", "") ts = j.get("timestamp") if ts: try: tt = datetime.datetime.fromisoformat(ts.replace("Z", "+00:00")) t0 = tt if t0 is None or tt < t0 else t0 t1 = tt if t1 is None or tt > t1 else t1 except Exception: pass if t == "system": system_subtypes[j.get("subtype", "?")] += 1 if j.get("subtype") == "compact_boundary" or "compact" in str(j.get("subtype", "")): c["compact"] += 1 hn = j.get("hookName") or j.get("hook_name") if hn: hook_names[hn] += 1 if t == "user": m = j.get("message", {}) cont = m.get("content", "") if isinstance(cont, str): if not cont.startswith("<") and j.get("userType", "external") == "external" and not j.get("isMeta"): human_prompts += 1 if not first_prompt: first_prompt = cont[:120].replace("\n", " ") elif isinstance(cont, list): has_tr = any(isinstance(b, dict) and b.get("type") == "tool_result" for b in cont) if not has_tr and not j.get("isMeta"): txt = text_of(cont) if txt and not txt.startswith("<"): human_prompts += 1 if not first_prompt: first_prompt = txt[:120].replace("\n", " ") for b in cont: if isinstance(b, dict) and b.get("type") == "tool_result": txt = text_of(b.get("content", "")) for mm in PAT_BLOCK.findall(txt): c["blocked"] += 1 key = re.sub(r"\s+", " ", mm)[:90] if len(blocked_examples[key]) < 2: blocked_examples[key].append(sid) if PAT_DENY.search(txt): c["denied"] += 1 if len(denied_examples) < 15: denied_examples.append((sid, re.sub(r"\s+", " ", txt)[:160])) # hook feedback in attachments / text txt = text_of(cont) if "BLOCKED" in txt and isinstance(cont, str): c["blocked_text"] += 1 if t == "assistant": m = j.get("message", {}) u = m.get("usage", {}) or {} usage_in += u.get("input_tokens", 0) or 0 usage_out += u.get("output_tokens", 0) or 0 cache_r += u.get("cache_read_input_tokens", 0) or 0 cache_c += u.get("cache_creation_input_tokens", 0) or 0 c["assistant"] += 1 for b in m.get("content", []) or []: if isinstance(b, dict) and b.get("type") == "tool_use": n = b.get("name", "") c["tool"] += 1 c["t:" + n] += 1 agg["t:" + n] += 1 inp = b.get("input", {}) or {} if n in ("Agent", "Task"): agent_models[inp.get("model", "default")] += 1 agent_types[inp.get("subagent_type", "?")] += 1 if n == "Skill": skills[inp.get("skill", "?")] += 1 dur = (t1 - t0).total_seconds() / 3600 if t0 and t1 else 0 rows.append(dict(sid=sid, date=t0.strftime("%m-%d") if t0 else "?", hours=round(dur, 1), title=title[:38], pm=pm, prompts=human_prompts, asst=c["assistant"], tools=c["tool"], agents=c["t:Agent"] + c["t:Task"], blocked=c["blocked"], denied=c["denied"], compact=c["compact"], out_k=usage_out // 1000, cache_r_m=round(cache_r / 1e6, 1), first=first_prompt[:70])) for k in ("assistant", "tool", "blocked", "denied", "compact"): agg[k] += c[k] agg["prompts"] += human_prompts agg["out"] += usage_out agg["cache_r"] += cache_r agg["cache_c"] += cache_c agg["hours"] += dur rows.sort(key=lambda r: r["date"]) hdr = ["sid", "date", "hours", "pm", "prompts", "asst", "tools", "agents", "blocked", "denied", "compact", "out_k", "cache_r_m", "title"] print("\t".join(hdr)) for r in rows: print("\t".join(str(r[h]) for h in hdr)) print("\n== Первые промпты сессий") for r in rows: print(f"{r['date']} {r['sid']} | {r['first']}") print("\n== Агрегаты") for k in ("prompts", "assistant", "tool", "blocked", "denied", "compact"): print(f"{k}: {agg[k]}") print(f"hours: {agg['hours']:.0f}; output tokens: {agg['out']/1e6:.1f}M; cache read: {agg['cache_r']/1e9:.2f}G; cache create: {agg['cache_c']/1e6:.0f}M") print("\n== Инструменты (top 25)") for k, v in sorted(((k, v) for k, v in agg.items() if k.startswith("t:")), key=lambda x: -x[1])[:25]: print(f"{k[2:]}: {v}") print("\n== Skills"); [print(f"{k}: {v}") for k, v in skills.most_common()] print("\n== Модели сабагентов"); [print(f"{k}: {v}") for k, v in agent_models.most_common()] print("\n== Типы сабагентов"); [print(f"{k}: {v}") for k, v in agent_types.most_common()] print("\n== system subtypes"); [print(f"{k}: {v}") for k, v in system_subtypes.most_common(15)] print("\n== hookName в system"); [print(f"{k}: {v}") for k, v in hook_names.most_common(15)] print("\n== Блокировки хуков (уникальные тексты, top 40)") for k, v in sorted(blocked_examples.items(), key=lambda x: -len(x[1]))[:40]: print(f"[{','.join(v)}] {k}") print("\n== Отказы (примеры)") for s, e in denied_examples: print(f"[{s}] {e}")