# -*- coding: utf-8 -*- """本地预览：UTF-8 静态资源 + 配置落盘 API。用法: python serve.py""" from __future__ import annotations import http.server import json import os import queue import sys from pathlib import Path from playback_hub import HUB ROOT = Path(__file__).resolve().parent AGV_DIR = (ROOT / "agv").resolve() PORT = int(os.environ.get("PORT", "8765")) UTF8_TYPES = { ".html": "text/html; charset=utf-8", ".js": "text/javascript; charset=utf-8", ".json": "application/json; charset=utf-8", ".css": "text/css; charset=utf-8", ".svg": "image/svg+xml; charset=utf-8", } class Handler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(ROOT), **kwargs) def end_headers(self): ext = Path(self.path.split("?", 1)[0]).suffix.lower() if ext in UTF8_TYPES: self.send_header("Content-Type", UTF8_TYPES[ext]) if ext in {".html", ".js", ".json", ".css"}: self.send_header("Cache-Control", "no-store, must-revalidate") super().end_headers() def do_OPTIONS(self): path = self.path.split("?", 1)[0] if path.startswith("/api/"): self.send_response(204) self._cors_headers() self.end_headers() return self.send_error(404, "Not Found") def do_GET(self): path = self.path.split("?", 1)[0] if path == "/api/catalog": self._json_response(200, self._build_catalog_response()) return if path == "/api/agv/state": state = HUB.get_state() self._json_response( 200, { "version": 1, "ok": True, "data": {**state, "subscribers": HUB.subscriber_count()}, "error": None, }, ) return if path == "/api/agv/events": self._agv_events_sse() return super().do_GET() def do_POST(self): path = self.path.split("?", 1)[0] if path == "/api/save-config": self._save_config() return if path == "/api/agv": self._agv_api() return if path == "/api/agv/report": self._agv_report() return self.send_error(404, "Not Found") def _save_config(self): try: length = int(self.headers.get("Content-Length", 0)) raw = self.rfile.read(length).decode("utf-8") payload = json.loads(raw) filename = payload["file"] config = payload["config"] except (KeyError, json.JSONDecodeError, UnicodeDecodeError) as e: self._json_response(400, {"ok": False, "error": f"请求体无效: {e}"}) return if not isinstance(filename, str) or not filename.endswith(".json"): self._json_response(400, {"ok": False, "error": "仅允许保存 agv/*.json"}) return if "/" in filename or "\\" in filename or ".." in filename: self._json_response(400, {"ok": False, "error": "非法文件名"}) return target = (AGV_DIR / filename).resolve() if not str(target).startswith(str(AGV_DIR)): self._json_response(403, {"ok": False, "error": "路径越界"}) return try: AGV_DIR.mkdir(parents=True, exist_ok=True) text = json.dumps(config, ensure_ascii=False, indent=2) + "\n" target.write_text(text, encoding="utf-8") except OSError as e: self._json_response(500, {"ok": False, "error": str(e)}) return self._json_response(200, {"ok": True, "file": filename, "path": str(target.relative_to(ROOT))}) def _read_manifest_models(self) -> list[dict]: manifest_path = AGV_DIR / "manifest.json" if not manifest_path.exists(): return [] try: data = json.loads(manifest_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return [] models = data.get("models") return models if isinstance(models, list) else [] def _map_actions(self, cfg: dict) -> list[dict]: actions = cfg.get("actions") or {} if not isinstance(actions, dict): return [] out = [] for action_id, act in actions.items(): if not isinstance(act, dict): continue out.append( { "id": action_id, "label": act.get("label") or action_id, "uiClass": act.get("uiClass") or "mode-idle", } ) return out def _build_catalog_response(self) -> dict: vehicles = [] for m in self._read_manifest_models(): if not isinstance(m, dict): continue model_id = m.get("id") or "" cfg_file = m.get("config") or (f"{model_id}.json" if model_id else "") item = { "modelId": model_id, "name": m.get("name") or model_id, "file": m.get("file"), "config": cfg_file, "root": m.get("root"), "actions": [], "configError": None, } if not cfg_file: item["configError"] = "缺少 config 字段" vehicles.append(item) continue cfg_path = (AGV_DIR / cfg_file).resolve() if not str(cfg_path).startswith(str(AGV_DIR)) or not cfg_path.exists(): item["configError"] = f"缺少配置 {cfg_file}" vehicles.append(item) continue try: cfg = json.loads(cfg_path.read_text(encoding="utf-8")) item["actions"] = self._map_actions(cfg) item["root"] = m.get("root") or (cfg.get("model") or {}).get("root") or item["root"] except (OSError, json.JSONDecodeError) as e: item["configError"] = str(e) vehicles.append(item) return {"version": 1, "ok": True, "data": {"vehicles": vehicles}, "error": None} def _agv_api(self): try: length = int(self.headers.get("Content-Length", 0)) raw = self.rfile.read(length).decode("utf-8") payload = json.loads(raw) except (json.JSONDecodeError, UnicodeDecodeError) as e: self._json_response(400, {"version": 1, "ok": False, "data": None, "error": {"code": "INVALID_REQUEST", "message": str(e)}}) return req_id = payload.get("id") version = payload.get("version") method = (payload.get("method") or "").strip() if version != 1: self._json_response(400, {"version": 1, "id": req_id, "ok": False, "data": None, "error": {"code": "INVALID_REQUEST", "message": "version 须为 1"}}) return if not method: self._json_response(400, {"version": 1, "id": req_id, "ok": False, "data": None, "error": {"code": "INVALID_REQUEST", "message": "缺少 method"}}) return if method not in {"playAction", "stopAction", "clearActions"}: self._json_response( 400, { "version": 1, "id": req_id, "ok": False, "data": None, "error": { "code": "UNKNOWN_METHOD", "message": "POST /api/agv 仅支持 playAction、stopAction、clearActions；查询请用 GET /api/catalog", }, }, ) return try: params = payload.get("params") or {} self._validate_remote(method, params) data = HUB.dispatch(method, params, req_id) hint = None if data.get("subscribers", 0) == 0: hint = "无播放页在线，命令已缓存；打开 play.html 后将自动同步" self._json_response( 200, { "version": 1, "id": req_id, "ok": True, "data": data, "hint": hint, "error": None, }, ) except ValueError as e: self._json_response( 400, { "version": 1, "id": req_id, "ok": False, "data": None, "error": {"code": "INVALID_REQUEST", "message": str(e)}, }, ) def _find_model(self, model_id: str) -> dict | None: for m in self._read_manifest_models(): if isinstance(m, dict) and m.get("id") == model_id: return m return None def _load_model_config(self, model_id: str) -> tuple[dict | None, str | None]: entry = self._find_model(model_id) if not entry: return None, f"模型不存在: {model_id}" cfg_file = entry.get("config") or f"{model_id}.json" cfg_path = (AGV_DIR / cfg_file).resolve() if not str(cfg_path).startswith(str(AGV_DIR)) or not cfg_path.exists(): return None, f"缺少配置 {cfg_file}" try: return json.loads(cfg_path.read_text(encoding="utf-8")), None except (OSError, json.JSONDecodeError) as e: return None, str(e) @staticmethod def _pick_action_ids(params: dict) -> list[str]: ids = params.get("actionIds") or params.get("actions") if isinstance(ids, list): return [str(x).strip() for x in ids if str(x).strip()] if isinstance(ids, str): return [s.strip() for s in ids.replace("+", ",").split(",") if s.strip()] one = params.get("actionId") or params.get("action") return [str(one).strip()] if one else [] def _validate_remote(self, method: str, params: dict) -> None: model_id = (params.get("modelId") or params.get("model") or "").strip() if not model_id: raise ValueError("params.modelId 必填") if not self._find_model(model_id): raise ValueError(f"模型不存在: {model_id}") if method == "clearActions": return cfg, err = self._load_model_config(model_id) if err: raise ValueError(err) if method in {"playAction", "stopAction"}: action_ids = self._pick_action_ids(params) if not action_ids: raise ValueError("params.actionId 必填") known = set((cfg.get("actions") or {}).keys()) for action_id in action_ids: if action_id not in known: raise ValueError(f"动作不存在: {action_id}（模型 {model_id}）") def _agv_report(self): try: length = int(self.headers.get("Content-Length", 0)) raw = self.rfile.read(length).decode("utf-8") payload = json.loads(raw) except (json.JSONDecodeError, UnicodeDecodeError) as e: self._json_response(400, {"ok": False, "error": str(e)}) return client_id = str(payload.get("clientId") or "").strip() if not client_id: self._json_response(400, {"ok": False, "error": "clientId 必填"}) return model_id = payload.get("modelId") if isinstance(model_id, str): model_id = model_id.strip() or None active = payload.get("activeActions") if active is not None and not isinstance(active, list): self._json_response(400, {"ok": False, "error": "activeActions 须为数组"}) return data = HUB.report_from_client(client_id, model_id, active) self._json_response(200, {"ok": True, "data": data}) def _cors_headers(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") def _sse_write(self, payload: dict): line = f"data: {json.dumps(payload, ensure_ascii=False)}\n\n" self.wfile.write(line.encode("utf-8")) self.wfile.flush() def _agv_events_sse(self): self.send_response(200) self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "keep-alive") self._cors_headers() self.end_headers() sub = HUB.subscribe() try: state = HUB.get_state() self._sse_write({"type": "connected", "revision": state.get("revision", 0), "state": state}) while True: try: msg = sub.get(timeout=15) self._sse_write(msg) except queue.Empty: self._sse_write({"type": "ping"}) except (BrokenPipeError, ConnectionResetError, OSError): pass finally: HUB.unsubscribe(sub) def _json_response(self, code: int, data: dict): body = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self._cors_headers() self.end_headers() self.wfile.write(body) def log_message(self, fmt, *args): sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args)) def main(): os.chdir(ROOT) with http.server.ThreadingHTTPServer(("", PORT), Handler) as httpd: print(f"Serving {ROOT}") print(f"配置页: http://127.0.0.1:{PORT}/viewer/index.html") print(f"播放页: http://127.0.0.1:{PORT}/viewer/play.html") print(f"远程API: GET /api/catalog | POST /api/agv (playAction / stopAction / clearActions)") try: httpd.serve_forever() except KeyboardInterrupt: print("\n已停止服务") if __name__ == "__main__": main()