but_usage_report.py

#!/usr/bin/env python3
"""
but_usage_report.py — Extract and visualize 'but' CLI usage from Claude Code sessions.

Scans all Claude Code JSONL session files (~/.claude/projects/**/*.jsonl), finds
every Bash tool invocation of the 'but' CLI, matches each command to its output,
and produces structured data (JSON + CSV) plus analysis charts.

PREREQUISITES:
    pip install matplotlib          # only needed for charts

USAGE:
    # Full pipeline: extract data + generate charts + print summary
    python3 but_usage_report.py

    # Extract data only (no matplotlib needed)
    python3 but_usage_report.py --extract-only

    # Regenerate charts from previously extracted but_usage.json
    python3 but_usage_report.py --charts-only

    # Custom output directory
    python3 but_usage_report.py --output-dir /tmp/but-analysis

OUTPUT FILES (in --output-dir, default ./but_usage_output/):
    but_usage.json                  Full extracted data
    but_usage.csv                   Same data, output truncated to 5000 chars
    but_subcommand_histogram.png    Subcommand popularity bar chart
    but_error_rate.png              Error rate by subcommand
    but_usage_over_time.png         Daily usage histogram
    but_subcommand_mix.png          Stacked subcommand composition over time

DATA SCHEMA (each record in but_usage.json):
    {
      "session_id":      "uuid",          # Claude Code session ID (filename stem)
      "project":         "string",        # Project directory name from ~/.claude/projects/
      "timestamp":       "string|number", # Message timestamp (epoch ms or ISO string)
      "command":         "string",        # Full shell command, e.g. "but status --json"
      "but_subcommand":  "string",        # Parsed subcommand, e.g. "status", "commit"
      "description":     "string",        # Claude's description of the command
      "output":          "string",        # Tool result text (stdout + stderr)
      "is_error":        true|false       # Whether the command errored
    }

DATA SOURCE — Claude Code JSONL format:
    Each line in a session file is a JSON object. Bash tool calls appear as:
        {"type": "assistant", "message": {"content": [
            {"type": "tool_use", "name": "Bash", "id": "toolu_...",
             "input": {"command": "but status --json", "description": "..."}}
        ]}}

    Tool results appear in the next 1-3 lines as either:
      Format 1 — tool_result in message.content:
        {"type": "tool_result", "tool_use_id": "toolu_...", "content": "...", "is_error": false}
      Format 2 — toolUseResult top-level field:
        {"toolUseResult": {"stdout": "...", "stderr": "...", "interrupted": false}}
"""

from __future__ import annotations

import argparse
import csv
import json
import sys
from collections import Counter
from datetime import datetime
from pathlib import Path

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# Configuration
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

CLAUDE_PROJECTS_DIR = Path.home() / ".claude" / "projects"
DEFAULT_OUTPUT_DIR = Path("but_usage_output")
MAX_CSV_OUTPUT_LEN = 5000  # truncate command output in CSV to this many chars


# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# Extraction helpers
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

def is_but_command(command: str) -> bool:
    """Return True if a shell command is a 'but' CLI invocation.

    Matches commands starting with 'but ' (the CLI) or 'but-' (e.g. but-engineering).
    Does NOT match embedded references like 'cargo check -p but'.
    """
    cmd = command.strip()
    return cmd.startswith("but ") or cmd.startswith("but-")


def extract_subcommand(command: str) -> str:
    """Parse the but subcommand from a full command string.

    Examples:
        'but status --json'            -> 'status'
        'but-engineering post ...'     -> 'but-engineering'
        'but commit branch -m ...'     -> 'commit'
        'but --help'                   -> '--help'
    """
    cmd = command.strip()
    if cmd.startswith("but-"):
        return cmd.split()[0]
    parts = cmd.split()
    return parts[1] if len(parts) >= 2 else ""


def parse_timestamp(ts) -> datetime | None:
    """Convert a timestamp (epoch ms, epoch seconds, or ISO string) to datetime."""
    if ts is None:
        return None
    if isinstance(ts, (int, float)):
        return datetime.fromtimestamp(ts / 1000) if ts > 1e12 else datetime.fromtimestamp(ts)
    if isinstance(ts, str):
        try:
            v = float(ts)
            return datetime.fromtimestamp(v / 1000) if v > 1e12 else datetime.fromtimestamp(v)
        except ValueError:
            try:
                return datetime.fromisoformat(ts)
            except ValueError:
                return None
    return None


def get_output_and_error(result_data: dict) -> tuple[str, bool]:
    """Extract output text and error flag from a tool result.

    Handles two JSONL result formats:
      1. tool_result block — has 'content' (str) and 'is_error' (bool)
      2. toolUseResult    — has 'stdout', 'stderr', 'interrupted'

    Also applies a heuristic: if output starts with 'Exit code N' where N != 0,
    mark as error even if the source didn't flag it.
    """
    if not result_data:
        return "", False

    output = ""
    is_error = False

    # Format 1: tool_result in message content
    if "content" in result_data and isinstance(result_data.get("content"), str):
        output = result_data["content"]
        is_error = result_data.get("is_error", False)
    # Format 2: toolUseResult with stdout/stderr
    elif "stdout" in result_data or "stderr" in result_data:
        stdout = result_data.get("stdout", "") or ""
        stderr = result_data.get("stderr", "") or ""
        output = stdout
        if stderr:
            output = f"{stdout}\n[stderr]: {stderr}" if stdout else f"[stderr]: {stderr}"
        is_error = result_data.get("interrupted", False)

    # Heuristic: non-zero exit code
    if not is_error and output:
        first_line = output.strip().split("\n")[0]
        if first_line.startswith("Exit code") and "Exit code 0" not in first_line:
            is_error = True

    return output, is_error


# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# Extraction
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

def process_session_file(fpath: Path) -> list[dict]:
    """Parse a single JSONL session file and return records for every 'but' invocation.

    For each Bash tool_use block whose command is a but invocation, searches forward
    up to 10 lines to find the matching tool result (by tool call ID or adjacency).

    Returns a list of dicts matching the DATA SCHEMA documented at the top of this file.
    """
    rel = fpath.relative_to(CLAUDE_PROJECTS_DIR)
    project = rel.parts[0] if rel.parts else "unknown"
    session_id = fpath.stem

    try:
        with open(fpath, "r") as f:
            lines = f.readlines()
    except Exception as e:
        print(f"  Warning: could not read {fpath}: {e}", file=sys.stderr)
        return []

    # Pass 1: find all but tool_use calls
    but_calls: list[tuple[int, str, str, str, object]] = []  # (line_idx, tool_id, command, description, timestamp)
    for i, line in enumerate(lines):
        try:
            obj = json.loads(line)
        except json.JSONDecodeError:
            continue

        content = obj.get("message", {}).get("content", [])
        if not isinstance(content, list):
            continue

        for block in content:
            if not isinstance(block, dict):
                continue
            if block.get("type") != "tool_use" or block.get("name") != "Bash":
                continue
            inp = block.get("input", {})
            command = inp.get("command", "")
            if is_but_command(command):
                but_calls.append((
                    i,
                    block.get("id", ""),
                    command,
                    inp.get("description", ""),
                    obj.get("timestamp"),
                ))

    if not but_calls:
        return []

    # Pass 2: match each call to its result
    records = []
    for line_idx, tool_id, command, description, timestamp in but_calls:
        output = ""
        is_error = False

        for j in range(line_idx + 1, min(line_idx + 10, len(lines))):
            try:
                obj2 = json.loads(lines[j])
            except json.JSONDecodeError:
                continue

            # Check toolUseResult (format 2)
            tr = obj2.get("toolUseResult")
            if isinstance(tr, dict) and (tr.get("stdout") is not None or tr.get("stderr") is not None):
                output, is_error = get_output_and_error(tr)
                break

            # Check message.content for tool_result (format 1)
            content2 = obj2.get("message", {}).get("content", [])
            if isinstance(content2, list):
                for b2 in content2:
                    if (isinstance(b2, dict)
                            and b2.get("type") == "tool_result"
                            and b2.get("tool_use_id") == tool_id):
                        output, is_error = get_output_and_error(b2)
                        break
                else:
                    continue
                break

        records.append({
            "session_id": session_id,
            "project": project,
            "timestamp": timestamp,
            "command": command,
            "but_subcommand": extract_subcommand(command),
            "description": description,
            "output": output,
            "is_error": is_error,
        })

    return records


def extract(output_dir: Path) -> list[dict]:
    """Scan all Claude Code sessions and extract but CLI usage.

    Walks ~/.claude/projects/**/*.jsonl, extracts every 'but' command invocation,
    and writes the results to but_usage.json and but_usage.csv in output_dir.

    Returns the full list of extracted records.
    """
    if not CLAUDE_PROJECTS_DIR.exists():
        print(f"Error: {CLAUDE_PROJECTS_DIR} does not exist", file=sys.stderr)
        sys.exit(1)

    jsonl_files = sorted(CLAUDE_PROJECTS_DIR.rglob("*.jsonl"))
    print(f"Found {len(jsonl_files)} JSONL files to scan")

    all_records: list[dict] = []
    files_with_hits = 0

    for i, fpath in enumerate(jsonl_files):
        if fpath.stat().st_size == 0:
            continue
        records = process_session_file(fpath)
        if records:
            files_with_hits += 1
            all_records.extend(records)
        if (i + 1) % 500 == 0:
            print(f"  Scanned {i + 1}/{len(jsonl_files)} files, {len(all_records)} but commands found so far")

    all_records.sort(key=lambda r: r.get("timestamp") or 0)
    print(f"\nDone! Found {len(all_records)} but command invocations across {files_with_hits} session files.")

    # Write JSON
    output_dir.mkdir(parents=True, exist_ok=True)
    json_path = output_dir / "but_usage.json"
    with open(json_path, "w") as f:
        json.dump(all_records, f, indent=2)
    print(f"Written {json_path} ({json_path.stat().st_size / 1024:.0f} KB)")

    # Write CSV
    csv_path = output_dir / "but_usage.csv"
    fieldnames = ["session_id", "project", "timestamp", "command", "but_subcommand",
                  "description", "output", "is_error"]
    with open(csv_path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for record in all_records:
            row = dict(record)
            if len(row["output"]) > MAX_CSV_OUTPUT_LEN:
                row["output"] = row["output"][:MAX_CSV_OUTPUT_LEN] + "...[truncated]"
            writer.writerow(row)
    print(f"Written {csv_path} ({csv_path.stat().st_size / 1024:.0f} KB)")

    return all_records


# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# Summary
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

def print_summary(data: list[dict]) -> None:
    """Print a terminal summary: totals, error rate, top subcommands, top projects."""
    print("\n=== Summary ===")
    print(f"Total but commands: {len(data)}")

    error_count = sum(1 for r in data if r["is_error"])
    if data:
        print(f"Errors: {error_count} ({100 * error_count / len(data):.1f}%)")

    subcmd_counts = Counter(r["but_subcommand"] for r in data)
    print(f"\nTop subcommands:")
    for cmd, count in subcmd_counts.most_common(20):
        errs = sum(1 for r in data if r["but_subcommand"] == cmd and r["is_error"])
        print(f"  {cmd:25s} {count:5d}  ({errs} errors)")

    proj_counts = Counter(r["project"] for r in data)
    print(f"\nTop projects:")
    for proj, count in proj_counts.most_common(10):
        print(f"  {proj:60s} {count:5d}")


# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# Charts
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

def generate_charts(data: list[dict], output_dir: Path) -> None:
    """Generate 4 analysis charts from extracted but-usage data.

    Filters out 'but-engineering' commands (coordination tool, not a but subcommand).

    Charts produced:
      1. but_subcommand_histogram.png  — horizontal bar chart of subcommand frequency
      2. but_error_rate.png            — error rate by subcommand (>=5 uses)
      3. but_usage_over_time.png       — daily usage bar chart
      4. but_subcommand_mix.png        — stacked daily composition (top 8 + other)

    Requires matplotlib (pip install matplotlib).
    """
    import matplotlib
    matplotlib.use("Agg")
    import matplotlib.pyplot as plt

    output_dir.mkdir(parents=True, exist_ok=True)

    # Filter to actual but subcommands (exclude but-engineering coordination tool)
    but_cmds = [r for r in data if not r["but_subcommand"].startswith("but-")]
    print(f"\nGenerating charts for {len(but_cmds)} but commands (excluding but-engineering)...")

    counts = Counter(r["but_subcommand"] for r in but_cmds)
    error_counts = Counter(r["but_subcommand"] for r in but_cmds if r["is_error"])

    # ── Chart 1: Subcommand popularity ─────────────────────────────────────
    top_cmds = [(cmd, cnt) for cmd, cnt in counts.most_common() if cnt >= 2]
    cmds = [c for c, _ in reversed(top_cmds)]
    cnts = [n for _, n in reversed(top_cmds)]
    errs = [error_counts.get(c, 0) for c in cmds]
    ok = [n - e for n, e in zip(cnts, errs)]

    fig, ax = plt.subplots(figsize=(10, 8))
    ax.barh(cmds, ok, color="#4C9BE8", label="Success")
    ax.barh(cmds, errs, left=ok, color="#E85454", label="Error")
    for i, (c, n, e) in enumerate(zip(cmds, cnts, errs)):
        label = f"{n}" + (f" ({e} err)" if e > 0 else "")
        ax.text(n + 2, i, label, va="center", fontsize=9)
    ax.set_xlabel("Invocation count")
    ax.set_title("but CLI — Subcommand Popularity\n(commands with >= 2 uses)")
    ax.legend(loc="lower right")
    ax.set_xlim(0, max(cnts) * 1.15)
    plt.tight_layout()
    path1 = output_dir / "but_subcommand_histogram.png"
    plt.savefig(path1, dpi=150)
    print(f"  Saved {path1}")
    plt.close()

    # ── Chart 2: Error rate by subcommand ──────────────────────────────────
    min_uses = 5
    freq_cmds = [(cmd, cnt) for cmd, cnt in counts.most_common() if cnt >= min_uses]
    err_rate_cmds = sorted(freq_cmds, key=lambda x: error_counts.get(x[0], 0) / x[1], reverse=True)

    cmds2 = [c for c, _ in reversed(err_rate_cmds)]
    rates = [100 * error_counts.get(c, 0) / n for c, n in reversed(err_rate_cmds)]
    totals = [n for _, n in reversed(err_rate_cmds)]

    fig, ax = plt.subplots(figsize=(10, 6))
    colors = ["#E85454" if r > 10 else "#F5A623" if r > 5 else "#4C9BE8" for r in rates]
    ax.barh(cmds2, rates, color=colors)
    for i, (r, t, c) in enumerate(zip(rates, totals, cmds2)):
        e = error_counts.get(c, 0)
        ax.text(r + 0.5, i, f"{r:.0f}% ({e}/{t})", va="center", fontsize=9)
    ax.set_xlabel("Error rate (%)")
    ax.set_title(f"but CLI — Error Rate by Subcommand\n(commands with >= {min_uses} uses)")
    ax.set_xlim(0, max(rates) * 1.3 if rates else 10)
    plt.tight_layout()
    path2 = output_dir / "but_error_rate.png"
    plt.savefig(path2, dpi=150)
    print(f"  Saved {path2}")
    plt.close()

    # ── Chart 3: Usage over time ───────────────────────────────────────────
    dated = []
    for r in but_cmds:
        dt = parse_timestamp(r.get("timestamp"))
        if dt:
            dated.append((r, dt))

    by_day = Counter(dt.strftime("%Y-%m-%d") for _, dt in dated)
    days = sorted(by_day.keys())
    day_counts = [by_day[d] for d in days]

    fig, ax = plt.subplots(figsize=(12, 4))
    ax.bar(range(len(days)), day_counts, color="#4C9BE8", width=1.0, edgecolor="white", linewidth=0.3)
    step = max(1, len(days) // 15)
    ax.set_xticks(range(0, len(days), step))
    ax.set_xticklabels([days[i] for i in range(0, len(days), step)], rotation=45, ha="right", fontsize=8)
    ax.set_ylabel("Commands / day")
    ax.set_title("but CLI — Usage Over Time")
    plt.tight_layout()
    path3 = output_dir / "but_usage_over_time.png"
    plt.savefig(path3, dpi=150)
    print(f"  Saved {path3}")
    plt.close()

    # ── Chart 4: Subcommand composition over time ──────────────────────────
    top8 = [c for c, _ in counts.most_common(8)]
    by_day_cmd: dict[str, Counter] = {}
    for r, dt in dated:
        day = dt.strftime("%Y-%m-%d")
        cmd = r["but_subcommand"] if r["but_subcommand"] in top8 else "other"
        by_day_cmd.setdefault(day, Counter())[cmd] += 1

    days_sorted = sorted(by_day_cmd.keys())
    categories = top8 + ["other"]
    stacks = {cat: [by_day_cmd.get(d, Counter()).get(cat, 0) for d in days_sorted] for cat in categories}

    fig, ax = plt.subplots(figsize=(12, 5))
    cmap = plt.cm.tab10
    bottom = [0] * len(days_sorted)
    for i, cat in enumerate(categories):
        vals = stacks[cat]
        ax.bar(range(len(days_sorted)), vals, bottom=bottom, label=cat,
               color=cmap(i), width=1.0, edgecolor="white", linewidth=0.2)
        bottom = [b + v for b, v in zip(bottom, vals)]
    step = max(1, len(days_sorted) // 15)
    ax.set_xticks(range(0, len(days_sorted), step))
    ax.set_xticklabels([days_sorted[i] for i in range(0, len(days_sorted), step)], rotation=45, ha="right", fontsize=8)
    ax.set_ylabel("Commands / day")
    ax.set_title("but CLI — Subcommand Mix Over Time")
    ax.legend(loc="upper left", fontsize=8, ncol=3)
    plt.tight_layout()
    path4 = output_dir / "but_subcommand_mix.png"
    plt.savefig(path4, dpi=150)
    print(f"  Saved {path4}")
    plt.close()

    print("All charts generated.")


# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# CLI
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

def main():
    parser = argparse.ArgumentParser(
        description="Extract and analyze 'but' CLI usage from Claude Code sessions.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python3 but_usage_report.py                       # full pipeline
  python3 but_usage_report.py --extract-only        # data only, no charts
  python3 but_usage_report.py --charts-only         # regenerate charts from existing JSON
  python3 but_usage_report.py --output-dir /tmp/out # custom output location
        """,
    )
    parser.add_argument("--extract-only", action="store_true",
                        help="Only extract data (JSON + CSV), skip chart generation")
    parser.add_argument("--charts-only", action="store_true",
                        help="Only generate charts from existing but_usage.json")
    parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR,
                        help=f"Output directory (default: {DEFAULT_OUTPUT_DIR})")
    args = parser.parse_args()

    if args.extract_only and args.charts_only:
        parser.error("Cannot use both --extract-only and --charts-only")

    output_dir = args.output_dir

    if args.charts_only:
        json_path = output_dir / "but_usage.json"
        if not json_path.exists():
            print(f"Error: {json_path} not found. Run without --charts-only first.", file=sys.stderr)
            sys.exit(1)
        with open(json_path) as f:
            data = json.load(f)
        print(f"Loaded {len(data)} records from {json_path}")
        generate_charts(data, output_dir)
        return

    data = extract(output_dir)
    print_summary(data)

    if not args.extract_only:
        generate_charts(data, output_dir)


if __name__ == "__main__":
    main()