eric/eink-push
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: E1002 新增公网图片服务（恒定 URL 现渲染）＋底部改为 BTC 时K

Browse Source 
- 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>
This commit is contained in:
杨建宽
2026-09-10 18:55:47 +08:00
parent 6dc89b504e
commit 85b5113ba4
14 changed files with 1192 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

213
data.py
View File

@@ -4,18 +4,35 @@
数据层 —— 汇总仪表盘所需的全部数据。
橘喵今日经营 + 近30天趋势来自 jm_api（jm-devops 后端统计接口，免鉴权 GET）。
天气来自 weather_api（和风天气，定位南京·秦淮）。日期/时间用真实系统时间。
天气来自 weather_api（和风天气，定位南京·江宁）。日期/时间用真实系统时间（统一按 config.TZ_NAME 时区）。
任何接口失败时相关数据以占位符 "--" 呈现，不回退 mock。
两种调用形态：
- 本机脚本（main.py）：get_dashboard_data() —— 带文件缓存（天气/趋势/BTC），用量读本地快照；
- 图片服务（server.py）：直接用各 `*_fetch()` 纯拉取函数 + `_usage(now, raw=<推送数据>)`，不落盘。
"""
import json
from datetime import datetime
from zoneinfo import ZoneInfo
import config
import jm_api
import weather_api
import usage_local
import btc_api
WEEK_CN = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
TZ = ZoneInfo(config.TZ_NAME)
def now_tz():
"""当前时刻（带 config.TZ_NAME 时区）。"""
return datetime.now(TZ)
def _local(ts):
"""epoch 秒 → 该时区的 datetime。"""
return datetime.fromtimestamp(ts, TZ)
TREND_DAYS = 30 # 近 N 天趋势
FIVE_HOUR, SEVEN_DAY = 5 * 3600, 7 * 86400 # 两个滚动窗口长度（秒），用于算「时间已流逝%」
@@ -85,7 +102,7 @@ def _trend(now):
def _weather(now):
"""和风天气（南京·秦淮），带 WEATHER_TTL 文件缓存。
"""和风天气（南京·江宁），带 WEATHER_TTL 文件缓存（本机脚本用）。
缓存未过期 → 直接复用，不发请求（每分钟刷新时省掉 14/15 次天气调用）。
过期 → 重新拉取并写缓存；全部接口失败时沿用上次缓存，仍无缓存才回退占位 "--"。
@@ -94,6 +111,18 @@ def _weather(now):
cached = _read_weather_cache()
if cached and ts - cached.get("ts", 0) < config.WEATHER_TTL:
return cached["data"]
out, ok = weather_fetch()
if ok:
_write_weather_cache(ts, out)
return out
if cached: # 全部失败：沿用上次缓存，避免天气区整片 "--"
print("[警告] 天气全部拉取失败，沿用上次缓存")
return cached["data"]
return out
def weather_fetch():
"""实时拉取天气（不读不写缓存）→ （weather dict，是否至少一个接口成功）。服务端每次请求直接调用。"""
out = {"loc": config.WEATHER_LOC_NAME, "cond": "--", "temp": "--", "range": "--", "icon": ""}
ok = False
try:
@@ -110,13 +139,7 @@ def _weather(now):
ok = True
except Exception as e:
print(f"[警告] 今日温区拉取失败：{e}")
if ok:
_write_weather_cache(ts, out)
return out
if cached: # 全部失败：沿用上次缓存，避免天气区整片 "--"
print("[警告] 天气全部拉取失败，沿用上次缓存")
return cached["data"]
return out
return out, ok
# ---------- 格式化 ----------
@@ -149,8 +172,11 @@ def _placeholder_col(k):
# ---------- 橘喵业务数据 ----------
def _mao(now):
def mao_fetch(now, with_trend=True):
"""橘喵今日经营（实时接口）；with_trend=False 时不拉近 30 天趋势（服务端与宽版都不用它，且趋势有文件缓存）。
返回（mao dict，实时接口是否成功）。"""
upd = now.strftime("%H:%M")
ok = True
# 今日经营（realtime）
try:
rt = jm_api.get_realtime()
@@ -163,26 +189,74 @@ def _mao(now):
except Exception as e:
print(f"[警告] 今日经营拉取失败：{e}")
cols = [_placeholder_col(k) for k in ("单量", "流水", "毛利")]
# 近30天趋势（按日期缓存，每天只拉一次）；宽版另用其最后 7 天做「近7天」数据表
trend = _trend(now)
return {"upd": upd, "cols": cols, "trend": trend, "week": _recent_days(trend, 7)}
ok = False
# 近30天趋势（按日期缓存，每天只拉一次）
trend = _trend(now) if with_trend else {"dates": [], "series": []}
return {"upd": upd, "cols": cols, "trend": trend}, ok
def _recent_days(trend, n):
"""趋势的最后 n 天 → 表格数据 {dates:[MM/DD…], rows:[(名, [文本…], 合计文本)]}，供宽版「橘喵·近7天」表使用。
recent-days 不含今日，故表头日期到昨天为止。单量按千分位、流水/毛利按紧凑金额格式；无数据返回空 dict。"""
dates, series = trend.get("dates") or [], trend.get("series") or []
if not dates or not series:
return {}
dates = dates[-n:]
rows = []
for s in series:
data = (s.get("data") or [])[-n:]
if len(data) != len(dates):
continue
fmt = _fmt_count if s["k"] == "单量" else _fmt_money
rows.append((s["k"], [fmt(v) for v in data], fmt(sum(data))))
return {"dates": dates, "rows": rows}
def _mao(now):
return mao_fetch(now)[0]
# ---------- 比特币 K 线 ----------
def _read_btc_cache():
try:
with open(config.BTC_CACHE_PATH, encoding="utf-8") as f:
return json.load(f) # {"ts": float, "data": {...}}
except Exception:
return None
def _write_btc_cache(ts, data):
try:
with open(config.BTC_CACHE_PATH, "w", encoding="utf-8") as f:
json.dump({"ts": ts, "data": data}, f, ensure_ascii=False)
except Exception as e:
print(f"[警告] BTC 缓存写入失败：{e}")
def _btc(now):
"""比特币 K 线（宽版底部面板；周期 BTC_BAR、根数 BTC_LIMIT），带 BTC_TTL 文件缓存；
拉取失败沿用旧缓存，仍无缓存回退空 candles。
结构契约（renderer 依赖）：{symbol, bar, src, candles:[{t, o, h, l, c}], last, chg, chg_label}，其中 t 为轴标签文本；
last 为最新一根（当前未收盘）的收盘价；chg 为涨跌幅（%）——时 K 取相对 24 根前（24h）收盘、不足则相对首根，
日 K 取相对上一根收盘，chg_label 说明口径（"24h"/"1d"）；涨=绿 / 跌=红 由 renderer 按 c≥o 判定。"""
ts = now.timestamp()
bar = config.BTC_BAR
cached = _read_btc_cache()
if cached and ts - cached.get("ts", 0) < config.BTC_TTL and cached.get("data", {}).get("bar") == bar:
return cached["data"]
try:
out = btc_fetch()
_write_btc_cache(ts, out)
return out
except Exception as e:
print(f"[警告] BTC K 线拉取失败：{e}")
if cached:
print("[警告] 沿用上次 BTC 缓存")
return cached["data"]
return btc_empty()
def btc_empty():
return {"symbol": "BTC/USDT", "bar": config.BTC_BAR, "src": "", "candles": [], "last": None, "chg": None, "chg_label": ""}
def btc_fetch():
"""实时拉取 BTC K 线并整理成渲染结构（不读不写缓存）；失败抛异常。服务端每次请求直接调用。"""
bar = config.BTC_BAR
src, rows = btc_api.get_candles(bar, config.BTC_LIMIT)
fmt = "%m/%d %H:%M" if bar == "1H" else "%m/%d"
candles = [{"t": _local(r["ts"]).strftime(fmt), "o": r["o"], "h": r["h"], "l": r["l"], "c": r["c"]} for r in rows]
last = rows[-1]["c"]
if bar == "1H":
base = rows[-25]["c"] if len(rows) >= 25 else rows[0]["c"]
chg_label = "24h" if len(rows) >= 25 else f"{len(rows) - 1}h"
else:
base, chg_label = rows[-2]["c"], "1d"
return {"symbol": "BTC/USDT", "bar": bar, "src": src, "candles": candles,
"last": last, "chg": (last - base) / base * 100 if base else 0.0, "chg_label": chg_label}
# ---------- Claude Code 用量 ----------
@@ -214,7 +288,7 @@ def _fmt_reset(resets_at, window):
ts = _parse_reset_ts(resets_at)
if ts is None:
return None
dt = datetime.fromtimestamp(ts)
dt = _local(ts)
return dt.strftime("%H:%M") if window <= 86400 else dt.strftime("%m/%d %H:%M")
@@ -227,26 +301,25 @@ USAGE_WINDOWS = (
("seven_day", "7d", "all 7d", SEVEN_DAY, False),
("seven_day_fable", "Fable 7d", "fable 7d", SEVEN_DAY, True),
)
# ChatGPT 用量窗口：键名沿用同一套约定（five_hour / seven_day / seven_day_codex），推送端照此结构提供原始数据即可。
GPT_WINDOWS = (
("five_hour", "5h", "all 5h", FIVE_HOUR, False),
("seven_day", "7d", "all 7d", SEVEN_DAY, False),
("seven_day_codex", "Codex 7d", "codex 7d", SEVEN_DAY, True),
)
def _usage(now):
"""Claude Code 用量（5h / 7d / Fable 7d）——只读本地 statusline 生产的快照（见 usage_local.py），**不发任何请求**。
utilization 来自 statusline 从 Claude Code stdin 落盘的权威实时值（Fable 行来自其转存的 API limits）；
time_pct(时间已流逝%) 与 pace 每次按当前时间**现算**（resets_at 为绝对时间）。
文件缺失/损坏 → 回退占位 pct=None（renderer 显示 "--"）；快照里没有某窗口（如旧版 statusline 无 Fable）同样 "--"。
def usage_from_raw(now, raw, windows, title):
"""原始快照 → 渲染结构（Claude / ChatGPT 通用）。
raw 形如 {five_hour:{utilization, resets_at}, seven_day:{...}, <scoped 键>:{...}|null, updated_at?}；
utilization 为已用百分比，resets_at 为 epoch 秒或 ISO；缺哪个窗口哪个显示 "--"。
time_pct（时间已流逝%）与 pace 按传入的 now **现算**——所以推送端只需推原始快照，服务端渲染时 pace 仍随时钟准确推进。
结构契约（renderer 依赖）：{title, warn, ok, pace_tol, updated?, bars:[{k, label, pct(0~100 或 None), time_pct?, reset?, scoped}]}；
pct ≥ warn 时该条进度条与百分比转红（宽版另有 pct ≤ ok 绿档）；pace = pct − time_pct（>0 超前↑ / <0 节余↓，宽版按 ±pace_tol 分三色）；
reset 为重置时刻文本；updated 为快照落盘时刻 HH:MM（活跃使用时持续刷新，空闲时停在最后一次）。
"""
raw = {}
try:
raw = usage_local.get_usage()
except Exception as e:
print(f"[警告] Claude 用量读取失败：{e}")
reset 为重置时刻文本；updated 为快照落盘时刻 HH:MM（活跃使用时持续刷新，空闲时停在最后一次）。"""
raw = raw or {}
bars = []
for key, k, label, win, scoped in USAGE_WINDOWS:
for key, k, label, win, scoped in windows:
seg = raw.get(key) or {}
util = seg.get("utilization")
pct = int(round(float(util))) if util is not None else None
@@ -258,32 +331,47 @@ def _usage(now):
if rs:
bar["reset"] = rs
bars.append(bar)
out = {"title": "Claude Usage", "warn": config.USAGE_WARN, "ok": config.USAGE_OK, "pace_tol": config.PACE_TOL, "bars": bars}
out = {"title": title, "warn": config.USAGE_WARN, "ok": config.USAGE_OK, "pace_tol": config.PACE_TOL, "bars": bars}
upd = _parse_reset_ts(raw.get("updated_at"))
if upd:
out["updated"] = datetime.fromtimestamp(upd).strftime("%H:%M")
out["updated"] = _local(upd).strftime("%H:%M")
return out
def _usage_gpt(now):
"""ChatGPT 用量——**目前为 mock 占位**（mock=True，renderer 在标题右侧标「示例数据」）。
结构与 _usage 完全一致，后续接入真实数据源时只需替换本函数、保持返回结构不变。
mock 取值刻意覆盖宽版三档配色：5h 88%（≥warn 红）、7d 71%（黑）、Codex 7d 22%（≤ok 绿，scoped，与左侧 fable 行对称）。"""
bars = []
for k, label, pct, win, remain, scoped in (("5h", "all 5h", 88, FIVE_HOUR, 2 * 3600, False),
("7d", "all 7d", 71, SEVEN_DAY, 3 * 86400, False),
("Codex 7d", "codex 7d", 22, SEVEN_DAY, 3 * 86400, True)):
reset_ts = now.timestamp() + remain
bars.append({"k": k, "label": label, "pct": pct, "scoped": scoped,
"time_pct": _window_time_pct(reset_ts, win, now),
"reset": _fmt_reset(reset_ts, win)})
return {"title": "ChatGPT Usage", "warn": config.USAGE_WARN, "ok": config.USAGE_OK, "pace_tol": config.PACE_TOL,
"bars": bars, "mock": True}
def _usage(now, raw=None):
"""Claude Code 用量（5h / 7d / Fable 7d）。raw 为 None 时读本地 statusline 快照（见 usage_local.py，**不发请求**）；
服务端传入推送来的原始快照。utilization 来自 statusline 从 Claude Code stdin 落盘的权威实时值（Fable 行来自其转存的 API limits）。"""
if raw is None:
raw = {}
try:
raw = usage_local.get_usage()
except Exception as e:
print(f"[警告] Claude 用量读取失败：{e}")
return usage_from_raw(now, raw, USAGE_WINDOWS, "Claude Usage")
def gpt_mock_raw(now):
"""ChatGPT 原始快照的 mock（尚无真实数据源时用）：刻意覆盖宽版三档配色——
5h 88%（≥warn 红）、7d 71%（黑）、Codex 7d 22%（≤ok 绿，scoped，与左侧 fable 行对称）。"""
t = now.timestamp()
return {"five_hour": {"utilization": 88, "resets_at": t + 2 * 3600},
"seven_day": {"utilization": 71, "resets_at": t + 3 * 86400},
"seven_day_codex": {"utilization": 22, "resets_at": t + 3 * 86400}}
def _usage_gpt(now, raw=None):
"""ChatGPT 用量。raw 为推送来的原始快照（结构同 gpt_mock_raw）；为 None 时用 mock 并标 mock=True
（renderer 在标题右侧标「示例数据」）。接入真实数据源时由推送端提供 raw，本函数无需改动。"""
mock = raw is None
out = usage_from_raw(now, gpt_mock_raw(now) if mock else raw, GPT_WINDOWS, "ChatGPT Usage")
if mock:
out["mock"] = True
return out
def get_dashboard_data(now=None):
"""返回渲染所需的完整数据字典。"""
now = now or datetime.now()
"""返回渲染所需的完整数据字典（本机脚本形态：带文件缓存，用量读本地快照）。"""
now = now or now_tz()
return {
"date": {
"greg": now.strftime("%Y/%m/%d"),
@@ -294,4 +382,5 @@ def get_dashboard_data(now=None):
"mao": _mao(now),
"usage": _usage(now),
"usage_gpt": _usage_gpt(now),
"btc": _btc(now),
}