YANG JIANKUAN
85b5113ba4
- server.py：三条恒定 URL 并行——/e1002.png 内存现渲染 PNG（主力）、 /e1002.html 现代网页、/e1002-web.png 网页经无头 Chrome 截图后固化六色； 天气/橘喵/BTC 按 WEATHER_TTL/MAO_TTL/BTC_TTL 走内存缓存，过期才实拉、 失败沿用旧值；用量不读本机文件，改由 POST /push/usage(_gpt) 推送暂存 - push_client.py：本机把 Claude 用量快照推送到服务（X-Push-Token 鉴权） - btc_api.py：OKX/Coinbase/Huobi 免费接口逐源兜底，时K 近 72 小时，涨绿跌红 - data.py 拆出 weather_fetch/mao_fetch/btc_fetch 与 usage_from_raw， 统一按 TZ_NAME 时区；宽版只用点阵字体整数倍字号 12/24/36/48 - probe_server.py：验证平台对同一 URL 是否周期重抓（结论：Image 控件 不重抓，HTML 控件重抓） Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
56 lines
2.3 KiB
Python
56 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
推送端（本机运行）—— 把本机的用量原始快照 POST 到 server.py，服务端暂存、渲染时现算 pace。
|
||
|
||
推什么：
|
||
- Claude：~/.claude/usage-snapshot.json 原文（statusline 生产，含 five_hour / seven_day / seven_day_fable / updated_at）→ POST /push/usage
|
||
- ChatGPT：暂无本机数据源；将来有了按 data.GPT_WINDOWS 的键（five_hour / seven_day / seven_day_codex，各含 utilization + resets_at）
|
||
组一个 dict → POST /push/usage_gpt。目前不推，服务端用 mock 并标「示例数据」。
|
||
鉴权：请求头 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 config
|
||
import usage_local
|
||
|
||
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",
|
||
})
|
||
last = None
|
||
for i in range(RETRY_TIMES + 1):
|
||
try:
|
||
with urllib.request.urlopen(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}")
|
||
|
||
|
||
def push_usage():
|
||
if not config.PUSH_URL or not config.PUSH_TOKEN:
|
||
raise SystemExit("[推送] 需在 .env 配置 PUSH_URL 与 PUSH_TOKEN")
|
||
raw = usage_local.get_usage() # 缺文件直接抛错——没有快照就没什么可推
|
||
resp = _post_json(config.PUSH_URL.rstrip("/") + "/push/usage", raw)
|
||
print(f"[推送] Claude 用量快照已推送到 {config.PUSH_URL}：{resp}")
|
||
return resp
|
||
|
||
|
||
if __name__ == "__main__":
|
||
push_usage()
|