From 850f003df1129df052035d394cf040158af2a691 Mon Sep 17 00:00:00 2001 From: Eagle Date: Thu, 19 Mar 2026 07:15:45 +0800 Subject: [PATCH] =?UTF-8?q?init:=20claude-usage.py=20=E2=80=94=20dual-path?= =?UTF-8?q?=20Claude=20usage=20monitor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Web API (curl_cffi + Chrome TLS fingerprint) as primary - OAuth API (stdlib urllib) as fallback - Burn rate analysis, daily marks, history logging --- README.md | 60 +++++++++ claude-usage.py | 337 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 397 insertions(+) create mode 100644 README.md create mode 100755 claude-usage.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..f73efa2 --- /dev/null +++ b/README.md @@ -0,0 +1,60 @@ +# claude-usage + +Claude API usage monitor with dual-path authentication. + +## Two API Paths + +### 1. Web API (Primary) โ€” curl_cffi + sessionKey cookie +- Endpoint: `claude.ai/api/organizations/{org_id}/usage` +- Uses curl_cffi `impersonate="chrome"` to bypass Cloudflare TLS fingerprint (JA3/JA4) detection +- Returns extra_usage, prepaid/credits balance +- sessionKey needs manual refresh from browser devtools + +### 2. OAuth API (Fallback) โ€” stdlib urllib + Bearer token +- Endpoint: `api.anthropic.com/api/oauth/usage` +- Standard Python TLS (no fingerprint needed) +- Token auto-refreshes via refresh_token +- No extra_usage/prepaid data + +## Setup + +1. Install dependency: `pip install curl_cffi` (or use `uv run`) +2. Set `WEB_TOKEN` to your claude.ai sessionKey cookie +3. Set org ID in the API URLs +4. Place OAuth credentials at `~/.claude/.credentials.json` + +## Usage + +```bash +# Show usage +uv run claude-usage.py + +# Mark daily baseline (run at 00:00) +uv run claude-usage.py --mark + +# Log snapshot to history +uv run claude-usage.py --log + +# Force OAuth API (skip Web API) +uv run claude-usage.py --oauth +``` + +## Output + +``` +๐Ÿ“Š Token Usage (claude.ai web @ 07:04:49) + +Session โ–‘โ–‘โ–‘โ–‘โ–‘โ–‘โ–‘โ–‘โ–‘โ–‘ 1% used ๐ŸŸข โ†ป4h55m +Weekly โ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–Šโ–‘โ–‘ 78% used ๐ŸŸข โ†ป1d3h +20% +Sonnet โ–โ–‘โ–‘โ–‘โ–‘โ–‘โ–‘โ–‘โ–‘โ–‘ 2% used ๐ŸŸข โ†ป1d5h + +--- ็‡ƒ็‡’้€Ÿ็Ž‡ --- +ๅฎŒ็พŽ้€Ÿ็Ž‡: 14.3%/ๅคฉ +็•ถๅ‰้€Ÿ็Ž‡: 13.4%/ๅคฉ๏ผˆ93% ratio๏ผ‰ +ๅฏไปฅ่ถ…้€Ÿ 141%ๅ–”! ๐ŸŽ๏ธ +``` + +## Credits + +Original: ๆ‹ๆ‹ ๐Ÿฆ… (Eagle) +Bash reference: ่ธ่ธ ๐Ÿพ (Tata) diff --git a/claude-usage.py b/claude-usage.py new file mode 100755 index 0000000..59a3975 --- /dev/null +++ b/claude-usage.py @@ -0,0 +1,337 @@ +#!/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("```")