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>
75 lines
2.9 KiB
Python
75 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
推送端（本机运行）—— 在本机采集两路账户级 AI 用量，POST 到 server.py；服务端暂存、渲染时现算 pace。
|
||
|
||
推什么（形态 A：凭证与采集都留在本机，服务端不持任何凭证、不读本机文件）：
|
||
- Claude：claude_usage.get_usage()（复用 Claude Code 钥匙串登录态调 /api/oauth/usage）→ POST /push/usage
|
||
- Codex：codex_usage.get_usage()（`codex app-server` RPC，独立 CODEX_HOME 的 ChatGPT 登录）→ POST /push/usage_gpt
|
||
两路彼此独立：一路失败不影响另一路；两路都失败才以非零码退出。采集器自带 ≥300s 最小间隔缓存，
|
||
所以与 main.py（小屏渲染）在同一分钟内先后运行也只打一次接口。
|
||
鉴权：请求头 X-Push-Token = PUSH_TOKEN。网络：直连（不走代理），连接类异常重试 2 次。
|
||
调度：与喵喵设备推送同节律，由 launchd 每 5 分钟跑一次 `main.py --target push`（run.sh 末尾）。
|
||
"""
|
||
import json
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
import claude_usage
|
||
import codex_usage
|
||
import config
|
||
|
||
RETRY_TIMES, RETRY_DELAY, TIMEOUT = 2, 2, 10
|
||
|
||
|
||
def _post_json(url, payload):
|
||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
req = urllib.request.Request(url, data=body, method="POST", headers={
|
||
"Content-Type": "application/json; charset=utf-8",
|
||
"X-Push-Token": config.PUSH_TOKEN,
|
||
"User-Agent": "eink-push/push_client",
|
||
})
|
||
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||
last = None
|
||
for i in range(RETRY_TIMES + 1):
|
||
try:
|
||
with opener.open(req, timeout=TIMEOUT) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as e: # 4xx/5xx 不重试，直接报出服务端的说明
|
||
raise RuntimeError(f"HTTP {e.code}: {e.read().decode('utf-8', 'replace')[:200]}") from e
|
||
except urllib.error.URLError as e: # 连接类：重试
|
||
last = e
|
||
if i < RETRY_TIMES:
|
||
time.sleep(RETRY_DELAY)
|
||
raise RuntimeError(f"连接失败：{last}")
|
||
|
||
|
||
SOURCES = (
|
||
("Claude", claude_usage.get_usage, "/push/usage"),
|
||
("Codex", codex_usage.get_usage, "/push/usage_gpt"),
|
||
)
|
||
|
||
|
||
def push_usage():
|
||
if not config.PUSH_URL or not config.PUSH_TOKEN:
|
||
raise SystemExit("[推送] 需在 .env 配置 PUSH_URL 与 PUSH_TOKEN")
|
||
base = config.PUSH_URL.rstrip("/")
|
||
ok = 0
|
||
for name, getter, path in SOURCES:
|
||
try:
|
||
raw = getter()
|
||
status = raw.pop("status", "")
|
||
resp = _post_json(base + path, raw)
|
||
print(f"[推送] {name} 用量（{status}）已推送到 {base}{path}：{resp}")
|
||
ok += 1
|
||
except Exception as e:
|
||
print(f"[推送] {name} 用量跳过：{e}")
|
||
if ok == 0:
|
||
raise SystemExit(1)
|
||
return ok
|
||
|
||
|
||
if __name__ == "__main__":
|
||
push_usage()
|