#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Manda el resumen del consumo de Claude Code de ESTA máquina al lugar central. python3 reportar-consumo.py # mide y envía lo de ayer python3 reportar-consumo.py --ver # muestra lo que enviaría, sin enviar python3 reportar-consumo.py --dias 7 # los últimos 7 días python3 reportar-consumo.py --instalar # lo deja corriendo solo, una vez al día **Qué se manda y qué NO.** Se mandan únicamente números agregados por día: cuántas peticiones, cuántos tokens de cada clase, el contexto medio y el reparto por modelo. **Nunca se manda el contenido de las conversaciones**, ni los nombres de los archivos, ni los comandos, ni el nombre de los proyectos. Eso ni siquiera se lee: el guion sólo mira el bloque `usage` de cada respuesta. Por qué existe: `medir-consumo.py` sólo ve la máquina donde corre, y el panel de la organización da el cuánto pero no el en qué. Sin esto no se puede saber si las reglas de REGLAS-EQUIPO.md §13 se están cumpliendo. Tiquete #23433. El envío es **saliente** por HTTPS a una API que ya está expuesta. No abre ningún puerto ni deja nada escuchando en esta máquina — que era el motivo por el que se descartó montar un colector (oci-dr PR #20 y #21). """ import argparse import collections import datetime import getpass import glob import json import os import platform import socket import ssl import subprocess import sys import urllib.error import urllib.request # La consola de Windows suele venir en cp1252, donde los emojis de los avisos no # existen: al imprimirlos Python corta con UnicodeEncodeError. Se pide que reemplace # lo que no pueda escribir en vez de fallar. (Reportado por María, 13-ago-2026.) def _salida_tolerante(): """Que stdout y stderr reemplacen lo que no puedan escribir, en vez de cortar.""" import io for nombre in ("stdout", "stderr"): canal = getattr(sys, nombre) try: canal.reconfigure(errors="replace") continue except (AttributeError, ValueError): pass # Python 3.6 no tiene reconfigure, y es justo donde más probable es toparse # con una consola en cp1252. try: setattr(sys, nombre, io.TextIOWrapper( canal.buffer, encoding=canal.encoding, errors="replace", line_buffering=True)) except Exception: pass _salida_tolerante() # A dónde se envía. Se puede cambiar sin tocar el código, con estas variables: # MB_CONSUMO_URL la dirección del receptor # MB_CONSUMO_CLAVE la clave compartida URL = os.environ.get( "MB_CONSUMO_URL", "https://bitacora.sistemasmb.com:4001/consumoClaude" ) CLAVE = os.environ.get("MB_CONSUMO_CLAVE", "") # Tarifa pública de la API, en dólares por millón de tokens: # (entrada, salida, escribir al caché, leer del caché). Sirve para comparar una # práctica contra otra, no es una factura. TARIFA = { "opus": (15.0, 75.0, 18.75, 1.50), "sonnet": (3.0, 15.0, 3.75, 0.30), "haiku": (0.80, 4.0, 1.00, 0.08), } # Lo que no se pudo enviar espera acá y se reintenta en la corrida siguiente. Sin # esto, un día sin internet es un día perdido para siempre: las transcripciones se # borran solas a los 30 días. PENDIENTES = os.path.join( os.path.expanduser("~"), ".claude", "consumo-pendiente.jsonl" ) def tarifa_de(modelo): m = (modelo or "").lower() for familia, precios in TARIFA.items(): if familia in m: return familia, precios return "sonnet", TARIFA["sonnet"] def medir(dias): """Agrega el consumo por día. Devuelve una lista de días, del más viejo al más nuevo. Sólo se lee el bloque `usage` de cada respuesta del modelo y su fecha. El resto de la transcripción —lo que se habló— no se toca. """ hoy = datetime.date.today() desde = hoy - datetime.timedelta(days=dias) por_dia = collections.defaultdict(collections.Counter) modelos_por_dia = collections.defaultdict(collections.Counter) sesiones_por_dia = collections.defaultdict(set) for ruta in glob.glob(os.path.expanduser("~/.claude/projects/*/*.jsonl")): sesion = os.path.basename(ruta)[:-6] try: fh = open(ruta, errors="ignore") except (IOError, OSError): continue with fh: for linea in fh: if '"usage"' not in linea: continue try: d = json.loads(linea) except ValueError: continue if d.get("type") != "assistant" or d.get("isSidechain"): continue uso = (d.get("message") or {}).get("usage") sello = d.get("timestamp") if not uso or not sello: continue dia = sello[:10] try: if datetime.date(*[int(x) for x in dia.split("-")]) < desde: continue except ValueError: continue familia, precios = tarifa_de((d.get("message") or {}).get("model")) entrada, salida, escribir, leer = precios lee = uso.get("cache_read_input_tokens", 0) escribe = uso.get("cache_creation_input_tokens", 0) ent = uso.get("input_tokens", 0) sal = uso.get("output_tokens", 0) c = por_dia[dia] c["peticiones"] += 1 c["tok_lee_cache"] += lee c["tok_escribe_cache"] += escribe c["tok_entrada"] += ent c["tok_salida"] += sal c["contexto_sumado"] += lee + escribe + ent # En centésimos de dólar, para que viaje como entero y no como # coma flotante: comparar decimales entre máquinas trae sorpresas. c["costo_centavos"] += round( (ent * entrada + sal * salida + escribe * escribir + lee * leer) / 1e6 * 100 ) if lee + escribe + ent >= 500000: c["peticiones_sobre_500k"] += 1 modelos_por_dia[dia][familia] += 1 sesiones_por_dia[dia].add(sesion) dias_ordenados = [] for dia in sorted(por_dia): c = por_dia[dia] n = c["peticiones"] dias_ordenados.append( { "dia": dia, "peticiones": n, "sesiones": len(sesiones_por_dia[dia]), "contexto_medio": c["contexto_sumado"] // n if n else 0, "peticiones_sobre_500k": c["peticiones_sobre_500k"], "tok_lee_cache": c["tok_lee_cache"], "tok_escribe_cache": c["tok_escribe_cache"], "tok_entrada": c["tok_entrada"], "tok_salida": c["tok_salida"], "costo_centavos": c["costo_centavos"], "peticiones_por_modelo": dict(modelos_por_dia[dia]), } ) return dias_ordenados def quien(): """Identifica la máquina y la persona, sin inventar nada. MB_CONSUMO_PERSONA permite poner el nombre real; si no está, se usa el usuario del sistema, que en estas máquinas alcanza para saber de quién es. """ return { "persona": os.environ.get("MB_CONSUMO_PERSONA") or getpass.getuser(), "maquina": socket.gethostname(), "sistema": platform.system(), } def revisar_destino(): """Devuelve un texto de error si no se puede enviar con seguridad, o None.""" if not URL: return "no hay a dónde enviar (MB_CONSUMO_URL vacío)" # Sin HTTPS la clave y los datos viajan en claro. Se permite localhost sólo # porque es como se prueba el circuito contra un receptor de mentira. if not URL.startswith("https://"): local = URL.startswith("http://localhost") or URL.startswith("http://127.0.0.1") if not local: return "la dirección no es https:// ({}); la clave viajaría en claro".format(URL) if not CLAVE: return "falta la clave (MB_CONSUMO_CLAVE vacía)" return None def enviar(paquetes): """Envía una lista de paquetes. Devuelve (enviados, error o None).""" problema = revisar_destino() if problema: return 0, problema enviados = 0 for p in paquetes: cuerpo = json.dumps(p).encode("utf-8") req = urllib.request.Request( URL, data=cuerpo, headers={"Content-Type": "application/json", "X-API-Key": CLAVE}, ) try: with urllib.request.urlopen(req, timeout=30) as r: if r.status >= 300: return enviados, "el receptor respondió {}".format(r.status) enviados += 1 except urllib.error.HTTPError as e: try: detalle = e.read().decode("utf-8", "replace")[:120] except Exception: detalle = "sin detalle" return enviados, "el receptor respondió {} ({})".format(e.code, detalle) except (urllib.error.URLError, socket.timeout, ssl.SSLError) as e: return enviados, "no se pudo llegar al receptor ({})".format(e) return enviados, None def leer_pendientes(): if not os.path.exists(PENDIENTES): return [] guardados = [] with open(PENDIENTES, encoding="utf-8") as f: for linea in f: linea = linea.strip() if not linea: continue try: guardados.append(json.loads(linea)) except ValueError: continue return guardados def guardar_pendientes(paquetes): """Guarda lo que no se pudo enviar, sin duplicar el mismo día dos veces.""" vistos = set() unicos = [] for p in paquetes: clave = (p.get("persona"), p.get("maquina"), p.get("dia")) if clave in vistos: continue vistos.add(clave) unicos.append(p) # Sólo se guardan los últimos 60 días: más viejo que eso ya no le sirve a nadie # y el archivo no puede crecer sin techo. unicos = sorted(unicos, key=lambda p: p.get("dia", ""))[-60:] try: os.makedirs(os.path.dirname(PENDIENTES), exist_ok=True) with open(PENDIENTES, "w", encoding="utf-8") as f: for p in unicos: f.write(json.dumps(p) + "\n") except (IOError, OSError) as e: # Se dice. Guardar en silencio lo que no se guardó es peor que perderlo: # nadie se entera hasta que faltan días en el reporte y ya no se recuperan. print("⚠️ No se pudo guardar lo pendiente ({}). Esos días se pierden.".format(e)) def instalar_agendado(): """Deja el envío corriendo solo, una vez al día. Devuelve un texto con qué hizo.""" yo = os.path.abspath(__file__) py = sys.executable sistema = platform.system() if sistema in ("Darwin", "Linux"): linea = "17 9 * * * {} {} >/dev/null 2>&1".format(py, yo) try: actual = subprocess.run( ["crontab", "-l"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL ).stdout.decode() except (OSError, subprocess.SubprocessError): actual = "" # Se quitan las entradas viejas en vez de darse por agendado con cualquiera # que mencione el guion: una comentada, o una que apunte a un checkout que # ya se movió, dejaría el envío muerto y pareciendo instalado. if linea in actual.splitlines(): return "Ya estaba agendado con esta misma ruta; no se duplicó." vivas = [l for l in actual.splitlines() if "reportar-consumo.py" not in l] nuevo = ("\n".join(vivas).rstrip("\n") + "\n" + linea + "\n").lstrip("\n") try: p = subprocess.run(["crontab", "-"], input=nuevo.encode()) if p.returncode != 0: return "No se pudo agendar. Agregá esta línea a mano con `crontab -e`:\n " + linea except (OSError, subprocess.SubprocessError) as e: return "No se pudo agendar ({}). La línea es:\n {}".format(e, linea) return "Agendado: corre todos los días a las 9:17." return ( "En Windows el agendado automático todavía no está probado.\n" "Pedile a Claude que te cree una tarea programada diaria que ejecute:\n" ' "{}" "{}"'.format(py, yo) ) def main(): ap = argparse.ArgumentParser(add_help=True) ap.add_argument("--ver", action="store_true", help="mostrar sin enviar") ap.add_argument("--dias", type=int, default=2, help="cuántos días medir") ap.add_argument("--instalar", action="store_true", help="dejarlo corriendo solo") args = ap.parse_args() if args.instalar: print(instalar_agendado()) return identidad = quien() hoy = datetime.date.today().isoformat() paquetes = [] for dia in medir(args.dias): # El día de hoy todavía no terminó: se manda igual, y el receptor se queda # con el último valor de cada día para que se corrija solo mañana. p = dict(identidad) p.update(dia) p["parcial"] = dia["dia"] == hoy paquetes.append(p) pendientes = leer_pendientes() if args.ver: # Los pendientes también se muestran: si no, `--ver` dice que se enviarían # 2 días y la corrida real manda 5, que es justo lo que uno quiere ver antes. print(json.dumps(pendientes + paquetes, indent=2, ensure_ascii=False)) if pendientes: print("\n({} de esos días venían pendientes de antes)".format(len(pendientes))) problema = revisar_destino() print("\n(no se envió nada: --ver sólo muestra)") if problema: print("⚠️ Y así como está, no se podría enviar: {}".format(problema)) return if pendientes: print("Había {} día(s) sin enviar de corridas anteriores.".format(len(pendientes))) todos = pendientes + paquetes if not todos: print("No hay nada que enviar: esta máquina no tiene conversaciones recientes.") return enviados, error = enviar(todos) if error: guardar_pendientes(todos[enviados:]) print("Se enviaron {} de {}. El resto quedó guardado y se reintenta mañana.".format( enviados, len(todos))) print("Motivo: {}".format(error)) sys.exit(1) if os.path.exists(PENDIENTES): try: os.remove(PENDIENTES) except OSError: pass print("Enviados {} día(s) de {} en {}.".format( enviados, identidad["persona"], identidad["maquina"])) if __name__ == "__main__": main()