Merge pull request #2266 from xinnan-tech/py_mqtt

update:增加单模块mqtt-gateway支持

main/xiaozhi-server/config.yaml

@@ -41,6 +41,12 @@ server:
     # 可选:设备白名单，如果设置了白名单，那么白名单的机器无论是什么token都可以连接。
     #allowed_devices:
     #  - "24:0A:C4:1D:3B:F0"  # MAC地址列表
+ # MQTT网关配置，用于通过OTA下发到设备，根据mqtt_gateway的.env文件配置，格式为host:port
+  mqtt_gateway: null
+  # MQTT签名密钥，用于生成MQTT连接密码，根据mqtt_gateway的.env文件配置
+  mqtt_signature_key: null
+  # UDP网关配置
+  udp_gateway: null
 log:
   # 设置控制台输出的日志格式，时间、日志级别、标签、消息
   log_format: "<green>{time:YYMMDD HH:mm:ss}</green>[{version}_{selected_module}][<light-blue>{extra[tag]}</light-blue>]-<level>{level}</level>-<light-green>{message}</light-green>"

main/xiaozhi-server/core/api/ota_handler.py

@@ -1,5 +1,8 @@
 import json
 import time
+import base64
+import hashlib
+import hmac
 from aiohttp import web
 from core.utils.util import get_local_ip
 from core.api.base_handler import BaseHandler
@@ -10,6 +13,24 @@ TAG = __name__
 class OTAHandler(BaseHandler):
     def __init__(self, config: dict):
         super().__init__(config)
+        
+    def generate_password_signature(self, content: str, secret_key: str) -> str:
+        """生成MQTT密码签名
+        
+        Args:
+            content: 签名内容 (clientId + '|' + username)
+            secret_key: 密钥
+            
+        Returns:
+            str: Base64编码的HMAC-SHA256签名
+        """
+        try:
+            hmac_obj = hmac.new(secret_key.encode('utf-8'), content.encode('utf-8'), hashlib.sha256)
+            signature = hmac_obj.digest()
+            return base64.b64encode(signature).decode('utf-8')
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"生成MQTT密码签名失败: {e}")
+            return ""
 
     def _get_websocket_url(self, local_ip: str, port: int) -> str:
         """获取websocket地址
@@ -58,10 +79,66 @@ class OTAHandler(BaseHandler):
                     "version": data_json["application"].get("version", "1.0.0"),
                     "url": "",
                 },
-                "websocket": {
-                    "url": self._get_websocket_url(local_ip, port),
-                },
             }
+            
+
+            mqtt_gateway_endpoint = server_config.get("mqtt_gateway")
+            
+            if mqtt_gateway_endpoint:  # 如果配置了非空字符串
+                # 尝试从请求数据中获取设备型号
+                device_model = "default"
+                try:
+                    if "device" in data_json and isinstance(data_json["device"], dict):
+                        device_model = data_json["device"].get("model", "default")
+                    elif "model" in data_json:
+                        device_model = data_json["model"]
+                    group_id = f"GID_{device_model}".replace(":", "_").replace(" ", "_")
+                except Exception as e:
+                    self.logger.bind(tag=TAG).error(f"获取设备型号失败: {e}")
+                    group_id = "GID_default"
+
+                mac_address_safe = device_id.replace(":", "_")
+                mqtt_client_id = f"{group_id}@@@{mac_address_safe}@@@{mac_address_safe}"
+
+                # 构建用户数据
+                user_data = {
+                    "ip": "unknown"
+                }
+                try:
+                    user_data_json = json.dumps(user_data)
+                    username = base64.b64encode(user_data_json.encode('utf-8')).decode('utf-8')
+                except Exception as e:
+                    self.logger.bind(tag=TAG).error(f"生成用户名失败: {e}")
+                    username = ""
+
+                # 生成密码
+                password = ""
+                signature_key = server_config.get("mqtt_signature_key", "")
+                if signature_key:
+                    password = self.generate_password_signature(mqtt_client_id + "|" + username, signature_key)
+                    if not password:
+                        password = ""  # 签名失败则留空，由设备决定是否允许无密码
+                else:
+                    self.logger.bind(tag=TAG).warning("缺少MQTT签名密钥，密码留空")
+
+                # 构建MQTT配置（直接使用 mqtt_gateway 字符串）
+                return_json["mqtt_gateway"] = {
+                    "endpoint": mqtt_gateway_endpoint,
+                    "client_id": mqtt_client_id,
+                    "username": username,
+                    "password": password,
+                    "publish_topic": "device-server",
+                    "subscribe_topic": f"devices/p2p/{mac_address_safe}"
+                }
+                self.logger.bind(tag=TAG).info(f"为设备 {device_id} 下发MQTT网关配置")
+                
+                
+            else:  # 未配置 mqtt_gateway，下发 WebSocket
+                return_json["websocket"] = {
+                    "url": self._get_websocket_url(local_ip, port),
+                }
+                self.logger.bind(tag=TAG).info(f"未配置MQTT网关，为设备 {device_id} 下发WebSocket配置")
+            
             response = web.Response(
                 text=json.dumps(return_json, separators=(",", ":")),
                 content_type="application/json",
@@ -90,4 +167,4 @@ class OTAHandler(BaseHandler):
             response = web.Response(text="OTA接口异常", content_type="text/plain")
         finally:
             self._add_cors_headers(response)
-            return response
+            return response