245 lines
7.6 KiB
Python
245 lines
7.6 KiB
Python
"""Consulta de crédito / saldo disponible vía la API de OpenAI (y compatibles).
|
|
|
|
OpenAI no documenta un endpoint público de saldo con secret key; el endpoint
|
|
histórico de credit_grants sigue siendo el más usado por herramientas de
|
|
monitorización y por APIs compatibles. Se consulta con la key del servidor
|
|
y se normaliza la respuesta para la UI.
|
|
|
|
Si EXTRACTOR_BASE_URL apunta a un mock/local, también se prueba la API oficial
|
|
de OpenAI con la misma key.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
from urllib.parse import urljoin, urlparse
|
|
|
|
import httpx
|
|
|
|
from .config import extractor_llm_config
|
|
|
|
_log = logging.getLogger("domain.openai_balance")
|
|
|
|
_TIMEOUT = httpx.Timeout(connect=5.0, read=15.0, write=10.0, pool=5.0)
|
|
_OPENAI_OFFICIAL = "https://api.openai.com/v1"
|
|
|
|
|
|
def _is_openai_host(base_url: str) -> bool:
|
|
host = (urlparse(base_url).hostname or "").lower()
|
|
return host in {"api.openai.com", "openai.com"}
|
|
|
|
|
|
def _is_local_or_mock(base_url: str) -> bool:
|
|
host = (urlparse(base_url).hostname or "").lower()
|
|
return host in {
|
|
"localhost",
|
|
"127.0.0.1",
|
|
"host.docker.internal",
|
|
"0.0.0.0",
|
|
} or host.endswith(".local")
|
|
|
|
|
|
def _candidate_roots(base_url: str) -> list[str]:
|
|
cleaned = (base_url or _OPENAI_OFFICIAL).rstrip("/") + "/"
|
|
roots = [cleaned]
|
|
if cleaned.rstrip("/").endswith("/v1"):
|
|
roots.append(cleaned.rstrip("/").removesuffix("/v1").rstrip("/") + "/")
|
|
else:
|
|
roots.append(urljoin(cleaned, "v1/"))
|
|
seen: set[str] = set()
|
|
out: list[str] = []
|
|
for root in roots:
|
|
if root not in seen:
|
|
seen.add(root)
|
|
out.append(root)
|
|
return out
|
|
|
|
|
|
def _candidate_urls(base_url: str) -> list[str]:
|
|
"""URLs a probar: primero el proveedor configurado, luego OpenAI oficial si aplica."""
|
|
paths = (
|
|
"dashboard/billing/credit_grants",
|
|
"v1/dashboard/billing/credit_grants",
|
|
"credits", # OpenRouter-style
|
|
"user/balance", # DeepSeek-style
|
|
)
|
|
urls: list[str] = []
|
|
seen: set[str] = set()
|
|
|
|
bases = [base_url or _OPENAI_OFFICIAL]
|
|
# Con mock/local u otro proveedor, el saldo de cuenta OpenAI sigue en api.openai.com
|
|
if not _is_openai_host(bases[0]):
|
|
bases.append(_OPENAI_OFFICIAL)
|
|
|
|
for base in bases:
|
|
for root in _candidate_roots(base):
|
|
for path in paths:
|
|
url = urljoin(root, path)
|
|
if url not in seen:
|
|
seen.add(url)
|
|
urls.append(url)
|
|
return urls
|
|
|
|
|
|
def _as_float(value: Any) -> float | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _normalize_payload(payload: Any) -> dict[str, float] | None:
|
|
"""Extrae total_available / granted / used de respuestas conocidas."""
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
|
|
# OpenAI credit_grants
|
|
total_available = _as_float(payload.get("total_available"))
|
|
total_granted = _as_float(payload.get("total_granted"))
|
|
total_used = _as_float(payload.get("total_used"))
|
|
if total_available is not None:
|
|
return {
|
|
"total_available": total_available,
|
|
"total_granted": total_granted if total_granted is not None else total_available,
|
|
"total_used": total_used if total_used is not None else 0.0,
|
|
}
|
|
|
|
# OpenRouter: { data: { total_credits, total_usage } }
|
|
data = payload.get("data")
|
|
if isinstance(data, dict):
|
|
credits = _as_float(data.get("total_credits"))
|
|
usage = _as_float(data.get("total_usage"))
|
|
if credits is not None:
|
|
used = usage or 0.0
|
|
return {
|
|
"total_available": credits - used,
|
|
"total_granted": credits,
|
|
"total_used": used,
|
|
}
|
|
|
|
# DeepSeek: { balance_infos: [{ total_balance }] }
|
|
infos = payload.get("balance_infos")
|
|
if isinstance(infos, list) and infos:
|
|
totals = [_as_float(item.get("total_balance")) for item in infos if isinstance(item, dict)]
|
|
totals = [t for t in totals if t is not None]
|
|
if totals:
|
|
available = sum(totals)
|
|
return {"total_available": available, "total_granted": available, "total_used": 0.0}
|
|
|
|
for key in ("balance", "available", "remaining", "credit_balance", "total_balance"):
|
|
value = _as_float(payload.get(key))
|
|
if value is not None:
|
|
return {"total_available": value, "total_granted": value, "total_used": 0.0}
|
|
|
|
return None
|
|
|
|
|
|
def _friendly_error(detail: str, status_code: int, *, candidate: str = "") -> str:
|
|
lower = detail.lower()
|
|
if "session key" in lower or "browser" in lower or (
|
|
status_code in {401, 403} and _is_openai_host(candidate)
|
|
):
|
|
return (
|
|
"OpenAI no permite consultar el crédito con una API key "
|
|
"(sk-proj / sk-...). Solo se ve en el panel web: "
|
|
"https://platform.openai.com/settings/organization/billing/overview"
|
|
)
|
|
if status_code in {401, 403}:
|
|
return detail or f"Acceso denegado al consultar el saldo (HTTP {status_code})."
|
|
return detail or f"HTTP {status_code}"
|
|
|
|
|
|
def fetch_openai_balance(
|
|
*,
|
|
api_key: str | None = None,
|
|
base_url: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Devuelve el crédito disponible consultando la API del proveedor LLM."""
|
|
config = extractor_llm_config()
|
|
key = (api_key if api_key is not None else str(config["api_key"])).strip()
|
|
url = (base_url if base_url is not None else str(config["base_url"])).strip()
|
|
|
|
if not key:
|
|
return {
|
|
"ok": False,
|
|
"available": None,
|
|
"currency": "USD",
|
|
"source": None,
|
|
"error": "API key de OpenAI no configurada (EXTRACTOR_API_KEY).",
|
|
}
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
last_error = "No se pudo obtener el saldo de crédito."
|
|
dead_hosts: set[str] = set()
|
|
|
|
with httpx.Client(timeout=_TIMEOUT, follow_redirects=True) as client:
|
|
for candidate in _candidate_urls(url):
|
|
host = (urlparse(candidate).hostname or "").lower()
|
|
if host in dead_hosts:
|
|
continue
|
|
|
|
try:
|
|
response = client.get(candidate, headers=headers)
|
|
except httpx.HTTPError as exc:
|
|
last_error = f"Error de red al consultar {candidate}: {exc}"
|
|
_log.warning(last_error)
|
|
# Si el mock/local está caído, no insistir en ese host
|
|
if _is_local_or_mock(candidate) or "Connection refused" in str(exc):
|
|
dead_hosts.add(host)
|
|
continue
|
|
|
|
if response.status_code == 404:
|
|
continue
|
|
|
|
if response.status_code >= 400:
|
|
detail = ""
|
|
try:
|
|
body = response.json()
|
|
if isinstance(body, dict):
|
|
err = body.get("error")
|
|
if isinstance(err, dict):
|
|
detail = str(err.get("message") or err)
|
|
else:
|
|
detail = str(body.get("detail") or body.get("message") or "")
|
|
except Exception:
|
|
detail = (response.text or "")[:300]
|
|
last_error = _friendly_error(detail, response.status_code, candidate=candidate)
|
|
# 401/403 en OpenAI billing suelen aplicar a todos los endpoints
|
|
if response.status_code in {401, 403} and _is_openai_host(candidate):
|
|
break
|
|
continue
|
|
|
|
try:
|
|
payload = response.json()
|
|
except Exception:
|
|
last_error = f"Respuesta no JSON desde {candidate}"
|
|
continue
|
|
|
|
normalized = _normalize_payload(payload)
|
|
if normalized is None:
|
|
last_error = f"Formato de saldo no reconocido desde {candidate}"
|
|
continue
|
|
|
|
return {
|
|
"ok": True,
|
|
"available": round(normalized["total_available"], 4),
|
|
"total_granted": round(normalized["total_granted"], 4),
|
|
"total_used": round(normalized["total_used"], 4),
|
|
"currency": "USD",
|
|
"source": candidate,
|
|
"error": None,
|
|
}
|
|
|
|
return {
|
|
"ok": False,
|
|
"available": None,
|
|
"currency": "USD",
|
|
"source": None,
|
|
"error": last_error,
|
|
}
|