""" 企微回调路由 处理安保人员在企微卡片上的操作（前往处理/已处理/误报忽略）。 提供告警详情接口供 H5 页面使用。 接收企微用户消息并路由到交互Agent。 """ import asyncio from datetime import datetime from fastapi import APIRouter, Depends, Query, Request from fastapi.responses import PlainTextResponse from pydantic import BaseModel from typing import Optional from app.yudao_compat import YudaoResponse from app.models import get_session, AlarmEvent, AlarmLlmAnalysis from app.services.alarm_event_service import get_alarm_event_service, AlarmEventService from app.utils.logger import logger router = APIRouter(prefix="/api/wechat", tags=["企微回调"]) class AlarmActionRequest(BaseModel): """企微卡片操作请求""" alarm_id: str action: str # confirm / complete / ignore operator_uid: str # 企微 userid remark: Optional[str] = None @router.get("/alarm_detail") async def get_alarm_detail(alarm_id: str = Query(..., description="告警ID")): """ 告警详情接口（供 H5 页面使用，无认证） 返回告警基本信息 + VLM 分析描述 """ db = get_session() try: alarm = db.query(AlarmEvent).filter(AlarmEvent.alarm_id == alarm_id).first() if not alarm: return YudaoResponse.error(404, "告警不存在") # 查 VLM 分析结果 vlm_desc = "" analysis = db.query(AlarmLlmAnalysis).filter( AlarmLlmAnalysis.alarm_id == alarm_id ).order_by(AlarmLlmAnalysis.id.desc()).first() if analysis: vlm_desc = analysis.summary or "" # snapshot_url 可能是 COS key，需转为可访问的预签名 URL snapshot_url = alarm.snapshot_url or "" if snapshot_url and not snapshot_url.startswith("http"): try: from app.services.oss_storage import get_oss_storage snapshot_url = get_oss_storage().get_presigned_url(snapshot_url) except Exception: pass return YudaoResponse.success({ "alarm_id": alarm.alarm_id, "alarm_type": alarm.alarm_type, "device_id": alarm.device_id, "scene_id": alarm.scene_id, "event_time": alarm.event_time.strftime('%Y-%m-%d %H:%M:%S') if alarm.event_time else "", "alarm_level": alarm.alarm_level, "alarm_status": alarm.alarm_status, "handle_status": alarm.handle_status, "snapshot_url": snapshot_url, "handler": alarm.handler or "", "handle_remark": alarm.handle_remark or "", "vlm_description": vlm_desc, }) finally: db.close() @router.post("/callback/alarm_action") async def alarm_action_callback( req: AlarmActionRequest, service: AlarmEventService = Depends(get_alarm_event_service), ): """ 企微告警操作回调（无认证，由 H5 页面调用） action: - confirm: 前往处理 → handle_status=HANDLING - complete: 已处理完成 → handle_status=DONE, alarm_status=CLOSED - ignore: 误报忽略 → alarm_status=FALSE, handle_status=DONE """ action_map = { "confirm": { "alarm_status": "CONFIRMED", "handle_status": "HANDLING", "remark": "前往处理", }, "complete": { "alarm_status": "CLOSED", "handle_status": "DONE", "remark": "手动结单", }, "ignore": { "alarm_status": "FALSE", "handle_status": "IGNORED", "remark": "标记误报", }, } action_cfg = action_map.get(req.action) if not action_cfg: return YudaoResponse.error(400, f"无效操作: {req.action}") result = service.handle_alarm( alarm_id=req.alarm_id, alarm_status=action_cfg["alarm_status"], handle_status=action_cfg["handle_status"], handler=req.operator_uid, remark=req.remark or action_cfg["remark"], ) if not result: return YudaoResponse.error(404, "告警不存在") logger.info( f"企微操作: alarm={req.alarm_id}, action={req.action}, " f"operator={req.operator_uid}" ) return YudaoResponse.success(True) # ==================== 交互Agent消息回调 ==================== @router.get("/agent/callback") async def wechat_agent_verify( msg_signature: str = Query(...), timestamp: str = Query(...), nonce: str = Query(...), echostr: str = Query(...), ): """企微回调URL验证（首次配置时调用）""" from app.services.wechat_crypto import WeChatCrypto try: crypto = WeChatCrypto() echo = crypto.verify_url(msg_signature, timestamp, nonce, echostr) return PlainTextResponse(content=echo) except Exception as e: logger.error(f"企微URL验证失败: {e}") return PlainTextResponse(content="", status_code=403) @router.post("/agent/callback") async def wechat_agent_message( request: Request, msg_signature: str = Query(...), timestamp: str = Query(...), nonce: str = Query(...), ): """接收企微用户消息，路由到交互Agent""" body = await request.body() from app.services.wechat_crypto import WeChatCrypto try: crypto = WeChatCrypto() msg = crypto.decrypt_message(body, msg_signature, timestamp, nonce) except Exception as e: logger.error(f"企微消息解密失败: {e}") return PlainTextResponse(content="success") msg_type = msg.get("MsgType", "") event = msg.get("Event", "") # ---- 模板卡片按钮点击事件 ---- if msg_type == "event" and event == "template_card_event": asyncio.create_task(_process_card_button_click(msg)) return PlainTextResponse(content="success") # ---- 文本消息 → Agent 对话 ---- if msg_type == "text": user_id = msg.get("FromUserName", "") content = msg.get("Content", "") logger.info(f"收到企微消息: user={user_id}, content={content[:50]}") asyncio.create_task(_process_agent_message(user_id, content)) return PlainTextResponse(content="success") return PlainTextResponse(content="success") async def _process_agent_message(user_id: str, content: str): """异步处理消息并主动回复""" try: from app.services.agent_dispatcher import get_agent_dispatcher from app.services.wechat_service import get_wechat_service dispatcher = get_agent_dispatcher() reply = await dispatcher.handle_message(user_id, content) wechat = get_wechat_service() await wechat.send_text_message(user_id, reply) except Exception as e: logger.error(f"Agent消息处理失败: user={user_id}, error={e}", exc_info=True) async def _process_card_button_click(msg: dict): """ 处理模板卡片按钮点击事件 企微回调 XML 解密后包含： - FromUserName: 点击者 userid - EventKey: 按钮 key (handle_{alarm_id} / ignore_{alarm_id}) - TaskId: 卡片的 task_id (alarm_id) - ResponseCode: 用于更新卡片状态（一次性） """ try: from app.services.wechat_service import get_wechat_service user_id = msg.get("FromUserName", "") event_key = msg.get("EventKey", "") task_id = msg.get("TaskId", "") response_code = msg.get("ResponseCode", "") logger.info( f"卡片按钮点击: user={user_id}, key={event_key}, " f"task={task_id}" ) # 解析 action 和 alarm_id if event_key.startswith("handle_"): action = "handle" alarm_id = event_key[len("handle_"):] elif event_key.startswith("ignore_"): action = "ignore" alarm_id = event_key[len("ignore_"):] elif event_key.startswith("done_"): # 已处理状态的按钮，忽略 return else: logger.warning(f"未知的按钮 key: {event_key}") return # 更新告警状态 action_map = { "handle": { "alarm_status": "CONFIRMED", "handle_status": "HANDLING", "remark": "企微卡片-前往处理", }, "ignore": { "alarm_status": "FALSE", "handle_status": "IGNORED", "remark": "企微卡片-标记误报", }, } action_cfg = action_map.get(action) if action_cfg: service = get_alarm_event_service() service.handle_alarm( alarm_id=alarm_id, alarm_status=action_cfg["alarm_status"], handle_status=action_cfg["handle_status"], handler=user_id, remark=action_cfg["remark"], ) logger.info(f"告警状态已更新: alarm={alarm_id}, action={action}, by={user_id}") # 更新卡片按钮状态（变灰 + 显示处理结果） if response_code: wechat = get_wechat_service() await wechat.update_alarm_card( response_code=response_code, user_ids=[user_id], alarm_id=alarm_id, action=action, operator_name=user_id, ) except Exception as e: logger.error(f"处理卡片按钮点击失败: {e}", exc_info=True) # ==================== Agent测试接口（开发用） ==================== @router.post("/agent/test") async def test_agent_message( user_id: str = Query(default="test_user"), content: str = Query(..., description="测试消息内容"), ): """测试Agent对话（开发用，无加密，直接返回回复）""" from app.services.agent_dispatcher import get_agent_dispatcher dispatcher = get_agent_dispatcher() reply = await dispatcher.handle_message(user_id, content) return YudaoResponse.success({"user_id": user_id, "content": content, "reply": reply}) # ==================== 群聊管理接口 ==================== class CreateGroupRequest(BaseModel): """创建群聊请求""" name: str # 群聊名称 owner: str # 群主 userid user_list: list # 成员 userid 列表（至少2人） chat_id: Optional[str] = "" # 自定义群聊ID class SendGroupCardRequest(BaseModel): """发送群聊卡片请求""" chat_id: str # 群聊ID alarm_id: str # 告警ID mention_user_ids: Optional[list] = [] # 要 @的人员 @router.post("/group/create") async def create_group_chat(req: CreateGroupRequest): """创建企微群聊""" from app.services.wechat_service import get_wechat_service wechat = get_wechat_service() chatid = await wechat.create_group_chat( name=req.name, owner=req.owner, user_list=req.user_list, chat_id=req.chat_id, ) if chatid: return YudaoResponse.success({"chatid": chatid}) return YudaoResponse.error(500, "创建群聊失败") @router.post("/group/send_alarm") async def send_group_alarm(req: SendGroupCardRequest): """发送告警卡片到群聊（测试用，根据 alarm_id 查库）""" from app.services.wechat_service import get_wechat_service from app.config import settings db = get_session() try: alarm = db.query(AlarmEvent).filter(AlarmEvent.alarm_id == req.alarm_id).first() if not alarm: return YudaoResponse.error(404, "告警不存在") # 查 VLM 描述 vlm_desc = "" analysis = db.query(AlarmLlmAnalysis).filter( AlarmLlmAnalysis.alarm_id == req.alarm_id ).order_by(AlarmLlmAnalysis.id.desc()).first() if analysis: vlm_desc = analysis.summary or "" wechat = get_wechat_service() service_base_url = settings.wechat.service_base_url or "" event_time_str = alarm.event_time.strftime('%Y-%m-%d %H:%M:%S') if alarm.event_time else "" sent = await wechat.send_group_card( chat_id=req.chat_id, alarm_id=alarm.alarm_id, alarm_type=alarm.alarm_type, area_name=alarm.scene_id or "未知区域", camera_name=alarm.device_id or "未知摄像头", description=vlm_desc or f"{alarm.alarm_type} 告警", event_time=event_time_str, alarm_level=alarm.alarm_level or 2, service_base_url=service_base_url, mention_user_ids=req.mention_user_ids, ) if sent: return YudaoResponse.success(True) return YudaoResponse.error(500, "发送失败") finally: db.close()