From 83ab65c10b15dc02215d35697b4ad3400851a74e Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Mon, 3 Aug 2026 16:56:56 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20=E6=8F=92=E4=BB=B6=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E8=AF=BB=E5=8F=96=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- aibox_sophon_adapter/__init__.py | 42 +++++++++++++++++--------------- doubao/__init__.py | 22 ++++++++++++++--- doubao/router/analyze.py | 22 ++++++++++++++++- 3 files changed, 62 insertions(+), 24 deletions(-) diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 2c1c8d0..783dd26 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/aibox_sophon_adapter/__init__.py @@ -90,20 +90,22 @@ async def create_workers(session, config, count=1, worker_manager=None): 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://127.0.0.1:8554") - link_path = worker_config.get("link_path", "/data/bm-app") + # 所有运行时配置均从 [workers.config] 及其子节读取 + data_dir = Path(config.get("data_dir", "./data")) + mediamtx_addr = config.get("mediamtx_addr", "rtsp://127.0.0.1:8554") + link_path = config.get("link_path", "/data/bm-app") - # 设定全局变量 - global heartbeat_timeout - heartbeat_timeout = hb_timeout + # 心跳配置 + heartbeat = config.get("heartbeat", {}) + hb_interval = heartbeat.get("interval", 5) + hb_timeout = heartbeat.get("timeout", _default_timeout) + check_interval = heartbeat.get("check_interval", 10) + + # 告警保留配置 + alarm_retention = config.get("alarm_retention", {}) + alarm_max = alarm_retention.get("max_records", 0) # 初始化数据库 - data_dir = Path(config.get("data_dir", "./data")) if isinstance(config, dict) else Path("./data") db_path = data_dir / "aibox_sophon_adapter.db" db_path.parent.mkdir(parents=True, exist_ok=True) @@ -113,21 +115,21 @@ async def create_workers(session, config, count=1, worker_manager=None): # 初始化视频流控制器 stream_manager = StreamManager(mediamtx_addr) - # 读取告警保留配置 - alarm_retention_config = worker_config.get("alarm_retention", {}) if isinstance(config, dict) else {} - alarm_max = alarm_retention_config.get("max_records", 0) - # upload 根目录（用于清理告警关联视频） - upload_root_path = config.get("data_dir", "./data/uploads") if isinstance(config, dict) else "./data/uploads" + upload_root_path = str(data_dir / "uploads") + + # 设定模块级全局变量（供 _check_offline_devices 使用） + global heartbeat_timeout + heartbeat_timeout = hb_timeout # 初始化外部 MQTT 转发客户端 - mqtt_fwd_config = worker_config.get("mqtt_forward", {}) if isinstance(config, dict) else {} - mqtt_fwd_enabled = mqtt_fwd_config.get("enabled", False) + mqtt_fwd = config.get("mqtt_forward", {}) + mqtt_fwd_enabled = mqtt_fwd.get("enabled", False) mqtt_fwd_client = None if mqtt_fwd_enabled: from .mqtt_forward import MQTTForwardClient - mqtt_fwd_uri = mqtt_fwd_config.get("uri", "") - mqtt_fwd_topic = mqtt_fwd_config.get("topic", "/board_ping") + mqtt_fwd_uri = mqtt_fwd.get("uri", "") + mqtt_fwd_topic = mqtt_fwd.get("topic", "/board_ping") if mqtt_fwd_uri: mqtt_fwd_client = MQTTForwardClient(uri=mqtt_fwd_uri, topic=mqtt_fwd_topic) try: diff --git a/doubao/__init__.py b/doubao/__init__.py index 701f1ae..9aa33c8 100644 --- a/doubao/__init__.py +++ b/doubao/__init__.py @@ -48,10 +48,26 @@ async def create_workers(session, config, count=1, worker_manager=None): 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) + # ra_url 为远程证明（安全令牌）端点地址，可能与推理 base_url 不同 + # 若不单独配置则默认使用 base_url 的域名 + /v1/security/token + ra_url = config.get("ra_url", base_url) + ak = config.get("ak", "") + sk = config.get("sk", "") + ra_service_name = config.get("ra_service_name", "AICC.ConfidentialInference") + ra_policy_id = config.get("ra_policy_id", "router_policy") - aicc_config = AICCConfig(ra_url=base_url, auth_token=api_key) + os.environ["DOUBAO_BASE_URL"] = base_url + logger.info("豆包 AICC base_url: %s, ra_url: %s, ra_service_name: %s, ra_policy_id: %s", + base_url, ra_url, ra_service_name, ra_policy_id) + + aicc_config = AICCConfig( + ra_url=ra_url, + auth_token=api_key, + ak=ak, + sk=sk, + aicc_llm_service_name=ra_service_name, + policy_id=ra_policy_id, + ) client = AsyncOpenAI( api_key=api_key, base_url=base_url, diff --git a/doubao/router/analyze.py b/doubao/router/analyze.py index 96cc5b8..df8f6a3 100644 --- a/doubao/router/analyze.py +++ b/doubao/router/analyze.py @@ -56,5 +56,25 @@ async def analyze(request: AnalyzePostRequest) -> AnalyzePostResponse: logger.info("分析完成: 结果长度=%d", len(result) if result else 0) return AnalyzePostResponse(result=result) # type: ignore except Exception as e: + error_msg = str(e) + # 针对 AICC 安全通道密钥缺失，给出明确提示 + # KeyMissingError 可能被包装在 APIConnectionError 的 __cause__ 链中 + is_key_missing = ( + "KeyMissingError" in type(e).__name__ + or "Can not get key" in error_msg + or any("KeyMissingError" in type(c).__name__ for c in [getattr(e, '__cause__', None), getattr(e, '__context__', None)] if c is not None) + ) + if is_key_missing: + logger.error( + "AICC 安全通道密钥获取失败，请检查: " + "1) api_key 是否有效 " + "2) ra_url（安全令牌地址）是否正确 " + "3) APIGateway /v1/security/token 端点是否可达" + ) + return AnalyzePostResponse( + result="", + error="AICC 安全通道初始化失败：无法获取加密密钥。" + "请检查 api_key 有效性及安全令牌服务 (/v1/security/token) 连通性。" + ) logger.error("调用豆包API失败: %s", e, exc_info=True) - return AnalyzePostResponse(result="", error=f"调用豆包API失败: {str(e)}") \ No newline at end of file + return AnalyzePostResponse(result="", error=f"调用豆包API失败: {error_msg}") \ No newline at end of file