77 lines
2.8 KiB
Python
77 lines
2.8 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 get_devices():
|
||
"""获取设备列表。"""
|
||
data = _request("GET", "/devices")
|
||
if data.get("code") != 0:
|
||
raise RuntimeError(f"获取设备失败: {data}")
|
||
return data.get("data", [])
|
||
|
||
|
||
def resolve_device_id():
|
||
"""返回目标设备 ID：优先用 config.DEVICE_ID，否则取列表第一个。"""
|
||
if config.DEVICE_ID:
|
||
return config.DEVICE_ID
|
||
devices = get_devices()
|
||
if not devices:
|
||
raise RuntimeError("账号下没有任何设备")
|
||
return devices[0]["deviceId"]
|
||
|
||
|
||
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
|