YANG JIANKUAN
3e0e6c73a1
- config.py 顶部新增零依赖 _load_dotenv()，凭证/参数统一从同目录 .env 读取， 真实环境变量（cron/命令行注入）优先；移除所有明文 fallback - 新增 .env.example 模板，.env 加入 .gitignore - pusher.py 删除 get_devices()，resolve_device_id() 直接取 ZECTRIX_DEVICE_ID， 不再调 GET /devices 列表 - jm_api.py 改为设备密钥授权（client_secret，免验证码），token 进程内缓存 + 401 重试 - 同步 CLAUDE.md / main.py 文档 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
推送层 —— 调用 Zectrix 极趣云平台 API（仅用标准库 urllib，无第三方依赖）。
|
||
文档：显示推送 / 设备管理。
|
||
"""
|
||
import json
|
||
import uuid
|
||
import urllib.request
|
||
import urllib.error
|
||
|
||
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)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
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 _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 push_image(image_path, device_id=None, page_id=config.PAGE_ID, dither=config.DITHER):
|
||
"""推送图片到设备显示。"""
|
||
device_id = device_id or resolve_device_id()
|
||
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
|