ai_usage_analyzer.py

gistfile1.txt

#!/usr/bin/env python3
"""
AI usage analyzer — parses Claude Code transcripts (~/.claude) and claude.ai
data exports into a metrics dataset for pattern analysis.

Stdlib only. Works on any machine with Python 3.8+.

Subcommands:
  extract           Parse a Claude Code data dir (~/.claude or .../projects)
                    -> metrics JSON (+ prompts.jsonl unless --sanitize)
  extract-claudeai  Parse a claude.ai export (conversations.json or the zip)
                    -> metrics JSON (+ prompts.jsonl unless --sanitize)
  report            Merge one or more metrics JSONs -> combined summary JSON

Sanitized mode (--sanitize) exports NO message content: only per-session
numeric/boolean metrics, hashed project identifiers, and aggregate counts.
Review the output JSON before moving it off the machine.
"""

import argparse
import hashlib
import json
import math
import os
import re
import sys
import zipfile
from collections import Counter, defaultdict
from datetime import datetime, timezone

SCHEMA_VERSION = 3

# ---------------------------------------------------------------- helpers

CORRECTION_RE = re.compile(
    r"^\s*(no\b|nope\b|not\s|wrong\b|actually\b|wait\b|stop\b|undo\b|"
    r"revert\b|instead\b|that'?s\s+not\b|don'?t\b|hmm+\b|why\s+did\s+you\b|"
    r"you\s+(didn'?t|missed|broke|forgot)\b|still\s+(not|doesn'?t|broken)\b|"
    r"i\s+said\b|i\s+meant\b|try\s+again\b|redo\b|go\s+back\b)",
    re.IGNORECASE,
)
QUESTION_RE = re.compile(r"\?\s*$")
INTERRUPT_MARKERS = (
    "[Request interrupted by user",
    "[Request cancelled by user",
)
CONTEXT_CUES_RE = re.compile(
    r"\b(because|so that|the goal|context:|background:|for example|e\.g\.|"
    r"constraint|must not|should not|make sure|don'?t\s+change)\b",
    re.IGNORECASE,
)
NORMALIZE_RE = re.compile(r"[^a-z ]+")
WS_RE = re.compile(r"\s+")


def parse_ts(ts):
    if not ts:
        return None
    try:
        return datetime.fromisoformat(ts.replace("Z", "+00:00"))
    except (ValueError, TypeError):
        return None


def short_hash(text, salt=""):
    return hashlib.sha256((salt + text).encode("utf-8", "replace")).hexdigest()[:10]


def text_of_user_content(content):
    """Extract the human-typed text from a user message's content field.
    Returns (text, has_tool_result, tool_error_count)."""
    if isinstance(content, str):
        return content, False, 0
    texts, has_tr, errs = [], False, 0
    if isinstance(content, list):
        for block in content:
            if not isinstance(block, dict):
                continue
            btype = block.get("type")
            if btype == "text":
                texts.append(block.get("text") or "")
            elif btype == "tool_result":
                has_tr = True
                if block.get("is_error"):
                    errs += 1
    return "\n".join(texts), has_tr, errs


def strip_injected(text):
    """Remove system-injected wrappers to isolate what the human typed."""
    text = re.sub(r"<system-reminder>.*?</system-reminder>", "", text, flags=re.DOTALL)
    text = re.sub(r"<local-command-std(out|err)>.*?</local-command-std(out|err)>",
                  "", text, flags=re.DOTALL)
    text = re.sub(r"<command-(name|message|args)>.*?</command-\1>", "", text,
                  flags=re.DOTALL)
    return text.strip()


def normalized_template(text):
    """Normalize a prompt so near-identical asks hash the same."""
    t = text.lower()
    t = re.sub(r"[/~][\w./-]+", " P ", t)          # paths
    t = re.sub(r"https?://\S+", " U ", t)          # urls
    t = re.sub(r"\d+", " N ", t)                   # numbers
    t = NORMALIZE_RE.sub(" ", t)
    t = WS_RE.sub(" ", t).strip()
    return t


def pct(values, p):
    if not values:
        return None
    vs = sorted(values)
    k = (len(vs) - 1) * p
    f, c = math.floor(k), math.ceil(k)
    if f == c:
        return vs[f]
    return vs[f] + (vs[c] - vs[f]) * (k - f)


def summarize_dist(values):
    if not values:
        return None
    return {
        "n": len(values),
        "mean": round(sum(values) / len(values), 2),
        "p50": round(pct(values, 0.5), 2),
        "p90": round(pct(values, 0.9), 2),
        "max": round(max(values), 2),
    }


# ------------------------------------------------- Claude Code extraction

def iter_jsonl(path):
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    yield json.loads(line)
                except json.JSONDecodeError:
                    continue
    except OSError:
        return


def classify_tool(name):
    if not name:
        return "other"
    if name.startswith("mcp__"):
        return "mcp"
    return name


def extract_claude_code_session(path, sanitize, salt, prompts_out, source):
    """Parse one session .jsonl -> session metrics dict (or None if empty)."""
    events = list(iter_jsonl(path))
    if not events:
        return None

    s = {
        "source": source,
        "kind": "claude-code",
        "session_file": os.path.basename(path),
        "n_user_turns": 0,          # human-typed turns only
        "n_assistant_msgs": 0,
        "n_tool_calls": 0,
        "n_tool_errors": 0,
        "n_interrupts": 0,
        "n_corrections": 0,
        "n_questions": 0,
        "n_context_rich_prompts": 0,
        "n_slash_commands": 0,
        "n_sidechain_events": 0,
        "slash_commands": Counter(),
        "tools": Counter(),
        "models": Counter(),
        "prompt_chars": [],
        "first_prompt_chars": None,
        "tokens_in": 0,
        "tokens_out": 0,
        "tokens_cache_read": 0,
        "used_plan_mode": False,
        "used_subagents": False,
        "used_skills": False,
        "used_todos": False,
        "used_mcp": False,
        "used_web": False,
        "used_thinking": False,
        "compacted": False,
        "git_branch": None,
        "n_git_commits": 0,
        "prompt_template_hashes": [],
        "ended_after_interrupt": False,
    }

    first_ts = last_ts = None
    cwd = None
    version = None
    last_human_was_interrupt = False

    for ev in events:
        if not isinstance(ev, dict):
            continue
        etype = ev.get("type")
        ts = parse_ts(ev.get("timestamp"))
        if ts:
            first_ts = first_ts or ts
            last_ts = ts
        cwd = ev.get("cwd") or cwd
        version = ev.get("version") or version
        if ev.get("gitBranch"):
            s["git_branch"] = ev["gitBranch"]
        if ev.get("isSidechain"):
            s["n_sidechain_events"] += 1

        if etype == "summary":
            continue
        if ev.get("isCompactSummary") or etype == "compact":
            s["compacted"] = True
            continue

        msg = ev.get("message") or {}
        role = msg.get("role")

        if etype == "user" and role == "user":
            raw_text, has_tr, errs = text_of_user_content(msg.get("content"))
            s["n_tool_errors"] += errs
            if ev.get("isMeta"):
                continue
            text = strip_injected(raw_text)
            # slash command?
            cmd = re.search(r"<command-name>([^<]+)</command-name>", raw_text)
            if cmd:
                s["n_slash_commands"] += 1
                s["slash_commands"][cmd.group(1).strip()] += 1
                last_human_was_interrupt = False
                continue
            if any(m in raw_text for m in INTERRUPT_MARKERS):
                s["n_interrupts"] += 1
                last_human_was_interrupt = True
                continue
            if has_tr and not text:
                continue
            if not text:
                continue
            last_human_was_interrupt = False
            s["n_user_turns"] += 1
            s["prompt_chars"].append(len(text))
            if s["first_prompt_chars"] is None:
                s["first_prompt_chars"] = len(text)
            if CORRECTION_RE.search(text):
                s["n_corrections"] += 1
            if QUESTION_RE.search(text):
                s["n_questions"] += 1
            if CONTEXT_CUES_RE.search(text) or len(text) > 400:
                s["n_context_rich_prompts"] += 1
            s["prompt_template_hashes"].append(
                short_hash(normalized_template(text), salt))
            if not sanitize and prompts_out is not None:
                prompts_out.write(json.dumps({
                    "source": source,
                    "session": os.path.basename(path),
                    "ts": ev.get("timestamp"),
                    "cwd": cwd,
                    "text": text[:4000],
                }) + "\n")

        elif etype == "assistant" and role == "assistant":
            s["n_assistant_msgs"] += 1
            model = msg.get("model")
            if model:
                s["models"][model] += 1
            usage = msg.get("usage") or {}
            s["tokens_in"] += usage.get("input_tokens") or 0
            s["tokens_out"] += usage.get("output_tokens") or 0
            s["tokens_cache_read"] += usage.get("cache_read_input_tokens") or 0
            content = msg.get("content")
            if isinstance(content, list):
                for block in content:
                    if not isinstance(block, dict):
                        continue
                    btype = block.get("type")
                    if btype == "thinking":
                        s["used_thinking"] = True
                    elif btype == "tool_use":
                        name = block.get("name") or ""
                        s["n_tool_calls"] += 1
                        s["tools"][classify_tool(name)] += 1
                        if name in ("ExitPlanMode", "EnterPlanMode",
                                    "exit_plan_mode"):
                            s["used_plan_mode"] = True
                        elif name in ("Task", "Agent"):
                            s["used_subagents"] = True
                        elif name == "Skill":
                            s["used_skills"] = True
                        elif name in ("TodoWrite", "TaskCreate"):
                            s["used_todos"] = True
                        elif name.startswith("mcp__"):
                            s["used_mcp"] = True
                        elif name in ("WebSearch", "WebFetch"):
                            s["used_web"] = True
                        if name == "Bash":
                            cmd_str = (block.get("input") or {}).get("command", "")
                            s["n_git_commits"] += len(
                                re.findall(r"\bgit\s+commit\b", cmd_str))

    if s["n_user_turns"] == 0 and s["n_assistant_msgs"] == 0:
        return None

    s["ended_after_interrupt"] = last_human_was_interrupt
    s["start_ts"] = first_ts.isoformat() if first_ts else None
    s["duration_min"] = (
        round((last_ts - first_ts).total_seconds() / 60, 1)
        if first_ts and last_ts else None)
    s["hour_local"] = first_ts.astimezone().hour if first_ts else None
    s["weekday"] = first_ts.astimezone().strftime("%a") if first_ts else None
    s["cc_version"] = version

    project = cwd or os.path.basename(os.path.dirname(path))
    s["project"] = short_hash(project, salt) if sanitize else project

    if sanitize:
        s["session_file"] = short_hash(s["session_file"], salt)
        s["git_branch"] = bool(s["git_branch"])
        s["slash_commands"] = dict(
            (short_hash(k, salt) if not k.startswith("/") else k, v)
            for k, v in s["slash_commands"].items())
    else:
        s["slash_commands"] = dict(s["slash_commands"])

    s["tools"] = dict(s["tools"])
    s["models"] = dict(s["models"])
    s["prompt_chars_dist"] = summarize_dist(s.pop("prompt_chars"))
    return s


def find_project_dirs(root):
    """Accept ~/.claude, ~/.claude/projects, or a copied projects dir."""
    candidates = []
    projects = os.path.join(root, "projects")
    base = projects if os.path.isdir(projects) else root
    for dirpath, _dirnames, filenames in os.walk(base):
        for fn in filenames:
            if fn.endswith(".jsonl"):
                candidates.append(os.path.join(dirpath, fn))
    return base, sorted(candidates)


def extract_env_features(root, sanitize, salt):
    """What's configured in ~/.claude besides transcripts."""
    feats = {}
    def listdir(p):
        try:
            return os.listdir(p)
        except OSError:
            return []
    cmds = [f for f in listdir(os.path.join(root, "commands"))
            if f.endswith(".md")]
    skills = [d for d in listdir(os.path.join(root, "skills"))
              if os.path.isdir(os.path.join(root, "skills", d))]
    agents = [f for f in listdir(os.path.join(root, "agents"))
              if f.endswith(".md")]
    feats["n_custom_commands"] = len(cmds)
    feats["n_custom_skills"] = len(skills)
    feats["n_custom_agents"] = len(agents)
    if not sanitize:
        feats["custom_commands"] = sorted(cmds)
        feats["custom_skills"] = sorted(skills)
        feats["custom_agents"] = sorted(agents)
    feats["has_global_claude_md"] = os.path.isfile(os.path.join(root, "CLAUDE.md"))
    settings = {}
    try:
        with open(os.path.join(root, "settings.json")) as f:
            settings = json.load(f)
    except (OSError, json.JSONDecodeError):
        pass
    feats["has_hooks"] = bool(settings.get("hooks"))
    feats["default_model"] = settings.get("model")
    feats["n_mcp_servers_global"] = len(settings.get("mcpServers") or {})
    return feats


def cmd_extract(args):
    root = os.path.expanduser(args.data_dir)
    base, files = find_project_dirs(root)
    if not files:
        sys.exit(f"No .jsonl transcripts found under {base}")
    salt = args.salt or ""
    prompts_out = None
    if not args.sanitize and args.prompts_out:
        prompts_out = open(args.prompts_out, "w", encoding="utf-8")
    sessions = []
    for i, path in enumerate(files):
        sess = extract_claude_code_session(
            path, args.sanitize, salt, prompts_out, args.source)
        if sess:
            sessions.append(sess)
        if (i + 1) % 200 == 0:
            print(f"  ...{i + 1}/{len(files)} files", file=sys.stderr)
    if prompts_out:
        prompts_out.close()

    # cross-session repetition (privacy-safe: hashes only)
    template_counts = Counter()
    for sess in sessions:
        template_counts.update(sess.pop("prompt_template_hashes"))
    repeated = {h: c for h, c in template_counts.items() if c >= 3}

    out = {
        "schema_version": SCHEMA_VERSION,
        "extractor": "claude-code",
        "source": args.source,
        "sanitized": bool(args.sanitize),
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "n_transcript_files": len(files),
        "env": extract_env_features(
            root if os.path.isdir(os.path.join(root, "projects")) else base,
            args.sanitize, salt),
        "repeated_prompt_templates": dict(
            sorted(repeated.items(), key=lambda kv: -kv[1])[:50]),
        "sessions": sessions,
    }
    with open(args.out, "w", encoding="utf-8") as f:
        json.dump(out, f, indent=1)
    print(f"Wrote {args.out}: {len(sessions)} sessions from {len(files)} files."
          f"{' (sanitized — review before exporting)' if args.sanitize else ''}")


# --------------------------------------------------- claude.ai extraction

def load_claudeai_conversations(path):
    if path.endswith(".zip"):
        with zipfile.ZipFile(path) as z:
            name = next((n for n in z.namelist()
                         if n.endswith("conversations.json")), None)
            if not name:
                sys.exit("No conversations.json inside zip")
            with z.open(name) as f:
                return json.load(f)
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def cmd_extract_claudeai(args):
    convs = load_claudeai_conversations(os.path.expanduser(args.export_file))
    salt = args.salt or ""
    prompts_out = None
    if not args.sanitize and args.prompts_out:
        prompts_out = open(args.prompts_out, "w", encoding="utf-8")
    sessions = []
    template_counts = Counter()
    for conv in convs:
        msgs = conv.get("chat_messages") or []
        human = [m for m in msgs if m.get("sender") == "human"]
        asst = [m for m in msgs if m.get("sender") == "assistant"]
        if not human:
            continue
        tss = [parse_ts(m.get("created_at")) for m in msgs]
        tss = [t for t in tss if t]
        first_ts = min(tss) if tss else None
        last_ts = max(tss) if tss else None
        pchars, ncorr, nq, nctx, nattach = [], 0, 0, 0, 0
        for m in human:
            text = m.get("text") or ""
            if not text and isinstance(m.get("content"), list):
                text = "\n".join(b.get("text", "") for b in m["content"]
                                 if isinstance(b, dict) and b.get("type") == "text")
            nattach += len(m.get("attachments") or []) + len(m.get("files") or [])
            if not text.strip():
                continue
            pchars.append(len(text))
            if CORRECTION_RE.search(text):
                ncorr += 1
            if QUESTION_RE.search(text):
                nq += 1
            if CONTEXT_CUES_RE.search(text) or len(text) > 400:
                nctx += 1
            template_counts[short_hash(normalized_template(text), salt)] += 1
            if prompts_out is not None:
                prompts_out.write(json.dumps({
                    "source": args.source,
                    "session": conv.get("uuid"),
                    "ts": m.get("created_at"),
                    "conversation_name": conv.get("name"),
                    "text": text[:4000],
                }) + "\n")
        name = conv.get("name") or ""
        sessions.append({
            "source": args.source,
            "kind": "claude-ai",
            "session_file": (short_hash(conv.get("uuid") or name, salt)
                             if args.sanitize else conv.get("uuid")),
            "conversation_name": None if args.sanitize else name,
            "start_ts": first_ts.isoformat() if first_ts else None,
            "duration_min": (round((last_ts - first_ts).total_seconds() / 60, 1)
                             if first_ts and last_ts else None),
            "hour_local": first_ts.astimezone().hour if first_ts else None,
            "weekday": first_ts.astimezone().strftime("%a") if first_ts else None,
            "n_user_turns": len(pchars),
            "n_assistant_msgs": len(asst),
            "n_corrections": ncorr,
            "n_questions": nq,
            "n_context_rich_prompts": nctx,
            "n_attachments": nattach,
            "first_prompt_chars": pchars[0] if pchars else None,
            "prompt_chars_dist": summarize_dist(pchars),
        })
    if prompts_out:
        prompts_out.close()
    repeated = {h: c for h, c in template_counts.items() if c >= 3}
    out = {
        "schema_version": SCHEMA_VERSION,
        "extractor": "claude-ai",
        "source": args.source,
        "sanitized": bool(args.sanitize),
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "repeated_prompt_templates": dict(
            sorted(repeated.items(), key=lambda kv: -kv[1])[:50]),
        "sessions": sessions,
    }
    with open(args.out, "w", encoding="utf-8") as f:
        json.dump(out, f, indent=1)
    print(f"Wrote {args.out}: {len(sessions)} conversations.")


# ------------------------------------------------------------- reporting

def cmd_report(args):
    all_sessions = []
    meta = []
    for path in args.metrics:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
        meta.append({k: data.get(k) for k in
                     ("source", "extractor", "sanitized", "env",
                      "repeated_prompt_templates", "n_transcript_files")})
        all_sessions.extend(data.get("sessions") or [])

    by_source = defaultdict(list)
    for s in all_sessions:
        by_source[s.get("source", "?")].append(s)

    def agg(sessions):
        n = len(sessions)
        if n == 0:
            return {}
        turns = [s["n_user_turns"] for s in sessions if s.get("n_user_turns")]
        durs = [s["duration_min"] for s in sessions
                if s.get("duration_min") is not None]
        corr = sum(s.get("n_corrections", 0) for s in sessions)
        allturns = sum(turns) or 1
        tools = Counter()
        models = Counter()
        slash = Counter()
        hours = Counter()
        days = Counter()
        for s in sessions:
            tools.update(s.get("tools") or {})
            models.update(s.get("models") or {})
            slash.update(s.get("slash_commands") or {})
            if s.get("hour_local") is not None:
                hours[s["hour_local"]] += 1
            if s.get("weekday"):
                days[s["weekday"]] += 1
        frac = lambda key: round(
            sum(1 for s in sessions if s.get(key)) / n, 3)
        return {
            "n_sessions": n,
            "total_user_turns": sum(turns),
            "turns_per_session": summarize_dist(turns),
            "duration_min": summarize_dist(durs),
            "correction_rate_per_turn": round(corr / allturns, 3),
            "interrupts_total": sum(s.get("n_interrupts", 0) for s in sessions),
            "question_rate_per_turn": round(
                sum(s.get("n_questions", 0) for s in sessions) / allturns, 3),
            "context_rich_rate_per_turn": round(
                sum(s.get("n_context_rich_prompts", 0) for s in sessions)
                / allturns, 3),
            "tool_errors_total": sum(s.get("n_tool_errors", 0) for s in sessions),
            "sessions_ended_after_interrupt": sum(
                1 for s in sessions if s.get("ended_after_interrupt")),
            "feature_usage_frac": {
                "plan_mode": frac("used_plan_mode"),
                "subagents": frac("used_subagents"),
                "skills": frac("used_skills"),
                "todos": frac("used_todos"),
                "mcp": frac("used_mcp"),
                "web": frac("used_web"),
                "thinking": frac("used_thinking"),
                "compacted": frac("compacted"),
            },
            "git_commits_total": sum(s.get("n_git_commits", 0) for s in sessions),
            "top_tools": dict(tools.most_common(20)),
            "models": dict(models.most_common()),
            "slash_commands": dict(slash.most_common(20)),
            "sessions_by_hour": dict(sorted(hours.items())),
            "sessions_by_weekday": dict(days.most_common()),
            "tokens_out_total": sum(s.get("tokens_out", 0) for s in sessions),
            "single_turn_sessions": sum(1 for s in sessions
                                        if s.get("n_user_turns") == 1),
            "long_sessions_20plus_turns": sum(
                1 for s in sessions if (s.get("n_user_turns") or 0) >= 20),
            "n_projects": len({s.get("project") for s in sessions
                               if s.get("project")}),
        }

    report = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "inputs": meta,
        "by_source": {src: agg(ss) for src, ss in sorted(by_source.items())},
        "combined": agg(all_sessions),
    }
    with open(args.out, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=1)
    print(f"Wrote {args.out}")


# ----------------------------------------------------------------- main

def main():
    ap = argparse.ArgumentParser(description=__doc__)
    sub = ap.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("extract", help="Extract Claude Code transcripts")
    p.add_argument("data_dir", help="~/.claude, ~/.claude/projects, or a copy")
    p.add_argument("--out", default="cc_metrics.json")
    p.add_argument("--source", default="claude-code")
    p.add_argument("--sanitize", action="store_true",
                   help="No content: hashed ids, numeric metrics only")
    p.add_argument("--salt", default="", help="Salt for hashing (sanitize mode)")
    p.add_argument("--prompts-out", default=None,
                   help="Also write raw prompts jsonl (non-sanitized only)")
    p.set_defaults(func=cmd_extract)

    p = sub.add_parser("extract-claudeai", help="Extract claude.ai export")
    p.add_argument("export_file", help="conversations.json or the export .zip")
    p.add_argument("--out", default="claudeai_metrics.json")
    p.add_argument("--source", default="claude-ai")
    p.add_argument("--sanitize", action="store_true")
    p.add_argument("--salt", default="")
    p.add_argument("--prompts-out", default=None)
    p.set_defaults(func=cmd_extract_claudeai)

    p = sub.add_parser("report", help="Merge metrics files into a summary")
    p.add_argument("metrics", nargs="+", help="metrics JSON files")
    p.add_argument("--out", default="combined_report.json")
    p.set_defaults(func=cmd_report)

    args = ap.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()