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

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

import json
import logging
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: asyncio.Queue = None
_queue_stats = {"enqueued": 0, "processed": 0, "dropped": 0}


def get_message_queue() -> asyncio.Queue:
    """获取消息队列实例"""
    global _message_queue
    if _message_queue is None:
        _message_queue = asyncio.Queue(maxsize=10000)
    return _message_queue


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

    Returns:
        bool: 是否入队成功
    """
    queue = get_message_queue()
    try:
        queue.put_nowait(msg)
        _queue_stats["enqueued"] += 1
        return True
    except asyncio.QueueFull:
        _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_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 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}")

        with _running_simulators_lock:
            if device_id in _running_simulators and _running_simulators[device_id].is_running():
                return True

            # 确保 MQTT 已连接
            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()

            # 创建并启动模拟器，传递多主题配置
            # 兼容旧数据：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",
                }

            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,
            )
            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:
                _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 batch_start_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.start_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 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 batch_delete_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.delete_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 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 推送。
    """
    from app.database import async_session_maker

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

    while True:
        try:
            # 从队列获取消息（阻塞等待）
            queue = get_message_queue()
            msg: QueueMessage = await queue.get()

            # 创建数据库 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.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()
                    _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()
    return {
        "queue_size": queue.qsize(),
        "enqueued": _queue_stats["enqueued"],
        "processed": _queue_stats["processed"],
        "dropped": _queue_stats["dropped"],
    }