124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
"""Usage Codex (ChatGPT) via l'endpoint `GET https://chatgpt.com/backend-api/wham/usage`.
|
|
|
|
La réponse expose `rate_limit.primary_window` (5 h) et `secondary_window` (hebdo) avec
|
|
`used_percent` + `reset_*`, soit l'équivalent du `/usage` de Claude → on en fait des
|
|
jauges « % restant ».
|
|
|
|
Le token d'accès est lu dans un `auth.json` maintenu frais par un processus externe (monté
|
|
en lecture seule). Monitorink ne fait que le relire (aucun refresh ici).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import binascii
|
|
import json
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
|
|
import httpx
|
|
|
|
from config import config
|
|
|
|
USAGE_URL = "https://chatgpt.com/backend-api/wham/usage"
|
|
|
|
|
|
@dataclass
|
|
class CodexGauge:
|
|
name: str
|
|
remaining: float
|
|
resets_in: str
|
|
|
|
|
|
@dataclass
|
|
class CodexStatus:
|
|
ok: bool
|
|
error: str | None = None
|
|
plan_type: str = ""
|
|
limited: bool = False
|
|
gauges: list[CodexGauge] = field(default_factory=list)
|
|
|
|
|
|
def _human_delay(seconds: int) -> str:
|
|
seconds = max(0, int(seconds))
|
|
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"
|
|
|
|
|
|
def _read_token() -> str | None:
|
|
"""Lit le token openai-codex dans le auth.json (pool actif, sinon provider)."""
|
|
try:
|
|
with open(config.codex_token_file) as f:
|
|
data = json.load(f)
|
|
except (OSError, json.JSONDecodeError):
|
|
return None
|
|
pool = (((data.get("credential_pool") or {}).get("openai-codex") or [{}]) or [{}])[0]
|
|
token = pool.get("access_token")
|
|
if not token:
|
|
token = ((data.get("providers") or {}).get("openai-codex") or {}).get("tokens", {}).get("access_token")
|
|
return token or None
|
|
|
|
|
|
def _account_id(jwt: str) -> str | None:
|
|
"""Extrait chatgpt_account_id du payload du JWT (header `chatgpt-account-id` requis)."""
|
|
try:
|
|
payload = jwt.split(".")[1]
|
|
payload += "=" * (-len(payload) % 4)
|
|
claims = json.loads(base64.urlsafe_b64decode(payload))
|
|
except (IndexError, ValueError, binascii.Error, json.JSONDecodeError):
|
|
return None
|
|
return (claims.get("https://api.openai.com/auth") or {}).get("chatgpt_account_id")
|
|
|
|
|
|
def _gauge(window: dict | None, name: str) -> CodexGauge | None:
|
|
if not window:
|
|
return None
|
|
used = float(window.get("used_percent", 0) or 0)
|
|
resets = window.get("reset_after_seconds")
|
|
if resets is None and window.get("reset_at"):
|
|
resets = int(window["reset_at"]) - int(time.time())
|
|
return CodexGauge(name=name, remaining=max(0.0, 100 - used), resets_in=_human_delay(resets or 0))
|
|
|
|
|
|
async def fetch_status() -> CodexStatus:
|
|
if not config.codex_token_file:
|
|
return CodexStatus(ok=False, error="Codex désactivé")
|
|
token = _read_token()
|
|
if not token:
|
|
return CodexStatus(ok=False, error="token codex introuvable")
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Accept": "application/json",
|
|
"User-Agent": "Monitorink/1.0",
|
|
}
|
|
acct = _account_id(token)
|
|
if acct:
|
|
headers["chatgpt-account-id"] = acct
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(USAGE_URL, headers=headers)
|
|
if resp.status_code != 200:
|
|
return CodexStatus(ok=False, error=f"HTTP {resp.status_code}")
|
|
data = resp.json()
|
|
except httpx.HTTPError:
|
|
return CodexStatus(ok=False, error="injoignable")
|
|
|
|
rl = data.get("rate_limit") or {}
|
|
gauges = [
|
|
g for g in (
|
|
_gauge(rl.get("primary_window"), "Session (5 h)"),
|
|
_gauge(rl.get("secondary_window"), "Hebdo (7 j)"),
|
|
) if g is not None
|
|
]
|
|
return CodexStatus(
|
|
ok=True,
|
|
plan_type=str(data.get("plan_type") or ""),
|
|
limited=bool(rl.get("limit_reached")),
|
|
gauges=gauges,
|
|
)
|