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>
143 lines
6.1 KiB
Python
143 lines
6.1 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
验证工具 —— SenseCraft HMI 的 Image/HTML 控件对**同一 URL** 是否每个刷新周期都重新抓取。
|
||
|
||
原理：本地起一个极简 HTTP 服务，固定路径 /probe.png 每 --every 秒重绘一张 800x480 探针图
|
||
（大号生成时刻 HH:MM:SS + 递增序号 + 六色校准块），响应头 Cache-Control: no-store；
|
||
每次有人请求就在终端打一行日志（时间、路径、User-Agent、来源 IP，含经隧道透传的 CF-Connecting-IP）。
|
||
|
||
判定：
|
||
- 终端日志每个设备刷新周期都出现一次 /probe.png 请求，且设备画面上的时刻/序号随之推进 → **每周期重抓，方案二成立**；
|
||
此后只要不断覆盖同一 URL 指向的文件即可，无需再经 HTML 页。
|
||
- 日志只在「部署」那一刻出现一次、之后再无请求 → 平台只在部署时抓一次并缓存，方案二不成立。
|
||
- 也可同时把 /index.html（HTML 控件方案）放到另一台画布对照。
|
||
|
||
用法：
|
||
python3 probe_server.py [--port 8787] [--every 60]
|
||
# 另开终端暴露成公网 HTTPS（云端抓不到局域网/明文 HTTP）。cloudflared 快速隧道无需账号：
|
||
brew install cloudflared
|
||
cloudflared tunnel --url http://localhost:8787
|
||
# 把打印出的 https://xxxx.trycloudflare.com/probe.png 填入 SenseCraft 画布的 Image 控件（铺满 800x480）→ 部署。
|
||
"""
|
||
import argparse
|
||
import io
|
||
import threading
|
||
import time
|
||
from datetime import datetime
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
||
from PIL import Image, ImageDraw, ImageFont
|
||
|
||
import config
|
||
|
||
BLACK, WHITE, RED = (0, 0, 0), (255, 255, 255), (255, 0, 0)
|
||
SIX = [BLACK, WHITE, (255, 255, 0), RED, (0, 255, 0), (0, 0, 255)]
|
||
|
||
_state = {"png": b"", "seq": 0, "stamp": ""}
|
||
_lock = threading.Lock()
|
||
|
||
|
||
def render_probe(seq, now):
|
||
"""800x480 探针图：序号（左上）/ 大号时刻（居中，F96）/ 日期（下）/ 底部六色块。全部纯色。"""
|
||
img = Image.new("RGB", (800, 480), WHITE)
|
||
d = ImageDraw.Draw(img)
|
||
d.fontmode = "1"
|
||
f24 = ImageFont.truetype(config.FONT_PATH, 24)
|
||
f96 = ImageFont.truetype(config.FONT_PATH, 96)
|
||
d.text((20, 16), f"probe #{seq:04d}", font=f24, fill=BLACK)
|
||
d.text((20, 48), "SenseCraft re-fetch test / 同一 URL 每周期是否重抓", font=f24, fill=BLACK)
|
||
t = now.strftime("%H:%M:%S")
|
||
w = d.textlength(t, font=f96)
|
||
for dx in range(3): # 伪粗体
|
||
d.text(((800 - w) / 2 + dx, 150), t, font=f96, fill=RED)
|
||
ds = now.strftime("%Y/%m/%d %A")
|
||
d.text(((800 - d.textlength(ds, font=f24)) / 2, 290), ds, font=f24, fill=BLACK)
|
||
d.text((20, 400), "画面上的时刻若随设备刷新周期推进 → 平台每周期重新抓取了同一 URL", font=f24, fill=BLACK)
|
||
n = len(SIX)
|
||
for i, c in enumerate(SIX):
|
||
d.rectangle([round(800 / n * i), 456, round(800 / n * (i + 1)) - 1, 479], fill=c)
|
||
buf = io.BytesIO()
|
||
img.save(buf, "PNG")
|
||
return buf.getvalue()
|
||
|
||
|
||
def regenerate(every):
|
||
while True:
|
||
now = datetime.now()
|
||
with _lock:
|
||
_state["seq"] += 1
|
||
_state["png"] = render_probe(_state["seq"], now)
|
||
_state["stamp"] = now.strftime("%H:%M:%S")
|
||
print(f"[{now:%H:%M:%S}] 重绘 probe.png seq={_state['seq']}", flush=True)
|
||
time.sleep(every)
|
||
|
||
|
||
_INDEX = """<!doctype html><html><head><meta charset="utf-8">
|
||
<meta http-equiv="Cache-Control" content="no-store"><title>probe</title>
|
||
<style>html,body{margin:0;width:800px;height:480px;overflow:hidden;background:#fff}
|
||
img{display:block;width:800px;height:480px;image-rendering:pixelated}</style></head>
|
||
<body><img src="/probe.png" width="800" height="480"></body></html>"""
|
||
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = "probe/1.0"
|
||
|
||
def log_message(self, fmt, *args): # 关闭默认日志，用自定义格式
|
||
pass
|
||
|
||
def _log(self, code):
|
||
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"[{datetime.now():%H:%M:%S}] {code} GET {self.path} ip={ip} ua={ua}", flush=True)
|
||
|
||
def _send(self, code, ctype, body):
|
||
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")
|
||
self.send_header("Access-Control-Allow-Origin", "*")
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def do_GET(self):
|
||
path = self.path.split("?", 1)[0]
|
||
if path == "/probe.png":
|
||
with _lock:
|
||
body = _state["png"]
|
||
self._log(200)
|
||
return self._send(200, "image/png", body)
|
||
if path in ("/", "/index.html"):
|
||
self._log(200)
|
||
return self._send(200, "text/html; charset=utf-8", _INDEX.encode("utf-8"))
|
||
if path == "/favicon.ico":
|
||
return self._send(204, "image/x-icon", b"")
|
||
self._log(404)
|
||
self._send(404, "text/plain; charset=utf-8", b"not found")
|
||
|
||
do_HEAD = do_GET
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="SenseCraft 同 URL 重抓验证服务")
|
||
ap.add_argument("--port", type=int, default=8787)
|
||
ap.add_argument("--every", type=int, default=60, help="探针图重绘间隔（秒）")
|
||
args = ap.parse_args()
|
||
threading.Thread(target=regenerate, args=(args.every,), daemon=True).start()
|
||
while not _state["png"]:
|
||
time.sleep(0.05)
|
||
srv = ThreadingHTTPServer(("0.0.0.0", args.port), Handler)
|
||
print(f"探针服务已启动：http://localhost:{args.port}/probe.png （HTML 对照页：/index.html）")
|
||
print("下一步：另开终端执行 cloudflared tunnel --url http://localhost:%d ，把得到的 https 地址 + /probe.png 填入 Image 控件。" % args.port)
|
||
print("观察本终端：每个设备刷新周期是否出现一条 GET /probe.png。Ctrl+C 退出。", flush=True)
|
||
try:
|
||
srv.serve_forever()
|
||
except KeyboardInterrupt:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|