"""Local Device Hub and control panel for the CoCo ESP32 remote."""

from __future__ import annotations

import argparse
import hmac
import json
import secrets
import socket
import threading
import time
from collections import deque
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, quote, unquote, urlparse

from coco_adapter import CocoAdapter, CocoAdapterError


PANEL_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = PANEL_DIR.parent
DATA_DIR = PANEL_DIR / "data"
CONFIG_PATH = DATA_DIR / "config.json"
AUDIO_DIR = DATA_DIR / "audio"
MAX_EVENT_LOG = 160
MAX_AUDIO_BYTES = 3 * 1024 * 1024
AUDIO_EXPIRY_SECONDS = 15 * 60
DISCOVERY_PORT = 8791
DEFAULT_PAIRING_PORT = 8792
DISCOVERY_TTL_SECONDS = 15


def _lan_ip() -> str:
    probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        probe.connect(("192.0.2.1", 80))
        return str(probe.getsockname()[0])
    except OSError:
        return "127.0.0.1"
    finally:
        probe.close()


def _now() -> float:
    return time.time()


def _new_token() -> str:
    return secrets.token_urlsafe(32)


def _default_config(port: int) -> dict[str, Any]:
    return {
        "device_token": _new_token(),
        "device_tokens": {},
        "revoked_devices": [],
        "selected_device_id": "",
        "panel_token": _new_token(),
        "public_url": f"http://{_lan_ip()}:{port}",
        "bridge_url": "http://127.0.0.1:8765",
        "agent_id": "default",
        "normal_overlay": "",
        "rage_overlay": (
            "进入狂暴模式：语气直接、有力量、回答简短。"
            "仍然遵守安全要求，并且不编造事实。"
        ),
    }


class DeviceHub:
    def __init__(self, port: int, public_url: str = "") -> None:
        DATA_DIR.mkdir(parents=True, exist_ok=True)
        AUDIO_DIR.mkdir(parents=True, exist_ok=True)
        self._lock = threading.RLock()
        self.config = self._load_config(port)
        if public_url:
            self.config["public_url"] = public_url.rstrip("/")
            self._save_config()
        self.devices: dict[str, dict[str, Any]] = {}
        self.discovered: dict[str, dict[str, Any]] = {}
        self.commands: dict[str, list[dict[str, Any]]] = {}
        self.audio_jobs: dict[str, dict[str, Any]] = {}
        self.events: deque[dict[str, Any]] = deque(maxlen=MAX_EVENT_LOG)
        self._processing_devices: set[str] = set()
        self._discovery_stop = threading.Event()
        self._discovery_thread = threading.Thread(
            target=self._listen_for_discovery,
            name="coco-hardware-discovery",
            daemon=True,
        )
        self._discovery_thread.start()

    def _load_config(self, port: int) -> dict[str, Any]:
        config = _default_config(port)
        try:
            stored = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            stored = {}
        if isinstance(stored, dict):
            config.update({key: value for key, value in stored.items() if value is not None})
        for key in ("device_token", "panel_token"):
            if not isinstance(config.get(key), str) or not config[key]:
                config[key] = _new_token()
        if not isinstance(config.get("device_tokens"), dict):
            config["device_tokens"] = {}
        if not isinstance(config.get("known_devices"), dict):
            config["known_devices"] = {}
        if not isinstance(config.get("revoked_devices"), list):
            config["revoked_devices"] = []
        if not isinstance(config.get("selected_device_id"), str):
            config["selected_device_id"] = ""
        self.config = config
        self._save_config()
        return config

    def _save_config(self) -> None:
        CONFIG_PATH.write_text(
            json.dumps(self.config, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )

    def panel_authorized(self, token: str) -> bool:
        expected = str(self.config.get("panel_token") or "")
        return bool(expected and token and hmac.compare_digest(expected, token))

    def device_authorized(self, device_id: str, token: str) -> bool:
        revoked = {str(value) for value in self.config.get("revoked_devices", []) if value}
        if device_id in revoked:
            return False
        device_tokens = self.config.get("device_tokens")
        if isinstance(device_tokens, dict):
            expected = str(device_tokens.get(device_id) or "")
            if expected:
                return bool(token and hmac.compare_digest(expected, token))
            if device_tokens:
                return False
        expected = str(self.config.get("device_token") or "")
        return bool(expected and token and hmac.compare_digest(expected, token))

    def stop(self) -> None:
        self._discovery_stop.set()

    def _listen_for_discovery(self) -> None:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            sock.bind(("0.0.0.0", DISCOVERY_PORT))
            sock.settimeout(0.5)
            while not self._discovery_stop.is_set():
                try:
                    raw, address = sock.recvfrom(2048)
                except socket.timeout:
                    continue
                except OSError:
                    break
                try:
                    payload = json.loads(raw.decode("utf-8", errors="replace"))
                except json.JSONDecodeError:
                    continue
                self.register_discovery(payload, address[0])
        except OSError as exc:
            self._log("hub", "error", f"设备发现监听失败：{exc}")
        finally:
            sock.close()

    def register_discovery(self, payload: dict[str, Any], address: str) -> None:
        if not isinstance(payload, dict) or payload.get("type") != "coco_hw_discovery":
            return
        device_id = str(payload.get("device_id") or "").strip()
        nonce = str(payload.get("pairing_nonce") or "").strip()
        if not device_id or len(device_id) > 64 or not nonce:
            return
        with self._lock:
            old = self.discovered.get(device_id) or {}
            item = {
                "id": device_id,
                "ip": address,
                "firmware": str(payload.get("firmware") or old.get("firmware") or ""),
                "pairing_nonce": nonce,
                "pairing_port": int(payload.get("pairing_port") or DEFAULT_PAIRING_PORT),
                "paired": bool(payload.get("paired")),
                "rssi": payload.get("rssi"),
                "last_seen": _now(),
                "capabilities": payload.get("capabilities") or old.get("capabilities") or [],
            }
            self.discovered[device_id] = item
            if device_id in self.devices:
                self.devices[device_id].update({"ip": address, "rssi": payload.get("rssi")})

    def pair_discovered_device(self, device_id: str) -> dict[str, Any]:
        device_id = str(device_id or "").strip()
        with self._lock:
            self._expire_discovered_locked()
            device = dict(self.discovered.get(device_id) or {})
            if not device:
                raise ValueError("设备未发现或已超时")
            token = _new_token()
            device_tokens = self.config.setdefault("device_tokens", {})
            if not isinstance(device_tokens, dict):
                device_tokens = {}
                self.config["device_tokens"] = device_tokens
            device_tokens[device_id] = token
            revoked = [str(value) for value in self.config.get("revoked_devices", []) if value]
            self.config["revoked_devices"] = [value for value in revoked if value != device_id]
            self.config["selected_device_id"] = device_id
            self._save_config()

        self._send_pairing_packet(
            device,
            {
                "type": "coco_hw_pair",
                "action": "pair",
                "device_id": device_id,
                "pairing_nonce": device["pairing_nonce"],
                "hub_url": str(self.config.get("public_url") or "").rstrip("/"),
                "device_token": token,
            },
        )
        with self._lock:
            self._device(device_id).update({
                "ip": device.get("ip"),
                "firmware": str(device.get("firmware") or ""),
                "capabilities": device.get("capabilities") or [],
                "last_seen": _now(),
                "indicator": "idle",
            })
            self._remember_device_locked(device_id, self.devices[device_id])
            self._log(device_id, "pair", "已发送配对信息")
            return self.panel_state()

    def delete_device(self, device_id: str) -> dict[str, Any]:
        device_id = str(device_id or "").strip()
        with self._lock:
            device = dict(self.discovered.get(device_id) or self.devices.get(device_id) or {})
            tokens = self.config.get("device_tokens")
            if isinstance(tokens, dict):
                tokens.pop(device_id, None)
            revoked = [str(value) for value in self.config.get("revoked_devices", []) if value]
            if device_id and device_id not in revoked:
                revoked.append(device_id)
            self.config["revoked_devices"] = revoked
            self.devices.pop(device_id, None)
            self.commands.pop(device_id, None)
            known_devices = self.config.get("known_devices")
            if isinstance(known_devices, dict):
                known_devices.pop(device_id, None)
            if self.config.get("selected_device_id") == device_id:
                self.config["selected_device_id"] = ""
            self._save_config()
            nonce = str(device.get("pairing_nonce") or "")
        if device and nonce:
            self._send_pairing_packet(
                device,
                {
                    "type": "coco_hw_pair",
                    "action": "forget",
                    "device_id": device_id,
                    "pairing_nonce": nonce,
                },
                ignore_errors=True,
            )
        with self._lock:
            self._log(device_id or "hub", "delete", "设备连接已删除")
            return self.panel_state()

    def _send_pairing_packet(self, device: dict[str, Any], payload: dict[str, Any], ignore_errors: bool = False) -> None:
        host = str(device.get("ip") or "").strip()
        port = int(device.get("pairing_port") or DEFAULT_PAIRING_PORT)
        if not host:
            if ignore_errors:
                return
            raise ValueError("设备 IP 不可用")
        raw = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            sock.settimeout(1.0)
            sock.sendto(raw, (host, port))
        except OSError:
            if not ignore_errors:
                raise
        finally:
            sock.close()

    def _expire_discovered_locked(self) -> None:
        now = _now()
        for device_id in list(self.discovered):
            if now - float(self.discovered[device_id].get("last_seen") or 0) > DISCOVERY_TTL_SECONDS:
                self.discovered.pop(device_id, None)

    def register_device(self, payload: dict[str, Any], address: str) -> dict[str, Any]:
        device_id = self._device_id(payload)
        with self._lock:
            device = self._device(device_id)
            device.update(
                {
                    "ip": address,
                    "firmware": str(payload.get("firmware") or device.get("firmware") or ""),
                    "capabilities": payload.get("capabilities") or device.get("capabilities") or [],
                    "last_seen": _now(),
                    "online": True,
                    "indicator": "idle" if device.get("indicator") == "offline" else device.get("indicator", "idle"),
                }
            )
            self._remember_device_locked(device_id, device)
            self._log(device_id, "connected", f"{address} 已连接")
            return self._device_response(device_id, payload)

    def heartbeat(self, payload: dict[str, Any], address: str) -> dict[str, Any]:
        device_id = self._device_id(payload)
        mode = str(payload.get("mode") or "").strip().lower()
        with self._lock:
            device = self._device(device_id)
            self._apply_ack(device_id, payload.get("ack"))
            device.update(
                {
                    "ip": address,
                    "last_seen": _now(),
                    "online": True,
                    "rssi": payload.get("rssi"),
                    "recording": bool(payload.get("recording")),
                    "playing": bool(payload.get("playing")),
                    "wifi": str(payload.get("wifi") or ""),
                }
            )
            if mode in {"normal", "rage"}:
                device["mode"] = mode
            if device.get("indicator") == "offline":
                device["indicator"] = "idle"
            self._remember_device_locked(device_id, device)
            return self._device_response(device_id, payload)

    def device_event(self, payload: dict[str, Any], address: str) -> dict[str, Any]:
        device_id = self._device_id(payload)
        event = str(payload.get("event") or "").strip().lower()
        action = str(payload.get("action") or "").strip().lower()
        button = str(payload.get("button") or "").strip().upper()
        with self._lock:
            device = self._device(device_id)
            self._apply_ack(device_id, payload.get("ack"))
            device.update({"ip": address, "last_seen": _now(), "online": True})
            if device.get("indicator") == "offline":
                device["indicator"] = "idle"
            if event == "button" and action in {"down", "up"}:
                device["button_seq"] = int(device.get("button_seq") or 0) + 1
                device["last_button"] = button
                device["last_button_action"] = action
                device["last_button_at"] = _now()
            if event == "button" and action == "down" and button in {"C", "D"}:
                device["mode"] = "normal" if button == "C" else "rage"
                self._log(device_id, "mode", f"已切换到 {device['mode']} 模式")
            elif event == "audio" and action == "playing":
                device["indicator"] = "busy"
                self._log(device_id, "audio", "正在播放 CoCo 回复")
            elif event == "audio" and action in {"finished", "stopped"}:
                device["indicator"] = "idle"
                self._log(device_id, "audio", "回复播放完成")
            elif event == "audio" and action == "failed":
                device["indicator"] = "error"
                self._log(device_id, "error", str(payload.get("detail") or "音频播放失败"))
            elif event == "button":
                self._log(device_id, "button", f"{button} {action}")
            else:
                self._log(device_id, event or "event", action or "设备事件")
            return self._device_response(device_id, payload)

    def accept_audio(self, device_id: str, audio_wav: bytes, address: str) -> dict[str, Any]:
        if not audio_wav:
            raise ValueError("音频内容为空")
        if len(audio_wav) > MAX_AUDIO_BYTES:
            raise ValueError("音频超过 3 MB 上限")
        with self._lock:
            device = self._device(device_id)
            if device_id != self._selected_device_id_locked():
                device.update({"ip": address, "last_seen": _now(), "online": True, "indicator": "idle"})
                self._log(device_id, "audio", "未选中，语音已忽略")
                return self._device_response(device_id, {})
            if device_id in self._processing_devices:
                raise ValueError("上一段语音仍在处理中")
            self._processing_devices.add(device_id)
            device.update(
                {
                    "ip": address,
                    "last_seen": _now(),
                    "online": True,
                    "indicator": "busy",
                }
            )
            self._log(device_id, "audio", f"收到 {len(audio_wav)} 字节语音")
        threading.Thread(
            target=self._process_audio,
            args=(device_id, audio_wav),
            name=f"coco-hardware-audio-{device_id}",
            daemon=True,
        ).start()
        return self._device_response(device_id, {})

    def _process_audio(self, device_id: str, audio_wav: bytes) -> None:
        progress_state = {"transcript": False, "reply": False}

        def on_progress(step: str, payload: dict) -> None:
            with self._lock:
                if step == "stt_start":
                    self._log(device_id, "stt", "正在识别")
                elif step == "transcript":
                    transcript = str(payload.get("text") or "").strip()
                    if transcript:
                        progress_state["transcript"] = True
                        self._log(device_id, "transcript", transcript)
                    else:
                        stats = payload.get("audio") if isinstance(payload.get("audio"), dict) else {}
                        progress_state["transcript"] = True
                        self._log(
                            device_id,
                            "stt",
                            f"未识别到语音（{stats.get('duration', '?')}s, "
                            f"rms={stats.get('rms', '?')}, peak={stats.get('peak', '?')}）",
                        )
                elif step == "reply":
                    response = str(payload.get("text") or "").strip()
                    if response:
                        progress_state["reply"] = True
                        self._log(device_id, "reply", response)
                elif step == "audio_ready":
                    timings = payload.get("timings_ms") if isinstance(payload.get("timings_ms"), dict) else {}
                    self._log(
                        device_id,
                        "audio",
                        "语音链路耗时 "
                        f"{timings.get('total_ms', '?')}ms"
                        f"（识别 {timings.get('stt_ms', '?')}ms，"
                        f"对话 {timings.get('chat_ms', '?')}ms，"
                        f"合成 {timings.get('tts_ms', '?')}ms）",
                    )

        try:
            with self._lock:
                device = dict(self._device(device_id))
                mode = str(device.get("mode") or "normal")
                agent_id = str(self.config.get("agent_id") or "default")
                overlay_key = "rage_overlay" if mode == "rage" else "normal_overlay"
                overlay = str(self.config.get(overlay_key) or "")
                bridge_url = str(self.config.get("bridge_url") or "")

            adapter = CocoAdapter(PROJECT_ROOT, bridge_url)
            result = adapter.process_audio(
                audio_wav,
                agent_id,
                overlay,
                progress=on_progress,
            )
            with self._lock:
                if result.agent_id != agent_id:
                    self.config["agent_id"] = result.agent_id
                    self._save_config()
                    self._log("hub", "config", f"已自动选择智能体 {result.agent_id}")
                if result.transcript and not progress_state["transcript"]:
                    self._log(device_id, "transcript", result.transcript)
                elif not result.transcript and not progress_state["transcript"]:
                    stats = result.audio_stats or {}
                    self._log(
                        device_id,
                        "stt",
                        f"未识别到语音（{stats.get('duration', '?')}s, "
                        f"rms={stats.get('rms', '?')}, peak={stats.get('peak', '?')}）",
                    )
                if result.response and not progress_state["reply"]:
                    self._log(device_id, "reply", result.response)
                if not result.audio_mp3:
                    self._device(device_id)["indicator"] = "idle"
                    return
                audio_url = self._store_audio_job(device_id, result.audio_mp3)
                self.queue_command(device_id, "play_audio", {"url": audio_url})
                self._device(device_id)["indicator"] = "busy"
        except CocoAdapterError as exc:
            self._set_error(device_id, str(exc))
        except Exception as exc:
            self._set_error(device_id, f"语音处理失败：{exc}")
        finally:
            with self._lock:
                self._processing_devices.discard(device_id)

    def _store_audio_stream_job(
        self,
        device_id: str,
        bridge_url: str,
        agent_id: str,
        text: str,
    ) -> str:
        self._expire_audio()
        audio_id = secrets.token_urlsafe(12)
        token = secrets.token_urlsafe(24)
        self.audio_jobs[audio_id] = {
            "device_id": device_id,
            "token": token,
            "stream": True,
            "bridge_url": bridge_url,
            "agent_id": agent_id,
            "text": text,
            "expires_at": _now() + AUDIO_EXPIRY_SECONDS,
        }
        base = str(self.config.get("public_url") or "").rstrip("/")
        if not base:
            raise ValueError("未配置 Hub 对 ESP32 可访问的 public_url")
        return f"{base}/api/device/audio/stream/{quote(audio_id)}?token={quote(token)}"

    def _store_audio_job(self, device_id: str, audio_mp3: bytes) -> str:
        self._expire_audio()
        audio_id = secrets.token_urlsafe(12)
        token = secrets.token_urlsafe(24)
        path = AUDIO_DIR / f"{audio_id}.mp3"
        path.write_bytes(audio_mp3)
        self.audio_jobs[audio_id] = {
            "device_id": device_id,
            "token": token,
            "path": path,
            "expires_at": _now() + AUDIO_EXPIRY_SECONDS,
        }
        base = str(self.config.get("public_url") or "").rstrip("/")
        if not base:
            raise ValueError("未配置 Hub 对 ESP32 可访问的 public_url")
        return f"{base}/api/device/audio/{quote(audio_id)}?token={quote(token)}"

    def get_audio_job(self, audio_id: str, token: str) -> dict[str, Any] | None:
        with self._lock:
            self._expire_audio()
            job = self.audio_jobs.get(audio_id)
            if not job or not hmac.compare_digest(str(job["token"]), token):
                return None
            return dict(job)

    def get_audio(self, audio_id: str, token: str) -> Path | None:
        job = self.get_audio_job(audio_id, token)
        if not job or not job.get("path"):
            return None
        return Path(job["path"])

    def _expire_audio(self) -> None:
        now = _now()
        expired = [
            audio_id
            for audio_id, job in self.audio_jobs.items()
            if float(job.get("expires_at") or 0) <= now
        ]
        for audio_id in expired:
            job = self.audio_jobs.pop(audio_id, None)
            if job:
                path = job.get("path")
                if path:
                    Path(path).unlink(missing_ok=True)

    def queue_command(
        self,
        device_id: str,
        command_type: str,
        payload: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        if command_type not in {
            "set_indicator",
            "beep",
            "set_mode",
            "start_recording",
            "stop_recording",
            "play_audio",
            "forget",
        }:
            raise ValueError("不支持的设备命令")
        command = {
            "id": secrets.token_urlsafe(10),
            "type": command_type,
            "payload": payload or {},
            "created_at": int(_now()),
        }
        with self._lock:
            if command_type != "forget" and not self._device_online_locked(device_id):
                raise ValueError("设备已断开，无法操作")
            self._device(device_id)
            self.commands.setdefault(device_id, []).append(command)
            if command_type == "set_indicator":
                indicator = str(command["payload"].get("indicator") or "")
                if indicator in {"idle", "busy", "error", "offline"}:
                    self._device(device_id)["indicator"] = indicator
            if command_type == "set_mode":
                mode = str(command["payload"].get("mode") or "")
                if mode in {"normal", "rage"}:
                    self._device(device_id)["mode"] = mode
            if command_type == "start_recording":
                self._device(device_id)["recording"] = True
                self._device(device_id)["indicator"] = "busy"
            if command_type == "stop_recording":
                self._device(device_id)["recording"] = False
            self._log(device_id, "command", command_type)
        return command

    def panel_state(self) -> dict[str, Any]:
        with self._lock:
            now = _now()
            self._expire_discovered_locked()
            selected_device_id = self._selected_device_id_locked()
            devices = []
            for device_id, device in self._known_devices_locked().items():
                live = self.devices.get(device_id, {})
                item = {**device, **live}
                item["id"] = device_id
                item["online"] = bool(item.get("last_seen") and now - item["last_seen"] < 8)
                item["pending_commands"] = len(self.commands.get(device_id, []))
                item["paired"] = True
                item["selected"] = device_id == selected_device_id
                if not item["online"] and item.get("indicator") != "error":
                    item["indicator"] = "offline"
                devices.append(item)
            devices.sort(key=lambda item: item["id"])
            discovered = []
            for device_id, device in self.discovered.items():
                if device_id in self.devices:
                    continue
                item = dict(device)
                item["online"] = bool(item.get("last_seen") and now - item["last_seen"] < DISCOVERY_TTL_SECONDS)
                item["selected"] = False
                discovered.append(item)
            discovered.sort(key=lambda item: item["id"])
            return {
                "ok": True,
                "devices": devices,
                "discovered": discovered,
                "events": list(self.events),
                "config": {
                    "public_url": self.config.get("public_url"),
                    "device_token": self.config.get("device_token"),
                    "agent_id": self.config.get("agent_id"),
                    "selected_device_id": selected_device_id,
                    "normal_overlay": self.config.get("normal_overlay"),
                    "rage_overlay": self.config.get("rage_overlay"),
                },
            }

    def update_panel_config(self, payload: dict[str, Any]) -> dict[str, Any]:
        allowed = {"public_url", "agent_id", "normal_overlay", "rage_overlay", "selected_device_id"}
        with self._lock:
            for key in allowed:
                if key in payload:
                    value = str(payload[key] or "").strip()
                    if key == "selected_device_id" and value:
                        if value not in self._known_devices_locked():
                            raise ValueError("设备不存在")
                        if not self._device_online_locked(value):
                            raise ValueError("设备已断开，无法选择")
                    self.config[key] = value
            self._save_config()
            self._log("hub", "config", "硬件设置已更新")
            return self.panel_state()

    def _selected_device_id_locked(self) -> str:
        selected = str(self.config.get("selected_device_id") or "").strip()
        if selected and self._device_online_locked(selected):
            return selected
        selected = next(
            (
                device_id
                for device_id in sorted(self._known_devices_locked())
                if self._device_online_locked(device_id)
            ),
            "",
        )
        if self.config.get("selected_device_id") != selected:
            self.config["selected_device_id"] = selected
            self._save_config()
        return selected

    def _device_online_locked(self, device_id: str) -> bool:
        device = self.devices.get(device_id)
        return bool(device and device.get("last_seen") and _now() - float(device["last_seen"]) < 8)

    def _known_devices_locked(self) -> dict[str, dict[str, Any]]:
        known = self.config.setdefault("known_devices", {})
        if not isinstance(known, dict):
            known = {}
            self.config["known_devices"] = known
        tokens = self.config.get("device_tokens")
        if isinstance(tokens, dict):
            for device_id in tokens:
                if device_id:
                    known.setdefault(str(device_id), {})
        for device_id in self.devices:
            known.setdefault(device_id, {})
        return known

    def _remember_device_locked(self, device_id: str, device: dict[str, Any]) -> None:
        known = self._known_devices_locked()
        item = known.setdefault(device_id, {})
        for key in ("id", "ip", "firmware", "capabilities", "last_seen", "rssi", "wifi", "mode"):
            if key in device and device[key] not in (None, ""):
                item[key] = device[key]
        item["id"] = device_id
        self._save_config()

    def _device_response(self, device_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        device = self._device(device_id)
        self._apply_ack(device_id, payload.get("ack"))
        return {
            "ok": True,
            "indicator": device.get("indicator", "idle"),
            "mode": device.get("mode", "normal"),
            "commands": list(self.commands.get(device_id, []))[:8],
        }

    def _device_id(self, payload: dict[str, Any]) -> str:
        device_id = str(payload.get("device_id") or "").strip()
        if not device_id or len(device_id) > 64:
            raise ValueError("device_id 无效")
        return device_id

    def _device(self, device_id: str) -> dict[str, Any]:
        return self.devices.setdefault(
            device_id,
            {
                "indicator": "offline",
                "mode": "normal",
                "online": False,
                "last_seen": 0,
                "rssi": None,
                "recording": False,
                "playing": False,
            },
        )

    def _apply_ack(self, device_id: str, values: Any) -> None:
        if not isinstance(values, list):
            return
        ack = {str(value) for value in values if value}
        if ack:
            self.commands[device_id] = [
                command for command in self.commands.get(device_id, [])
                if command["id"] not in ack
            ]

    def _set_error(self, device_id: str, message: str) -> None:
        with self._lock:
            device = self._device(device_id)
            device["indicator"] = "error"
            self.queue_command(device_id, "beep", {"duration_ms": 180})
            self._log(device_id, "error", message)

    def _log(self, device_id: str, kind: str, message: str) -> None:
        self.events.appendleft(
            {
                "at": time.strftime("%H:%M:%S"),
                "device_id": device_id,
                "kind": kind,
                "message": str(message)[:600],
            }
        )


class HubHandler(BaseHTTPRequestHandler):
    hub: DeviceHub
    server_version = "CoCoHardwareHub/1.0"

    def do_GET(self) -> None:
        try:
            path, _, query = self._request_parts()
            if path.startswith("/api/device/audio/"):
                audio_id = path.rsplit("/", 1)[-1]
                self._send_audio(audio_id, query.get("token", [""])[0])
                return
            if path == "/api/panel/state":
                self._require_panel()
                self._send_json(self.hub.panel_state())
                return
            self._send_error(HTTPStatus.NOT_FOUND, "接口不存在")
        except Exception as exc:
            self._send_exception(exc)

    def do_POST(self) -> None:
        try:
            path, _, _ = self._request_parts()
            if path == "/api/panel/command":
                self._require_panel()
                data = self._read_json()
                command = self.hub.queue_command(
                    str(data.get("device_id") or ""),
                    str(data.get("type") or ""),
                    data.get("payload") if isinstance(data.get("payload"), dict) else {},
                )
                self._send_json({"ok": True, "command": command})
                return
            if path == "/api/panel/config":
                self._require_panel()
                self._send_json(self.hub.update_panel_config(self._read_json()))
                return
            if path == "/api/panel/device/pair":
                self._require_panel()
                data = self._read_json()
                self._send_json(self.hub.pair_discovered_device(str(data.get("device_id") or "")))
                return
            if path == "/api/panel/device/delete":
                self._require_panel()
                data = self._read_json()
                self._send_json(self.hub.delete_device(str(data.get("device_id") or "")))
                return
            if path.startswith("/api/device/"):
                self._require_device()
                if path == "/api/device/hello":
                    self._send_json(self.hub.register_device(self._read_json(), self.client_address[0]))
                    return
                if path == "/api/device/heartbeat":
                    self._send_json(self.hub.heartbeat(self._read_json(), self.client_address[0]))
                    return
                if path == "/api/device/event":
                    self._send_json(self.hub.device_event(self._read_json(), self.client_address[0]))
                    return
                if path == "/api/device/audio":
                    query = parse_qs(urlparse(self.path).query)
                    device_id = str(query.get("device_id", [""])[0]).strip()
                    self._send_json(
                        self.hub.accept_audio(device_id, self._read_body(), self.client_address[0]),
                        HTTPStatus.ACCEPTED,
                    )
                    return
            self._send_error(HTTPStatus.NOT_FOUND, "接口不存在")
        except Exception as exc:
            self._send_exception(exc)

    def do_OPTIONS(self) -> None:
        self.send_response(HTTPStatus.NO_CONTENT)
        self._send_common_headers()
        self.end_headers()

    def _request_parts(self) -> tuple[str, list[str], dict[str, list[str]]]:
        parsed = urlparse(self.path)
        path = parsed.path.rstrip("/") or "/"
        parts = [unquote(part) for part in path.split("/") if part]
        return path, parts, parse_qs(parsed.query)

    def _require_panel(self) -> None:
        if not self.hub.panel_authorized(self.headers.get("X-CoCo-Panel-Token", "")):
            raise PermissionError("硬件设置 token 无效")

    def _require_device(self) -> None:
        device_id = ""
        try:
            path, _, query = self._request_parts()
            if path == "/api/device/audio":
                device_id = str(query.get("device_id", [""])[0]).strip()
            else:
                device_id = str(self._peek_json_device_id()).strip()
        except Exception:
            device_id = ""
        if not self.hub.device_authorized(device_id, self.headers.get("X-CoCo-Device-Token", "")):
            raise PermissionError("设备配对 token 无效")

    def _peek_json_device_id(self) -> str:
        raw = self._read_body()
        try:
            data = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            return ""
        return str(data.get("device_id") or "") if isinstance(data, dict) else ""

    def _send_audio(self, audio_id: str, token: str) -> None:
        job = self.hub.get_audio_job(audio_id, token)
        if not job:
            self._send_error(HTTPStatus.NOT_FOUND, "音频已过期或不存在")
            return
        if job.get("stream"):
            self._send_audio_stream(job)
            return
        path = Path(job.get("path") or "")
        if not path.is_file():
            self._send_error(HTTPStatus.NOT_FOUND, "音频已过期或不存在")
            return
        self._send_bytes(path.read_bytes(), "audio/mpeg")

    def _send_audio_stream(self, job: dict[str, Any]) -> None:
        adapter = CocoAdapter(
            PROJECT_ROOT,
            str(job.get("bridge_url") or ""),
        )
        response = None
        started_at = time.perf_counter()
        first_chunk_at = None
        try:
            response = adapter.open_tts_stream(
                str(job.get("agent_id") or ""),
                str(job.get("text") or ""),
            )
            if getattr(response, "status", HTTPStatus.OK) != HTTPStatus.OK:
                detail = response.read(240).decode("utf-8", errors="replace")
                self._send_error(HTTPStatus.BAD_GATEWAY, detail or "TTS 流式请求失败")
                return

            self.close_connection = True
            self.send_response(HTTPStatus.OK)
            self._send_common_headers()
            self.send_header("Content-Type", "audio/mpeg")
            self.send_header("Connection", "close")
            self.end_headers()
            while True:
                chunk = response.read(4096)
                if not chunk:
                    break
                if first_chunk_at is None:
                    first_chunk_at = time.perf_counter()
                    with self.hub._lock:
                        self.hub._log(
                            str(job.get("device_id") or "hub"),
                            "audio",
                            f"音频首包 {int((first_chunk_at - started_at) * 1000)}ms",
                        )
                self.wfile.write(chunk)
                self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError):
            pass
        except Exception as exc:
            print(f"[hub] TTS 流式转发失败: {exc}")
        finally:
            if response is not None:
                response.close()

    def _read_body(self) -> bytes:
        cached = getattr(self, "_cached_body", None)
        if cached is not None:
            return cached
        length = int(self.headers.get("Content-Length", "0") or 0)
        if length < 0 or length > MAX_AUDIO_BYTES + 64 * 1024:
            raise ValueError("请求体过大")
        body = self.rfile.read(length) if length else b""
        self._cached_body = body
        return body

    def _read_json(self) -> dict[str, Any]:
        raw = self._read_body()
        if not raw:
            return {}
        try:
            data = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise ValueError("JSON 格式错误") from exc
        if not isinstance(data, dict):
            raise ValueError("JSON 必须是对象")
        return data

    def _send_json(self, data: dict[str, Any], status: HTTPStatus = HTTPStatus.OK) -> None:
        self._send_bytes(
            json.dumps(data, ensure_ascii=False).encode("utf-8"),
            "application/json; charset=utf-8",
            status,
        )

    def _send_error(self, status: HTTPStatus, message: str) -> None:
        self._send_json({"ok": False, "error": message}, status)

    def _send_exception(self, exc: Exception) -> None:
        if isinstance(exc, PermissionError):
            self._send_error(HTTPStatus.FORBIDDEN, str(exc))
        elif isinstance(exc, FileNotFoundError):
            self._send_error(HTTPStatus.NOT_FOUND, str(exc))
        elif isinstance(exc, ValueError):
            self._send_error(HTTPStatus.BAD_REQUEST, str(exc))
        else:
            self._send_error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))

    def _send_bytes(self, content: bytes, content_type: str, status: HTTPStatus = HTTPStatus.OK) -> None:
        self.send_response(status)
        self._send_common_headers()
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(content)))
        self.end_headers()
        self.wfile.write(content)

    def _send_common_headers(self) -> None:
        self.send_header("Cache-Control", "no-store")
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Access-Control-Allow-Headers", "Content-Type, X-CoCo-Panel-Token")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")

    def log_message(self, fmt: str, *args: Any) -> None:
        print(f"[hub] {self.address_string()} - {fmt % args}")


def main() -> None:
    parser = argparse.ArgumentParser(description="CoCo ESP32 Hardware Hub")
    parser.add_argument("--host", default="0.0.0.0", help="监听地址")
    parser.add_argument("--port", type=int, default=8790, help="监听端口")
    parser.add_argument(
        "--public-url",
        default="",
        help="ESP32 可访问的 Hub URL，例如 http://192.168.1.10:8790",
    )
    args = parser.parse_args()

    hub = DeviceHub(args.port, args.public_url)

    class Handler(HubHandler):
        pass

    Handler.hub = hub
    server = ThreadingHTTPServer((args.host, args.port), Handler)
    print(f"CoCo Hardware Hub: {hub.config['public_url']}")
    print("设备管理入口: CoCo 设置 > 设备管理")
    print("ESP32 首次配网: 长按 B 3 秒后连接 CoCo-Remote-xxxxxx 热点")
    print("按 Ctrl+C 停止")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n正在停止 Hardware Hub...")
    finally:
        hub.stop()
        server.server_close()


if __name__ == "__main__":
    main()
