update:兼容全模块部署，不启动自带的OTA 服务

main/xiaozhi-server/app.py

@@ -5,8 +5,11 @@ from config.settings import load_config, check_config_file
 from core.websocket_server import WebSocketServer
 from core.ota_server import SimpleOtaServer
 from core.utils.util import check_ffmpeg_installed
+from config.logger import setup_logging
+from core.utils.util import get_local_ip
 
 TAG = __name__
+logger = setup_logging()
 
 
 async def wait_for_exit():
@@ -35,10 +38,41 @@ async def main():
     # 启动 WebSocket 服务器
     ws_server = WebSocketServer(config)
     ws_task = asyncio.create_task(ws_server.start())
+    ota_task = None
 
-    # 启动 Simple OAT 服务器
+    read_config_from_api = config.get("read_config_from_api", False)
-    ota_server = SimpleOtaServer(config)
+    if not read_config_from_api:
-    ota_task = asyncio.create_task(ota_server.start())
+        # 启动 Simple OAT 服务器
+        ota_server = SimpleOtaServer(config)
+        ota_task = asyncio.create_task(ota_server.start())
+
+        logger.bind(tag=TAG).info(
+            "OTA接口是\t\thttp://{}:{}/xiaozhi/ota/",
+            get_local_ip(),
+            config["server"]["ota_port"],
+        )
+
+    # 获取WebSocket配置，使用安全的默认值
+    websocket_port = 8000
+    server_config = config.get("server", {})
+    if isinstance(server_config, dict):
+        websocket_port = int(server_config.get("port", 8000))
+
+    logger.bind(tag=TAG).info(
+        "Websocket地址是\tws://{}:{}/xiaozhi/v1/",
+        get_local_ip(),
+        websocket_port,
+    )
+
+    logger.bind(tag=TAG).info(
+        "=======上面的地址是websocket协议地址，请勿用浏览器访问======="
+    )
+    logger.bind(tag=TAG).info(
+        "如想测试websocket请用谷歌浏览器打开test目录下的test_page.html"
+    )
+    logger.bind(tag=TAG).info(
+        "=============================================================\n"
+    )
 
     try:
         await wait_for_exit()  # 监听退出信号
@@ -46,10 +80,12 @@ async def main():
         print("任务被取消，清理资源中...")
     finally:
         ws_task.cancel()
-        ota_task.cancel()
+        if ota_task:
+            ota_task.cancel()
         try:
             await ws_task
-            await ota_task
+            if ota_task:
+                await ota_task
         except asyncio.CancelledError:
             pass
         print("服务器已关闭，程序退出。")

main/xiaozhi-server/config.yaml

@@ -9,9 +9,14 @@ server:
   # 服务器监听地址和端口(Server listening address and port)
   ip: 0.0.0.0
   port: 8000
-  # 最简化安装环供OTA基础信息接口
+  # OTA接口的端口号
   ota_port: 8002
-  # OAT返回信息时区偏移量
+  # 这个websocket配置是指ota接口向设备发送的websocket地址
+  # 如果按默认的写法，ota接口会自动生成websocket地址。这个地址你可以直接用浏览器访问ota接口确认一下
+  # 当你使用docker部署或使用公网部署(使用ssl、域名)时，不一定准确
+  # 所以如果你使用docker部署或使用公网部署时，请设置正确的websocket地址
+  websocket: ws://你的ip或者域名:端口号/xiaozhi/v1/
+  # OTA返回信息时区偏移量
   timezone_offset: +8
   # 认证配置
   auth:

main/xiaozhi-server/core/ota_server.py

@@ -14,32 +14,39 @@ class SimpleOtaServer:
         self.config = config
         self.logger = setup_logging()
 
+    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 websocket_config and "你" not in websocket_config:
+            return websocket_config
+        else:
+            return f"ws://{local_ip}:{port}/xiaozhi/v1/"
+
     async def start(self):
         server_config = self.config["server"]
         host = server_config.get("ip", "0.0.0.0")
         port = int(server_config.get("ota_port"))
 
         if port:
-            self.logger.bind(tag=TAG).info(
-                "Simple OTA Server is running at http://{}:{}/xiaozhi/ota/", get_local_ip(), port
-            )
-            self.logger.bind(tag=TAG).info(
-                "=======上面的地址为最简化安装环境提供OTA基础信息，可用作小智固件的自定义OTA地址======="
-            )
-            self.logger.bind(tag=TAG).info(
-                "如想测试OTA地址请用谷歌浏览器打开test目录下的test_page.html"
-            )
-            self.logger.bind(tag=TAG).info(
-                "=============================================================\n"
-            )
-
             app = web.Application()
-
             # 添加路由
-            app.add_routes([
+            app.add_routes(
-                web.post("/xiaozhi/ota/", self._handle_ota_request),
+                [
-                web.options("/xiaozhi/ota/", self._handle_ota_request)
+                    web.get("/xiaozhi/ota/", self._handle_ota_get_request),
-            ])
+                    web.post("/xiaozhi/ota/", self._handle_ota_request),
+                    web.options("/xiaozhi/ota/", self._handle_ota_request),
+                ]
+            )
 
             # 运行服务
             runner = web.AppRunner(app)
@@ -74,26 +81,55 @@ class SimpleOtaServer:
 
             # OTA基础信息
             return_json = {
-                "server_time":{
+                "server_time": {
                     "timestamp": int(round(time.time() * 1000)),
-                    "timezone_offset": server_config.get("timezone_offset", 8) * 60
+                    "timezone_offset": server_config.get("timezone_offset", 8) * 60,
                 },
                 "firmware": {
                     "version": data_json["application"].get("version", "1.0.0"),
-                    "url": ""
+                    "url": "",
+                },
+                "websocket": {
+                    "url": self._get_websocket_url(local_ip, port),
                 },
-                "websocket":{
-                    "url": f"ws://{local_ip}:{port}/xiaozhi/v1/",
-                }
             }
-            response = web.Response(text=json.dumps(return_json, separators=(',', ':')), content_type="application/json")
+            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": "request error."}
-            response = web.Response(text=json.dumps(return_json, separators=(',', ':')), content_type="application/json")
+            response = web.Response(
+                text=json.dumps(return_json, separators=(",", ":")),
+                content_type="application/json",
+            )
         finally:
             # 添加header，允许跨域访问
-            response.headers["Access-Control-Allow-Headers"] = "client-id, content-type, device-id"
+            response.headers["Access-Control-Allow-Headers"] = (
+                "client-id, content-type, device-id"
+            )
+            response.headers["Access-Control-Allow-Credentials"] = "true"
+            response.headers["Access-Control-Allow-Origin"] = "*"
+            return response
+
+    async def _handle_ota_get_request(self, request):
+        """处理 /xiaozhi/ota/ 的 GET 请求"""
+        try:
+            server_config = self.config["server"]
+            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:
+            # 添加header，允许跨域访问
+            response.headers["Access-Control-Allow-Headers"] = (
+                "client-id, content-type, device-id"
+            )
             response.headers["Access-Control-Allow-Credentials"] = "true"
             response.headers["Access-Control-Allow-Origin"] = "*"
             return response

main/xiaozhi-server/core/websocket_server.py

@@ -27,18 +27,6 @@ class WebSocketServer:
         host = server_config.get("ip", "0.0.0.0")
         port = int(server_config.get("port", 8000))
 
-        self.logger.bind(tag=TAG).info(
-            "Web Socket Server is running at ws://{}:{}/xiaozhi/v1/", get_local_ip(), port
-        )
-        self.logger.bind(tag=TAG).info(
-            "=======上面的地址是websocket协议地址，请勿用浏览器访问======="
-        )
-        self.logger.bind(tag=TAG).info(
-            "如想测试websocket请用谷歌浏览器打开test目录下的test_page.html"
-        )
-        self.logger.bind(tag=TAG).info(
-            "=============================================================\n"
-        )
         async with websockets.serve(self._handle_connection, host, port):
             await asyncio.Future()