modify: 视频推流

aibox_sophon_adapter/__init__.py

@@ -88,12 +88,15 @@ async def create_workers(session, config, count=1, worker_manager=None):
     """
     from .global_var import set_global_var, heartbeat_timeout as _default_timeout
     from chongming_database import Database
+    from chongming_stream_manager import StreamManager
 
     # 读取心跳配置
     worker_config = config.get("worker", {}).get("heartbeat", {}) if isinstance(config, dict) else {}
     hb_interval = worker_config.get("interval", 5)
     hb_timeout = worker_config.get("timeout", _default_timeout)
     check_interval = worker_config.get("check_interval", 10)
+    mediamtx_addr = worker_config.get("mediamtx_addr", "rtsp://172.21.110.223:8554")
+    link_path = worker_config.get("link_path", "/data/bm-app")
 
     # 设定全局变量
     global heartbeat_timeout
@@ -107,9 +110,14 @@ async def create_workers(session, config, count=1, worker_manager=None):
     db = Database(db_path)
     await db.initialize()
 
+    # 初始化视频流控制器
+    stream_manager = StreamManager(mediamtx_addr)
+
     set_global_var("db", db)
     set_global_var("heartbeat_timeout", hb_timeout)
     set_global_var("heartbeat_interval", hb_interval)
+    set_global_var("stream_manager", stream_manager)
+    set_global_var("link_path", link_path)
 
     logger.info(
         f"[aibox_sophon_adapter] 已初始化: "

aibox_sophon_adapter/global_var.py

@@ -7,10 +7,15 @@ from typing import Any
 
 from chongming_database import Database
 from .models.orm import *
+from chongming_stream_manager import StreamManager
 
 # 数据库实例
 db: Database | None = None
 
+# 视频流管理器
+stream_manager: StreamManager | None = None
+link_path: str | None = None
+
 # 心跳超时配置（秒）
 heartbeat_timeout: float = 15.0
 

aibox_sophon_adapter/models/board_models.py

@@ -130,4 +130,3 @@ class DeviceSummary(BaseModel):
     last_ping_time: Optional[str] = Field(default=None, description="最后心跳时间")
     grant_code: int = Field(..., description="授权码")
     grant_desc: str = Field(..., description="授权描述")
-

aibox_sophon_adapter/router/board_ping.py

@@ -27,7 +27,7 @@ async def board_ping(request: BoardPingRequest, worker_id: str = "") -> dict:
     设备通过 MQTT 主题 /board_ping 定期（默认 5 秒）上报系统状态。
     收到心跳后更新数据库中的设备记录，标记为在线。
     """
-    from ..global_var import db, heartbeat_timeout
+    from ..global_var import db, heartbeat_timeout, stream_manager, link_path
 
     board_id = request.BoardId
 
@@ -82,6 +82,17 @@ async def board_ping(request: BoardPingRequest, worker_id: str = "") -> dict:
 
             await session.commit()
 
+        if stream_manager is None:
+            raise RuntimeError("StreamManager not initialized")
+        
+        ping_data = json.loads(ping_json)
+        medias = ping_data.get("Medias", {})
+        for media in medias:
+            if media.get("MediaStatus", {}).get("label", "") == "正常":
+                stream_id = media["MediaName"]
+                source_url = media["MediaUrl"].replace("/Link", link_path)
+                await stream_manager.start_stream(stream_id, source_url)
+
         logger.debug(
             f"[Heartbeat] {board_id}: temp={request.BoardTemp}, "
             f"disk={request.HostDisk.Used}/{request.HostDisk.Total} kB, "

auth/__init__.py

@@ -21,13 +21,22 @@ async def create_workers(session, config, count=1, worker_manager=None):
     db = Database(db_path)
     await db.initialize()
 
-    # 初始化管理员
+    # 初始化管理员（多进程并发启动时必须幂等，避免 UNIQUE 约束冲突）
+    from sqlalchemy.exc import IntegrityError
+
     async with db.get_session() as sess:
         select_stmt = select(Users) \
             .where(Users.username == "admin") # type: ignore
         result = await sess.execute(select_stmt)
         user: Users | None = result.scalar_one_or_none()
 
+        # 按用户名未命中时，再按邮箱兜底，避免历史数据中已存在同邮箱用户
+        if user is None:
+            select_stmt = select(Users) \
+                .where(Users.email == "admin@example.com") # type: ignore
+            result = await sess.execute(select_stmt)
+            user = result.scalar_one_or_none()
+
         if user is None:
             user = Users(
                 username="admin",
@@ -38,7 +47,11 @@ async def create_workers(session, config, count=1, worker_manager=None):
                 is_superuser=True
             )
             sess.add(user)
+            try:
                 await sess.commit()
+            except IntegrityError:
+                # 其他并发进程已抢先创建 admin，当前进程忽略即可
+                await sess.rollback()
     
     set_global_var("db", db)
 

doubao/__init__.py

@@ -2,6 +2,7 @@
 豆包视觉理解 Worker 插件
 """
 import os
+import logging
 from openai import AsyncOpenAI
 from volcenginesdkarkruntime import AsyncArk
 from chongming_worker import create_workers as _create_workers
@@ -12,6 +13,8 @@ from .global_var import (
 )
 from .router import *
 
+logger = logging.getLogger("plugins.doubao")
+
 # 自动绑定 gen_models 生成的 Pydantic 模型
 auto_bind_models("doubao")
 
@@ -19,24 +22,34 @@ async def create_workers(session, config, count=1, worker_manager=None):
     """
     工厂函数：由 app 生命周期调用，创建 Worker 协程。
     """
+    logger.info("豆包插件初始化开始 (count=%d, aicc=%s)", count, config.get("is_aicc", False))
+
     is_aicc = config.get("is_aicc", False)
     set_is_aicc(is_aicc)
 
     api_key = config.get("api_key") or os.environ.get("VOLC_ACCESSKEY")
     if not api_key:
+        logger.error("豆包插件初始化失败: 缺少 API_KEY")
         raise ValueError("缺少 API_KEY，请在配置中设置 api_key 或环境变量 VOLC_ACCESSKEY")
 
     model = config.get("model", "doubao-seed-evolving")
     os.environ["DOUBAO_MODEL"] = model
+    logger.info("豆包模型: %s", model)
 
     if is_aicc:
+        logger.info("豆包使用 AICC 安全通信模式")
         # AICC 模式：懒加载，避免非 AICC 环境导入失败
+        try:
             from bytedance.volcengine_aicc_sdk.secure_http_client import (
                 AICCConfig, AsyncSecureHttpClient,
             )
+        except ImportError as e:
+            logger.error("AICC SDK 导入失败: %s，请确认已安装 bytedance_volcengine_aicc_sdk", e)
+            raise
 
         base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3")
         os.environ["DOUBAO_BASE_URL"] = base_url
+        logger.info("豆包 AICC base_url: %s", base_url)
 
         aicc_config = AICCConfig(ra_url=base_url, auth_token=api_key)
         client = AsyncOpenAI(
@@ -45,16 +58,22 @@ async def create_workers(session, config, count=1, worker_manager=None):
             http_client=AsyncSecureHttpClient(aicc_config=aicc_config),
         )
         set_aicc_client(client)
+        logger.info("豆包 AICC 客户端已创建")
     else:
+        logger.info("豆包使用标准 Ark SDK 模式")
         # Ark 模式：使用 AsyncArk
         base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3")
+        os.environ["DOUBAO_BASE_URL"] = base_url
         client = AsyncArk(base_url=base_url, api_key=api_key)
         set_ark_client(client)
+        logger.info("豆包 Ark 客户端已创建 (base_url=%s)", base_url)
 
     # 委托给通用运行时创建 Worker 任务
+    logger.info("豆包插件初始化完成，创建 %d 个 Worker 副本", count)
     return _create_workers(session, config, count, worker_manager)
 
 async def cleanup(worker_item: str):
     """清理资源"""
+    logger.info("豆包插件清理: %s", worker_item)
     cleanup_all()
-    print(f"{worker_item} 清理完毕")
+    logger.info("豆包插件清理完毕: %s", worker_item)

doubao/router/analyze.py

@@ -1,8 +1,11 @@
 import os
+import logging
 from chongming_worker import rpc_method
 from ..models.pydantic import AnalyzePostRequest, AnalyzePostResponse
 from ..global_var import get_client, get_is_aicc
 
+logger = logging.getLogger("plugins.doubao.analyze")
+
 @rpc_method(
     subject="doubao.analyze",
     method="POST",
@@ -10,10 +13,11 @@ from ..global_var import get_client, get_is_aicc
     response_model=AnalyzePostResponse,
 )
 async def analyze(request: AnalyzePostRequest) -> AnalyzePostResponse:
-    print(f"[DEBUG] analyze called with prompt={request.prompt}, image_base64 length={len(request.image_base64)}")
+    logger.info("收到分析请求: prompt=%s, image_len=%d", request.prompt, len(request.image_base64))
     client = get_client()
     if client is None:
-        return AnalyzePostResponse(result="", error="客户端未初始化")
+        logger.error("豆包客户端未初始化（is_aicc=%s）", get_is_aicc())
+        return AnalyzePostResponse(result="", error="客户端未初始化，请检查插件启动日志确认 API_KEY 和 AICC 配置是否正确")
 
     # 获取图片数据
     image_base64 = request.image_base64
@@ -49,6 +53,8 @@ async def analyze(request: AnalyzePostRequest) -> AnalyzePostResponse:
             **create_kwargs # type: ignore
         )
         result = response.choices[0].message.content # type: ignore
+        logger.info("分析完成: 结果长度=%d", len(result) if result else 0)
         return AnalyzePostResponse(result=result) # type: ignore
     except Exception as e:
+        logger.error("调用豆包API失败: %s", e, exc_info=True)
         return AnalyzePostResponse(result="", error=f"调用豆包API失败: {str(e)}")