import os import httpx from packages.core.logger import app_logger # 告警 Webhook 配置(从环境变量读取) FEISHU_WEBHOOK_URL = os.getenv("FEISHU_WEBHOOK_URL") DINGTALK_WEBHOOK_URL = os.getenv("DINGTALK_WEBHOOK_URL") WECOM_WEBHOOK_URL = os.getenv("WECOM_WEBHOOK_URL") async def send_alert(title: str, message: str, level: str = "warning"): """ 通用告警模块:接入钉钉/飞书/企业微信机器人的 Webhook。 支持同时发送到多个渠道。 Args: title: 告警标题 message: 告警详情 level: 告警级别 (warning/error/critical) """ alert_msg = f"🚨 [ALERT - {level.upper()}] {title}: {message}" # 记录日志 if level == "critical" or level == "error": app_logger.error(alert_msg) else: app_logger.warning(alert_msg) # 确定告警颜色和图标 if level == "critical": color = "red" emoji = "🔴" elif level == "error": color = "orange" emoji = "🟠" else: color = "yellow" emoji = "🟡" # 发送到飞书 if FEISHU_WEBHOOK_URL: await _send_to_feishu(title, message, level, emoji) # 发送到钉钉 if DINGTALK_WEBHOOK_URL: await _send_to_dingtalk(title, message, level, emoji) # 发送到企业微信 if WECOM_WEBHOOK_URL: await _send_to_wecom(title, message, level, emoji) async def _send_to_feishu(title: str, message: str, level: str, emoji: str): """发送告警到飞书机器人""" try: payload = { "msg_type": "interactive", "card": { "header": { "title": { "content": f"{emoji} {title}", "tag": "plain_text" }, "template": "red" if level == "critical" else "orange" if level == "error" else "yellow" }, "elements": [ { "tag": "div", "text": { "content": f"**级别**: {level.upper()}\n\n**详情**:\n{message}", "tag": "lark_md" } }, { "tag": "hr" }, { "tag": "note", "elements": [ { "tag": "plain_text", "content": f"海关数据系统告警 - {level.upper()}" } ] } ] } } async with httpx.AsyncClient(timeout=10.0) as client: response = await client.post(FEISHU_WEBHOOK_URL, json=payload) if response.status_code == 200: app_logger.info(f"Alert sent to Feishu successfully: {title}") else: app_logger.error(f"Failed to send alert to Feishu: {response.text}") except Exception as e: app_logger.error(f"Error sending alert to Feishu: {e}") async def _send_to_dingtalk(title: str, message: str, level: str, emoji: str): """发送告警到钉钉机器人""" try: payload = { "msgtype": "markdown", "markdown": { "title": f"{emoji} {title}", "text": f"### {emoji} {title}\n\n" f"**级别**: {level.upper()}\n\n" f"**详情**:\n\n{message}\n\n" f"---\n\n" f"*海关数据系统告警*" } } async with httpx.AsyncClient(timeout=10.0) as client: response = await client.post(DINGTALK_WEBHOOK_URL, json=payload) if response.status_code == 200: app_logger.info(f"Alert sent to DingTalk successfully: {title}") else: app_logger.error(f"Failed to send alert to DingTalk: {response.text}") except Exception as e: app_logger.error(f"Error sending alert to DingTalk: {e}") async def _send_to_wecom(title: str, message: str, level: str, emoji: str): """发送告警到企业微信机器人""" try: payload = { "msgtype": "markdown", "markdown": { "content": f"## {emoji} {title}\n\n" f"**级别**: {level.upper()}\n\n" f"**详情**:\n{message}\n\n" f"> 海关数据系统告警" } } async with httpx.AsyncClient(timeout=10.0) as client: response = await client.post(WECOM_WEBHOOK_URL, json=payload) if response.status_code == 200: app_logger.info(f"Alert sent to WeCom successfully: {title}") else: app_logger.error(f"Failed to send alert to WeCom: {response.text}") except Exception as e: app_logger.error(f"Error sending alert to WeCom: {e}") def send_alert_sync(title: str, message: str, level: str = "warning"): """ 同步版本的告警发送(用于非异步上下文) """ import asyncio try: asyncio.create_task(send_alert(title, message, level)) except RuntimeError: # 如果没有运行中的事件循环,只记录日志 alert_msg = f"🚨 [ALERT - {level.upper()}] {title}: {message}" if level == "critical" or level == "error": app_logger.error(alert_msg) else: app_logger.warning(alert_msg)