117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
"""Usage Codex (OpenAI via ChatGPT) exposé par l'agent maison Hermes.
|
|
|
|
Le plan ChatGPT/Codex ne fournit pas de quota « % restant » récupérable (le rate-limit
|
|
n'arrive que dans les headers d'un appel génératif). On affiche donc, depuis l'API du
|
|
dashboard Hermes :
|
|
- la conso Codex sur 7 jours (`/api/analytics/models`, provider `openai-codex`) ;
|
|
- un statut limite déduit des 429 `usage_limit_reached` loggés (`/api/logs`).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
import httpx
|
|
|
|
from config import config
|
|
|
|
_TOKEN_RE = re.compile(r'__HERMES_SESSION_TOKEN__="([^"]+)"')
|
|
_RESETS_RE = re.compile(r"'resets_at':\s*(\d+)")
|
|
_CODEX_PROVIDER = "openai-codex"
|
|
|
|
|
|
@dataclass
|
|
class CodexStatus:
|
|
ok: bool
|
|
error: str | None = None
|
|
api_calls: int = 0
|
|
total_tokens: int = 0
|
|
cost: float = 0.0
|
|
days: int = 7
|
|
limited: bool = False
|
|
resets_in_human: str = ""
|
|
|
|
@property
|
|
def tokens_human(self) -> str:
|
|
t = self.total_tokens
|
|
if t >= 1_000_000:
|
|
return f"{t / 1_000_000:.1f}".replace(".", ",") + " M"
|
|
if t >= 1_000:
|
|
return f"{t // 1000} k"
|
|
return str(t)
|
|
|
|
|
|
def _human_delay(seconds: int) -> str:
|
|
h, m = seconds // 3600, (seconds % 3600) // 60
|
|
if h >= 24:
|
|
return f"{h // 24}j {h % 24}h"
|
|
if h:
|
|
return f"{h}h{m:02d}"
|
|
return f"{m}min"
|
|
|
|
|
|
async def _session_token(client: httpx.AsyncClient, base: str) -> str | None:
|
|
"""Récupère le jeton de session injecté dans la page du dashboard Hermes."""
|
|
resp = await client.get(base + "/")
|
|
m = _TOKEN_RE.search(resp.text)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
async def fetch_status() -> CodexStatus:
|
|
base = config.hermes_url
|
|
if not base:
|
|
return CodexStatus(ok=False, error="Hermes désactivé")
|
|
base = base.rstrip("/")
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
token = await _session_token(client, base)
|
|
if not token:
|
|
return CodexStatus(ok=False, error="jeton Hermes introuvable")
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
models_resp = await client.get(
|
|
f"{base}/api/analytics/models", params={"days": 7}, headers=headers,
|
|
)
|
|
logs_resp = await client.get(
|
|
f"{base}/api/logs",
|
|
params={"file": "agent", "search": "usage_limit_reached", "lines": 300},
|
|
headers=headers,
|
|
)
|
|
if models_resp.status_code != 200:
|
|
return CodexStatus(ok=False, error=f"HTTP {models_resp.status_code}")
|
|
models = models_resp.json()
|
|
except httpx.HTTPError:
|
|
return CodexStatus(ok=False, error="injoignable")
|
|
|
|
calls = tokens = 0
|
|
cost = 0.0
|
|
for m in models.get("models") or []:
|
|
if m.get("provider") != _CODEX_PROVIDER:
|
|
continue
|
|
calls += int(m.get("api_calls", 0) or 0)
|
|
tokens += int(m.get("input_tokens", 0) or 0) + int(m.get("output_tokens", 0) or 0)
|
|
cost += float(m.get("estimated_cost", 0) or 0)
|
|
|
|
# Statut limite : un 429 dont le reset est encore dans le futur => limité.
|
|
limited = False
|
|
resets_in_human = ""
|
|
if logs_resp.status_code == 200:
|
|
now = int(time.time())
|
|
future_resets = [
|
|
int(ts) for ts in _RESETS_RE.findall("\n".join(logs_resp.json().get("lines") or []))
|
|
if int(ts) > now
|
|
]
|
|
if future_resets:
|
|
limited = True
|
|
resets_in_human = _human_delay(max(future_resets) - now)
|
|
|
|
return CodexStatus(
|
|
ok=True,
|
|
api_calls=calls,
|
|
total_tokens=tokens,
|
|
cost=round(cost, 2),
|
|
limited=limited,
|
|
resets_in_human=resets_in_human,
|
|
)
|