YANG JIANKUAN
7378d6bb88
- Claude：复用 Claude Code 钥匙串登录态直调 /api/oauth/usage，零依赖 statusline 与 Claude Code 进程；只读不刷新 token - Codex：经 codex app-server JSON-RPC 读 ChatGPT 订阅的 Codex 额度池（普通聊天额度无账户级途径，右块改名 Codex Usage） - 两路采集共用 TTL 缓存＋失败退避（CLAUDE_USAGE_TTL／CODEX_USAGE_TTL，语义同 WEATHER_TTL），推送端两路独立上报 - E1002：Claude 7d 与 fable 7d 合并为一行（单边框内上下对半实心，数字取较大者）、标签去 all 前缀、空值按 00% 00 占位等宽 - E1002／网页版：今日实时环比同比改涨绿跌红，与 K 线一致 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
63 lines
2.8 KiB
Python
63 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
用量采集的公共骨架 —— 本地缓存 + 有效期（TTL）+ 失败退避（Claude / Codex 两个采集器共用）。
|
||
|
||
为什么需要：launchd 每 5 分钟起两个独立进程（小屏渲染 + 上报），且用户可能手动连跑几次；
|
||
账户级用量接口对突发调用敏感（Anthropic 的 /api/oauth/usage 一分钟内几十次即 429 并封约 24 小时，
|
||
连带 Claude Code `/usage` 与 claude.ai 一起看不到用量）。故：
|
||
- 距上次成功拉取不足 TTL 秒（CLAUDE_USAGE_TTL / CODEX_USAGE_TTL，语义同 WEATHER_TTL/MAO_TTL/BTC_TTL，0=每次实拉）→ 直接用缓存，不发请求；
|
||
- 拉取失败 → 沿用旧缓存（标 stale），并按 BACKOFF 秒退避，退避期内不再尝试；异常对象带 `backoff` 属性时按它退避
|
||
（claude_usage.TokenUnavailable = 一个 TTL：token 过期是本机状态、Claude Code 随时可能刷新，不必等满 BACKOFF）；
|
||
- 从未成功且本次失败 → 抛出，由上层回退占位 "--"。
|
||
缓存文件只存快照与时间戳，**不存任何凭证**。
|
||
"""
|
||
import json
|
||
import os
|
||
import time
|
||
|
||
|
||
def cached_snapshot(path, ttl, backoff, fetch, force=False):
|
||
"""返回 (snapshot, status)。status ∈ {"fetched", "cache", "stale", "backoff"}。
|
||
fetch() 需返回可 JSON 序列化的快照 dict（含 updated_at）。"""
|
||
state = _load(path)
|
||
now = time.time()
|
||
snap = state.get("snapshot")
|
||
if not force and snap and now - state.get("fetched_at", 0) < ttl:
|
||
return snap, "cache"
|
||
if not force and state.get("backoff_until", 0) > now:
|
||
if snap:
|
||
return snap, "backoff"
|
||
raise RuntimeError(f"退避期内（至 {time.strftime('%H:%M:%S', time.localtime(state['backoff_until']))}），且无旧快照")
|
||
try:
|
||
snap_new = fetch()
|
||
except Exception as e:
|
||
state["backoff_until"] = now + getattr(e, "backoff", backoff) # 异常可自带更短退避（如本机 token 过期，只需等一个 TTL）
|
||
state["last_error"] = f"{type(e).__name__}: {e}"[:300]
|
||
_save(path, state)
|
||
if snap:
|
||
print(f"[用量] 拉取失败，沿用旧快照：{e}")
|
||
return snap, "stale"
|
||
raise
|
||
state = {"fetched_at": now, "snapshot": snap_new}
|
||
_save(path, state)
|
||
return snap_new, "fetched"
|
||
|
||
|
||
def _load(path):
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def _save(path, state):
|
||
tmp = path + ".tmp"
|
||
try:
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(state, f, ensure_ascii=False)
|
||
os.replace(tmp, path)
|
||
except Exception as e: # 缓存写不进去不算致命
|
||
print(f"[用量] 缓存写入失败：{e}")
|