Custo por consumidor: quanto cada serviço gasta de modelo

{"ts": "2026-10-20T09:12:00Z", "consumer": "bot-telegram", "model": "gpt-4o-mini", "calls": 1, "prompt_tokens": 45000, "completion_tokens": 3000, "cost_usd": 0.0090}

#!/usr/bin/env python3
"""Agrupa gasto de modelo por consumidor, modelo e dia.

Lê um arquivo JSONL com um registro por chamada (ou por lote) e imprime
o ranking de consumidores e o detalhe por dia e modelo.

Campos aceitos em cada linha:
  ts ou date        instante (ISO 8601 ou epoch)
  consumer          nome do serviço ou tarefa que fez a chamada
  model             nome do modelo
  calls             quantas chamadas (opcional, padrão 1)
  prompt_tokens     tokens de entrada
  completion_tokens tokens de saída
  cost_usd          custo da chamada em dólares (opcional)

Uso:
  python3 custo_por_consumidor.py uso.jsonl
  python3 custo_por_consumidor.py            # roda com exemplo embutido
"""
import json
import sys
from collections import defaultdict
from datetime import datetime, timezone


def para_data(valor):
    if isinstance(valor, (int, float)):
        dt = datetime.fromtimestamp(valor, tz=timezone.utc)
    else:
        s = str(valor).strip()
        try:
            dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
        except ValueError:
            dt = datetime.strptime(s[:10], "%Y-%m-%d").replace(tzinfo=timezone.utc)
    return dt.date().isoformat()


def carregar(caminho):
    with open(caminho, encoding="utf-8") as f:
        for linha in f:
            linha = linha.strip()
            if not linha:
                continue
            yield json.loads(linha)


EXEMPLO = [
    {"ts": "2026-10-20T09:12:00Z", "consumer": "bot-telegram", "model": "gpt-4o-mini",
     "calls": 1, "prompt_tokens": 45000, "completion_tokens": 3000, "cost_usd": 0.0090},
    {"ts": "2026-10-20T10:00:00Z", "consumer": "resumo-diario", "model": "gpt-4o-mini",
     "calls": 1, "prompt_tokens": 12000, "completion_tokens": 800, "cost_usd": 0.0024},
    {"ts": "2026-10-21T08:00:00Z", "consumer": "resumo-diario", "model": "gpt-4o-mini",
     "calls": 1, "prompt_tokens": 11000, "completion_tokens": 700, "cost_usd": 0.0022},
    {"ts": "2026-10-21T11:30:00Z", "consumer": "transcricao", "model": "whisper-1",
     "calls": 1, "prompt_tokens": 0, "completion_tokens": 0, "cost_usd": 0.0120},
    {"ts": "2026-10-22T14:00:00Z", "consumer": "bot-telegram", "model": "gpt-4o-mini",
     "calls": 1, "prompt_tokens": 60000, "completion_tokens": 4000, "cost_usd": 0.0120},
]


def main():
    if len(sys.argv) > 1:
        registros = list(carregar(sys.argv[1]))
    else:
        registros = EXEMPLO

    grupos = defaultdict(lambda: {"chamadas": 0, "prompt": 0,
                                 "completion": 0, "custo": 0.0})
    for r in registros:
        chave = (r.get("consumer", "?"), r.get("model", "?"),
                 para_data(r.get("ts") or r.get("date")))
        g = grupos[chave]
        g["chamadas"] += int(r.get("calls", 1) or 1)
        g["prompt"] += int(r.get("prompt_tokens", 0) or 0)
        g["completion"] += int(r.get("completion_tokens", 0) or 0)
        g["custo"] += float(r.get("cost_usd", 0) or 0)

    por_consumidor = defaultdict(lambda: {"chamadas": 0, "prompt": 0,
                                         "completion": 0, "custo": 0.0})
    for (consumidor, modelo, dia), g in grupos.items():
        c = por_consumidor[consumidor]
        c["chamadas"] += g["chamadas"]
        c["prompt"] += g["prompt"]
        c["completion"] += g["completion"]
        c["custo"] += g["custo"]

    print("CONSUMO POR CONSUMIDOR (total da janela)")
    print(f"{'consumidor':20} {'chamadas':>9} {'prompt':>10} "
          f"{'completion':>11} {'custo_usd':>10}")
    for consumidor, c in sorted(por_consumidor.items(),
                               key=lambda x: -x[1]["custo"]):
        print(f"{consumidor:20} {c['chamadas']:9} {c['prompt']:10} "
              f"{c['completion']:11} {c['custo']:10.4f}")

    print("\nDETALHE POR DIA E MODELO")
    for (consumidor, modelo, dia), g in sorted(grupos.items()):
        print(f"{dia}  {consumidor:20} {modelo:14} chamadas={g['chamadas']:4} "
              f"prompt={g['prompt']:8} completion={g['completion']:7} "
              f"custo={g['custo']:.4f}")

    total = sum(c["custo"] for c in por_consumidor.values())
    print(f"\nTOTAL: {len(registros)} registros, "
          f"{len(por_consumidor)} consumidores, "
          f"custo total = {total:.4f} USD")


if __name__ == "__main__":
    main()

python3 custo_por_consumidor.py uso.jsonl

import json, time

def chamar(consumer, model, fn, *args, **kwargs):
    inicio = time.time()
    resp = fn(*args, **kwargs)
    uso = getattr(resp, "usage", None)
    registro = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "consumer": consumer,
        "model": model,
        "calls": 1,
        "prompt_tokens": getattr(uso, "prompt_tokens", 0) if uso else 0,
        "completion_tokens": getattr(uso, "completion_tokens", 0) if uso else 0,
        "cost_usd": 0.0,
    }
    with open("uso.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps(registro) + "\n")
    return resp