From ca400d8ea7a5c42cd9dd79b5695b3490dacb968f Mon Sep 17 00:00:00 2001
From: FAN-yeB <1442100690@qq.com>
Date: Tue, 23 Sep 2025 11:29:05 +0800
Subject: [PATCH] =?UTF-8?q?Revert=20"=E6=9B=B4=E6=94=B9=E4=B8=8B=E5=8F=91?=
=?UTF-8?q?=E9=85=8D=E7=BD=AE"?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
This reverts commit 8891a967315bc325e8f9cd4eda0e0369201e0423.
---
main/xiaozhi-server/config.yaml | 20 ++-
main/xiaozhi-server/core/api/ota_handler.py | 161 +++++++++++++-------
2 files changed, 126 insertions(+), 55 deletions(-)
diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml
index 807cbd2d..a946579c 100644
--- a/main/xiaozhi-server/config.yaml
+++ b/main/xiaozhi-server/config.yaml
@@ -41,10 +41,24 @@ server:
# 可选:设备白名单，如果设置了白名单，那么白名单的机器无论是什么token都可以连接。
#allowed_devices:
# - "24:0A:C4:1D:3B:F0" # MAC地址列表
- # MQTT网关配置，用于通过OTA下发到设备，与智控台要求格式相同，与mqtt_gateway的配置相同
- mqtt_gateway: null
- # MQTT签名密钥，用于生成MQTT连接密钥，与mqtt_gateway的配置相同
+ # MQTT网关配置，用于通过OTA下发到设备
+ mqtt_gateway:
+ # MQTT服务器地址以及端口,根据mqtt_gateway的.env文件配置，为null或者不填写时使用websocket协议
+ host: null
+ port: null
+ # group_id会从设备型号动态生成，格式为GID_设备型号
+ # 此配置仅作为当获取设备型号失败时可以填写的回退值
+ # group_id:
+
+ # MQTT签名密钥，用于生成MQTT连接密码，根据mqtt_gateway的.env文件配置
mqtt_signature_key: null
+
+ # UDP网关配置，根据mqtt_gateway的.env文件配置
+ udp_gateway:
+ # UDP服务器地址
+ host: null
+ # UDP服务器端口
+ port: null
log:
# 设置控制台输出的日志格式，时间、日志级别、标签、消息
log_format: "{time:YYMMDD HH:mm:ss}[{version}_{selected_module}][{extra[tag]}]-{level}-{message}"
diff --git a/main/xiaozhi-server/core/api/ota_handler.py b/main/xiaozhi-server/core/api/ota_handler.py
index 8262c629..f24c3f8f 100644
--- a/main/xiaozhi-server/core/api/ota_handler.py
+++ b/main/xiaozhi-server/core/api/ota_handler.py
@@ -32,8 +32,26 @@ class OTAHandler(BaseHandler):
self.logger.bind(tag=TAG).error(f"生成MQTT密码签名失败: {e}")
return ""
+ def _get_websocket_url(self, local_ip: str, port: int) -> str:
+ """获取websocket地址
+
+ Args:
+ local_ip: 本地IP地址
+ port: 端口号
+
+ Returns:
+ str: websocket地址
+ """
+ server_config = self.config["server"]
+ websocket_config = server_config.get("websocket", "")
+
+ if "你的" not in websocket_config:
+ return websocket_config
+ else:
+ return f"ws://{local_ip}:{port}/xiaozhi/v1/"
+
async def handle_post(self, request):
- """处理 OTA POST 请求 - 仅下发 MQTT 配置"""
+ """处理 OTA POST 请求"""
try:
data = await request.text()
self.logger.bind(tag=TAG).debug(f"OTA请求方法: {request.method}")
@@ -41,52 +59,17 @@ class OTAHandler(BaseHandler):
self.logger.bind(tag=TAG).debug(f"OTA请求数据: {data}")
device_id = request.headers.get("device-id", "")
- if not device_id:
+ if device_id:
+ self.logger.bind(tag=TAG).info(f"OTA请求设备ID: {device_id}")
+ else:
raise Exception("OTA请求设备ID为空")
data_json = json.loads(data)
server_config = self.config["server"]
- mqtt_endpoint = server_config.get("mqtt_gateway", "")
- signature_key = server_config.get("mqtt_signature_key", "")
+ port = int(server_config.get("port", 8000))
+ local_ip = get_local_ip()
- if not mqtt_endpoint:
- raise Exception("未配置 mqtt_gateway")
-
- # 从设备型号获取 group_id
- 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}"
-
- # 构建用户名（Base64编码的JSON）
- 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 = ""
- if signature_key:
- password = self.generate_password_signature(mqtt_client_id + "|" + username, signature_key)
- if not password:
- raise Exception("MQTT密码签名生成失败")
- else:
- raise Exception("未配置 mqtt_signature_key")
-
- # 构建返回的 MQTT 配置
return_json = {
"server_time": {
"timestamp": int(round(time.time() * 1000)),
@@ -96,25 +79,97 @@ class OTAHandler(BaseHandler):
"version": data_json["application"].get("version", "1.0.0"),
"url": "",
},
- "mqtt": {
- "endpoint": mqtt_endpoint,
+ }
+
+ mqtt_gateway_config = server_config.get("mqtt_gateway", {})
+ mqtt_gateway_host = mqtt_gateway_config.get("host", "")
+ mqtt_gateway_port = mqtt_gateway_config.get("port", "")
+
+
+ if mqtt_gateway_host: # 配置了mqtt_gateway，使用MQTT和UDP协议传输
+ # 从设备型号获取group_id，与manager-api保持一致
+ # 客户端ID格式：groupId@@@macAddress@@@macAddress
+ # 尝试从请求数据中获取设备型号
+ device_model = "default"
+ try:
+ # 假设设备型号在request数据的某个字段中
+ 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"]
+ # 为了保证格式一致性，进行与manager-api相同的处理
+ group_id = f"GID_{device_model}".replace(":", "_").replace(" ", "_")
+ except Exception as e:
+ self.logger.bind(tag=TAG).error(f"获取设备型号失败: {e}")
+ # 如果获取失败，使用配置文件中的默认值
+ group_id = mqtt_gateway_config.get("group_id", "GID_default").replace(":", "_")
+
+ mac_address_safe = device_id.replace(":", "_")
+ mqtt_client_id = f"{group_id}@@@{mac_address_safe}@@@{mac_address_safe}"
+
+ # 构建用户数据（包含IP等信息）
+ user_data = {
+ "ip": "unknown"
+ }
+
+ # 将用户数据编码为Base64 JSON
+ 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 = ""
+
+ # 获取MQTT签名密钥
+ 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 = mqtt_gateway_config.get("password", "")
+ else:
+ # 如果没有签名密钥，使用配置文件中的密码
+ password = mqtt_gateway_config.get("password", "")
+ self.logger.bind(tag=TAG).warning("缺少MQTT签名密钥，使用配置文件中的密码")
+
+ # 构建MQTT配置
+ endpoint = f"{mqtt_gateway_host}:{mqtt_gateway_port}"
+ return_json["mqtt_gateway"] = {
+ "endpoint": 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配置")
-
+ self.logger.bind(tag=TAG).info(f"为设备 {device_id} 下发MQTT网关配置")
+
+ # 添加UDP网关配置
+ if "udp_gateway" in server_config:
+ udp_config = server_config["udp_gateway"]
+ udp_host = udp_config.get("host", "")
+ udp_port = udp_config.get("port", "")
+ if udp_host:
+ return_json["udp_gateway"] = {
+ "host": udp_host,
+ "port": udp_port
+ }
+ self.logger.bind(tag=TAG).info(f"为设备 {device_id} 下发UDP网关配置")
+
+ 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",
)
except Exception as e:
- self.logger.bind(tag=TAG).error(f"处理OTA请求失败: {e}")
- return_json = {"success": False, "message": str(e)}
+ return_json = {"success": False, "message": "request error."}
response = web.Response(
text=json.dumps(return_json, separators=(",", ":")),
content_type="application/json",
@@ -127,12 +182,14 @@ class OTAHandler(BaseHandler):
"""处理 OTA GET 请求"""
try:
server_config = self.config["server"]
- mqtt_endpoint = server_config.get("mqtt_gateway", "未配置")
- message = f"OTA接口运行正常，MQTT网关地址：{mqtt_endpoint}"
+ local_ip = get_local_ip()
+ port = int(server_config.get("port", 8000))
+ websocket_url = self._get_websocket_url(local_ip, port)
+ message = f"OTA接口运行正常，向设备发送的websocket地址是：{websocket_url}"
response = web.Response(text=message, content_type="text/plain")
except Exception as e:
self.logger.bind(tag=TAG).error(f"OTA GET请求异常: {e}")
response = web.Response(text="OTA接口异常", content_type="text/plain")
finally:
self._add_cors_headers(response)
- return response
\ No newline at end of file
+ return response