134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
"""Récupération de l'usage de l'abonnement Claude via l'endpoint OAuth `/usage`.
|
|
|
|
Validé empiriquement (compte Max 5x) :
|
|
GET https://api.anthropic.com/api/oauth/usage
|
|
headers: Authorization: Bearer <token>, anthropic-beta: oauth-2025-04-20,
|
|
User-Agent: claude-code/<version>, Content-Type: application/json
|
|
réponse: { five_hour:{utilization,resets_at}, seven_day:{...},
|
|
seven_day_opus, seven_day_sonnet, extra_usage }
|
|
|
|
Le token utilisé est un token longue durée dédié généré par `claude setup-token`
|
|
(env MONITORINK_CLAUDE_TOKEN). Aucun refresh/écriture n'est effectué ici : en cas de 401,
|
|
on remonte un état d'erreur pour affichage (« token à régénérer »).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
|
|
from config import config
|
|
|
|
USAGE_URL = "https://api.anthropic.com/api/oauth/usage"
|
|
|
|
|
|
@dataclass
|
|
class Window:
|
|
"""Une fenêtre glissante d'usage (5h ou 7j)."""
|
|
|
|
utilization: float # 0..100
|
|
resets_at: datetime | None
|
|
|
|
@property
|
|
def remaining_pct(self) -> float:
|
|
return max(0.0, 100.0 - self.utilization)
|
|
|
|
@property
|
|
def resets_in_human(self) -> str:
|
|
if not self.resets_at:
|
|
return ""
|
|
delta = self.resets_at - datetime.now(timezone.utc)
|
|
secs = int(delta.total_seconds())
|
|
if secs <= 0:
|
|
return "bientôt"
|
|
h, m = divmod(secs // 60, 60)
|
|
if h >= 24:
|
|
return f"{h // 24}j {h % 24}h"
|
|
if h:
|
|
return f"{h}h{m:02d}"
|
|
return f"{m}min"
|
|
|
|
|
|
@dataclass
|
|
class ClaudeUsage:
|
|
ok: bool
|
|
error: str | None = None
|
|
five_hour: Window | None = None
|
|
seven_day: Window | None = None
|
|
seven_day_opus: Window | None = None
|
|
seven_day_sonnet: Window | None = None
|
|
burn_rate: float | None = None # tokens/min (ccusage, optionnel)
|
|
|
|
|
|
def _parse_window(raw: dict | None) -> Window | None:
|
|
if not raw:
|
|
return None
|
|
util = float(raw.get("utilization", 0) or 0)
|
|
resets = raw.get("resets_at")
|
|
dt = None
|
|
if resets:
|
|
try:
|
|
dt = datetime.fromisoformat(str(resets).replace("Z", "+00:00"))
|
|
except ValueError:
|
|
dt = None
|
|
return Window(utilization=util, resets_at=dt)
|
|
|
|
|
|
def _burn_rate_from_ccusage() -> float | None:
|
|
"""Burn rate du bloc 5h actif via `ccusage blocks --json` (best-effort)."""
|
|
try:
|
|
proc = subprocess.run(
|
|
["ccusage", "blocks", "--active", "--json"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=20,
|
|
)
|
|
if proc.returncode != 0:
|
|
return None
|
|
data = json.loads(proc.stdout)
|
|
blocks = data.get("blocks") or []
|
|
for b in blocks:
|
|
if b.get("isActive"):
|
|
bp = b.get("burnRate") or {}
|
|
rate = bp.get("tokensPerMinute") or bp.get("tokensPerMinuteForIndicator")
|
|
return float(rate) if rate is not None else None
|
|
except (subprocess.SubprocessError, json.JSONDecodeError, ValueError, FileNotFoundError):
|
|
return None
|
|
return None
|
|
|
|
|
|
async def fetch_usage() -> ClaudeUsage:
|
|
if not config.claude_token:
|
|
return ClaudeUsage(ok=False, error="MONITORINK_CLAUDE_TOKEN manquant")
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {config.claude_token}",
|
|
"anthropic-beta": "oauth-2025-04-20",
|
|
"User-Agent": config.claude_ua,
|
|
"Content-Type": "application/json",
|
|
}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(USAGE_URL, headers=headers)
|
|
except httpx.HTTPError as exc:
|
|
return ClaudeUsage(ok=False, error=f"réseau: {exc}")
|
|
|
|
if resp.status_code == 401:
|
|
return ClaudeUsage(ok=False, error="token expiré — relancer `claude setup-token`")
|
|
if resp.status_code != 200:
|
|
return ClaudeUsage(ok=False, error=f"HTTP {resp.status_code}")
|
|
|
|
data = resp.json()
|
|
burn = _burn_rate_from_ccusage() if config.ccusage_enabled else None
|
|
return ClaudeUsage(
|
|
ok=True,
|
|
five_hour=_parse_window(data.get("five_hour")),
|
|
seven_day=_parse_window(data.get("seven_day")),
|
|
seven_day_opus=_parse_window(data.get("seven_day_opus")),
|
|
seven_day_sonnet=_parse_window(data.get("seven_day_sonnet")),
|
|
burn_rate=burn,
|
|
)
|