为最简化安装环境提供OTA基础信息接口

main/xiaozhi-server/app.py

@@ -3,6 +3,7 @@ import sys
 import signal
 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
 
 TAG = __name__
@@ -35,14 +36,20 @@ async def main():
     ws_server = WebSocketServer(config)
     ws_task = asyncio.create_task(ws_server.start())
 
+    # 启动 Simple OAT 服务器
+    ota_server = SimpleOtaServer(config)
+    ota_task = asyncio.create_task(ota_server.start())
+
     try:
         await wait_for_exit()  # 监听退出信号
     except asyncio.CancelledError:
         print("任务被取消，清理资源中...")
     finally:
         ws_task.cancel()
+        ota_task.cancel()
         try:
             await ws_task
+            await ota_task
         except asyncio.CancelledError:
             pass
         print("服务器已关闭，程序退出。")

main/xiaozhi-server/config.yaml

@@ -9,6 +9,10 @@ server:
   # 服务器监听地址和端口(Server listening address and port)
   ip: 0.0.0.0
   port: 8000
+  # 最简化安装环供OTA基础信息接口
+  ota_port: 8002
+  # OAT返回信息时区偏移量
+  timezone_offset: +8
   # 认证配置
   auth:
     # 是否启用认证
@@ -239,8 +243,8 @@ LLM:
     api_key: 你的deepseek web key
     temperature: 0.7  # 温度值
     max_tokens: 500   # 最大生成token数
-    top_p: 1         
-    top_k: 50        
+    top_p: 1
+    top_k: 50
     frequency_penalty: 0  # 频率惩罚
   AliAppLLM:
     # 定义LLM API类型
@@ -249,7 +253,7 @@ LLM:
     app_id: 你的app_id
     # 可在这里找到你的 api_key https://bailian.console.aliyun.com/?apiKey=1#/api-key
     api_key: 你的api_key
-    # 是否不使用本地prompt：true|false (默不用请在百练应用中设置prompt) 
+    # 是否不使用本地prompt：true|false (默不用请在百练应用中设置prompt)
     is_no_prompt: true
     # Ali_memory_id：false（不使用）|你的memory_id（请在百练应用中设置中获取）
     # Tips！：Ali_memory未实现多用户存储记忆(记忆按id调用)

main/xiaozhi-server/core/ota_server.py

@@ -0,0 +1,99 @@
+import json
+import time
+import asyncio
+from aiohttp import web
+from config.logger import setup_logging
+from core.connection import ConnectionHandler
+from core.utils.util import get_local_ip, initialize_modules
+
+TAG = __name__
+
+
+class SimpleOtaServer:
+    def __init__(self, config: dict):
+        self.config = config
+        self.logger = setup_logging()
+
+    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([
+                web.post("/xiaozhi/ota/", self._handle_ota_request),
+                web.options("/xiaozhi/ota/", self._handle_ota_request)
+            ])
+
+            # 运行服务
+            runner = web.AppRunner(app)
+            await runner.setup()
+            site = web.TCPSite(runner, host, port)
+            await site.start()
+
+            # 保持服务运行
+            while True:
+                await asyncio.sleep(3600)  # 每隔 1 小时检查一次
+
+    async def _handle_ota_request(self, request):
+        """处理 /xiaozhi/ota/ 的 POST 请求"""
+        try:
+            data = await request.text()
+            self.logger.bind(tag=TAG).debug(f"OTA请求方法: {request.method}")
+            self.logger.bind(tag=TAG).debug(f"OTA请求头: {request.headers}")
+            self.logger.bind(tag=TAG).debug(f"OTA请求数据: {data}")
+
+            device_id = request.headers.get("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"]
+            host = server_config.get("ip", "0.0.0.0")
+            port = int(server_config.get("port", 8000))
+            local_ip = get_local_ip()
+
+            # OTA基础信息
+            return_json = {
+                "server_time":{
+                    "timestamp": int(round(time.time() * 1000)),
+                    "timezone_offset": server_config.get("timezone_offset", 8) * 60
+                },
+                "firmware": {
+                    "version": data_json["application"].get("version", "1.0.0"),
+                    "url": ""
+                },
+                "websocket":{
+                    "url": f"ws://{local_ip}:{port}/xiaozhi/v1/",
+                }
+            }
+            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")
+        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

@@ -28,7 +28,7 @@ class WebSocketServer:
         port = int(server_config.get("port", 8000))
 
         self.logger.bind(tag=TAG).info(
-            "Server is running at ws://{}:{}/xiaozhi/v1/", get_local_ip(), port
+            "Web Socket Server is running at ws://{}:{}/xiaozhi/v1/", get_local_ip(), port
         )
         self.logger.bind(tag=TAG).info(
             "=======上面的地址是websocket协议地址，请勿用浏览器访问======="