#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ hjyiot.py - 贺简云物联 IoT 设备接入 SDK（Python 3.8+，仅标准库） 对应协议文档：docs/设备接入与公开应用.md（HTTP 设备接入 + 云盘 + report.fs + OTA） 基本用法： from hjyiot import HjyIotDevice dev = HjyIotDevice("https://www.hjyiot.cn/api.php", "", "") print(dev.heartbeat()) # 心跳 print(dev.report(props={"temperature": 27.5})) # 属性上报 print(dev.pull_commands()) # 拉取指令（取走即 sent） print(dev.disk_stats()) # 云盘空间概览 约定： * 设备鉴权与后台登录态完全独立，使用 HTTP Basic(device_id:device_token)； * 服务端统一返回 {"code":0,"msg":"ok","data":{...}}； * 本 SDK 所有方法返回响应中的 data 字段（dict）； 当 code != 0 时抛出 HjyIotError（可调用 raw_call 获取完整响应排查）。 * 云盘写类指令（report.fs）要求租户已在 Web 端签署《云盘使用协议》， 未开通时服务端会拒绝并返回提示，属于业务错误而非网络错误。 安全提醒：真实 device_token 请勿提交到公开仓库或写入配置文件随包分发。 """ from __future__ import annotations import base64 import json import time import urllib.error import urllib.parse import urllib.request from typing import Any, Dict, List, Optional class HjyIotError(Exception): """平台返回 code != 0，或网络/协议异常。""" def __init__(self, code: int, msg: str, raw: Optional[dict] = None): super().__init__(f"[{code}] {msg}") self.code = code self.msg = msg self.raw = raw class HjyIotDevice: """贺简云物联 IoT 设备接入客户端。 :param base_url: 平台 API 入口，如 "https://www.hjyiot.cn/api.php" :param device_id: 设备 ID（后台设备管理生成） :param device_token: 设备令牌（后台设备管理生成/重置） :param timeout: 网络超时（秒） """ def __init__(self, base_url: str, device_id: str, device_token: str, timeout: float = 10.0): self.base_url = base_url.rstrip("/") self.device_id = device_id self.device_token = device_token self.timeout = timeout # ------------------------------------------------------------------ 底层 def raw_call(self, action: str, method: str = "GET", body: Optional[dict] = None) -> Dict[str, Any]: """发起一次设备 API 请求，返回完整响应字典。 若网络层失败（DNS/超时/连接拒绝等）抛出 HjyIotError(code=-1)。 HTTP 4xx/5xx 也尽量读取平台 JSON 体，按平台业务错误返回。 """ sep = "&" if "?" in self.base_url else "?" url = f"{self.base_url}{sep}action={urllib.parse.quote(action)}" request = urllib.request.Request(url, method=method) token_raw = f"{self.device_id}:{self.device_token}".encode("utf-8") request.add_header("Authorization", "Basic " + base64.b64encode(token_raw).decode("ascii")) if body is not None: request.add_header("Content-Type", "application/json") payload = json.dumps(body, ensure_ascii=False).encode("utf-8") else: payload = None try: with urllib.request.urlopen(request, data=payload, timeout=self.timeout) as response: raw_text = response.read().decode("utf-8", errors="replace") except urllib.error.HTTPError as exc: raw_text = exc.read().decode("utf-8", errors="replace") except urllib.error.URLError as exc: raise HjyIotError(-1, f"网络错误: {exc.reason}") from exc if not raw_text.strip(): raise HjyIotError(-1, "空响应") try: parsed = json.loads(raw_text) except ValueError as exc: raise HjyIotError(-1, f"响应不是合法 JSON: {raw_text[:200]}") from exc if not isinstance(parsed, dict): raise HjyIotError(-1, f"响应结构异常: {raw_text[:200]}") return parsed def _data(self, resp: Dict[str, Any]) -> Any: """校验业务错误码并返回 data 字段。""" code = resp.get("code") if code != 0: raise HjyIotError(int(code or -1), str(resp.get("msg", "未知错误")), raw=resp) return resp.get("data") def call(self, action: str, method: str = "GET", body: Optional[dict] = None) -> Dict[str, Any]: return self._data(self.raw_call(action, method=method, body=body)) # ------------------------------------------------------------ 设备基础 def heartbeat(self) -> Dict[str, Any]: """心跳/上线（由离线变在线时触发一次 device_status 规则）。""" return self.call("heartbeat", method="POST", body={}) def report(self, props: Optional[Dict[str, Any]] = None, fs: Optional[List[Dict[str, Any]]] = None, ts: Optional[int] = None) -> Dict[str, Any]: """属性上报（可选携带云盘 report.fs 指令）。 :param props: 属性字典，值必须为标量或 None（嵌套对象/数组会被平台丢弃） :param fs: report.fs 指令数组，每条约 {"op": "write"|"delete"|"delete_content", ...} :param ts: Unix 秒级时间戳，缺省用本机时间 """ body: Dict[str, Any] = {} if props is not None: if not isinstance(props, dict): raise ValueError("props 必须是 dict") body["props"] = props if fs is not None: body["fs"] = fs body["ts"] = int(ts if ts is not None else time.time()) return self.call("report", method="POST", body=body) def pull_commands(self) -> Dict[str, Any]: """拉取待执行指令。 注意：拉取成功即标记 sent（取走即 sent），且不再进入后续拉取结果； 设备必须执行后再通过 report() 回写实际结果。 """ return self.call("command", method="GET") # ------------------------------------------------------------ OTA 固件 def ota_download(self, job_id: str, dest_path: str, timeout: float = 120.0) -> Dict[str, Any]: """下载本设备待升级的 OTA 固件到本地文件，并按响应头 X-OTA-SHA256 校验。 :param job_id: 来自指令 type=ota 的 payload.job_id :param dest_path: 本地保存路径（绝对/相对均可） :param timeout: 下载超时（秒），固件较大时应适当放宽 :return: {"version", "size", "sha256", "sha256_match", "path"} 说明： * 成功时平台返回二进制流（Content-Type: application/octet-stream）， 响应头含 X-OTA-Version（URL 编码，本文已自动 decode）、X-OTA-SHA256、 X-OTA-Size、X-OTA-Job； * 失败时平台返回 JSON 错误包（code=1002 功能停用/云盘封禁/已冻结、 1003 任务不存在、1004 缺少 job_id/未就绪/窗口过期、1006 固件缺失）， 本方法会抛出 HjyIotError； * 下载窗口为推送后 24 小时，过期需在平台重新推送。 """ if not job_id: raise ValueError("ota_download 需要 job_id") sep = "&" if "?" in self.base_url else "?" url = (f"{self.base_url}{sep}action=ota_download" f"&job_id={urllib.parse.quote(job_id)}") request = urllib.request.Request(url, method="GET") token_raw = f"{self.device_id}:{self.device_token}".encode("utf-8") request.add_header("Authorization", "Basic " + base64.b64encode(token_raw).decode("ascii")) try: response = urllib.request.urlopen(request, timeout=timeout) except urllib.error.HTTPError as exc: raw_text = exc.read().decode("utf-8", errors="replace") self._raise_ota_error(raw_text) raise # 极端情况：非 JSON 错误体，交由上层感知 except urllib.error.URLError as exc: raise HjyIotError(-1, f"网络错误: {exc.reason}") from exc with response: ctype = (response.headers.get("Content-Type") or "").lower() if "json" in ctype: raw_text = response.read().decode("utf-8", errors="replace") self._raise_ota_error(raw_text) sha256_expect = (response.headers.get("X-OTA-SHA256") or "").lower() version = urllib.parse.unquote(response.headers.get("X-OTA-Version") or "") size_hdr = response.headers.get("X-OTA-Size") or "" import hashlib # 局部导入，避免模块级多余依赖 digest = hashlib.sha256() written = 0 with open(dest_path, "wb") as fh: while True: chunk = response.read(65536) if not chunk: break fh.write(chunk) digest.update(chunk) written += len(chunk) local_hex = digest.hexdigest() return { "version": version, "size": written, "sha256": local_hex, "sha256_match": bool(sha256_expect) and local_hex == sha256_expect, "path": dest_path, "size_header": int(size_hdr) if size_hdr.isdigit() else None, } @staticmethod def _raise_ota_error(raw_text: str) -> None: """解析 OTA 下载的 JSON 错误包并抛出 HjyIotError。""" try: parsed = json.loads(raw_text) except ValueError: raise HjyIotError(-1, f"OTA 响应异常: {raw_text[:200]}") if isinstance(parsed, dict) and parsed.get("code") not in (0, None): raise HjyIotError(int(parsed.get("code")), str(parsed.get("msg", "OTA 下载失败")), raw=parsed) raise HjyIotError(-1, f"OTA 下载失败: {raw_text[:200]}", raw=parsed) # ------------------------------------------------------------ 云盘只读 def disk_stats(self) -> Dict[str, Any]: """云盘空间概览（只读，无需签署协议）。""" return self.call("disk_stats", method="GET") def disk_list(self, path: str = "") -> Dict[str, Any]: """浏览云盘目录（只读）；path 为空表示根目录。""" return self.call("disk_list", method="POST", body={"path": path or ""}) def disk_read(self, path: str) -> Dict[str, Any]: """读取云盘文件（只读；仅 txt/jpg，txt 返回 content，jpg 返回 base64 data）。""" if not path: raise ValueError("disk_read 需要 path") return self.call("disk_read", method="POST", body={"path": path}) # ------------------------------------------------- report.fs 写类指令 # 需租户已签署《云盘使用协议》，否则平台逐条拒绝并提示先开通。 def fs_write(self, path: str, content: str, mode: str = "append", create: bool = False) -> Dict[str, Any]: """写入/追加 txt 文本（单条 content 上限 128KB，单文件上限 8MB）。""" return self._fs_run({ "op": "write", "path": path, "content": content, "mode": mode, "create": bool(create), }) def fs_delete(self, path: str) -> Dict[str, Any]: """删除云盘文件（仅文件；目录删除请由平台人工处理）。""" return self._fs_run({"op": "delete", "path": path}) def fs_delete_content(self, path: str, needle: str, all_: bool = False) -> Dict[str, Any]: """删除 txt 文件中的内容片段（needle 子串匹配；all_=True 删除全部匹配）。""" return self._fs_run({ "op": "delete_content", "path": path, "needle": needle, "all": bool(all_), }) def _fs_run(self, op: Dict[str, Any]) -> Dict[str, Any]: resp = self.report(props={}, fs=[op], ts=int(time.time())) results = resp.get("fs") if isinstance(resp, dict) else None if not results: raise HjyIotError(-1, "服务端未返回 fs 逐条结果", raw=resp) item = results[0] if not item.get("ok"): raise HjyIotError(int(item.get("code") or -1), str(item.get("error") or "云盘指令执行失败"), raw=item) return item