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

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

import json
import logging
import threading
from datetime import datetime, timedelta
from typing import Optional, List, Tuple

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__)

# 密码加密密钥（使用 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:
            item.status = mqtt_manager.get_connection_status(item.id)

        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: 测试结果
        """
        config = await self.get_env_config(config_id)
        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,
        )

        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.metadata or {},
            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,
                topics=topics,
            )
            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()
                await self.db.flush()

            return success

    def _sync_report_log(self, kwargs: dict) -> None:
        """同步记录上报日志（在线程中执行）"""
        import asyncio
        try:
            # 使用 asyncio.run 创建新的事件循环
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            loop.run_until_complete(self._create_report_log(**kwargs))
            loop.close()
        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}")

    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 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