dzxj_dtu/backend/modules/mqtt_client.py

import paho.mqtt.client as mqtt
import threading
import json
import time
import logging
import ssl
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple, Any

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('mqtt_client')

@dataclass
class MQTTConfig:
    """MQTT配置数据类"""
    broker: str
    port: int = 1883
    username: str = ""
    password: str = ""
    client_id: str = ""
    keepalive: int = 60
    qos: int = 0
    # TLS/SSL 配置
    use_tls: bool = False
    tls_version: int = ssl.PROTOCOL_TLSv1_2
    ca_certs: Optional[str] = None
    certfile: Optional[str] = None
    keyfile: Optional[str] = None
    # 重连配置
    reconnect_delay: float = 1.0
    reconnect_delay_max: float = 120.0
    reconnect_exponential_backoff: bool = True
    max_reconnect_attempts: int = 0  # 0 表示无限尝试
    # 消息配置
    will_topic: Optional[str] = None
    will_payload: Optional[str] = None
    will_qos: int = 0
    will_retain: bool = False

class MQTTClient:
    """MQTT客户端封装类，提供高级MQTT通信功能"""
    
    def __init__(self):
        self.client = None
        self.is_connected = False
        self.config = None
        self.topics = []  # 存储主题和QoS的元组列表
        self.data_callback = None
        self.status_callback = None
        self.error_callback = None
        self.connected_event = threading.Event()
        self.lock = threading.RLock()
        self.reconnect_attempts = 0
        self.last_reconnect_time = 0
        self.last_will_set = False
    
    def connect(self, broker: str, port: int = 1883, username: str = "", 
                password: str = "", client_id: str = "", use_tls: bool = False, 
                qos: int = 0, **kwargs) -> Tuple[bool, str]:
        """
        连接到MQTT服务器
        
        Args:
            broker: MQTT服务器地址
            port: MQTT服务器端口
            username: 用户名
            password: 密码
            client_id: 客户端ID，为空时自动生成
            use_tls: 是否使用TLS加密
            qos: 默认QoS级别
            **kwargs: 其他配置参数
            
        Returns:
            tuple: (是否成功, 消息)
        """
        try:
            # 构建配置
            config = MQTTConfig(
                broker=broker,
                port=port,
                username=username,
                password=password,
                client_id=client_id or f"serial_mqtt_gateway_{time.time()}",
                use_tls=use_tls,
                qos=qos,
                **kwargs
            )
            
            with self.lock:
                # 断开现有连接
                if self.is_connected:
                    self.disconnect()
                
                logger.info(f"尝试连接MQTT服务器: {broker}:{port}")
                
                # 创建MQTT客户端
                self.client = mqtt.Client(
                    client_id=config.client_id,
                    clean_session=True,
                    userdata=None
                )
                
                # 设置回调函数
                self.client.on_connect = self._on_connect
                self.client.on_disconnect = self._on_disconnect
                self.client.on_message = self._on_message
                self.client.on_publish = self._on_publish
                self.client.on_subscribe = self._on_subscribe
                
                # 设置重连参数（兼容 paho-mqtt 1.x 与 2.x）
                try:
                    self.client.reconnect_delay_set(
                        min_delay=config.reconnect_delay,
                        max_delay=config.reconnect_delay_max,
                        exponential_backoff=config.reconnect_exponential_backoff
                    )
                except TypeError:
                    # paho-mqtt 1.x 的 reconnect_delay_set 不支持 exponential_backoff
                    self.client.reconnect_delay_set(
                        min_delay=config.reconnect_delay,
                        max_delay=config.reconnect_delay_max
                    )
                
                # 设置用户名密码
                if config.username and config.password:
                    self.client.username_pw_set(config.username, config.password)
                    logger.debug(f"已设置MQTT用户名密码认证")
                
                # 配置TLS
                if config.use_tls:
                    try:
                        self.client.tls_set(
                            ca_certs=config.ca_certs,
                            certfile=config.certfile,
                            keyfile=config.keyfile,
                            tls_version=config.tls_version
                        )
                        logger.debug(f"已配置MQTT TLS连接")
                    except Exception as e:
                        error_msg = f"配置TLS失败: {str(e)}"
                        logger.error(error_msg)
                        return False, error_msg
                
                # 设置遗嘱消息
                if config.will_topic:
                    will_payload = config.will_payload or "{\"status\":\"offline\"}"
                    self.client.will_set(
                        topic=config.will_topic,
                        payload=will_payload,
                        qos=config.will_qos,
                        retain=config.will_retain
                    )
                    self.last_will_set = True
                    logger.debug(f"已设置MQTT遗嘱消息: {config.will_topic}")
                
                # 连接到服务器
                self.client.connect(
                    host=config.broker,
                    port=config.port,
                    keepalive=config.keepalive
                )
                
                # 启动客户端循环
                self.client.loop_start()
                
                # 保存配置
                self.config = config
                
                # 等待连接成功或超时
                connected = self.connected_event.wait(timeout=10)
                
                if connected:
                    logger.info(f"已连接到MQTT服务器 {broker}:{port}")
                    self.reconnect_attempts = 0
                    self.last_reconnect_time = time.time()
                    
                    if self.status_callback:
                        self.status_callback(True)
                    
                    # 重新订阅之前的主题
                    if self.topics:
                        self._resubscribe_topics()
                    
                    return True, f"已连接到MQTT服务器 {broker}:{port}"
                else:
                    error_msg = "连接MQTT服务器超时"
                    logger.error(error_msg)
                    self.client.loop_stop()
                    self.client = None
                    
                    if self.status_callback:
                        self.status_callback(False)
                    
                    return False, error_msg
                
        except Exception as e:
            error_msg = f"连接失败: {str(e)}"
            logger.error(error_msg)
            
            if self.status_callback:
                self.status_callback(False)
            if self.error_callback:
                self.error_callback(error_msg)
            
            return False, error_msg
    
    def disconnect(self) -> Tuple[bool, str]:
        """
        断开MQTT连接
        
        Returns:
            tuple: (是否成功, 消息)
        """
        try:
            with self.lock:
                logger.info("断开MQTT连接")
                
                if self.client:
                    # 发布在线状态（如果设置了遗嘱消息）
                    if self.last_will_set and self.is_connected:
                        try:
                            self.client.publish(
                                topic=self.config.will_topic,
                                payload="{\"status\":\"online\"}",
                                qos=self.config.will_qos,
                                retain=self.config.will_retain
                            )
                            time.sleep(0.1)  # 给发布消息一些时间
                        except Exception as e:
                            logger.warning(f"发布离线状态失败: {str(e)}")
                    
                    # 停止循环并断开连接
                    self.client.loop_stop()
                    try:
                        self.client.disconnect()
                    except Exception as e:
                        logger.error(f"断开连接时出错: {str(e)}")
                    finally:
                        self.client = None
                
                self.is_connected = False
                self.connected_event.clear()
                
                if self.status_callback:
                    self.status_callback(False)
                
            return True, "已断开MQTT连接"
            
        except Exception as e:
            error_msg = f"断开连接失败: {str(e)}"
            logger.error(error_msg)
            return False, error_msg
    
    def subscribe(self, topics, qos: Optional[int] = None) -> Tuple[bool, str]:
        """
        订阅主题
        
        Args:
            topics: 主题字符串或列表
            qos: QoS级别，如果为None则使用配置中的默认值
            
        Returns:
            tuple: (是否成功, 消息)
        """
        try:
            with self.lock:
                if not self.client or not self.is_connected:
                    return False, "MQTT未连接"
                
                # 确定QoS级别
                qos_level = qos if qos is not None else (self.config.qos if self.config else 0)
                
                # 格式化主题列表
                if isinstance(topics, str):
                    topic_list = [(topics, qos_level)]
                    topic_names = [topics]
                else:
                    topic_list = [(topic, qos_level) for topic in topics]
                    topic_names = topics
                
                # 订阅主题
                result, _ = self.client.subscribe(topic_list)
                
                if result == mqtt.MQTT_ERR_SUCCESS:
                    # 更新本地主题列表
                    self.topics = topic_list
                    logger.info(f"已订阅主题: {', '.join(topic_names)}, QoS: {qos_level}")
                    return True, f"已订阅主题: {', '.join(topic_names)}"
                else:
                    error_msg = f"订阅失败: {mqtt.error_string(result)}"
                    logger.error(error_msg)
                    return False, error_msg
                    
        except Exception as e:
            error_msg = f"订阅失败: {str(e)}"
            logger.error(error_msg)
            return False, error_msg
    
    def publish(self, topic: str, message: Any, qos: Optional[int] = None, 
                retain: bool = False) -> Tuple[bool, str]:
        """
        发布消息
        
        Args:
            topic: 发布主题
            message: 消息内容（自动转换为JSON字符串）
            qos: QoS级别，如果为None则使用配置中的默认值
            retain: 是否为保留消息
            
        Returns:
            tuple: (是否成功, 消息)
        """
        try:
            with self.lock:
                if not self.client or not self.is_connected:
                    return False, "MQTT未连接"
                
                # 确定QoS级别
                qos_level = qos if qos is not None else (self.config.qos if self.config else 0)
                
                # 序列化消息
                if isinstance(message, (dict, list)):
                    payload = json.dumps(message)
                else:
                    payload = str(message)
                
                # 发布消息
                result = self.client.publish(
                    topic=topic,
                    payload=payload,
                    qos=qos_level,
                    retain=retain
                )
                
                # 检查发布是否成功
                if result.rc == mqtt.MQTT_ERR_SUCCESS:
                    logger.debug(f"消息已发布到主题 {topic}: {payload[:50]}...")
                    # 上行消息也转发到前端 MQTT 数据监控页面显示
                    if self.data_callback:
                        try:
                            self.data_callback({
                                'topic': topic,
                                'payload': payload,
                                'qos': qos_level,
                                'retain': retain,
                                'timestamp': time.time(),
                                'direction': 'uplink'
                            })
                        except Exception as e:
                            logger.debug(f"转发上行消息到数据回调失败: {e}")
                    return True, f"消息已发布到主题 {topic}"
                else:
                    error_msg = f"发布失败: {mqtt.error_string(result.rc)}"
                    logger.error(error_msg)
                    return False, error_msg
                    
        except Exception as e:
            error_msg = f"发布失败: {str(e)}"
            logger.error(error_msg)
            return False, error_msg
    
    def unsubscribe(self, topics) -> Tuple[bool, str]:
        """
        取消订阅主题
        
        Args:
            topics: 主题字符串或列表
            
        Returns:
            tuple: (是否成功, 消息)
        """
        try:
            with self.lock:
                if not self.client or not self.is_connected:
                    return False, "MQTT未连接"
                
                # 格式化主题列表
                if isinstance(topics, str):
                    topic_list = [topics]
                else:
                    topic_list = topics
                
                # 取消订阅
                result = self.client.unsubscribe(topic_list)
                
                if result.rc == mqtt.MQTT_ERR_SUCCESS:
                    # 更新本地主题列表
                    self.topics = [(t, q) for t, q in self.topics if t not in topic_list]
                    logger.info(f"已取消订阅主题: {', '.join(topic_list)}")
                    return True, f"已取消订阅主题: {', '.join(topic_list)}"
                else:
                    error_msg = f"取消订阅失败: {mqtt.error_string(result.rc)}"
                    logger.error(error_msg)
                    return False, error_msg
                    
        except Exception as e:
            error_msg = f"取消订阅失败: {str(e)}"
            logger.error(error_msg)
            return False, error_msg
    
    def _resubscribe_topics(self):
        """重新订阅所有已保存的主题"""
        if self.topics:
            try:
                topic_names = [topic for topic, _ in self.topics]
                logger.info(f"重新订阅之前的主题: {', '.join(topic_names)}")
                self.client.subscribe(self.topics)
            except Exception as e:
                logger.error(f"重新订阅主题失败: {str(e)}")
    
    def _on_connect(self, client, userdata, flags, rc):
        """连接回调函数"""
        if rc == 0:
            self.is_connected = True
            self.connected_event.set()
            logger.info(f"成功连接到MQTT服务器: {self.config.broker}:{self.config.port}")
            
            # 重置重连计数
            self.reconnect_attempts = 0
            self.last_reconnect_time = time.time()
        else:
            self.is_connected = False
            self.connected_event.clear()
            error_msg = f"连接MQTT服务器失败: {mqtt.connack_string(rc)} (代码: {rc})"
            logger.error(error_msg)
            
            if self.error_callback:
                self.error_callback(error_msg)
    
    def _on_disconnect(self, client, userdata, rc):
        """断开连接回调函数"""
        self.is_connected = False
        self.connected_event.clear()
        
        # 区分主动断开和意外断开
        if rc != 0:
            error_msg = f"意外断开MQTT连接 (代码: {rc})"
            logger.warning(error_msg)
            
            # 处理重连逻辑
            self._handle_reconnect()
            
            if self.error_callback:
                self.error_callback(error_msg)
        else:
            logger.info("主动断开MQTT连接")
        
        if self.status_callback:
            self.status_callback(False)
    
    def _on_message(self, client, userdata, msg):
        """消息接收回调函数"""
        try:
            # 尝试多种编码解码
            encodings = ['utf-8', 'latin-1', 'ascii']
            payload = None
            for encoding in encodings:
                try:
                    payload = msg.payload.decode(encoding)
                    break
                except UnicodeDecodeError:
                    continue
            
            # 如果都失败，转为十六进制
            if payload is None:
                payload = msg.payload.hex()
                logger.warning(f"收到无法解码的二进制消息，已转为十六进制: {payload[:50]}...")
            
            # 构建消息数据结构
            data = {
                'topic': msg.topic,
                'payload': payload,
                'qos': msg.qos,
                'retain': msg.retain,
                'timestamp': time.time()
            }
            
            logger.debug(f"收到MQTT消息: 主题={msg.topic}, 长度={len(msg.payload)}字节")
            
            if self.data_callback:
                self.data_callback(data)
        except Exception as e:
            error_msg = f"处理MQTT消息错误: {str(e)}"
            logger.error(error_msg)
            if self.error_callback:
                self.error_callback(error_msg)
    
    def _on_publish(self, client, userdata, mid):
        """发布回调函数"""
        logger.debug(f"消息发布成功，消息ID: {mid}")
    
    def _on_subscribe(self, client, userdata, mid, granted_qos):
        """订阅回调函数"""
        logger.debug(f"主题订阅成功，消息ID: {mid}, 授权QoS: {granted_qos}")
    
    def _handle_reconnect(self):
        """处理重连逻辑"""
        if not self.config or self.config.max_reconnect_attempts == 0:
            # 如果未设置最大重连次数或为0，则无限重连
            return
        
        self.reconnect_attempts += 1
        if self.reconnect_attempts > self.config.max_reconnect_attempts:
            logger.error(f"已达到最大重连次数 ({self.config.max_reconnect_attempts})，停止重连")
            # 可以在这里调用断开连接或通知上层
            self.disconnect()
    
    def set_data_callback(self, callback):
        """设置数据接收回调函数"""
        self.data_callback = callback
    
    def set_status_callback(self, callback):
        """设置状态变化回调函数"""
        self.status_callback = callback
    
    def set_error_callback(self, callback):
        """设置错误回调函数"""
        self.error_callback = callback
    
    def get_status(self) -> Dict[str, Any]:
        """
        获取当前连接状态
        
        Returns:
            dict: 包含连接状态和详细信息
        """
        with self.lock:
            status = {
                'connected': self.is_connected,
                'broker': self.config.broker if self.config else None,
                'port': self.config.port if self.config else None,
                'client_id': self.config.client_id if self.config else None,
                'topics': [topic for topic, _ in self.topics],
                'reconnect_attempts': self.reconnect_attempts,
                'last_reconnect_time': self.last_reconnect_time
            }
            return status
    
    def get_config(self) -> Optional[MQTTConfig]:
        """获取当前MQTT配置"""
        return self.config

# 创建全局MQTT实例
global_mqtt = MQTTClient()