0733ffd9c9
Находки RT-1…RT-20, бэклог B1…B12, решения владельцу; скрипты подсчёта по транскриптам (retro_stats.py, retro_blocks.py) как основа регулярного ретро. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
160 lines
7.4 KiB
Python
160 lines
7.4 KiB
Python
#!/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}")
|