Учет средств (Usage)

def read_usage(usage: dict) -> dict:
    """Приводит usage обоих форматов к одному виду. Суммы не дублируются."""
    if "prompt_tokens" in usage:  # формат чата
        prompt = usage["prompt_tokens"]
        completion = usage["completion_tokens"]
        cached = usage.get("prompt_tokens_details", {}).get("cached_read_tokens", 0)
        reasoning = usage.get("completion_tokens_details", {}).get("reasoning_tokens", 0)
        cost = usage.get("cost", {}).get("total_cost")
    else:  # формат Anthropic
        cached = usage.get("cache_read_input_tokens", 0)
        prompt = usage["input_tokens"] + cached
        completion = usage["output_tokens"]
        reasoning = usage.get("output_tokens_details", {}).get("thinking_tokens", 0)
        cost = None

    return {
        "prompt_tokens": prompt,
        "completion_tokens": completion,
        "total_tokens": usage.get("total_tokens", prompt + completion),
        "cached_read_tokens": cached,      # уже учтено в prompt_tokens
        "reasoning_tokens": reasoning,    # уже учтено в completion_tokens
        "total_cost": cost,
    }

curl -s https://speshu.ai/api/v1/balance \
  -H "Authorization: Bearer $SPESHU_API_KEY"

import os
import uuid
from datetime import datetime, timezone
from decimal import Decimal

import requests

# свой идентификатор запроса, который вы передаёте в X-Request-Id или храните рядом
REQUEST_ID = str(uuid.uuid4())


def log_usage(conn, request_id: str, *, endpoint: str, model: str, usage: dict,
              status: int, user_id: str | None) -> None:
    """Одна строка на один HTTP-запрос к API."""
    parsed = read_usage(usage)

    conn.execute(
        """
        INSERT INTO api_usage (
            request_id, user_id, endpoint, model, status,
            prompt_tokens, completion_tokens,
            cached_read_tokens, reasoning_tokens,
            total_cost, created_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            request_id,
            user_id,
            endpoint,
            model,
            status,
            parsed["prompt_tokens"],
            parsed["completion_tokens"],
            parsed["cached_read_tokens"],
            parsed["reasoning_tokens"],
            # сумма приходит строкой или числом — храните десятичным числом
            Decimal(str(parsed["total_cost"])) if parsed["total_cost"] is not None else None,
            datetime.now(timezone.utc),
        ),
    )
    conn.commit()

# user_id берётся из вашего сеанса, а не из тела запроса к API
log_usage(
    conn,
    request_id=REQUEST_ID,
    endpoint="/api/v1/chat/completions",
    model=MODEL,
    usage=response["usage"],
    status=response.status_code,
    user_id=session.user_id,   # ваш собственный идентификатор
)

import os

BUDGET_PER_USER_PER_MONTH = Decimal("500.00")
ALERT_SHARE = Decimal("0.8")


def check_threshold(conn, user_id: str, month_start) -> Decimal:
    """Возвращает расход пользователя с начала месяца."""
    row = conn.execute(
        "SELECT COALESCE(SUM(total_cost), 0) FROM api_usage "
        "WHERE user_id = ? AND created_at >= ?",
        (user_id, month_start),
    ).fetchone()

    spent = Decimal(str(row[0]))
    if spent >= BUDGET_PER_USER_PER_MONTH * ALERT_SHARE:
        notify(user_id, spent=spent, budget=BUDGET_PER_USER_PER_MONTH)

    return spent