Eagle
850f003df1
- Web API (curl_cffi + Chrome TLS fingerprint) as primary - OAuth API (stdlib urllib) as fallback - Burn rate analysis, daily marks, history logging
337 lines
13 KiB
Python
Executable file
337 lines
13 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
# /// script
|
||
# requires-python = ">=3.10"
|
||
# dependencies = ["curl_cffi"]
|
||
# ///
|
||
"""Claude API Usage — uses curl_cffi for TLS fingerprint impersonation.
|
||
|
||
Two API paths for fetching usage data:
|
||
|
||
1. Web API (Primary) — curl_cffi + sessionKey cookie
|
||
- Endpoint: claude.ai/api/organizations/{org_id}/usage
|
||
- Auth: browser sessionKey cookie (WEB_TOKEN)
|
||
- TLS: curl_cffi impersonate="chrome" to bypass Cloudflare JA3/JA4 fingerprint check
|
||
- Pros: returns extra_usage, prepaid/credits balance
|
||
- Cons: sessionKey expires, needs manual refresh from browser devtools
|
||
- Cloudflare blocks standard Python HTTP clients (urllib/httpx/requests) → 403
|
||
|
||
2. OAuth API (Fallback) — stdlib urllib + Bearer token
|
||
- Endpoint: api.anthropic.com/api/oauth/usage
|
||
- Auth: OAuth Bearer token (auto-refresh via refresh_token)
|
||
- TLS: standard Python TLS (no fingerprint needed, Anthropic API doesn't use Cloudflare bot detection)
|
||
- Pros: token auto-refreshes, no Cloudflare issues
|
||
- Cons: no extra_usage/prepaid balance data
|
||
|
||
Priority: Web API → OAuth fallback. Use --oauth flag to skip Web API.
|
||
"""
|
||
import json, os, sys, time, urllib.request, urllib.parse
|
||
from curl_cffi import requests as cffi_requests
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
TOKEN_FILE = SCRIPT_DIR / "claude-token.json"
|
||
MARK_FILE = SCRIPT_DIR / "claude-usage.json"
|
||
HISTORY_DIR = SCRIPT_DIR.parent / "data" / "usage-history"
|
||
CLAUDE_CREDS = Path.home() / ".claude" / ".credentials.json"
|
||
CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e"
|
||
REFRESH_URL = "https://platform.claude.com/v1/oauth/token"
|
||
USAGE_URL = "https://api.anthropic.com/api/oauth/usage"
|
||
BETA = "oauth-2025-04-20"
|
||
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36"
|
||
# WEB_TOKEN can be acquired from browser devtools by page https://claude.ai/settings/usage with cookie "sessionKey"
|
||
WEB_TOKEN = "YOUR_SESSION_KEY_HERE"
|
||
MULTIPLIER = 1 # Both now 20x Max plan (踏踏 upgraded 2026-02-11)
|
||
|
||
def load_json(p):
|
||
with open(p) as f: return json.load(f)
|
||
|
||
def save_json(p, d):
|
||
with open(p, "w") as f: json.dump(d, f, indent=2)
|
||
|
||
def extract_from_creds():
|
||
if not CLAUDE_CREDS.exists(): return None
|
||
o = load_json(CLAUDE_CREDS).get("claudeAiOauth", {})
|
||
t = {k: o.get(k, "") for k in ("accessToken", "refreshToken", "expiresAt")}
|
||
if not t["accessToken"] and not t["refreshToken"]: return None
|
||
save_json(TOKEN_FILE, t)
|
||
return t
|
||
|
||
def do_refresh(rt):
|
||
body = urllib.parse.urlencode({"grant_type": "refresh_token", "refresh_token": rt, "client_id": CLIENT_ID}).encode()
|
||
req = urllib.request.Request(REFRESH_URL, data=body, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json", "User-Agent": UA})
|
||
try:
|
||
r = json.loads(urllib.request.urlopen(req).read())
|
||
except urllib.error.HTTPError as e:
|
||
print(f"⚠️ Token refresh failed: {e.code} {e.reason}", file=sys.stderr)
|
||
return None
|
||
if "access_token" not in r: return None
|
||
t = {"accessToken": r["access_token"], "refreshToken": r.get("refresh_token", rt), "expiresAt": (time.time() + r.get("expires_in", 3600)) * 1000}
|
||
save_json(TOKEN_FILE, t)
|
||
return t["accessToken"]
|
||
|
||
def get_token():
|
||
for attempt in range(2):
|
||
if not TOKEN_FILE.exists():
|
||
if not extract_from_creds(): break
|
||
t = load_json(TOKEN_FILE)
|
||
if int(t.get("expiresAt", 0)) / 1000 > time.time() and t.get("accessToken"):
|
||
return t["accessToken"]
|
||
if t.get("refreshToken"):
|
||
at = do_refresh(t["refreshToken"])
|
||
if at: return at
|
||
if attempt == 0:
|
||
TOKEN_FILE.unlink(missing_ok=True)
|
||
if not extract_from_creds(): break
|
||
return None
|
||
|
||
def fetch_usage(token):
|
||
req = urllib.request.Request(USAGE_URL, headers={"Authorization": f"Bearer {token}", "Accept": "application/json", "anthropic-beta": BETA, "User-Agent": UA})
|
||
try:
|
||
resp = urllib.request.urlopen(req)
|
||
return json.loads(resp.read())
|
||
except urllib.error.HTTPError as e:
|
||
print(f"⚠️ API Error: {e.code} {e.reason}"); sys.exit(1)
|
||
|
||
def fetch_usage_web():
|
||
"""Primary: curl_cffi (Chrome TLS fingerprint) + WEB_TOKEN cookie."""
|
||
if not WEB_TOKEN:
|
||
return None
|
||
try:
|
||
resp = cffi_requests.get(
|
||
"https://claude.ai/api/organizations/YOUR_ORG_ID_HERE/usage",
|
||
headers={"User-Agent": UA},
|
||
cookies={"sessionKey": WEB_TOKEN},
|
||
impersonate="chrome",
|
||
timeout=15,
|
||
)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
data["_source"] = "claude.ai web"
|
||
return data
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
def bar(pct, w=10, scale=100):
|
||
"""Draw bar. pct is USED amount out of `scale`. More filled = more used = danger."""
|
||
clamped = max(0, min(pct, scale))
|
||
te = clamped * w * 8 // scale; fb, rem = te // 8, te % 8
|
||
b = "█" * fb + ("▏▎▍▌▋▊▉"[rem-1] if rem else "")
|
||
return b + "░" * (w - len(b))
|
||
|
||
def parse_reset_dt(s):
|
||
"""Parse ISO reset timestamp to datetime."""
|
||
if not s: return None
|
||
try:
|
||
return datetime.fromisoformat(s.replace("Z", "+00:00"))
|
||
except: return None
|
||
|
||
def burn_rate_emoji(used_pct, reset_at_str, period_hours):
|
||
"""Burn-rate based emoji: actual/expected ratio.
|
||
🟢 ratio < 1.0 (under budget), 🟡 1.0~1.5 (slightly over), 🔴 >1.5 (danger)"""
|
||
reset_at = parse_reset_dt(reset_at_str)
|
||
if not reset_at or period_hours <= 0:
|
||
return "🟢", 0.0
|
||
now_utc = datetime.now(timezone.utc)
|
||
remaining_s = (reset_at - now_utc).total_seconds()
|
||
elapsed_h = period_hours - remaining_s / 3600
|
||
if elapsed_h <= 0:
|
||
return "🟢", 0.0
|
||
expected_pct = (elapsed_h / period_hours) * 100
|
||
if expected_pct <= 0:
|
||
return "🟢", 0.0
|
||
ratio = used_pct / expected_pct
|
||
if ratio > 1.5:
|
||
return "🔴", ratio
|
||
elif ratio > 1.0:
|
||
return "🟡", ratio
|
||
else:
|
||
return "🟢", ratio
|
||
|
||
def fmt_reset(s):
|
||
if not s: return ""
|
||
try:
|
||
secs = max(0, (datetime.fromisoformat(s.replace("Z", "+00:00")) - datetime.now(timezone.utc)).total_seconds())
|
||
h, m = int(secs // 3600), int(secs % 3600 // 60)
|
||
return f"{h//24}d{h%24}h" if h > 24 else f"{h}h{m}m" if h else f"{m}m"
|
||
except: return ""
|
||
|
||
def get_weekly_delta(current_used_pct, scale):
|
||
"""Get delta of USED % from today's 00:00 mark. Positive = used more today.
|
||
Mark file stores REMAINING %, so we convert: baseline_used = scale - baseline_remain."""
|
||
if not MARK_FILE.exists(): return ""
|
||
try:
|
||
marks = load_json(MARK_FILE)
|
||
tw_tz = timezone(timedelta(hours=8))
|
||
today_tw = datetime.now(tw_tz).strftime("%Y-%m-%d")
|
||
today_utc = datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
||
baseline_remain = marks.get(today_tw) or marks.get(today_utc)
|
||
if baseline_remain is None: return ""
|
||
baseline_used = scale - baseline_remain
|
||
delta = current_used_pct - baseline_used
|
||
if delta == 0: return "-"
|
||
sign = "+" if delta > 0 else ""
|
||
return f" {sign}{delta}%"
|
||
except: return ""
|
||
|
||
def do_mark(data):
|
||
"""Save current weekly remaining % to mark file."""
|
||
w = data.get("seven_day")
|
||
if not w: return
|
||
pct = max(0, int(100 - w.get("utilization", 0)))
|
||
scaled = pct * MULTIPLIER
|
||
tw_tz = timezone(timedelta(hours=8))
|
||
today = datetime.now(tw_tz).strftime("%Y-%m-%d")
|
||
marks = {}
|
||
if MARK_FILE.exists():
|
||
try: marks = load_json(MARK_FILE)
|
||
except: pass
|
||
marks[today] = scaled
|
||
# Keep only last 14 days
|
||
keys = sorted(marks.keys())
|
||
if len(keys) > 14:
|
||
for k in keys[:-14]: del marks[k]
|
||
save_json(MARK_FILE, marks)
|
||
print(f"✅ Marked weekly: {scaled}% (of {MULTIPLIER*100}%) at {today}")
|
||
|
||
def do_log(data):
|
||
"""Append current usage snapshot to daily JSONL file."""
|
||
HISTORY_DIR.mkdir(parents=True, exist_ok=True)
|
||
now = datetime.now(timezone.utc)
|
||
today = now.strftime("%Y%m%d")
|
||
path = HISTORY_DIR / f"usage-history-{today}.jsonl"
|
||
|
||
snapshot = {"ts": now.isoformat(), "raw": {}}
|
||
for key in ("five_hour", "seven_day", "seven_day_sonnet", "seven_day_opus"):
|
||
w = data.get(key)
|
||
if w:
|
||
snapshot["raw"][key] = {"utilization": w.get("utilization", 0), "resets_at": w.get("resets_at", "")}
|
||
ex = data.get("extra_usage")
|
||
if ex and ex.get("is_enabled"):
|
||
snapshot["raw"]["extra_usage"] = {"used_credits": ex.get("used_credits", 0), "monthly_limit": ex.get("monthly_limit", 0)}
|
||
|
||
with open(path, "a") as f:
|
||
f.write(json.dumps(snapshot) + "\n")
|
||
|
||
# --- main ---
|
||
is_mark = "--mark" in sys.argv
|
||
is_log = "--log" in sys.argv
|
||
is_oauth = "--oauth" in sys.argv
|
||
|
||
# Priority: WEB API (curl_chrome) → OAuth fallback
|
||
# --oauth: skip WEB API, force OAuth
|
||
data = None if is_oauth else fetch_usage_web()
|
||
if not data:
|
||
token = get_token()
|
||
if not token:
|
||
print("❌ No valid token. Check WEB_TOKEN or run `claude` to login."); sys.exit(1)
|
||
data = fetch_usage(token)
|
||
|
||
if is_mark:
|
||
do_mark(data)
|
||
sys.exit(0)
|
||
|
||
if is_log:
|
||
do_log(data)
|
||
sys.exit(0)
|
||
|
||
scale = MULTIPLIER * 100
|
||
source = data.get("_source", "cc api")
|
||
now_ts = datetime.now(timezone(timedelta(hours=8))).strftime("%H:%M:%S")
|
||
print(f"📊 Token Usage ({source} @ {now_ts})\n\n```")
|
||
|
||
PERIOD_MAP = {"five_hour": 5, "seven_day": 168, "seven_day_sonnet": 168, "seven_day_opus": 168}
|
||
burn_warnings = []
|
||
|
||
for name, key in [("Session", "five_hour"), ("Weekly", "seven_day"), ("Sonnet", "seven_day_sonnet"), ("Opus", "seven_day_opus")]:
|
||
w = data.get(key)
|
||
if not w: continue
|
||
used_pct = max(0, int(w.get("utilization", 0)))
|
||
r = fmt_reset(w.get("resets_at"))
|
||
e, ratio = burn_rate_emoji(used_pct, w.get("resets_at"), PERIOD_MAP.get(key, 168))
|
||
reset_str = f" ↻{r}" if r else ""
|
||
if key == "seven_day":
|
||
scaled_remain = max(0, int(100 - used_pct)) * MULTIPLIER
|
||
scaled_used = max(0, 100 * MULTIPLIER - scaled_remain)
|
||
delta = get_weekly_delta(scaled_used, scale)
|
||
print(f"{name:<7} {bar(scaled_used, scale=scale)} {scaled_used:>3d}% used {e}{reset_str}{delta}")
|
||
else:
|
||
print(f"{name:<7} {bar(used_pct)} {used_pct:>3d}% used {e}{reset_str}")
|
||
# Collect burn warnings for 🟡 and 🔴
|
||
if e in ("🟡", "🔴"):
|
||
reset_dt = parse_reset_dt(w.get("resets_at"))
|
||
if reset_dt:
|
||
elapsed_h = PERIOD_MAP.get(key, 168) - (reset_dt - datetime.now(timezone.utc)).total_seconds() / 3600
|
||
expected = (elapsed_h / PERIOD_MAP.get(key, 168)) * 100 if elapsed_h > 0 else 0
|
||
burn_warnings.append(f"⚠️ {name} {ratio:.1f}x burn（預期 {expected:.0f}%，實際 {used_pct}%）")
|
||
|
||
# Extra usage
|
||
ex = data.get("extra_usage")
|
||
if ex and ex.get("is_enabled"):
|
||
spent = (ex.get("used_credits") or 0) / 100
|
||
limit_raw = ex.get("monthly_limit")
|
||
if limit_raw:
|
||
limit_usd = limit_raw / 100
|
||
used_pct = int(spent / limit_usd * 100) if limit_usd else 0
|
||
e, _ = burn_rate_emoji(used_pct, None, 0) # Extra has no meaningful burn rate
|
||
print(f"{'Extra':<7} {bar(used_pct)} {used_pct:>3d}% used {e} ${spent:.2f}/${limit_usd:.0f}")
|
||
else:
|
||
print(f"{'Extra':<7} ${spent:.2f} spent / unlimited")
|
||
# Fetch balance via prepaid/credits endpoint
|
||
try:
|
||
cr = cffi_requests.get(
|
||
"https://claude.ai/api/organizations/YOUR_ORG_ID_HERE/prepaid/credits",
|
||
headers={"User-Agent": UA},
|
||
cookies={"sessionKey": WEB_TOKEN},
|
||
impersonate="chrome",
|
||
timeout=15,
|
||
)
|
||
if cr.status_code == 200:
|
||
cdata = cr.json()
|
||
bal = (cdata.get("amount") or 0) / 100
|
||
print(f"{'ExBal':<7} ${bal:.2f}")
|
||
except:
|
||
pass
|
||
|
||
# Weekly burn rate analysis
|
||
w_data = data.get("seven_day")
|
||
if w_data:
|
||
w_used = max(0, w_data.get("utilization", 0))
|
||
w_reset_dt = parse_reset_dt(w_data.get("resets_at"))
|
||
if w_reset_dt:
|
||
now_utc = datetime.now(timezone.utc)
|
||
remaining_s = max(0, (w_reset_dt - now_utc).total_seconds())
|
||
remaining_h = remaining_s / 3600
|
||
elapsed_h = 168 - remaining_h # 7 days = 168 hours
|
||
remaining_pct = max(0, 100 - w_used)
|
||
|
||
if elapsed_h > 0:
|
||
perfect_rate = 100 / 7 # 14.29%/day
|
||
current_rate = w_used / (elapsed_h / 24)
|
||
ratio_pct = int(current_rate / perfect_rate * 100)
|
||
|
||
# Remaining days
|
||
remaining_d = remaining_h / 24
|
||
max_daily_rate = remaining_pct / remaining_d if remaining_d > 0 else 0
|
||
|
||
print(f"\n--- 燃燒速率 ---")
|
||
print(f"完美速率: {perfect_rate:.1f}%/天")
|
||
print(f"當前速率: {current_rate:.1f}%/天（{ratio_pct}% ratio）")
|
||
rem_d_str = f"{int(remaining_h//24)}d{int(remaining_h%24)}h"
|
||
print(f"剩下 {remaining_pct:.0f}% / {rem_d_str}")
|
||
|
||
if current_rate > 0 and max_daily_rate > current_rate:
|
||
turbo_pct = int(max_daily_rate / current_rate * 100)
|
||
print(f"最高可用 {max_daily_rate:.1f}%/天燒完")
|
||
print(f"可以超速 {turbo_pct}%喔! 🏎️")
|
||
elif current_rate > 0 and max_daily_rate <= current_rate:
|
||
brake_pct = int(max_daily_rate / current_rate * 100)
|
||
print(f"需降至 {max_daily_rate:.1f}%/天才能撐到重置")
|
||
print(f"請保持速限 {brake_pct}%...小心駕駛 🚗")
|
||
|
||
# Burn rate warnings (🟡 and 🔴 only)
|
||
for w in burn_warnings:
|
||
print(w)
|
||
|
||
print("```")
|