YANG JIANKUAN
04fa31cb94
- btc_api：新增 gateio／huobi_aws／binance_vision 三源，config 增 BTC_SOURCES 控制尝试顺序（境内 okx／coinbase／huobi 主域名不通） - server：IMAGE_TOKEN 非空时三个渲染端点须带 ?token=…，否则与未知路径同样 404（防爬虫扫描）；推送令牌改常量时间比较 - 文档：CLAUDE.md／README／.env.example 记录 NGCQ 部署（systemd eink-server、1Panel API 建站坑）、本机 server／cloudflared 已卸载 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
343 lines
17 KiB
Python
343 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
图片/网页服务 —— 给 SenseCraft HMI 的 HTML 控件周期抓取的**恒定 URL**，每次请求都**现取数据、现渲染**，全程内存、不落盘。
|
||
|
||
真机验证结论（2026-09）：SenseCraft 的 Image 控件只在填写 URL 时抓一次快照、此后不再更新；
|
||
HTML 控件填同一个 URL 则**每个刷新周期都重抓**。所以对外只暴露 HTML 控件用的 URL，两种形态并行供对比：
|
||
方案一 GET <IMAGE_PATH>（默认 /e1002.png）→ image/png：与 main.py --target e1002 同一套 Pillow 点阵渲染（renderer_e1002）
|
||
方案二 GET <HTML_PATH> （默认 /e1002.html）→ text/html：800x480 的现代网页（html_renderer，系统/网络字体 + 内联 SVG K 线），
|
||
由平台无头浏览器截图后量化（真机实拍：文字抗锯齿灰边被平台抖动成毛边、小字糊）
|
||
方案三 GET <HTML_PNG_PATH>（默认 /e1002-web.png）→ image/png：同一网页由**服务端**无头 Chrome 截图，再按最近色（不抖动）
|
||
固化为六色 PNG（html_shot），平台拿到的已是纯色图；加 ?raw=1 返回未量化的原始截图供对照。需服务器装 Chrome。
|
||
两者共用同一份 build_data()：
|
||
- 日期时间：服务器按 config.TZ_NAME 时区现算
|
||
- 天气 / 橘喵今日实时 / BTC K 线：**内存缓存 + 有效期**（WEATHER_TTL / MAO_TTL / BTC_TTL，.env 可配，0=每次实拉）：
|
||
收到请求先查该路缓存是否过期，过期才实拉并更新时间戳，否则直接用缓存渲染；拉取失败沿用旧值、不更新时间戳（下次再试），
|
||
从未成功则占位 "--"。整个服务**不读本机任何文件**（部署在公网服务器）。
|
||
- Claude Usage / Codex Usage：**不主动拉取、不持凭证**，由本机 `main.py --target push` 采集后推送原始快照到 POST /push/usage、/push/usage_gpt，
|
||
服务端暂存最新一份（内存；PUSH_STATE_PATH 非空时同时落盘，重启恢复）；渲染时按当前时刻现算 pace，
|
||
- 三路上游过期时**并行**实拉，整体最多等 FETCH_BUDGET 秒（默认 5），超预算的一路本次用旧值、后台拉完更新缓存（见 build_data）；
|
||
所以推送时机与设备抓取时机不必对齐。未推送过：Claude 显示 "--"，ChatGPT 用 mock 并标「示例数据」。
|
||
- 推送鉴权：请求头 X-Push-Token 必须等于 PUSH_TOKEN；PUSH_TOKEN 为空时拒绝一切推送（服务在公网，务必配置）。
|
||
- 页面鉴权：IMAGE_TOKEN 非空时，三个渲染端点必须带 ?token=<IMAGE_TOKEN>，否则返回与未知路径一样的 404（防爬虫扫描）。
|
||
其他端点：GET /health → JSON 运行状态；GET /push/usage、/push/usage_gpt → 查看暂存的原始快照（需同样令牌）。
|
||
部署：python3 server.py（监听 SERVER_HOST:SERVER_PORT），前置 Caddy/Nginx 做 HTTPS，或 cloudflared tunnel。
|
||
"""
|
||
import hmac
|
||
import io
|
||
import json
|
||
import os
|
||
import threading
|
||
import time
|
||
import urllib.parse
|
||
from concurrent.futures import ThreadPoolExecutor, wait
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
||
import config
|
||
import data
|
||
from renderer_e1002 import E1002Renderer
|
||
import html_renderer
|
||
import html_shot
|
||
|
||
|
||
class State:
|
||
"""服务端内存状态：推送来的原始用量快照 + 上游数据的「上一次成功值」+ 计数。"""
|
||
|
||
def __init__(self):
|
||
self.lock = threading.Lock()
|
||
self.raw = {"usage": None, "usage_gpt": None}
|
||
self.pushed_at = {"usage": None, "usage_gpt": None}
|
||
self.last_good = {"weather": None, "mao": None, "btc": None, "device": None} # 各路最近一次成功的数据
|
||
self.fetched_at = {"weather": 0.0, "mao": 0.0, "btc": 0.0, "device": 0.0} # 各路最近一次成功拉取的时间戳（缓存有效期起点）
|
||
self.inflight = {} # 进行中的拉取 future（key → Future），防同一路重复实拉
|
||
self.renders = {"png": 0, "html": 0}
|
||
self.last_render = None
|
||
self.started = time.time()
|
||
self._load()
|
||
|
||
# ---- 推送数据持久化（可选）----
|
||
def _load(self):
|
||
p = config.PUSH_STATE_PATH
|
||
if not p or not os.path.exists(p):
|
||
return
|
||
try:
|
||
with open(p, encoding="utf-8") as f:
|
||
st = json.load(f)
|
||
self.raw.update(st.get("raw") or {})
|
||
self.pushed_at.update(st.get("pushed_at") or {})
|
||
print(f"[状态] 已从 {p} 恢复推送数据")
|
||
except Exception as e:
|
||
print(f"[警告] 推送状态恢复失败：{e}")
|
||
|
||
def _save(self):
|
||
p = config.PUSH_STATE_PATH
|
||
if not p:
|
||
return
|
||
try:
|
||
tmp = p + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump({"raw": self.raw, "pushed_at": self.pushed_at}, f, ensure_ascii=False)
|
||
os.replace(tmp, p)
|
||
except Exception as e:
|
||
print(f"[警告] 推送状态落盘失败：{e}")
|
||
|
||
def push(self, kind, raw):
|
||
with self.lock:
|
||
self.raw[kind] = raw
|
||
self.pushed_at[kind] = time.time()
|
||
self._save()
|
||
|
||
|
||
STATE = State()
|
||
_RENDER_LOCK = threading.Lock() # Pillow 渲染器持有 self.img/self.d，非线程安全
|
||
_PNG_RENDERER = E1002Renderer()
|
||
|
||
|
||
_POOL = ThreadPoolExecutor(max_workers=6, thread_name_prefix="fetch")
|
||
|
||
|
||
def _refresh(key, fetch):
|
||
"""在线程池里实拉一路上游：成功则更新 last_good/fetched_at；无论成败都清掉 inflight。返回 (data, ok, 耗时秒)。"""
|
||
t0 = time.time()
|
||
try:
|
||
d, ok = fetch()
|
||
except Exception as e: # noqa: BLE001 —— 上游任何异常都按失败处理
|
||
print(f"[警告] {key} 拉取异常：{e}")
|
||
d, ok = None, False
|
||
with STATE.lock:
|
||
if ok:
|
||
STATE.last_good[key], STATE.fetched_at[key] = d, time.time()
|
||
STATE.inflight.pop(key, None)
|
||
return d, ok, time.time() - t0
|
||
|
||
|
||
def _cached(key, ttl, fetch):
|
||
"""带有效期的内存缓存（提交阶段）：未过期 → (缓存, 状态串, None)；过期 → 提交到线程池实拉，返回 (None, None, future)。
|
||
同一路已有进行中的拉取（上一请求超预算后仍在跑）则复用其 future，不重复打上游。"""
|
||
now = time.time()
|
||
with STATE.lock:
|
||
cached, ts = STATE.last_good[key], STATE.fetched_at[key]
|
||
if cached is not None and ttl > 0 and now - ts < ttl:
|
||
return cached, f"{key}:cache({int(now - ts)}s)", None
|
||
fut = STATE.inflight.get(key)
|
||
if fut is None:
|
||
fut = _POOL.submit(_refresh, key, fetch)
|
||
STATE.inflight[key] = fut
|
||
return None, None, fut
|
||
|
||
|
||
def _settle(key, fut, fallback):
|
||
"""预算等待结束后结算一路：完成 → fetched(耗时)/stale/none；未完成 → 用上一次成功值（slow→stale）或占位（slow→none），
|
||
拉取线程继续跑完并写入缓存供下次请求使用（stale-while-revalidate）。"""
|
||
with STATE.lock:
|
||
cached = STATE.last_good[key]
|
||
if fut.done():
|
||
d, ok, dt = fut.result()
|
||
if ok:
|
||
return d, f"{key}:fetched({dt:.2f}s)"
|
||
return (cached, f"{key}:stale") if cached is not None else (fallback, f"{key}:none")
|
||
return (cached, f"{key}:slow→stale") if cached is not None else (fallback, f"{key}:slow→none")
|
||
|
||
|
||
def build_data(now=None):
|
||
"""组装渲染数据：三路上游按各自 TTL 走内存缓存；过期的**并行**实拉，整体最多等 FETCH_BUDGET 秒，
|
||
超预算的那路本次用旧值/占位、后台继续拉完更新缓存——响应时间上限 ≈ 预算 + 渲染，任何一路上游卡死都不再拖垮整图。
|
||
（2026-09-14 前为串行实拉、各上游 socket 超时 30s，单路卡住即整体超时。）用量取推送暂存。返回 (data, 各路状态列表)。"""
|
||
now = now or data.now_tz()
|
||
with STATE.lock:
|
||
raw_usage, raw_gpt = STATE.raw["usage"], STATE.raw["usage_gpt"]
|
||
|
||
fallbacks = {
|
||
"weather": {"loc": config.WEATHER_LOC_NAME, "cond": "--", "temp": "--", "range": "--", "icon": ""},
|
||
"mao": {"upd": now.strftime("%H:%M"), "cols": [data._placeholder_col(k) for k in ("单量", "流水", "毛利")],
|
||
"trend": {"dates": [], "series": []}},
|
||
"btc": data.btc_empty(),
|
||
"device": data.device_empty(),
|
||
}
|
||
fetchers = {
|
||
"weather": (config.WEATHER_TTL, data.weather_fetch),
|
||
"mao": (config.MAO_TTL, lambda: data.mao_fetch(now, with_trend=False)),
|
||
"btc": (config.BTC_TTL, lambda: (data.btc_fetch(), True)),
|
||
"device": (config.DEVICE_TTL, data.device_fetch),
|
||
}
|
||
result, status, pending = {}, {}, {}
|
||
for key, (ttl, fetch) in fetchers.items():
|
||
d, st, fut = _cached(key, ttl, fetch)
|
||
if fut is None:
|
||
result[key], status[key] = d, st
|
||
else:
|
||
pending[key] = fut
|
||
if pending:
|
||
wait(pending.values(), timeout=config.FETCH_BUDGET)
|
||
for key, fut in pending.items():
|
||
result[key], status[key] = _settle(key, fut, fallbacks[key])
|
||
|
||
d = {
|
||
"date": {"greg": now.strftime("%Y/%m/%d"), "week": data.WEEK_CN[now.weekday()], "time": now.strftime("%H:%M")},
|
||
"weather": result["weather"],
|
||
"mao": result["mao"],
|
||
"usage": data._usage(now, raw=raw_usage if raw_usage is not None else {}), # 未推送 → 全部 "--"（不读本机文件）
|
||
"usage_gpt": data._usage_gpt(now, raw=raw_gpt, allow_local=False), # 未推送 → mock（标「示例数据」），服务端不本地采集
|
||
"btc": result["btc"],
|
||
"device": result["device"],
|
||
}
|
||
return d, [status[k] for k in ("weather", "mao", "btc", "device")]
|
||
|
||
|
||
def render_png_bytes(d):
|
||
"""渲染 PNG 到内存字节。"""
|
||
with _RENDER_LOCK:
|
||
img = _PNG_RENDERER.render(d)
|
||
buf = io.BytesIO()
|
||
img.save(buf, "PNG", optimize=True)
|
||
return buf.getvalue()
|
||
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = "eink-push/1.0"
|
||
protocol_version = "HTTP/1.1"
|
||
|
||
def log_message(self, fmt, *args):
|
||
pass
|
||
|
||
def _log(self, code, note=""):
|
||
ua = self.headers.get("User-Agent", "-")
|
||
ip = self.headers.get("CF-Connecting-IP") or self.headers.get("X-Forwarded-For") or self.client_address[0]
|
||
print(f"[{time.strftime('%H:%M:%S')}] {code} {self.command} {self.path} ip={ip} ua={ua[:60]} {note}", flush=True)
|
||
|
||
def _send(self, code, ctype, body, extra=None):
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", ctype)
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.send_header("Cache-Control", "no-store, no-cache, must-revalidate, max-age=0")
|
||
self.send_header("Pragma", "no-cache")
|
||
self.send_header("Expires", "0")
|
||
for k, v in (extra or {}).items():
|
||
self.send_header(k, v)
|
||
self.end_headers()
|
||
if self.command != "HEAD":
|
||
self.wfile.write(body)
|
||
|
||
def _json(self, code, obj):
|
||
self._send(code, "application/json; charset=utf-8", json.dumps(obj, ensure_ascii=False).encode("utf-8"))
|
||
|
||
def _authed(self):
|
||
tok = config.PUSH_TOKEN
|
||
return bool(tok) and hmac.compare_digest(self.headers.get("X-Push-Token", ""), tok)
|
||
|
||
def _page_authed(self):
|
||
"""页面令牌：IMAGE_TOKEN 为空则放行；否则查询串里 token 必须相等（常量时间比较）。"""
|
||
tok = config.IMAGE_TOKEN
|
||
if not tok:
|
||
return True
|
||
qs = urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query)
|
||
return any(hmac.compare_digest(v, tok) for v in qs.get("token", []))
|
||
|
||
# ---------- GET ----------
|
||
def do_GET(self):
|
||
path = self.path.split("?", 1)[0]
|
||
try:
|
||
if path in (config.IMAGE_PATH, config.HTML_PATH, config.HTML_PNG_PATH) and not self._page_authed():
|
||
# 令牌不对与路径不存在返回完全一样的 404，不暴露端点存在
|
||
self._log(404, "bad page token")
|
||
return self._json(404, {"ok": False, "error": "not found"})
|
||
if path == config.IMAGE_PATH:
|
||
t0 = time.time()
|
||
d, st = build_data()
|
||
body = render_png_bytes(d)
|
||
with STATE.lock:
|
||
STATE.renders["png"] += 1
|
||
STATE.last_render = time.time()
|
||
self._log(200, f"png {len(body)}B {time.time() - t0:.2f}s [{' '.join(st)}]")
|
||
return self._send(200, "image/png", body)
|
||
if path == config.HTML_PATH:
|
||
t0 = time.time()
|
||
d, st = build_data()
|
||
body = html_renderer.render_html(d).encode("utf-8")
|
||
with STATE.lock:
|
||
STATE.renders["html"] += 1
|
||
STATE.last_render = time.time()
|
||
self._log(200, f"html {len(body)}B {time.time() - t0:.2f}s [{' '.join(st)}]")
|
||
return self._send(200, "text/html; charset=utf-8", body)
|
||
if path == config.HTML_PNG_PATH:
|
||
t0 = time.time()
|
||
raw = "raw=1" in (self.path.split("?", 1) + [""])[1]
|
||
d, st = build_data()
|
||
html = html_renderer.render_html(d)
|
||
img = html_shot.screenshot_html(html)
|
||
if not raw:
|
||
img = html_shot.quantize6(img)
|
||
buf = io.BytesIO(); img.save(buf, "PNG", optimize=True); body = buf.getvalue()
|
||
with STATE.lock:
|
||
STATE.renders["web_png"] = STATE.renders.get("web_png", 0) + 1
|
||
STATE.last_render = time.time()
|
||
self._log(200, f"web-png{'(raw)' if raw else ''} {len(body)}B {time.time() - t0:.2f}s [{' '.join(st)}]")
|
||
return self._send(200, "image/png", body)
|
||
if path == "/health":
|
||
with STATE.lock:
|
||
st = {"ok": True, "uptime_s": int(time.time() - STATE.started), "renders": dict(STATE.renders),
|
||
"last_render": STATE.last_render, "pushed_at": dict(STATE.pushed_at),
|
||
"has_last_good": {k: v is not None for k, v in STATE.last_good.items()},
|
||
"cache_age_s": {k: (int(time.time() - v) if v else None) for k, v in STATE.fetched_at.items()},
|
||
"ttl_s": {"weather": config.WEATHER_TTL, "mao": config.MAO_TTL, "btc": config.BTC_TTL, "device": config.DEVICE_TTL},
|
||
"fetch_budget_s": config.FETCH_BUDGET, "inflight": sorted(STATE.inflight.keys()),
|
||
"image_path": config.IMAGE_PATH, "html_path": config.HTML_PATH,
|
||
"html_png_path": config.HTML_PNG_PATH, "tz": config.TZ_NAME}
|
||
self._log(200)
|
||
return self._json(200, st)
|
||
if path in ("/push/usage", "/push/usage_gpt"):
|
||
if not self._authed():
|
||
self._log(401); return self._json(401, {"ok": False, "error": "bad token"})
|
||
kind = path.rsplit("/", 1)[1]
|
||
with STATE.lock:
|
||
raw, at = STATE.raw[kind], STATE.pushed_at[kind]
|
||
self._log(200)
|
||
return self._json(200, {"ok": True, "kind": kind, "pushed_at": at, "raw": raw})
|
||
if path == "/favicon.ico":
|
||
return self._send(204, "image/x-icon", b"")
|
||
self._log(404)
|
||
self._json(404, {"ok": False, "error": "not found"})
|
||
except Exception as e: # noqa: BLE001 —— 任何渲染异常都返回 500 而不是断开连接
|
||
self._log(500, repr(e))
|
||
self._json(500, {"ok": False, "error": repr(e)})
|
||
|
||
do_HEAD = do_GET
|
||
|
||
# ---------- POST：推送 ----------
|
||
def do_POST(self):
|
||
path = self.path.split("?", 1)[0]
|
||
if path not in ("/push/usage", "/push/usage_gpt"):
|
||
self._log(404); return self._json(404, {"ok": False, "error": "not found"})
|
||
if not self._authed():
|
||
self._log(401); return self._json(401, {"ok": False, "error": "bad token"})
|
||
try:
|
||
n = int(self.headers.get("Content-Length") or 0)
|
||
if n <= 0 or n > 64 * 1024:
|
||
raise ValueError(f"bad content-length {n}")
|
||
raw = json.loads(self.rfile.read(n).decode("utf-8"))
|
||
if not isinstance(raw, dict):
|
||
raise ValueError("body must be a JSON object")
|
||
except Exception as e: # noqa: BLE001
|
||
self._log(400, repr(e)); return self._json(400, {"ok": False, "error": f"bad json: {e}"})
|
||
kind = path.rsplit("/", 1)[1]
|
||
STATE.push(kind, raw)
|
||
self._log(200, f"push {kind} keys={sorted(raw)[:6]}")
|
||
self._json(200, {"ok": True, "kind": kind, "received_at": time.time()})
|
||
|
||
|
||
def main():
|
||
if not config.PUSH_TOKEN:
|
||
print("[警告] PUSH_TOKEN 未配置：将拒绝一切推送，用量区会一直显示 \"--\"")
|
||
srv = ThreadingHTTPServer((config.SERVER_HOST, config.SERVER_PORT), Handler)
|
||
print(f"服务已启动 http://{config.SERVER_HOST}:{config.SERVER_PORT} 方案一 {config.IMAGE_PATH} 方案二 {config.HTML_PATH} "
|
||
f"方案三 {config.HTML_PNG_PATH} /health", flush=True)
|
||
try:
|
||
srv.serve_forever()
|
||
except KeyboardInterrupt:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|