257 lines
8.6 KiB
Python
257 lines
8.6 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 }
|
|
|
|
⚠️ L'endpoint exige le scope OAuth `user:profile`. Le token `claude setup-token` ne l'a PAS
|
|
(403). On utilise donc les credentials d'un **login Claude isolé dédié** (CLAUDE_CONFIG_DIR
|
|
séparé) : structure `{ claudeAiOauth: { accessToken, refreshToken, expiresAt(ms) } }`.
|
|
|
|
Le token expire (~8 h) ; on le rafraîchit via platform.claude.com et on réécrit le fichier
|
|
isolé de façon atomique. On ne touche jamais le ~/.claude partagé.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
|
|
from config import config
|
|
|
|
USAGE_URL = "https://api.anthropic.com/api/oauth/usage"
|
|
REFRESH_URL = "https://platform.claude.com/v1/oauth/token"
|
|
CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e" # client public Claude Code
|
|
EXPIRY_BUFFER_MS = 120_000 # rafraîchit 2 min avant expiration
|
|
|
|
_refresh_lock = asyncio.Lock()
|
|
|
|
|
|
@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 ExtraUsage:
|
|
"""Crédits « extra usage » (pay-as-you-go au-delà de l'abonnement)."""
|
|
|
|
used: float
|
|
limit: float | None
|
|
currency: str = "EUR"
|
|
|
|
@property
|
|
def remaining(self) -> float | None:
|
|
"""Crédits extra encore disponibles avant le plafond mensuel."""
|
|
if self.limit is None:
|
|
return None
|
|
return max(0.0, self.limit - self.used)
|
|
|
|
@property
|
|
def label(self) -> str:
|
|
sym = "€" if self.currency == "EUR" else self.currency
|
|
if self.limit:
|
|
return (
|
|
f"Extra : {self.remaining:.0f}{sym} restant"
|
|
f" · {self.used:.0f}{sym} utilisé / {self.limit:.0f}{sym}"
|
|
)
|
|
return f"Extra : {self.used:.0f}{sym} utilisé"
|
|
|
|
|
|
@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
|
|
extra: ExtraUsage | 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 _load_creds() -> dict:
|
|
with open(config.claude_creds_path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def _write_creds(data: dict) -> None:
|
|
"""Écriture atomique du fichier de credentials isolé (mode 0600)."""
|
|
path = config.claude_creds_path
|
|
fd, tmp = tempfile.mkstemp(dir=os.path.dirname(path) or ".")
|
|
try:
|
|
with os.fdopen(fd, "w") as f:
|
|
json.dump(data, f)
|
|
os.chmod(tmp, 0o600)
|
|
os.replace(tmp, path)
|
|
except BaseException:
|
|
if os.path.exists(tmp):
|
|
os.unlink(tmp)
|
|
raise
|
|
|
|
|
|
async def _refresh(data: dict) -> str:
|
|
"""Rafraîchit l'access token, réécrit le fichier isolé, renvoie le nouvel access token."""
|
|
o = data["claudeAiOauth"]
|
|
body = {
|
|
"grant_type": "refresh_token",
|
|
"refresh_token": o["refreshToken"],
|
|
"client_id": CLIENT_ID,
|
|
}
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
resp = await client.post(
|
|
REFRESH_URL, json=body,
|
|
headers={"Content-Type": "application/json", "User-Agent": config.claude_ua},
|
|
)
|
|
resp.raise_for_status()
|
|
tok = resp.json()
|
|
o["accessToken"] = tok["access_token"]
|
|
o["refreshToken"] = tok["refresh_token"]
|
|
o["expiresAt"] = int(time.time() * 1000) + int(tok["expires_in"]) * 1000
|
|
_write_creds(data)
|
|
return o["accessToken"]
|
|
|
|
|
|
async def _valid_token() -> str:
|
|
"""Renvoie un access token valide, en rafraîchissant si nécessaire (sérialisé)."""
|
|
data = _load_creds()
|
|
o = data["claudeAiOauth"]
|
|
if int(o.get("expiresAt", 0)) - int(time.time() * 1000) > EXPIRY_BUFFER_MS:
|
|
return o["accessToken"]
|
|
async with _refresh_lock:
|
|
# Re-lecture après acquisition du lock : un autre coroutine a peut-être déjà rafraîchi.
|
|
data = _load_creds()
|
|
o = data["claudeAiOauth"]
|
|
if int(o.get("expiresAt", 0)) - int(time.time() * 1000) > EXPIRY_BUFFER_MS:
|
|
return o["accessToken"]
|
|
return await _refresh(data)
|
|
|
|
|
|
async def _get_usage(token: str) -> httpx.Response:
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"anthropic-beta": "oauth-2025-04-20",
|
|
"User-Agent": config.claude_ua,
|
|
"Content-Type": "application/json",
|
|
}
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
return await client.get(USAGE_URL, headers=headers)
|
|
|
|
|
|
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)
|
|
for b in data.get("blocks") or []:
|
|
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 os.path.exists(config.claude_creds_path):
|
|
return ClaudeUsage(ok=False, error="credentials Claude absents — login isolé requis")
|
|
|
|
try:
|
|
token = await _valid_token()
|
|
except (OSError, KeyError, json.JSONDecodeError) as exc:
|
|
return ClaudeUsage(ok=False, error=f"credentials illisibles: {exc}")
|
|
except httpx.HTTPError as exc:
|
|
return ClaudeUsage(ok=False, error=f"refresh échoué: {exc}")
|
|
|
|
try:
|
|
resp = await _get_usage(token)
|
|
# Token rejeté malgré le buffer -> une tentative de refresh forcé.
|
|
if resp.status_code == 401:
|
|
data = _load_creds()
|
|
async with _refresh_lock:
|
|
token = await _refresh(data)
|
|
resp = await _get_usage(token)
|
|
except httpx.HTTPError as exc:
|
|
return ClaudeUsage(ok=False, error=f"réseau: {exc}")
|
|
|
|
if resp.status_code == 401:
|
|
return ClaudeUsage(ok=False, error="auth invalide — relancer le login isolé")
|
|
if resp.status_code == 403:
|
|
return ClaudeUsage(ok=False, error="scope insuffisant (login complet requis)")
|
|
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
|
|
|
|
extra = None
|
|
eu = data.get("extra_usage") or {}
|
|
if eu.get("is_enabled") and eu.get("used_credits"):
|
|
extra = ExtraUsage(
|
|
used=float(eu.get("used_credits", 0)),
|
|
limit=float(eu["monthly_limit"]) if eu.get("monthly_limit") else None,
|
|
currency=eu.get("currency", "EUR"),
|
|
)
|
|
|
|
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")),
|
|
extra=extra,
|
|
burn_rate=burn,
|
|
)
|