#!/usr/bin/env python3 """ Claude Code Leaderboard — Stats Push Script Parses ~/.claude/ session data and POSTs stats to the leaderboard. Runs as a Claude Code hook (Stop event) or standalone. Usage: python3 push_stats.py # parse + push python3 push_stats.py --force # ignore throttle, push immediately Env vars: PLAYER_NAME — your display name on the leaderboard (required) LEADERBOARD_URL — leaderboard endpoint (default: https://hadis-mac-mini.tailf8f871.ts.net:10000) PUSH_INTERVAL — seconds between pushes (default: 300 = 5 min) """ import json import os import sys import ssl import time import urllib.request from pathlib import Path # ── Config ── PLAYER_NAME = os.environ.get("PLAYER_NAME", "") LEADERBOARD_URL = os.environ.get("LEADERBOARD_URL", "https://hadis-mac-mini.tailf8f871.ts.net:10000") PUSH_INTERVAL = int(os.environ.get("PUSH_INTERVAL", "300")) THROTTLE_FILE = Path.home() / ".claude" / ".leaderboard_last_push" CLAUDE_DIR = Path.home() / ".claude" PROJECTS_DIR = CLAUDE_DIR / "projects" IDLE_THRESHOLD = 600 # 10 min # ── Pricing per token (USD) ── PRICING = { "opus": {"input": 5/1e6, "output": 25/1e6, "cache_read": 0.50/1e6, "cache_write": 6.25/1e6}, "sonnet": {"input": 3/1e6, "output": 15/1e6, "cache_read": 0.30/1e6, "cache_write": 3.75/1e6}, "haiku": {"input": 1/1e6, "output": 5/1e6, "cache_read": 0.10/1e6, "cache_write": 1.25/1e6}, } def should_push(): if "--force" in sys.argv: return True if THROTTLE_FILE.exists(): try: last = float(THROTTLE_FILE.read_text().strip()) if time.time() - last < PUSH_INTERVAL: return False except (ValueError, OSError): pass return True def mark_pushed(): try: THROTTLE_FILE.write_text(str(time.time())) except OSError: pass def model_family(model): m = (model or "").lower() if "haiku" in m: return "haiku" if "sonnet" in m: return "sonnet" return "opus" def cost_for(inp, out, cr, cw, model): p = PRICING[model_family(model)] return inp * p["input"] + out * p["output"] + cr * p["cache_read"] + cw * p["cache_write"] def parse_jsonl(filepath): """Parse a single JSONL file, return stats.""" timestamps = [] human = api = inp_t = out_t = cr_t = cw_t = lines = 0 cost = 0.0 models = {} try: with open(filepath, "r", errors="replace") as f: for line in f: try: msg = json.loads(line) except (json.JSONDecodeError, ValueError): continue ts_str = msg.get("timestamp") if ts_str: try: from datetime import datetime ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) timestamps.append(ts) except (ValueError, TypeError): pass inner = msg.get("message") or {} role = inner.get("role", "") msg_type = msg.get("type", "") if msg_type == "user" and not msg.get("toolUseResult"): content = inner.get("content", "") is_human = False if isinstance(content, str) and content.strip(): is_human = True elif isinstance(content, list): for c in content: if isinstance(c, dict) and c.get("type") == "text" and c.get("text", "").strip(): is_human = True break if is_human: human += 1 if msg_type == "assistant": api += 1 model = inner.get("model", "unknown") models[model] = models.get(model, 0) + 1 usage = inner.get("usage") or {} i = usage.get("input_tokens", 0) or 0 o = usage.get("output_tokens", 0) or 0 cr = usage.get("cache_read_input_tokens", 0) or 0 cw = usage.get("cache_creation_input_tokens", 0) or 0 inp_t += i; out_t += o; cr_t += cr; cw_t += cw cost += cost_for(i, o, cr, cw, model) # Count lines written for block in (inner.get("content") or []): if isinstance(block, dict) and block.get("type") == "tool_use": name = block.get("name", "") bi = block.get("input") or {} if name == "Write" and bi.get("content"): lines += bi["content"].count("\n") + 1 elif name == "Edit" and bi.get("new_string"): lines += bi["new_string"].count("\n") + 1 except OSError: pass # Active time timestamps.sort() active = 0 for i in range(1, len(timestamps)): gap = (timestamps[i] - timestamps[i - 1]).total_seconds() if gap <= IDLE_THRESHOLD: active += gap return { "human_prompts": human, "api_calls": api, "input_tokens": inp_t, "output_tokens": out_t, "cache_read": cr_t, "cache_write": cw_t, "lines_written": lines, "cost": cost, "active_seconds": active, "model_usage": models, "first_ts": timestamps[0].isoformat() if timestamps else None, "last_ts": timestamps[-1].isoformat() if timestamps else None, } def collect_all_stats(): """Walk ~/.claude/projects/ and aggregate stats.""" totals = { "total_prompts": 0, "total_api_calls": 0, "total_active_hours": 0, "total_input_tokens": 0, "total_output_tokens": 0, "total_cache_read": 0, "total_cache_write": 0, "total_lines_written": 0, "total_cost": 0, "total_sessions": 0, "total_projects": 0, "model_usage": {}, "earliest_session": None, "latest_session": None, "projects_data": [], } if not PROJECTS_DIR.exists(): return totals project_names = set() for proj_dir in PROJECTS_DIR.iterdir(): if not proj_dir.is_dir(): continue for f in proj_dir.iterdir(): if f.suffix != ".jsonl" or f.stat().st_size == 0: continue s = parse_jsonl(f) totals["total_sessions"] += 1 totals["total_prompts"] += s["human_prompts"] totals["total_api_calls"] += s["api_calls"] totals["total_input_tokens"] += s["input_tokens"] totals["total_output_tokens"] += s["output_tokens"] totals["total_cache_read"] += s["cache_read"] totals["total_cache_write"] += s["cache_write"] totals["total_lines_written"] += s["lines_written"] totals["total_cost"] += s["cost"] totals["total_active_hours"] += s["active_seconds"] / 3600 for m, c in s["model_usage"].items(): totals["model_usage"][m] = totals["model_usage"].get(m, 0) + c if s["first_ts"]: if not totals["earliest_session"] or s["first_ts"] < totals["earliest_session"]: totals["earliest_session"] = s["first_ts"] if s["last_ts"]: if not totals["latest_session"] or s["last_ts"] > totals["latest_session"]: totals["latest_session"] = s["last_ts"] # Parse subagents session_sub = proj_dir / f.stem / "subagents" if session_sub.exists(): for sf in session_sub.iterdir(): if sf.suffix == ".jsonl": sub = parse_jsonl(sf) totals["total_api_calls"] += sub["api_calls"] totals["total_input_tokens"] += sub["input_tokens"] totals["total_output_tokens"] += sub["output_tokens"] totals["total_cache_read"] += sub["cache_read"] totals["total_cache_write"] += sub["cache_write"] totals["total_lines_written"] += sub["lines_written"] totals["total_cost"] += sub["cost"] project_names.add(proj_dir.name) totals["total_cost"] = round(totals["total_cost"], 2) totals["total_active_hours"] = round(totals["total_active_hours"], 2) totals["total_projects"] = len(project_names) return totals def push(stats): """POST stats to the leaderboard.""" stats["name"] = PLAYER_NAME data = json.dumps(stats).encode() # Allow self-signed / Tailscale certs ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE req = urllib.request.Request( f"{LEADERBOARD_URL.rstrip('/')}/api/leaderboard/submit", data=data, headers={"Content-Type": "application/json"}, method="POST", ) urllib.request.urlopen(req, timeout=15, context=ctx) def main(): if not PLAYER_NAME: return if not should_push(): return try: stats = collect_all_stats() push(stats) mark_pushed() except Exception: pass # silent — don't break Claude Code if __name__ == "__main__": main()