← back to Cc Command Center

build_stats.py

193 lines

#!/usr/bin/env python3
"""Aggregate real Claude Code usage from ~/.claude/projects/**/*.jsonl into
public/data/stats.json — powers the command-center web viewer.

Recreates the "What's up next?" dashboard from the LLMJunky tweet using Steve's
own session data: stat cards, a 53-week activity heatmap, and a recents list
rendered as draggable floating session windows.
"""
import os, json, glob, datetime as dt
from collections import defaultdict, Counter

HOME = os.path.expanduser("~")
ROOT = os.path.join(HOME, ".claude", "projects")
OUT  = os.path.join(os.path.dirname(__file__), "public", "data", "stats.json")
USER = "Steve"

def parse_ts(s):
    if not s:
        return None
    try:
        return dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
    except Exception:
        return None

sessions = []                 # one summary dict per session file
day_counts = Counter()        # date -> assistant-message count (activity)
hour_counts = Counter()       # hour -> count
model_counts = Counter()      # model -> assistant messages
total_tokens = 0
total_user_msgs = 0
active_dates = set()

files = glob.glob(os.path.join(ROOT, "**", "*.jsonl"), recursive=True)

for fp in files:
    title = None
    first_user = None
    last_snippet = None
    first_ts = None
    last_ts = None
    user_msgs = 0
    sess_tokens = 0
    sess_id = os.path.splitext(os.path.basename(fp))[0]
    project = os.path.basename(os.path.dirname(fp)).lstrip("-").replace("-", "/")
    try:
        with open(fp, "r", errors="ignore") as fh:
            for line in fh:
                line = line.strip()
                if not line:
                    continue
                try:
                    d = json.loads(line)
                except Exception:
                    continue
                t = d.get("type")
                if t == "ai-title":
                    title = (d.get("title") or d.get("text") or title)
                elif t == "last-prompt":
                    if not title:
                        title = (d.get("prompt") or d.get("text") or title)
                ts = parse_ts(d.get("timestamp"))
                if ts:
                    first_ts = first_ts or ts
                    last_ts = ts
                msg = d.get("message") if isinstance(d.get("message"), dict) else {}
                if t == "user" and msg.get("role") == "user":
                    user_msgs += 1
                    content = msg.get("content")
                    if isinstance(content, str) and content.strip() and not first_user:
                        first_user = content.strip()[:240]
                    if isinstance(content, str) and content.strip():
                        last_snippet = content.strip()[:240]
                elif t == "assistant" and msg.get("role") == "assistant":
                    u = msg.get("usage") or {}
                    tok = (u.get("input_tokens", 0) + u.get("output_tokens", 0)
                           + u.get("cache_creation_input_tokens", 0)
                           + u.get("cache_read_input_tokens", 0))
                    sess_tokens += tok
                    m = msg.get("model")
                    if m:
                        model_counts[m] += 1
                    if ts:
                        day_counts[ts.date().isoformat()] += 1
                        hour_counts[ts.hour] += 1
                        active_dates.add(ts.date())
    except Exception:
        continue

    if user_msgs == 0 and sess_tokens == 0:
        continue  # skip empty/metadata-only sessions

    total_tokens += sess_tokens
    total_user_msgs += user_msgs
    title = (title or first_user or "Untitled session").strip().replace("\n", " ")[:90]
    sessions.append({
        "id": sess_id,
        "project": project,
        "title": title,
        "snippet": (first_user or "")[:240],
        "last": (last_snippet or first_user or "")[:400],
        "user_msgs": user_msgs,
        "tokens": sess_tokens,
        "ts": last_ts.isoformat() if last_ts else None,
        "mtime": os.path.getmtime(fp),
    })

# ---- streaks ----
def streaks(dates):
    if not dates:
        return 0, 0
    ds = sorted(dates)
    longest = cur = 1
    for i in range(1, len(ds)):
        if (ds[i] - ds[i-1]).days == 1:
            cur += 1
            longest = max(longest, cur)
        elif (ds[i] - ds[i-1]).days == 0:
            continue
        else:
            cur = 1
    # current streak counts back from the most recent active day
    today = max(ds)
    current = 1
    i = len(ds) - 1
    while i > 0 and (ds[i] - ds[i-1]).days in (0, 1):
        if (ds[i] - ds[i-1]).days == 1:
            current += 1
        i -= 1
    return current, longest

current_streak, longest_streak = streaks(active_dates)
peak_hour = hour_counts.most_common(1)[0][0] if hour_counts else 0
fav_model = model_counts.most_common(1)[0][0] if model_counts else "—"

MODEL_PRETTY = {
    "claude-opus-4-8": "Opus 4.8", "claude-opus-4-8[1m]": "Opus 4.8",
    "claude-sonnet-4-6": "Sonnet 4.6", "claude-haiku-4-5-20251001": "Haiku 4.5",
    "claude-fable-5": "Fable 5",
}
def pretty_model(m):
    return MODEL_PRETTY.get(m, m.replace("claude-", "").replace("-", " ").title())

def fmt_hour(h):
    ap = "AM" if h < 12 else "PM"
    hh = h % 12 or 12
    return f"{hh} {ap}"

# ---- heatmap: last 53 weeks, GitHub-style ----
today = dt.date.today()
start = today - dt.timedelta(days=today.weekday() + 7*52)  # align to Monday a year back
heat = []
d = start
while d <= today:
    heat.append({"date": d.isoformat(), "count": day_counts.get(d.isoformat(), 0)})
    d += dt.timedelta(days=1)

sessions.sort(key=lambda s: s["mtime"], reverse=True)
recents = sessions[:24]

# token comparison flavor line (The Little Prince ~ 22k tokens)
LITTLE_PRINCE = 22000
mult = round(total_tokens / LITTLE_PRINCE) if total_tokens else 0

stats = {
    "user": USER,
    "generated_at": dt.datetime.now().isoformat(timespec="seconds"),
    "cards": {
        "sessions": len(sessions),
        "messages": total_user_msgs,
        "total_tokens": total_tokens,
        "active_days": len(active_dates),
        "current_streak": current_streak,
        "longest_streak": longest_streak,
        "peak_hour": fmt_hour(peak_hour),
        "favorite_model": pretty_model(fav_model),
    },
    "flavor": f"You've used ~{mult:,}× more tokens than The Little Prince.",
    "models": [{"model": pretty_model(m), "count": c} for m, c in model_counts.most_common(8)],
    "heatmap": heat,
    "recents": recents,
}

os.makedirs(os.path.dirname(OUT), exist_ok=True)
with open(OUT, "w") as fh:
    json.dump(stats, fh)

c = stats["cards"]
print(f"Wrote {OUT}")
print(f"  sessions={c['sessions']}  messages={c['messages']:,}  tokens={c['total_tokens']:,}")
print(f"  active_days={c['active_days']}  current_streak={c['current_streak']}  longest={c['longest_streak']}")
print(f"  peak_hour={c['peak_hour']}  fav_model={c['favorite_model']}")
print(f"  {stats['flavor']}")