ai_fwt_pro/main/xingxing-server/core/http_server.py

import asyncio
from aiohttp import web
from config.logger import setup_logging
from core.api.app_gps_handler import AppGpsHandler
from core.api.ota_handler import OTAHandler
from core.api.vision_handler import VisionHandler

TAG = __name__


class SimpleHttpServer:
    def __init__(self, config: dict, ws_server=None):
        self.config = config
        self.logger = setup_logging()
        self.ws_server = ws_server
        self.app_gps_handler = AppGpsHandler(config, ws_server) if ws_server else None
        self.ota_handler = OTAHandler(config)
        self.vision_handler = VisionHandler(config)

    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):
        try:
            server_config = self.config["server"]
            read_config_from_api = self.config.get("read_config_from_api", False)
            host = server_config.get("ip", "0.0.0.0")
            port = int(server_config.get("http_port", 8003))

            if port:
                app = web.Application()

                if not read_config_from_api:
                    # 如果没有开启智控台，只是单模块运行，就需要再添加简单OTA接口，用于下发websocket接口
                    app.add_routes(
                        [
                            web.get("/xiaozhi/ota/", self.ota_handler.handle_get),
                            web.post("/xiaozhi/ota/", self.ota_handler.handle_post),
                            web.options(
                                "/xiaozhi/ota/", self.ota_handler.handle_options
                            ),
                            # 下载接口，仅提供 data/bin/*.bin 下载
                            web.get(
                                "/xiaozhi/ota/download/{filename}",
                                self.ota_handler.handle_download,
                            ),
                            web.options(
                                "/xiaozhi/ota/download/{filename}",
                                self.ota_handler.handle_options,
                            ),
                        ]
                    )
                # 添加路由
                app.add_routes(
                    [
                        web.get("/mcp/vision/explain", self.vision_handler.handle_get),
                        web.post(
                            "/mcp/vision/explain", self.vision_handler.handle_post
                        ),
                        web.options(
                            "/mcp/vision/explain", self.vision_handler.handle_options
                        ),
                    ]
                )

                # 运行服务
                if self.app_gps_handler:
                    app.add_routes(
                        [
                            web.get("/app/gps/latest", self.app_gps_handler.handle_get_latest),
                            web.get("/app/gps/ws", self.app_gps_handler.handle_ws),
                            web.get("/app/risk-alerts", self.app_gps_handler.handle_get_risk_alerts),
                            web.post("/app/risk-alerts", self.app_gps_handler.handle_post_risk_alerts),
                            web.delete("/app/risk-alerts/{alert_id}", self.app_gps_handler.handle_delete_risk_alert),
                            web.options("/app/gps/latest", self.app_gps_handler.handle_options),
                            web.options("/app/gps/ws", self.app_gps_handler.handle_options),
                            web.options("/app/risk-alerts", self.app_gps_handler.handle_options),
                            web.options("/app/risk-alerts/{alert_id}", self.app_gps_handler.handle_options),
                        ]
                    )

                runner = web.AppRunner(app)
                await runner.setup()
                site = web.TCPSite(runner, host, port)
                await site.start()

                # 保持服务运行
                while True:
                    await asyncio.sleep(3600)  # 每隔 1 小时检查一次
        except Exception as e:
            self.logger.bind(tag=TAG).error(f"HTTP服务器启动失败: {e}")
            import traceback

            self.logger.bind(tag=TAG).error(f"错误堆栈: {traceback.format_exc()}")
            raise