#!/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， 所以推送时机与设备抓取时机不必对齐。未推送过：Claude 显示 "--"，ChatGPT 用 mock 并标「示例数据」。 - 推送鉴权：请求头 X-Push-Token 必须等于 PUSH_TOKEN；PUSH_TOKEN 为空时拒绝一切推送（服务在公网，务必配置）。 其他端点：GET /health → JSON 运行状态；GET /push/usage、/push/usage_gpt → 查看暂存的原始快照（需同样令牌）。 部署：python3 server.py（监听 SERVER_HOST:SERVER_PORT），前置 Caddy/Nginx 做 HTTPS，或 cloudflared tunnel。 """ import io import json import os import threading import time 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} # 各路最近一次成功的数据 self.fetched_at = {"weather": 0.0, "mao": 0.0, "btc": 0.0} # 各路最近一次成功拉取的时间戳（缓存有效期起点） 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() def _cached(key, ttl, fetch, fallback): """带有效期的内存缓存：未过期 → 直接返回缓存；过期 → 实拉，成功则更新数据与时间戳，失败沿用旧值（无旧值用 fallback）。 fetch() 返回 (data, ok)。返回 (data, 状态字串) 供日志。""" 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)" try: d, ok = fetch() except Exception as e: # noqa: BLE001 —— 上游任何异常都按失败处理 print(f"[警告] {key} 拉取异常：{e}") d, ok = None, False if ok: with STATE.lock: STATE.last_good[key], STATE.fetched_at[key] = d, now return d, f"{key}:fetched" if cached is not None: return cached, f"{key}:stale" return fallback, f"{key}:none" def build_data(now=None): """组装渲染数据：三路上游按各自 TTL 走内存缓存，用量取推送暂存。返回 (data, 各路状态列表)。""" now = now or data.now_tz() with STATE.lock: raw_usage, raw_gpt = STATE.raw["usage"], STATE.raw["usage_gpt"] weather, s1 = _cached("weather", config.WEATHER_TTL, data.weather_fetch, {"loc": config.WEATHER_LOC_NAME, "cond": "--", "temp": "--", "range": "--", "icon": ""}) mao, s2 = _cached("mao", config.MAO_TTL, lambda: data.mao_fetch(now, with_trend=False), {"upd": now.strftime("%H:%M"), "cols": [data._placeholder_col(k) for k in ("单量", "流水", "毛利")], "trend": {"dates": [], "series": []}}) btc, s3 = _cached("btc", config.BTC_TTL, lambda: (data.btc_fetch(), True), data.btc_empty()) d = { "date": {"greg": now.strftime("%Y/%m/%d"), "week": data.WEEK_CN[now.weekday()], "time": now.strftime("%H:%M")}, "weather": weather, "mao": 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": btc, } return d, [s1, s2, s3] 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 self.headers.get("X-Push-Token", "") == tok # ---------- GET ---------- def do_GET(self): path = self.path.split("?", 1)[0] try: 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}, "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()