Mqtt_Send.py

import json
import logging
import csv
import os
import re
import threading
import time
from time import sleep
from datetime import datetime
import paho.mqtt.client as mqtt

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

class Mqtt:
    def __init__(self, broker_address, port):
        """
        初始化 MQTT 客户端

        :param broker_address: MQTT 代理地址
        :param port: MQTT 代理端口
        """
        self.lock = None
        self.broker_address = broker_address
        self.port = port
        self.client = None
        self._received_message_lock = threading.Lock()
        self.received_message = None
        self.message_type = None

    def connect(self):
        """
        连接到 MQTT 服务器
        """
        try:
            self.client = mqtt.Client()
            self.client.on_connect = self.on_connect  # 设置连接回调
            self.client.on_message = self.on_message  # 设置消息回调
            self.client.connect(self.broker_address, self.port)  # 连接到代理
            self.client.loop_start()  # 启动网络循环
        except Exception as e:
            logging.error(f"连接到MQTT服务器时发生错误: {e}")
            raise

    def disconnect(self):
        """
        断开与 MQTT 服务器的连接
        """
        if self.client:
            try:
                self.client.loop_stop()  # 停止网络循环
                self.client.disconnect()  # 断开连接
                logging.info("已断开与MQTT 服务器的连接")
            except Exception as e:
                logging.error(f"断开与MQTT 服务器的连接时发生错误: {e}")
            finally:
                self.client = None  # 确保资源被完全释放
        else:
            logging.warning("尝试断开连接时，客户端已不存在")

    def on_connect(self, client, userdata, flags, rc):
        """
        连接成功或失败的回调函数

        :param client: 客户端实例
        :param userdata: 用户数据
        :param flags: 连接标志
        :param rc: 返回码
        """
        try:
            if rc == 0:
                logging.info("连接成功")
            else:
                logging.error(f"连接失败，返回码: {rc}")
                # 根据不同的返回码采取不同的措施
                if rc == 1:
                    logging.error("错误：错误的协议版本")
                elif rc == 2:
                    logging.error("错误：无效的客户端标识符")
                elif rc == 3:
                    logging.error("错误：服务器不可用")
                elif rc == 4:
                    logging.error("错误：错误的用户名或密码")
                elif rc == 5:
                    logging.error("错误：未授权")
                else:
                    logging.error("未知错误")
        except Exception as e:
            logging.exception(f"在处理连接结果时发生异常: {e}")

    def on_message(self, client, userdata, msg):
        """
        接收到消息的回调函数

        :param client: 客户端实例
        :param userdata: 用户数据
        :param msg: 消息对象
        """
        try:
            payload = msg.payload.decode('utf-8', errors='replace')  # 处理解码错误
            logging.info(f"收到消息: {msg.topic} {payload[:50]}...")  # 脱敏日志记录
        except UnicodeDecodeError as e:
            logging.error(f"解码错误: {e}")
            payload = "无法解码的消息"

        with self._received_message_lock:
            self.received_message = payload  # 线程安全地存储接收到的消息

    def subscribe(self, topic):
        """
        订阅指定的主题

        :param topic: 主题名称
        """
        if self.client is None:
            logging.error("客户端未初始化，无法订阅主题")
            raise ValueError("客户端未初始化，无法订阅主题")

        # 输入验证
        if not isinstance(topic, str) or not topic.strip():
            logging.error("无效的主题名称")
            raise ValueError("无效的主题名称")

        try:
            self.client.subscribe(topic)
            logging.info(f"已订阅主题: {topic}")
        except ConnectionError as ce:
            logging.error(f"连接错误，无法订阅主题: {topic}, 错误信息: {str(ce)}")
            raise
        except TimeoutError as te:
            logging.error(f"超时错误，无法订阅主题: {topic}, 错误信息: {str(te)}")
            raise
        except ValueError as ve:
            logging.error(f"值错误，无法订阅主题: {topic}, 错误信息: {str(ve)}")
            raise
        except Exception as e:
            logging.error(f"未知错误，无法订阅主题: {topic}, 错误信息: {str(e)}")
            raise

    def set_message_type(self, message_type):
        """
        设置消息类型
        此方法用于设置或更改消息类型属性，允许对象根据需要处理不同类型的消息
        参数:
        message_type: 要设置的消息类型，可以是任何数据类型，但通常应该是字符串、整数或枚举类型
        返回:
        无
        """
        self.message_type = message_type

    def publish(self, topic, message):
        """
        发布消息到指定的主题

        :param topic: 主题名称
        :param message: 消息内容
        """
        if self.client:
            try:
                # 将消息转换为JSON字符串
                if self.message_type == dict and isinstance(message, dict):
                    message = json.dumps(message)
                elif message is None:
                    message = ""
                else:
                    message = str(message)
            except (TypeError, ValueError) as e:
                logging.error(f"{datetime.now()} - 消息转换失败: {e} - 调用者: {self.__class__.__name__}.publish")
                raise

            try:
                self.client.publish(topic, message)
                logging.info(f"{datetime.now()} - 已发布消息到主题: {topic} - 调用者: {self.__class__.__name__}.publish")
            except Exception as e:
                logging.error(f"{datetime.now()} - 消息发布失败: {e} - 失败的主题: {topic}, 消息: {message} - 调用者: {self.__class__.__name__}.publish")
                raise

    def wait_for_message(self, topic, timeout=5):
        """
        等待指定主题的消息

        :param topic: 主题名称
        :param timeout: 超时时间（秒）
        :return: 接收到的消息或 None
        """
        if self.client is None:
            logging.warning("Client is not initialized")
            return None

        if timeout < 0:
            logging.warning("Timeout cannot be negative")
            return None

        start_time = time.monotonic()
        while (time.monotonic() - start_time) < timeout:
            try:
                with self.lock:
                    if self.received_message is not None:
                        return self.received_message
            except Exception as e:
                logging.error(f"Error accessing received_message: {e}")
            try:
                time.sleep(0.1)
            except Exception as e:
                logging.error(f"Error in sleep: {e}")
        return None

    @staticmethod
    def read_config_from_csv(file_path):
        """
        从 CSV 文件读取配置

        :param file_path: CSV 文件路径
        :param allowed_directory: 允许访问的目录
        :return: 配置列表
        """
        try:
            # 验证文件路径
            if not os.path.isfile(file_path):
                raise FileNotFoundError(f"文件 {file_path} 不存在")

            with open(file_path, mode='r', encoding='utf-8') as file:
                reader = csv.DictReader(file)
                return [row for row in reader]
        except FileNotFoundError as e:
            print(f"错误: {e}")
            return []
        except PermissionError as e:
            print(f"错误: {e}")
            return []
        except Exception as e:
            print(f"未知错误: {e}")
            return []

    @staticmethod
    def wait_for_message(self, topic, timeout=5):
        """
        等待指定主题的消息

        :param topic: 主题名称
        :param timeout: 超时时间（秒）
        :return: 接收到的消息或 None
        """
        if not isinstance(topic, str) or not re.match(r'^[a-zA-Z0-9_\-]+$', topic):
            raise ValueError("Invalid topic format")

        if timeout < 0:
            return None

        try:
            if self.client:
                start_time = time.time()
                while (time.time() - start_time) < timeout:
                    if self.has_received_message():
                        return self.received_message
                    sleep(0.1)
                return None
        except AttributeError:
            return None
        return None

    def has_received_message(self):
        return hasattr(self, 'received_message')

    @staticmethod
    def read_config_from_csv(file_path):
        """
        从 CSV 文件读取配置

        :param file_path: CSV 文件路径
        :return: 配置列表
        """
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"文件 {file_path} 不存在")

        try:
            with open(file_path, mode='r', encoding='utf-8') as file:
                reader = csv.DictReader(file)
                config_list = [row for row in reader]
                if not config_list:
                    raise ValueError("CSV 文件内容为空或格式不正确")
                logging.info(f"成功读取文件 {file_path}")
                return config_list
        except Exception as e:
            logging.error(f"读取文件 {file_path} 时发生错误: {e}")
            return []

    @staticmethod
    def build_message(config, current_time,topic):
        """
        构建消息内容

        :param config: 配置字典
        :param current_time: 当前时间
        :return: 消息字典
        """
        #安卓信息设备上报
        if topic == "rebootResponseTopic":
            return {
                "method": "/system/readSystemInfo",
                "clientId": config['clientId'],
                "result": json.dumps({
                    "result": {
                        "buildInfo": {
                            "appToken": config['appToken'],
                            "companyNumber": config['companyNumber'],
                            "cnum": config['cnum'],
                            "conferenceName": "测试会议室",
                            "conferenceId": int(config['conferenceId']),
                            "defaultQrCodeUrl": "http://192.168.5.218:8888/group1/M00/00/21/wKgFyGNBWZmADtnNAAAwrxR0X8s016.png",
                            "aliasName": "zt",
                            "serverBaseUrl": "http://192.168.5.218:8996/",
                            "localBindTime": current_time,
                            "generalField": "{\"conferencePhone\":\"\",\"chooseTimeType\":1,\"meetingTopicSwitch\":\"1\",\"meetingContentSwitch\":\"1\",\"meetingReverseTypeSwitch\":\"1\",\"seatArrangeSwitch\":\"1\",\"meetingVoteSwitch\":\"1\",\"floorPlanPath\":\"\",\"jumpToPaperless\":2,\"approvalList\":[],\"isLeaderJoin\":false,\"meetingPublishScreenSwitch\":\"1\"}"
                        },
                        "faceVersion": "4.2.12021020201.1",
                        "wgtVersion": "0.0.81",
                        "deviceModel": "yx_rk3288",
                        "abiList": ["armeabi-v7a", "armeabi"],
                        "androidId": "48134e6047a19aaf",
                        "appName": "UBAINS",
                        "appPackageName": "com.ubains.local.gviewer",
                        "appVersion": 78,
                        "appVersionName": "1.1.78",
                        "battery": 0,
                        "bluetoothEnabled": False,
                        "camerasCount": 1,
                        "charging": True,
                        "density": 1,
                        "freeAndTotalMemory": "1176M/1997M",
                        "internalAvailableSize": 4306395136,
                        "internalTotalSize": 4877451264,
                        "ipAddress": "192.168.5.129",
                        "macAddress": config['macAddress'],
                        "networkType": "NETWORK_ETHERNET",
                        "processCpuUsage": "0.82%",
                        "resolution": "1280x800",
                        "romName": "rockchip",
                        "rooted": True,
                        "sdkVersionCode": 25,
                        "sdkVersionName": "7.1.2",
                        "sysDate": "Tue Oct 22 18:24:52 GMT+08:00 2024",
                        "sysDateStr": current_time,
                        "sysElapsedRealtime": "342:26:11",
                        "sysLanguage": "zh",
                        "sysSupportedSensors": ["Accelerometer sensor", "Gyroscope sensor",
                                                "Game Rotation Vector Sensor",
                                                "Gravity Sensor"],
                        "authCode": config['authCode'],
                        "conferenceName": "测试会议室"
                    }
                })
            }
        #安卓信息心跳上报
        elif topic == "/uams/android/broadcast":
            return json.dumps({
                "type":"heartbeat",
                "clientId" : config['clientId'],
                "appId":"com.ubains.uniplatform",
                "deviceId": config['deviceId']
            })

        #毫米波雷达数据上报
        elif "/properties/upload" in topic or "/event/upload" in topic:
            return json.dumps({
                "properties":{
                    "client_id" : config['client_id'],
                    "presence_state" : config['presence_state'],
                    "kaiguan" : config['kaiguan'],
                    "julishezhi" : config['julishezhi'],
                    "lingmindushezhi" : config['lingmindushezhi'],
                    "led":1,
                    "wifi_mac" : config['wifi_mac'],
                    "ble_mac" : config['ble_mac'],
                    "last_connection_time": current_time,
                    "current_time":current_time,
                    "device_model" : "c1_100_wifi_u",
                    "fw_version":"0.0.6",
                    "sn" : config['sn'],
                    "ip" : config['ip']
                }
            })

        # 北京富创项目的消息体与主题
        elif topic ==  "/meeting/message/sync":
            return json.dumps({
                    "action": config['action'],
                    "thirdMessageDTO": [{
                        "thirdPartyMeetingId": config['thirdPartyMeetingId'],
                        "messageCompere": "张三",
                        "thirdPartyUserId": "jiaojiao",
                        "conferenceName": config['conferenceName'],
                        "thirdPartyRoomId": config['thirdPartyRoomId'],
                        "messageName": config['messageName'],
                        "startTime": config['startTime'],
                        "endTime": config['endTime'],
                        "companyNumber": config['companyNumber'],
                        "participantList": ["JiaoJiao", "JiaYu", "DuiFangZhengZaiZhangTouFa", "DuoTangMaLaBan"]
            }]
    })


    def send_and_receive_messages(self, topic: str, message: str, num_times: int = 1, timeout: int = 5,
                                  interval: float = 0.2):
        """
        发送并接收消息

        :param topic: 主题名称
        :param message: 消息内容
        :param num_times: 发送次数，默认为1
        :param timeout: 超时时间（秒），默认为5秒
        :param interval: 每次发送之间的间隔时间（秒），默认为0.2秒
        """
        if not isinstance(topic, str) or not isinstance(message, str):
            raise ValueError("主题和消息必须是字符串类型")

        for i in range(num_times):
            try:
                self.publish(topic, message)
                received_message = self.wait_for_message(topic, timeout=timeout)

                if received_message:
                    logging.info("消息接收成功！")
                else:
                    logging.warning("超时时间内未接收到消息。")

                sleep(interval)
            except (ConnectionError, TimeoutError) as e:
                logging.error(f"网络连接或超时错误: {e}")
            except ValueError as e:
                logging.error(f"值错误: {e}")
            except Exception as e:
                logging.error(f"未知错误: {e}")