refactor: 推送改为局域网直连设备,推送前清空画廊旧图

原走 Zectrix 公网 API + 设备 MAC,现改为直连局域网墨水屏设备(喵喵固件)
的 HTTP 接口:POST /upload 推送 PNG 触发刷屏。设备画廊按追加保存,故推送
前先 GET /images 清空旧图,避免图片越推越多、且避免 slideshow 模式轮播
到过期图。

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-07-07 10:19:10 +08:00
parent 21426e13bc
commit 2121b1eff4
5 changed files with 39 additions and 66 deletions

View File

@@ -1,11 +1,12 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
推送层 —— 调用 Zectrix 极趣云平台 API仅用标准库 urllib无第三方依赖
文档:显示推送 / 设备管理。
推送层 —— 直连局域网墨水屏设备(喵喵固件,仅用标准库 urllib无第三方依赖
固件开源https://gitee.com/gxp666111/miaomiaoweb/ 设备自带管理页,可在浏览器打开
http://<EPD_HOST>/ 调试同一套接口)。
"""
import json
import uuid
import urllib.parse
import urllib.request
import urllib.error
@@ -13,53 +14,33 @@ import config
def _request(method, path, headers=None, body=None):
url = f"{config.API_BASE}{path}"
h = {"X-API-Key": config.API_KEY}
if headers:
h.update(headers)
req = urllib.request.Request(url, data=body, headers=h, method=method)
url = f"http://{config.EPD_HOST}{path}"
req = urllib.request.Request(url, data=body, headers=headers or {}, method=method)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
return json.loads(resp.read().decode("utf-8"))
return resp.read().decode("utf-8", "ignore")
except urllib.error.HTTPError as e:
raise RuntimeError(f"HTTP {e.code}: {e.read().decode('utf-8', 'ignore')}") from e
def resolve_device_id():
"""返回目标设备 ID直接取 config.DEVICE_ID硬件 MAC未配置则报错。"""
if not config.DEVICE_ID:
raise RuntimeError("未配置设备 ID请在 .env 设置 ZECTRIX_DEVICE_ID=<硬件 MAC>")
return config.DEVICE_ID
def _list_gallery_names():
text = _request("GET", "/images")
return [item["name"] for item in json.loads(text).get("items", [])]
def _multipart(fields, files):
"""构造 multipart/form-data 请求体。fields: dict[str,str]; files: list[(name, filename, bytes, mime)]"""
boundary = uuid.uuid4().hex
crlf = b"\r\n"
buf = bytearray()
for k, v in fields.items():
buf += b"--" + boundary.encode() + crlf
buf += f'Content-Disposition: form-data; name="{k}"'.encode() + crlf + crlf
buf += str(v).encode() + crlf
for name, filename, content, mime in files:
buf += b"--" + boundary.encode() + crlf
buf += f'Content-Disposition: form-data; name="{name}"; filename="{filename}"'.encode() + crlf
buf += f"Content-Type: {mime}".encode() + crlf + crlf
buf += content + crlf
buf += b"--" + boundary.encode() + b"--" + crlf
return bytes(buf), f"multipart/form-data; boundary={boundary}"
def _clear_gallery():
"""设备画廊按追加保存,用完即删,使其始终只留最新一张。单张删除失败不影响后续上传。"""
for name in _list_gallery_names():
try:
_request("POST", f"/delete_image?name={urllib.parse.quote(name)}")
except Exception as e:
print(f"[推送] 清理画廊旧图 {name} 失败:{e}")
def push_image(image_path, device_id=None, page_id=config.PAGE_ID, dither=config.DITHER):
"""推送图片到设备显示"""
device_id = device_id or resolve_device_id()
def push_image(image_path):
"""推送图片到局域网墨水屏设备:先清空画廊旧图,再上传新图触发显示刷新"""
_clear_gallery()
with open(image_path, "rb") as fp:
content = fp.read()
fields = {"dither": "true" if dither else "false", "pageId": str(page_id)}
files = [("images", "dashboard.png", content, "image/png")]
body, content_type = _multipart(fields, files)
data = _request("POST", f"/devices/{device_id}/display/image",
headers={"Content-Type": content_type}, body=body)
if data.get("code") != 0:
raise RuntimeError(f"推送失败: {data}")
return device_id, data
resp = _request("POST", "/upload", headers={"Content-Type": "image/png"}, body=content)
return config.EPD_HOST, resp