diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 7d4a832..e0fbc2c 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/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] 已初始化: " diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py index 057459e..8c8f95e 100644 --- a/aibox_sophon_adapter/global_var.py +++ b/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 diff --git a/aibox_sophon_adapter/models/board_models.py b/aibox_sophon_adapter/models/board_models.py index 7d27579..d36c159 100644 --- a/aibox_sophon_adapter/models/board_models.py +++ b/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="授权描述") - diff --git a/aibox_sophon_adapter/router/board_ping.py b/aibox_sophon_adapter/router/board_ping.py index 305b16f..24aaff0 100644 --- a/aibox_sophon_adapter/router/board_ping.py +++ b/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, " diff --git a/auth/__init__.py b/auth/__init__.py index 115a1ef..2d82da9 100644 --- a/auth/__init__.py +++ b/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) - await sess.commit() + try: + await sess.commit() + except IntegrityError: + # 其他并发进程已抢先创建 admin，当前进程忽略即可 + await sess.rollback() set_global_var("db", db) diff --git a/doubao/__init__.py b/doubao/__init__.py index 620fbfb..701f1ae 100644 --- a/doubao/__init__.py +++ b/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 环境导入失败 - from bytedance.volcengine_aicc_sdk.secure_http_client import ( - AICCConfig, AsyncSecureHttpClient, - ) + 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} 清理完毕") \ No newline at end of file + logger.info("豆包插件清理完毕: %s", worker_item) \ No newline at end of file diff --git a/doubao/router/analyze.py b/doubao/router/analyze.py index bd50ebe..96cc5b8 100644 --- a/doubao/router/analyze.py +++ b/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)}") \ No newline at end of file