diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 081c0b4..6cfc2e7 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/aibox_sophon_adapter/__init__.py @@ -105,6 +105,16 @@ async def create_workers(session, config, count=1, worker_manager=None): alarm_retention = config.get("alarm_retention", {}) alarm_max = alarm_retention.get("max_records", 0) + # 豆包 AI 分析开关 + doubao_cfg = config.get("doubao_analysis", {}) + doubao_enabled = doubao_cfg.get("enabled", True) + + # 告警转发配置 + alarm_fwd_cfg = config.get("alarm_forward", {}) + alarm_fwd_enabled = alarm_fwd_cfg.get("enabled", False) + alarm_fwd_url = alarm_fwd_cfg.get("url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receive") + alarm_fwd_timeout = alarm_fwd_cfg.get("timeout", 10.0) + # 初始化数据库 db_path = data_dir / "aibox_sophon_adapter.db" db_path.parent.mkdir(parents=True, exist_ok=True) @@ -150,6 +160,10 @@ async def create_workers(session, config, count=1, worker_manager=None): set_global_var("mqtt_forward", mqtt_fwd_client) set_global_var("upload_root", upload_root_path) set_global_var("alarm_max_records", alarm_max) + set_global_var("doubao_analysis_enabled", doubao_enabled) + set_global_var("alarm_forward_enabled", alarm_fwd_enabled) + set_global_var("alarm_forward_url", alarm_fwd_url) + set_global_var("alarm_forward_timeout", alarm_fwd_timeout) logger.info( f"[aibox_sophon_adapter] 已初始化: " diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py index c9c1c21..a42419d 100644 --- a/aibox_sophon_adapter/global_var.py +++ b/aibox_sophon_adapter/global_var.py @@ -28,6 +28,14 @@ upload_root: str = "./data/uploads" # 告警保留配置 alarm_max_records: int = 0 # 0 表示不限制 +# 豆包 AI 分析开关 +doubao_analysis_enabled: bool = True + +# 告警转发配置 +alarm_forward_enabled: bool = False +alarm_forward_url: str = "http://dg.datagoo.cn:60014/open-api/event-alarm/receive" +alarm_forward_timeout: float = 10.0 + # 心跳超时配置（秒） heartbeat_timeout: float = 15.0 @@ -53,5 +61,9 @@ __all__ = [ "mqtt_forward", "upload_root", "alarm_max_records", + "doubao_analysis_enabled", + "alarm_forward_enabled", + "alarm_forward_url", + "alarm_forward_timeout", "set_global_var", ] diff --git a/aibox_sophon_adapter/routes.py b/aibox_sophon_adapter/routes.py index cec91c2..40f2a4c 100644 --- a/aibox_sophon_adapter/routes.py +++ b/aibox_sophon_adapter/routes.py @@ -14,6 +14,7 @@ import hashlib import logging import aiofiles +import httpx from fastapi import APIRouter, UploadFile, File, Form, Request, Body, HTTPException from fastapi.responses import FileResponse @@ -186,26 +187,33 @@ async def alarm_report( # 获取完整报文（含 extra 字段，排除 None 值以生成整洁 JSON） full_data = body.model_dump(exclude_none=True) - async with Client(session=zenoh_session) as client: - try: - result = await client.call( - worker_type="doubao", - subject="doubao.analyze", - payload={ - "image_base64": full_data.get("ImageData", ""), - "prompt": full_data.get("Summary", ""), - }, - timeout=30.0, - retries=1 - ) - analysis = result.get("result", "") - error = result.get("error", "") - if error: - logger.warning(f"Doubao 分析失败: {error}") - else: - logger.info(f"Doubao 分析结果: {analysis}") - except Exception as e: - logger.error(f"Doubao 分析调用失败: {type(e).__name__}: {e}") + # 豆包 AI 分析（可通过配置开关关闭） + from .global_var import doubao_analysis_enabled + if doubao_analysis_enabled: + async with Client(session=zenoh_session) as client: + try: + result = await client.call( + worker_type="doubao", + subject="doubao.analyze", + payload={ + "image_base64": full_data.get("ImageData", ""), + "prompt": f"判断是否存在{full_data.get("Summary", "")}，直接返回True或False，不要输出其他内容", + }, + timeout=30.0, + retries=1 + ) + analysis = result.get("result", "") + error = result.get("error", "") + if error: + logger.warning(f"Doubao 分析失败: {error}") + if analysis == "False": + return AlarmReportResponse( + Result=AlarmReportResult(Code=0, Desc="Doubao 分析判定为 False，忽略告警") + ) + except Exception as e: + logger.error(f"Doubao 分析调用失败: {type(e).__name__}: {e}") + else: + logger.info("[AlarmReport] 豆包分析已关闭 (doubao_analysis.enabled=false)") board_id = full_data.get("BoardId", "") board_ip = full_data.get("BoardIp", "") @@ -265,6 +273,38 @@ async def alarm_report( except Exception as e: logger.error(f"[AlarmReport] 数据库写入失败: {e}") + # 告警转发到外部平台 + from .global_var import alarm_forward_enabled, alarm_forward_url, alarm_forward_timeout + if alarm_forward_enabled and alarm_forward_url: + try: + async with httpx.AsyncClient(timeout=alarm_forward_timeout) as http_client: + forward_resp = await http_client.post( + alarm_forward_url, + json=full_data, + headers={"Content-Type": "application/json"}, + ) + if forward_resp.is_success: + logger.info( + f"[AlarmReport] 转发成功: AlarmId={alarm_id}, " + f"URL={alarm_forward_url}, Status={forward_resp.status_code}" + ) + else: + logger.warning( + f"[AlarmReport] 转发失败: AlarmId={alarm_id}, " + f"URL={alarm_forward_url}, Status={forward_resp.status_code}, " + f"Body={forward_resp.text[:500]}" + ) + except httpx.TimeoutException: + logger.error( + f"[AlarmReport] 转发超时: AlarmId={alarm_id}, " + f"URL={alarm_forward_url}, Timeout={alarm_forward_timeout}s" + ) + except Exception as e: + logger.error( + f"[AlarmReport] 转发异常: AlarmId={alarm_id}, " + f"URL={alarm_forward_url}, Error={type(e).__name__}: {e}" + ) + return AlarmReportResponse( Result=AlarmReportResult(Code=0, Desc="") )