""" dashboard.py - Local web dashboard served on localhost:8080. """ import json import sqlite3 from http.server import HTTPServer, BaseHTTPRequestHandler from pathlib import Path from datetime import datetime DB_PATH = Path.home() / ".claude" / "usage.db" def _migrate_db(conn): """Asegura que las columnas nuevas existan (idempotente).""" turns_cols = {row[1] for row in conn.execute("PRAGMA table_info(turns)").fetchall()} sessions_cols = {row[1] for row in conn.execute("PRAGMA table_info(sessions)").fetchall()} changed = False for col in ("agent_name", "skill_name"): if col not in turns_cols: conn.execute(f"ALTER TABLE turns ADD COLUMN {col} TEXT") changed = True if "session_name" not in sessions_cols: conn.execute("ALTER TABLE sessions ADD COLUMN session_name TEXT") changed = True if changed: conn.commit() def get_dashboard_data(db_path=DB_PATH): if not db_path.exists(): return {"error": "Base de datos no encontrada. Ejecuta: python cli.py scan"} conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row _migrate_db(conn) # ── All models (for filter UI) ──────────────────────────────────────────── model_rows = conn.execute(""" SELECT COALESCE(model, 'unknown') as model FROM turns GROUP BY model ORDER BY SUM(input_tokens + output_tokens) DESC """).fetchall() all_models = [r["model"] for r in model_rows] # ── Daily per-model, ALL history (client filters by range) ──────────────── daily_rows = conn.execute(""" SELECT substr(timestamp, 1, 10) as day, COALESCE(model, 'unknown') as model, SUM(input_tokens) as input, SUM(output_tokens) as output, SUM(cache_read_tokens) as cache_read, SUM(cache_creation_tokens) as cache_creation, COUNT(*) as turns FROM turns GROUP BY day, model ORDER BY day, model """).fetchall() daily_by_model = [{ "day": r["day"], "model": r["model"], "input": r["input"] or 0, "output": r["output"] or 0, "cache_read": r["cache_read"] or 0, "cache_creation": r["cache_creation"] or 0, "turns": r["turns"] or 0, } for r in daily_rows] # ── All sessions (client filters by range and model) ────────────────────── session_rows = conn.execute(""" SELECT session_id, project_name, first_timestamp, last_timestamp, total_input_tokens, total_output_tokens, total_cache_read, total_cache_creation, model, turn_count, session_name FROM sessions ORDER BY last_timestamp DESC """).fetchall() sessions_all = [] for r in session_rows: try: t1 = datetime.fromisoformat(r["first_timestamp"].replace("Z", "+00:00")) t2 = datetime.fromisoformat(r["last_timestamp"].replace("Z", "+00:00")) duration_min = round((t2 - t1).total_seconds() / 60, 1) except Exception: duration_min = 0 sessions_all.append({ "session_id": r["session_id"][:8], "session_name": r["session_name"] or "", "project": r["project_name"] or "unknown", "last": (r["last_timestamp"] or "")[:16].replace("T", " "), "last_date": (r["last_timestamp"] or "")[:10], "duration_min": duration_min, "model": r["model"] or "unknown", "turns": r["turn_count"] or 0, "input": r["total_input_tokens"] or 0, "output": r["total_output_tokens"] or 0, "cache_read": r["total_cache_read"] or 0, "cache_creation": r["total_cache_creation"] or 0, }) # ── Top agentes SWL (columna agent_name, añadida en v5.3.3) ───────────────── try: agent_rows = conn.execute(""" SELECT agent_name, COUNT(*) as calls, SUM(input_tokens) as input, SUM(output_tokens) as output FROM turns WHERE agent_name IS NOT NULL GROUP BY agent_name ORDER BY input + output DESC LIMIT 15 """).fetchall() top_agents = [{"name": r["agent_name"], "calls": r["calls"], "input": r["input"] or 0, "output": r["output"] or 0} for r in agent_rows] except Exception: top_agents = [] # ── Top skills SWL (columna skill_name, añadida en v5.3.3) ────────────────── try: skill_rows = conn.execute(""" SELECT skill_name, COUNT(*) as calls, SUM(input_tokens) as input, SUM(output_tokens) as output FROM turns WHERE skill_name IS NOT NULL GROUP BY skill_name ORDER BY input + output DESC LIMIT 15 """).fetchall() top_skills = [{"name": r["skill_name"], "calls": r["calls"], "input": r["input"] or 0, "output": r["output"] or 0} for r in skill_rows] except Exception: top_skills = [] conn.close() return { "all_models": all_models, "daily_by_model": daily_by_model, "sessions_all": sessions_all, "top_agents": top_agents, "top_skills": top_skills, "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } HTML_TEMPLATE = r""" Panel de uso — Claude Code

Panel de uso — Claude Code

Cargando...
Modelos
Período

Uso diario de tokens

Por modelo

Top proyectos por tokens

Sesiones recientes
SesiónNombreProyectoÚltima actividadDuración ModeloTurnosEntradaSalidaCosto est.
Costo por modelo
ModeloTurnosEntradaSalida Caché lecturaCaché creaciónCosto est.
Top agentes SWL (histórico)
AgenteLlamadasTokens entradaTokens salida
Cargando...
Top skills SWL (histórico)
SkillLlamadasTokens entradaTokens salida
Cargando...
""" class DashboardHandler(BaseHTTPRequestHandler): def log_message(self, format, *args): pass def do_GET(self): if self.path in ("/", "/index.html"): self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(HTML_TEMPLATE.encode("utf-8")) elif self.path == "/api/data": data = get_dashboard_data() body = json.dumps(data).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) else: self.send_response(404) self.end_headers() def serve(port=8080): server = HTTPServer(("localhost", port), DashboardHandler) print(f"Dashboard running at http://localhost:{port}") print("Press Ctrl+C to stop.") try: server.serve_forever() except KeyboardInterrupt: print("\nStopped.") if __name__ == "__main__": serve()