#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 图片/网页服务 —— 给 SenseCraft HMI 的 HTML 控件周期抓取的**恒定 URL**，每次请求都**现取数据、现渲染**，全程内存、不落盘。 真机验证结论（2026-09）：SenseCraft 的 Image 控件只在填写 URL 时抓一次快照、此后不再更新； HTML 控件填同一个 URL 则**每个刷新周期都重抓**。所以对外只暴露 HTML 控件用的 URL，两种形态并行供对比： 方案一 GET （默认 /e1002.png）→ image/png：与 main.py --target e1002 同一套 Pillow 点阵渲染（renderer_e1002） 方案二 GET （默认 /e1002.html）→ text/html：800x480 的现代网页（html_renderer，系统/网络字体 + 内联 SVG K 线）， 由平台无头浏览器截图后量化（真机实拍：文字抗锯齿灰边被平台抖动成毛边、小字糊） 方案三 GET （默认 /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=，否则返回与未知路径一样的 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()