#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称：device_sim_service.py
模块描述：设备模拟模块业务服务层

作者：czj
创建日期：2026-07-29
最后修改：2026-08-05
"""

import json
import logging
import queue as thread_queue
import threading
import asyncio
from datetime import datetime, timedelta
from typing import Optional, List, Tuple
from dataclasses import dataclass

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, delete, and_

from app.models.device_sim import EnvConfig, DeviceSimulator, ReportLog
from app.schemas.device_sim import (
    EnvConfigCreate, EnvConfigUpdate,
    SimulatorCreate, SimulatorUpdate,
    TestConnectionResponse,
)
from app.services.mqtt_manager import MqttManager, mqtt_manager
from app.simulators import create_simulator
from app.utils.id_generator import generate_id
from app.config import settings

logger = logging.getLogger(__name__)

# ==================== 消息队列 ====================

@dataclass
class QueueMessage:
    """消息队列数据结构"""
    device_id: str
    device_name: str
    device_type: str
    topic: str
    payload: dict
    direction: str
    status: str
    error_message: Optional[str]
    timestamp: datetime


# 全局消息队列（最大 10000 条，线程安全）
_message_queue: thread_queue.Queue = None
_queue_stats = {"enqueued": 0, "processed": 0, "dropped": 0}
_queue_stats_lock = threading.Lock()


def get_message_queue() -> thread_queue.Queue:
    """获取消息队列实例（线程安全）"""
    global _message_queue
    if _message_queue is None:
        _message_queue = thread_queue.Queue(maxsize=10000)
    return _message_queue


def enqueue_message(msg: QueueMessage) -> bool:
    """
    将消息入队（非阻塞，在模拟器线程中调用）

    Returns:
        bool: 是否入队成功
    """
    queue = get_message_queue()
    try:
        queue.put_nowait(msg)
        with _queue_stats_lock:
            _queue_stats["enqueued"] += 1
        return True
    except thread_queue.Full:
        with _queue_stats_lock:
            _queue_stats["dropped"] += 1
        logger.warning(f"消息队列已满，丢弃消息: device_id={msg.device_id}")
        return False


# ==================== 消息推送回调 ====================

# 消息推送回调函数（由 WebSocket 路由设置）
_message_callback = None


def set_message_callback(callback):
    """设置消息推送回调函数"""
    global _message_callback
    _message_callback = callback

# 密码加密密钥（使用 settings 中的密钥，实际生产环境应使用密钥管理服务）
# 简化实现：使用固定的 Fernet 密钥，生产环境应替换
_ENCRYPTION_KEY = None


def _get_encryption_key() -> bytes:
    """获取加密密钥"""
    global _ENCRYPTION_KEY
    if _ENCRYPTION_KEY is None:
        from cryptography.fernet import Fernet
        # 使用 settings 中配置的密钥，如果没有则使用固定密钥
        key = getattr(settings, 'ENCRYPTION_KEY', None)
        if key:
            _ENCRYPTION_KEY = key.encode() if isinstance(key, str) else key
        else:
            # 开发环境使用固定密钥
            _ENCRYPTION_KEY = Fernet.generate_key()
    return _ENCRYPTION_KEY


def _encrypt_password(password: str) -> str:
    """加密密码"""
    from cryptography.fernet import Fernet
    f = Fernet(_get_encryption_key())
    return f.encrypt(password.encode()).decode()


def _decrypt_password(password_encrypted: str) -> str:
    """解密密码"""
    from cryptography.fernet import Fernet
    try:
        f = Fernet(_get_encryption_key())
        return f.decrypt(password_encrypted.encode()).decode()
    except Exception as e:
        logger.error(f"密码解密失败: {e}")
        return ""


# 模拟设备运行状态管理（内存中）
_running_simulators: dict = {}
_running_simulators_lock = threading.Lock()


class DeviceSimService:
    """设备模拟服务类"""

    def __init__(self, db: AsyncSession):
        self.db = db

    # ==================== 环境配置 CRUD ====================

    async def list_env_configs(self, skip: int = 0, limit: int = 100) -> Tuple[List[EnvConfig], int]:
        """
        获取环境配置列表

        Args:
            skip: 跳过的记录数
            limit: 返回记录数上限

        Returns:
            Tuple[list, int]: (配置列表, 总数)
        """
        # 查询总数
        count_query = select(func.count()).select_from(EnvConfig)
        total_result = await self.db.execute(count_query)
        total = total_result.scalar() or 0

        # 查询列表
        query = select(EnvConfig).order_by(EnvConfig.created_at.desc()).offset(skip).limit(limit)
        result = await self.db.execute(query)
        items = list(result.scalars().all())

        # 更新实时状态：仅当内存中有活跃连接时覆盖数据库状态
        for item in items:
            mem_status = mqtt_manager.get_connection_status(item.id)
            if mem_status == "connected":
                item.status = "connected"

        return items, total

    async def get_env_config(self, config_id: str) -> Optional[EnvConfig]:
        """
        获取环境配置详情

        Args:
            config_id: 配置 ID

        Returns:
            EnvConfig: 配置对象
        """
        query = select(EnvConfig).where(EnvConfig.id == config_id)
        result = await self.db.execute(query)
        config = result.scalar_one_or_none()
        if config:
            config.status = mqtt_manager.get_connection_status(config.id)
        return config

    async def create_env_config(self, data: EnvConfigCreate) -> EnvConfig:
        """
        创建环境配置

        Args:
            data: 创建数据

        Returns:
            EnvConfig: 创建的配置对象
        """
        # 自动生成默认多主题映射（如果未传入 topics）
        topics = data.topics
        if not topics:
            topics = {
                "door": f"{data.topic_prefix}/door",
                "paperless": f"{data.topic_prefix}/paperless",
                "central": f"{data.topic_prefix}/central",
                "client": f"{data.topic_prefix}/client",
            }

        config = EnvConfig(
            id=generate_id("env"),
            name=data.name,
            broker_host=data.broker_host,
            broker_port=data.broker_port,
            username=data.username,
            topic_prefix=data.topic_prefix,
            topics=topics,
            use_tls=data.use_tls,
            client_id_prefix=data.client_id_prefix,
            status="disconnected",
        )
        # 加密密码
        if data.password:
            config.password_encrypted = _encrypt_password(data.password)

        self.db.add(config)
        await self.db.flush()
        return config

    async def update_env_config(self, config_id: str, data: EnvConfigUpdate) -> Optional[EnvConfig]:
        """
        更新环境配置

        Args:
            config_id: 配置 ID
            data: 更新数据

        Returns:
            EnvConfig: 更新后的配置对象
        """
        config = await self.get_env_config(config_id)
        if not config:
            return None

        update_data = data.model_dump(exclude_unset=True)
        if "password" in update_data:
            if update_data["password"]:
                config.password_encrypted = _encrypt_password(update_data["password"])
            del update_data["password"]
        # 如果 topics 有更新，同步更新 topic_prefix（向后兼容）
        if "topics" in update_data and update_data["topics"]:
            # 取第一个设备类型的前缀作为 topic_prefix 的兼容值
            first_type = next(iter(update_data["topics"].values()), "")
            if first_type:
                # 提取公共前缀部分（如 "device/door" -> "device"）
                parts = first_type.split("/")
                if len(parts) >= 1:
                    config.topic_prefix = parts[0]

        for key, value in update_data.items():
            setattr(config, key, value)

        config.updated_at = datetime.now()
        await self.db.flush()
        return config

    async def delete_env_config(self, config_id: str) -> bool:
        """
        删除环境配置

        Args:
            config_id: 配置 ID

        Returns:
            bool: 是否删除成功
        """
        # 检查是否有设备关联
        query = select(func.count()).select_from(DeviceSimulator).where(
            DeviceSimulator.env_config_id == config_id
        )
        result = await self.db.execute(query)
        count = result.scalar() or 0
        if count > 0:
            raise ValueError(f"环境配置 {config_id} 下有 {count} 个模拟设备，请先删除设备")

        # 断开 MQTT 连接
        mqtt_manager.disconnect(config_id)

        config = await self.get_env_config(config_id)
        if not config:
            return False

        await self.db.delete(config)
        await self.db.flush()
        return True

    async def test_env_connection(self, config_id: str) -> TestConnectionResponse:
        """
        测试环境配置的 MQTT 连接

        Args:
            config_id: 配置 ID

        Returns:
            TestConnectionResponse: 测试结果
        """
        # 直接查询数据库，不用 get_env_config（避免内存状态覆盖）
        query = select(EnvConfig).where(EnvConfig.id == config_id)
        result = await self.db.execute(query)
        config = result.scalar_one_or_none()
        if not config:
            return TestConnectionResponse(success=False, message="环境配置不存在")

        password = ""
        if config.password_encrypted:
            password = _decrypt_password(config.password_encrypted)

        success, message, latency = mqtt_manager.test_connection(
            host=config.broker_host,
            port=config.broker_port,
            username=config.username,
            password=password,
            use_tls=config.use_tls,
        )

        # 更新数据库状态
        config.status = "connected" if success else "error"
        if success:
            config.last_connected_at = datetime.now()
        await self.db.commit()

        return TestConnectionResponse(
            success=success,
            message=message,
            latency_ms=latency
        )

    # ==================== 模拟设备 CRUD ====================

    async def list_simulators(
        self,
        skip: int = 0,
        limit: int = 100,
        device_type: Optional[str] = None,
        env_config_id: Optional[str] = None,
        status: Optional[str] = None,
    ) -> Tuple[List[DeviceSimulator], int]:
        """
        获取模拟设备列表

        Args:
            skip: 跳过的记录数
            limit: 返回记录数上限
            device_type: 设备类型筛选
            env_config_id: 环境配置筛选
            status: 状态筛选

        Returns:
            Tuple[list, int]: (设备列表, 总数)
        """
        conditions = []
        if device_type:
            conditions.append(DeviceSimulator.device_type == device_type)
        if env_config_id:
            conditions.append(DeviceSimulator.env_config_id == env_config_id)
        if status:
            conditions.append(DeviceSimulator.status == status)

        # 查询总数
        count_query = select(func.count()).select_from(DeviceSimulator)
        if conditions:
            count_query = count_query.where(and_(*conditions))
        total_result = await self.db.execute(count_query)
        total = total_result.scalar() or 0

        # 查询列表
        query = select(DeviceSimulator).order_by(DeviceSimulator.created_at.desc())
        if conditions:
            query = query.where(and_(*conditions))
        query = query.offset(skip).limit(limit)
        result = await self.db.execute(query)
        items = list(result.scalars().all())

        # 更新实时运行状态
        with _running_simulators_lock:
            for item in items:
                sim = _running_simulators.get(item.id)
                if sim and sim.is_running():
                    item.status = "running"
                elif item.status == "running":
                    item.status = "stopped"

        return items, total

    async def get_device_stats(
        self,
        device_type: Optional[str] = None,
        env_config_id: Optional[str] = None,
    ) -> dict:
        """
        获取设备统计信息（总设备/运行中/已停止，支持按类型分组）

        Args:
            device_type: 设备类型筛选（不传则统计全部类型）
            env_config_id: 环境配置筛选

        Returns:
            dict: {
                "total": int, "running": int, "stopped": int,
                "by_type": {door: {total, running, stopped}, ...}
            }
        """
        conditions = []
        if device_type:
            conditions.append(DeviceSimulator.device_type == device_type)
        if env_config_id:
            conditions.append(DeviceSimulator.env_config_id == env_config_id)

        # 查询全部设备（不分页，仅取统计所需列）
        query = select(DeviceSimulator.id, DeviceSimulator.device_type, DeviceSimulator.status)
        if conditions:
            query = query.where(and_(*conditions))
        result = await self.db.execute(query)
        rows = result.all()

        # 以内存 _running_simulators 为准修正实时运行状态
        running_ids = set()
        with _running_simulators_lock:
            for rid, sim in _running_simulators.items():
                if sim.is_running():
                    running_ids.add(rid)

        # 聚合统计
        total = len(rows)
        running = 0
        stopped = 0
        by_type: dict = {}
        type_order = ["door", "paperless", "central", "client"]
        for rid, rtype, rstatus in rows:
            is_running = rid in running_ids
            if is_running:
                running += 1
            else:
                stopped += 1
            t = rtype or "unknown"
            group = by_type.setdefault(t, {"total": 0, "running": 0, "stopped": 0})
            group["total"] += 1
            if is_running:
                group["running"] += 1
            else:
                group["stopped"] += 1

        return {
            "total": total,
            "running": running,
            "stopped": stopped,
            "by_type": by_type,
        }

    async def get_simulator(self, device_id: str) -> Optional[DeviceSimulator]:
        """
        获取模拟设备详情

        Args:
            device_id: 设备 ID

        Returns:
            DeviceSimulator: 设备对象
        """
        query = select(DeviceSimulator).where(DeviceSimulator.id == device_id)
        result = await self.db.execute(query)
        device = result.scalar_one_or_none()
        if device:
            with _running_simulators_lock:
                sim = _running_simulators.get(device.id)
                if sim and sim.is_running():
                    device.status = "running"
        return device

    async def create_simulator(self, data: SimulatorCreate) -> DeviceSimulator:
        """
        创建模拟设备

        Args:
            data: 创建数据

        Returns:
            DeviceSimulator: 创建的设备对象
        """
        # 检查环境配置是否存在
        config = await self.get_env_config(data.env_config_id)
        if not config:
            raise ValueError(f"环境配置不存在: {data.env_config_id}")

        # 检查设备 ID 是否已存在（同一环境下）
        check_query = select(DeviceSimulator).where(
            and_(
                DeviceSimulator.env_config_id == data.env_config_id,
                DeviceSimulator.device_id == data.device_id
            )
        )
        check_result = await self.db.execute(check_query)
        if check_result.scalar_one_or_none():
            raise ValueError(f"该环境下设备 ID 已存在: {data.device_id}")

        simulator = DeviceSimulator(
            id=generate_id("sim"),
            env_config_id=data.env_config_id,
            device_type=data.device_type,
            device_name=data.device_name,
            device_id=data.device_id,
            auto_reconnect=data.auto_reconnect,
            extra_attrs=data.extra_attrs or {},
            topic_params=data.topic_params or {},
            report_config=data.report_config,
            status="stopped",
        )

        self.db.add(simulator)
        await self.db.flush()
        return simulator

    async def import_simulators_from_excel(
        self,
        file_bytes: bytes,
        device_type: str,
    ) -> dict:
        """
        从 Excel 文件批量导入模拟设备

        Args:
            file_bytes: Excel 文件字节内容
            device_type: 设备类型（由前端页面决定）

        Returns:
            dict: 导入结果，符合 DeviceImportResponse 结构
        """
        from io import BytesIO
        from openpyxl import load_workbook

        # 表头 → 字段名映射
        COLUMN_MAP = {
            "环境配置名称": "env_config_name",
            "设备名称": "device_name",
            "设备ID": "device_id",
            "设备编号": "device_id",
            "设备id": "device_id",
            "授权码": "app_token",
            "授权码(app_token)": "app_token",
            "会议室编号": "room_id",
            "主题类型": "central_device_type",
            "设备类型": "central_device_type",  # 兼容旧版
            "自动重连": "auto_reconnect",
            "启用定时上报": "report_enabled",
            "上报间隔(秒)": "report_interval",
            "上报间隔": "report_interval",
        }

        # 中控设备类型中文 → 标识映射
        CENTRAL_DEVICE_TYPES = {
            "会议室在线": "room_online",
            "设备在线": "device_online",
            "音频系统": "audio",
            "视频系统": "video",
            "控制系统": "control",
            "网络系统": "network",
            "电源系统": "power",
        }

        # 解析 Excel
        wb = load_workbook(BytesIO(file_bytes), read_only=True)
        ws = wb.active

        # 读取表头，构建列索引 → 字段名映射
        header_row = next(ws.iter_rows(min_row=1, max_row=1, values_only=True))
        col_map = {}  # col_index -> field_name
        for idx, header in enumerate(header_row):
            if header is None:
                continue
            header_str = str(header).strip()
            field = COLUMN_MAP.get(header_str)
            if not field:
                field = COLUMN_MAP.get(header_str.lower())
            if field:
                col_map[idx] = field

        # 检查必填列是否存在
        required_fields = {"env_config_name", "device_name", "device_id"}
        found_fields = set(col_map.values())
        missing = required_fields - found_fields
        if missing:
            wb.close()
            raise ValueError(f"Excel 缺少必要列: {', '.join(missing)}")

        # 预加载环境配置，用于名称 → ID 查找
        env_configs = {}
        env_query = select(EnvConfig)
        env_result = await self.db.execute(env_query)
        for env in env_result.scalars().all():
            env_configs[env.name] = env.id

        # 逐行处理
        results = []
        success_count = 0
        failed_count = 0

        for row_idx, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
            # 提取本行数据
            row_data = {}
            for col_idx, field_name in col_map.items():
                if col_idx < len(row):
                    row_data[field_name] = row[col_idx]

            # 构建结果对象
            result = {
                "row": row_idx,
                "device_name": str(row_data.get("device_name", "") or ""),
                "device_id": str(row_data.get("device_id", "") or ""),
                "success": False,
                "error": "",
                "created_id": None,
            }

            # 跳过空行
            if not row_data.get("device_name") and not row_data.get("device_id"):
                continue

            # 校验必填字段
            if not row_data.get("env_config_name"):
                result["error"] = "环境配置名称为空"
                results.append(result)
                failed_count += 1
                continue
            if not row_data.get("device_name"):
                result["error"] = "设备名称为空"
                results.append(result)
                failed_count += 1
                continue
            if not row_data.get("device_id"):
                result["error"] = "设备ID为空"
                results.append(result)
                failed_count += 1
                continue

            # 环境配置名称 → ID
            env_name = str(row_data["env_config_name"]).strip()
            env_config_id = env_configs.get(env_name)
            if not env_config_id:
                result["error"] = f"环境配置不存在: {env_name}"
                results.append(result)
                failed_count += 1
                continue

            # 解析可选字段
            auto_reconnect = _parse_bool(row_data.get("auto_reconnect"), True)
            report_enabled = _parse_bool(row_data.get("report_enabled"), True)
            report_interval = _parse_int(row_data.get("report_interval"), 30)

            # 构建 topic_params
            topic_params = {}
            # 门口屏、无纸化：授权码（无纸化 rebootResponseTopic 的 authCode 来源）
            app_token = row_data.get("app_token")
            if app_token:
                topic_params["app_token"] = str(app_token).strip()
            # 中控：会议室编号（用于主题和消息体）
            room_id = row_data.get("room_id")
            if room_id:
                topic_params["room_id"] = str(room_id).strip()
            # 中控：设备编号（用于消息体 client_udid）
            # 设备编号 = device_id，存入 topic_params.device_number
            if device_type == "central":
                topic_params["device_number"] = str(row_data["device_id"]).strip()
            # 中控：主题类型
            central_device_type = row_data.get("central_device_type")
            if central_device_type:
                device_type_key = CENTRAL_DEVICE_TYPES.get(str(central_device_type).strip())
                if device_type_key:
                    topic_params["device_type"] = device_type_key
                else:
                    # 无效的主题类型，使用默认值
                    topic_params["device_type"] = "device_online"
            elif device_type == "central":
                # 中控设备但没有指定类型，默认为设备在线
                topic_params["device_type"] = "device_online"

            # 构建创建数据
            try:
                create_data = SimulatorCreate(
                    env_config_id=env_config_id,
                    device_type=device_type,
                    device_name=str(row_data["device_name"]).strip(),
                    device_id=str(row_data["device_id"]).strip(),
                    auto_reconnect=auto_reconnect,
                    topic_params=topic_params if topic_params else None,
                    report_config={
                        "enabled": report_enabled,
                        "interval": report_interval,
                    },
                )
            except Exception as e:
                result["error"] = f"数据验证失败: {str(e)}"
                results.append(result)
                failed_count += 1
                continue

            # 调用现有 create_simulator 创建设备
            try:
                device = await self.create_simulator(create_data)
                result["success"] = True
                result["created_id"] = device.id
                success_count += 1
            except ValueError as e:
                result["error"] = str(e)
                failed_count += 1
            except Exception as e:
                result["error"] = f"创建失败: {str(e)}"
                failed_count += 1

            results.append(result)

        wb.close()

        return {
            "total": success_count + failed_count,
            "success_count": success_count,
            "failed_count": failed_count,
            "results": results,
            "message": f"导入完成: 成功 {success_count}, 失败 {failed_count}",
        }

    async def update_simulator(self, device_id: str, data: SimulatorUpdate) -> Optional[DeviceSimulator]:
        """
        更新模拟设备

        Args:
            device_id: 设备 ID
            data: 更新数据

        Returns:
            DeviceSimulator: 更新后的设备对象
        """
        simulator = await self.get_simulator(device_id)
        if not simulator:
            return None

        update_data = data.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(simulator, key, value)

        simulator.updated_at = datetime.now()
        await self.db.flush()
        return simulator

    async def delete_simulator(self, device_id: str) -> bool:
        """
        删除模拟设备

        Args:
            device_id: 设备 ID

        Returns:
            bool: 是否删除成功
        """
        # 先停止设备
        await self.stop_simulator(device_id)

        simulator = await self.get_simulator(device_id)
        if not simulator:
            return False

        # 删除关联的上报记录
        delete_logs = delete(ReportLog).where(ReportLog.device_id == device_id)
        await self.db.execute(delete_logs)

        await self.db.delete(simulator)
        await self.db.flush()
        return True

    # ==================== 模拟设备控制 ====================

    async def start_simulator(self, device_id: str) -> bool:
        """
        启动模拟设备

        Args:
            device_id: 设备 ID

        Returns:
            bool: 是否启动成功
        """
        simulator = await self.get_simulator(device_id)
        if not simulator:
            raise ValueError(f"模拟设备不存在: {device_id}")

        config = await self.get_env_config(simulator.env_config_id)
        if not config:
            raise ValueError(f"关联环境配置不存在: {simulator.env_config_id}")

        # 先检查是否已在运行（用短锁快速判读，避免 IO 操作占用锁）
        with _running_simulators_lock:
            if device_id in _running_simulators and _running_simulators[device_id].is_running():
                return True

        # 确保 MQTT 已连接（IO 操作，放在锁外，避免批量启动时长锁竞争）
        if not mqtt_manager.is_connected(config.id):
            password = ""
            if config.password_encrypted:
                password = _decrypt_password(config.password_encrypted)

            client_id = f"{config.client_id_prefix}_{simulator.device_id[:8]}"
            connected = mqtt_manager.connect(
                env_config_id=config.id,
                host=config.broker_host,
                port=config.broker_port,
                username=config.username,
                password=password,
                client_id=client_id,
                use_tls=config.use_tls,
            )
            if not connected:
                raise ConnectionError("MQTT 连接失败，请检查环境配置")

            # 更新环境配置状态
            config.status = "connected"
            config.last_connected_at = datetime.now()

        # 创建并启动模拟器（IO 操作，放在锁外）
        # 兼容旧数据：topics 为空时，从 topic_prefix 自动生成默认映射
        topics = config.topics
        if not topics:
            prefix = config.topic_prefix or "device"
            topics = {
                "door": f"{prefix}/door",
                "paperless": f"{prefix}/paperless",
                "central": f"{prefix}/central",
                "client": f"{prefix}/client",
            }

        # 获取 token_api_host（从环境配置的 default_topic_params 中）
        token_api_host = None
        if config.default_topic_params:
            token_api_host = config.default_topic_params.get("token_api_host")

        sim = create_simulator(
            device_type=simulator.device_type,
            device_id=simulator.device_id,
            env_config_id=config.id,
            mqtt_manager=mqtt_manager,
            report_config=simulator.report_config,
            topics=topics,
            topic_params=simulator.topic_params,
            token_api_host=token_api_host,
        )
        if not sim:
            raise ValueError(f"不支持的设备类型: {simulator.device_type}")

        # 设置上报回调（记录日志）
        async def report_callback(**kwargs):
            try:
                await self._create_report_log(**kwargs)
            except Exception as e:
                logger.error(f"记录上报日志异常: {e}")

        sim.set_report_callback(lambda **kw: threading.Thread(
            target=lambda: self._sync_report_log(kw)
        ).start())

        success = sim.start()
        if success:
            # 启动成功后，用短锁写入内存表
            with _running_simulators_lock:
                _running_simulators[device_id] = sim
            simulator.status = "running"
            simulator.updated_at = datetime.now()
            try:
                await self.db.flush()
            except Exception as e:
                logger.warning(f"更新设备状态失败（模拟器已启动）: {device_id}, error={e}")
                # 不抛出异常，模拟器已在运行

        return success

    def _sync_report_log(self, kwargs: dict) -> None:
        """
        同步记录上报日志（在线程中执行）

        现在只做入队操作，实际的数据库写入在消费循环中完成。
        """
        try:
            # 从 kwargs 获取设备信息
            device_id = kwargs.get("device_id", "")

            # 从全局运行中的模拟器获取设备名称和类型
            with _running_simulators_lock:
                sim = _running_simulators.get(device_id)
                device_name = sim.device_name if sim else device_id
                device_type = sim.device_type if sim else "unknown"

            # 创建队列消息
            msg = QueueMessage(
                device_id=device_id,
                device_name=device_name,
                device_type=device_type,
                topic=kwargs.get("topic", ""),
                payload=kwargs.get("payload", {}),
                direction=kwargs.get("direction", "publish"),
                status=kwargs.get("status", "success"),
                error_message=kwargs.get("error_message"),
                timestamp=datetime.now(),
            )

            # 入队（非阻塞）
            enqueue_message(msg)

        except Exception as e:
            logger.error(f"消息入队异常: {e}")

    async def _create_report_log(self, device_id: str, topic: str,
                                  payload: dict, direction: str = "publish",
                                  status: str = "success",
                                  error_message: Optional[str] = None) -> None:
        """创建上报记录"""
        try:
            log = ReportLog(
                id=generate_id("log"),
                device_id=device_id,
                topic=topic,
                payload=payload,
                direction=direction,
                status=status,
                error_message=error_message,
            )
            self.db.add(log)

            # 更新设备统计
            query = select(DeviceSimulator).where(DeviceSimulator.id == device_id)
            result = await self.db.execute(query)
            sim = result.scalar_one_or_none()
            if sim:
                sim.total_reports = (sim.total_reports or 0) + 1
                sim.last_reported_at = datetime.now()

            await self.db.flush()

        except Exception as e:
            logger.error(f"创建上报记录异常: {e}")

        # TODO: WebSocket 推送回调在模拟器线程中执行有问题，暂时禁用
        # 需要重构为使用消息队列或异步任务来推送

    async def stop_simulator(self, device_id: str) -> bool:
        """
        停止模拟设备

        Args:
            device_id: 设备 ID

        Returns:
            bool: 是否停止成功
        """
        simulator = await self.get_simulator(device_id)
        if not simulator:
            return False

        with _running_simulators_lock:
            sim = _running_simulators.pop(device_id, None)
            if sim:
                sim.stop()

            simulator.status = "stopped"
            simulator.updated_at = datetime.now()
            await self.db.flush()

        return True

    async def manual_report(self, device_id: str, topic: str, payload: dict) -> bool:
        """
        手动触发上报

        Args:
            device_id: 设备 ID
            topic: 上报主题
            payload: 上报数据

        Returns:
            bool: 是否成功
        """
        simulator = await self.get_simulator(device_id)
        if not simulator:
            raise ValueError(f"模拟设备不存在: {device_id}")

        with _running_simulators_lock:
            sim = _running_simulators.get(device_id)
            if not sim or not sim.is_running():
                raise ValueError("设备未运行，请先启动")

            return sim.report(topic, payload)

    async def switch_device_type(self, device_id: str, device_type_name: str) -> dict:
        """
        切换中控设备上报类型

        Args:
            device_id: 设备 ID
            device_type_name: 设备类型标识（room_online/device_online/audio/video/control/network/power）

        Returns:
            dict: 切换结果
        """
        # 校验主题类型
        valid_types = ["room_online", "device_online", "audio", "video", "control", "network", "power"]
        if device_type_name not in valid_types:
            raise ValueError(f"无效的主题类型: {device_type_name}，支持的类型: {', '.join(valid_types)}")

        simulator = await self.get_simulator(device_id)
        if not simulator:
            raise ValueError(f"模拟设备不存在: {device_id}")

        if simulator.device_type != "central":
            raise ValueError("仅中控设备支持切换类型")

        # 检查是否正在运行
        was_running = False
        with _running_simulators_lock:
            sim = _running_simulators.get(device_id)
            if sim and sim.is_running():
                was_running = True

        # 如果正在运行，先停止
        if was_running:
            await self.stop_simulator(device_id)

        # 更新 topic_params
        topic_params = dict(simulator.topic_params or {})
        topic_params["device_type"] = device_type_name
        simulator.topic_params = topic_params
        simulator.updated_at = datetime.now()
        await self.db.flush()

        # 如果之前在运行，重新启动
        if was_running:
            await self.start_simulator(device_id)

        return {
            "device_id": device_id,
            "device_type": device_type_name,
            "was_running": was_running,
            "message": f"设备类型已切换为 {device_type_name}" + ("，设备已重启" if was_running else "")
        }

    # ==================== 批量操作 ====================

    async def batch_start_simulators(self, device_ids: List[str], max_concurrency: int = 50) -> dict:
        """
        批量启动模拟设备（线程池并行执行，提高大批量启动效率）

        使用 run_in_executor + ThreadPoolExecutor 实现真正的并行启动，
        每个设备在独立线程中运行，拥有独立的 asyncio 事件循环和数据库会话。
        避免 `asyncio.gather` 在同步 HTTP 调用（token API）时阻塞事件循环的问题。

        Args:
            device_ids: 设备 ID 列表
            max_concurrency: 最大并行数（默认 50，防止 Broker/Token API 过载）

        Returns:
            dict: {"success_count": int, "failed_count": int, "failed_ids": list, "message": str}
        """
        from concurrent.futures import ThreadPoolExecutor
        from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
        from app.config import settings

        if not device_ids:
            return {
                "success_count": 0,
                "failed_count": 0,
                "failed_ids": [],
                "message": "没有选择要启动的设备",
            }

        # 去重，避免同一设备重复启动
        unique_ids = list(dict.fromkeys(device_ids))

        # 关键：不能复用全局 async_session_maker/engine！
        # 全局 engine 的连接池内部是 asyncio.Queue（AsyncAdaptedQueuePool），
        # 在应用启动时（主事件循环 init_db）被首次使用后即绑定到主循环。
        # worker 线程如果用自己的新事件循环去 await 这个池子，会报
        # "Queue is bound to a different event loop"，导致批量启动大面积失败。
        # 因此每个 worker 线程必须使用独立 engine + 独立事件循环，
        # 线程内复用（ThreadPoolExecutor 会复用线程），批结束时统一释放。
        _thread_db = threading.local()          # 每个线程的 (loop, engine, maker)
        _created_db_ctxs = []                   # 记录本次批量启动创建的所有上下文
        _db_ctx_lock = threading.Lock()

        def _get_thread_db_ctx():
            """获取当前线程独立的事件循环 + 数据库引擎（首次创建，之后复用）"""
            ctx = getattr(_thread_db, "ctx", None)
            if ctx is None:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
                engine = create_async_engine(
                    settings.DATABASE_URL,
                    future=True,
                    connect_args={"check_same_thread": False} if "sqlite" in settings.DATABASE_URL else {},
                    pool_pre_ping="sqlite" in settings.DATABASE_URL,
                    pool_recycle=3600,
                    # 每个线程串行使用其引擎，pool_size=1 足够，避免连接数爆炸
                    pool_size=1,
                    max_overflow=0,
                )
                maker = async_sessionmaker(
                    engine, class_=AsyncSession, expire_on_commit=False, autoflush=False
                )
                ctx = {"loop": loop, "engine": engine, "maker": maker}
                _thread_db.ctx = ctx
                with _db_ctx_lock:
                    _created_db_ctxs.append(ctx)
            else:
                # 线程被复用：重新把该线程所属循环设为当前循环
                asyncio.set_event_loop(ctx["loop"])
            return ctx

        def _release_thread_db_ctxs():
            """批量启动结束后，在各自循环上释放引擎并关闭循环（线程池线程已退出）"""
            with _db_ctx_lock:
                ctxs = _created_db_ctxs[:]
                _created_db_ctxs.clear()
            for ctx in ctxs:
                loop, engine = ctx["loop"], ctx["engine"]
                try:
                    if not loop.is_closed():
                        loop.run_until_complete(engine.dispose())
                except Exception as e:
                    logger.warning(f"释放批量启动数据库引擎失败: {e}")
                finally:
                    try:
                        loop.close()
                    except Exception:
                        pass

        def _start_one_sync(device_id: str) -> bool:
            """
            在独立线程中启动单个设备（同步包装器）

            每个线程持有一个独立的事件循环和数据库引擎（见 _get_thread_db_ctx），
            整个线程生命周期内复用，规避全局连接池跨事件循环的绑定问题。
            """
            ctx = _get_thread_db_ctx()
            loop = ctx["loop"]
            try:
                async def _run():
                    async with ctx["maker"]() as session:
                        svc = DeviceSimService(session)
                        result = await svc.start_simulator(device_id)
                        await session.commit()
                        return result
                return loop.run_until_complete(_run())
            except Exception as e:
                logger.warning(f"批量启动设备失败: {device_id}, error={e}")
                return False

        loop = asyncio.get_event_loop()
        semaphore = asyncio.Semaphore(max_concurrency)

        # 使用线程池 + 信号量控制并发
        try:
            with ThreadPoolExecutor(max_workers=max_concurrency, thread_name_prefix="sim_start") as executor:
                async def _start_one(device_id: str) -> Tuple[str, bool]:
                    async with semaphore:
                        result = await loop.run_in_executor(executor, _start_one_sync, device_id)
                        return device_id, result

                results = await asyncio.gather(*[_start_one(did) for did in unique_ids])
        finally:
            # 无论成败都释放线程独立引擎，避免连接泄漏
            _release_thread_db_ctxs()

        success_count = 0
        failed_ids = []
        for device_id, success in results:
            if success:
                success_count += 1
            else:
                failed_ids.append(device_id)

        return {
            "success_count": success_count,
            "failed_count": len(failed_ids),
            "failed_ids": failed_ids,
            "message": f"批量启动完成: 成功 {success_count}, 失败 {len(failed_ids)}"
        }

    async def batch_stop_simulators(self, device_ids: List[str]) -> dict:
        """
        批量停止模拟设备

        Args:
            device_ids: 设备 ID 列表

        Returns:
            dict: {"success_count": int, "failed_count": int, "failed_ids": list, "message": str}
        """
        success_count = 0
        failed_ids = []

        for device_id in device_ids:
            try:
                result = await self.stop_simulator(device_id)
                if result:
                    success_count += 1
                else:
                    failed_ids.append(device_id)
            except Exception as e:
                logger.warning(f"批量停止设备失败: {device_id}, error={e}")
                failed_ids.append(device_id)

        return {
            "success_count": success_count,
            "failed_count": len(failed_ids),
            "failed_ids": failed_ids,
            "message": f"批量停止完成: 成功 {success_count}, 失败 {len(failed_ids)}"
        }

    async def start_all_by_device_type(self, device_type: str, env_config_id: Optional[str] = None) -> dict:
        """
        按设备类型批量启动所有设备（不分页，全量）

        Args:
            device_type: 设备类型筛选
            env_config_id: 可选，环境配置筛选

        Returns:
            dict: {"success_count": int, "failed_count": int, "failed_ids": list, "message": str}
        """
        # 查询该类型所有设备 ID
        conditions = [DeviceSimulator.device_type == device_type]
        if env_config_id:
            conditions.append(DeviceSimulator.env_config_id == env_config_id)

        query = select(DeviceSimulator.id).where(and_(*conditions))
        result = await self.db.execute(query)
        all_ids = [row[0] for row in result.all()]

        if not all_ids:
            return {
                "success_count": 0,
                "failed_count": 0,
                "failed_ids": [],
                "message": f"没有找到 {device_type} 类型的设备"
            }

        return await self.batch_start_simulators(all_ids)

    async def stop_all_by_device_type(self, device_type: str, env_config_id: Optional[str] = None) -> dict:
        """
        按设备类型批量停止所有设备（不分页，全量）

        Args:
            device_type: 设备类型筛选
            env_config_id: 可选，环境配置筛选

        Returns:
            dict: {"success_count": int, "failed_count": int, "failed_ids": list, "message": str}
        """
        # 查询该类型所有设备 ID
        conditions = [DeviceSimulator.device_type == device_type]
        if env_config_id:
            conditions.append(DeviceSimulator.env_config_id == env_config_id)

        query = select(DeviceSimulator.id).where(and_(*conditions))
        result = await self.db.execute(query)
        all_ids = [row[0] for row in result.all()]

        if not all_ids:
            return {
                "success_count": 0,
                "failed_count": 0,
                "failed_ids": [],
                "message": f"没有找到 {device_type} 类型的设备"
            }

        return await self.batch_stop_simulators(all_ids)

    async def batch_delete_simulators(self, device_ids: List[str]) -> dict:
        """
        批量删除模拟设备。

        先停止内存中的运行实例，再使用批量 SQL 删除上报记录和设备记录，
        避免逐台执行 MQTT 操作和数据库 flush 导致大批量删除长时间阻塞。
        """
        if not device_ids:
            return {
                "success_count": 0,
                "failed_count": 0,
                "failed_ids": [],
                "message": "没有选择要删除的设备",
            }

        # 去重，避免同一设备重复处理
        unique_ids = list(dict.fromkeys(device_ids))

        # 先确认数据库中实际存在的设备，避免把不存在的 ID 计为成功
        result = await self.db.execute(
            select(DeviceSimulator.id).where(DeviceSimulator.id.in_(unique_ids))
        )
        existing_ids = [row[0] for row in result.all()]
        existing_id_set = set(existing_ids)
        failed_ids = [device_id for device_id in unique_ids if device_id not in existing_id_set]

        # 停止运行中的实例。MQTT 停止异常不能阻断数据库删除。
        for device_id in existing_ids:
            with _running_simulators_lock:
                sim = _running_simulators.pop(device_id, None)
            if sim:
                try:
                    sim.stop()
                except Exception as e:
                    logger.warning(f"批量删除停止设备失败，继续删除数据库记录: {device_id}, error={e}")

        if existing_ids:
            # ReportLog.device_id 保存的是模拟器主键，和 DeviceSimulator.id 一致。
            await self.db.execute(
                delete(ReportLog).where(ReportLog.device_id.in_(existing_ids))
            )
            await self.db.execute(
                delete(DeviceSimulator).where(DeviceSimulator.id.in_(existing_ids))
            )
            await self.db.flush()

        success_count = len(existing_ids)
        return {
            "success_count": success_count,
            "failed_count": len(failed_ids),
            "failed_ids": failed_ids,
            "message": f"批量删除完成: 成功 {success_count}, 失败 {len(failed_ids)}",
        }

    # ==================== 上报记录 ====================

    async def list_report_logs(
        self,
        device_id: Optional[str] = None,
        direction: Optional[str] = None,
        skip: int = 0,
        limit: int = 100,
        start_time: Optional[datetime] = None,
        end_time: Optional[datetime] = None,
    ) -> Tuple[List[ReportLog], int]:
        """
        获取上报记录列表

        Args:
            device_id: 设备 ID 筛选
            direction: 方向筛选 (publish/subscribe)
            skip: 跳过的记录数
            limit: 返回记录数上限
            start_time: 开始时间
            end_time: 结束时间

        Returns:
            Tuple[list, int]: (记录列表, 总数)
        """
        conditions = []
        if device_id:
            conditions.append(ReportLog.device_id == device_id)
        if direction:
            conditions.append(ReportLog.direction == direction)
        if start_time:
            conditions.append(ReportLog.created_at >= start_time)
        if end_time:
            conditions.append(ReportLog.created_at <= end_time)

        # 查询总数
        count_query = select(func.count()).select_from(ReportLog)
        if conditions:
            count_query = count_query.where(and_(*conditions))
        total_result = await self.db.execute(count_query)
        total = total_result.scalar() or 0

        # 查询列表
        query = select(ReportLog).order_by(ReportLog.created_at.desc())
        if conditions:
            query = query.where(and_(*conditions))
        query = query.offset(skip).limit(limit)
        result = await self.db.execute(query)
        items = list(result.scalars().all())

        return items, total

    async def cleanup_report_logs(self, days: int = 7) -> int:
        """
        清理过期上报记录

        Args:
            days: 保留天数

        Returns:
            int: 清理的记录数
        """
        cutoff = datetime.now() - timedelta(days=days)
        query = delete(ReportLog).where(ReportLog.created_at < cutoff)
        result = await self.db.execute(query)
        return result.rowcount or 0


# ==================== 消息消费循环 ====================

async def message_consumer_loop():
    """
    消息消费循环（在主线程异步循环中运行）

    从线程安全的消息队列中取出消息，写入数据库并触发 WebSocket 推送。
    使用 poll 模式（get_nowait + sleep）替代阻塞 await queue.get()，
    因为队列是 thread_queue.Queue，不支持 asyncio 的 await。
    """
    from app.database import async_session_maker

    logger.info("消息消费循环已启动")

    while True:
        try:
            # 从队列获取消息（非阻塞 poll）
            queue = get_message_queue()
            try:
                msg: QueueMessage = queue.get_nowait()
            except thread_queue.Empty:
                await asyncio.sleep(0.05)
                continue

            # 创建数据库 session
            async with async_session_maker() as db:
                try:
                    # 写入上报记录
                    log = ReportLog(
                        id=generate_id("log"),
                        device_id=msg.device_id,
                        topic=msg.topic,
                        payload=msg.payload,
                        direction=msg.direction,
                        status=msg.status,
                        error_message=msg.error_message,
                    )
                    db.add(log)

                    # 更新设备统计
                    query = select(DeviceSimulator).where(DeviceSimulator.device_id == msg.device_id)
                    result = await db.execute(query)
                    sim = result.scalar_one_or_none()
                    if sim:
                        sim.total_reports = (sim.total_reports or 0) + 1
                        sim.last_reported_at = datetime.now()
                        # 更新设备名称和类型（用于 WebSocket 推送）
                        msg.device_name = sim.device_name
                        msg.device_type = sim.device_type

                    await db.commit()
                    with _queue_stats_lock:
                        _queue_stats["processed"] += 1

                except Exception as e:
                    await db.rollback()
                    logger.error(f"写入上报记录失败: {e}")

            # WebSocket 推送（不阻塞消费循环）
            if _message_callback:
                try:
                    await _message_callback(
                        device_id=msg.device_id,
                        device_name=msg.device_name,
                        device_type=msg.device_type,
                        topic=msg.topic,
                        direction=msg.direction,
                        payload=msg.payload,
                    )
                except Exception as e:
                    logger.debug(f"WebSocket 推送失败: {e}")

        except asyncio.CancelledError:
            logger.info("消息消费循环已停止")
            break
        except Exception as e:
            logger.error(f"消息消费循环异常: {e}")
            await asyncio.sleep(1)  # 避免快速失败循环


def get_queue_stats() -> dict:
    """获取队列统计信息"""
    queue = get_message_queue()
    with _queue_stats_lock:
        return {
            "queue_size": queue.qsize(),
            "enqueued": _queue_stats["enqueued"],
            "processed": _queue_stats["processed"],
            "dropped": _queue_stats["dropped"],
        }


# ==================== Excel 导入辅助函数 ====================

def _parse_bool(value, default: bool = True) -> bool:
    """Parse boolean from Excel cell value (supports Chinese 是/否)"""
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    s = str(value).strip().lower()
    if s in ("是", "true", "1", "yes", "y"):
        return True
    if s in ("否", "false", "0", "no", "n"):
        return False
    return default


def _parse_int(value, default: int = 0) -> int:
    """Parse integer from Excel cell value"""
    if value is None:
        return default
    try:
        return int(value)
    except (ValueError, TypeError):
        return default