"""通知发送器 — 支持钉钉、飞书、邮件的实际发送。
使用 httpx 进行异步 HTTP 请求,aiosmtplib 发送邮件。
支持消息模板渲染,根据不同事件类型发送不同格式的消息。
"""
import time
import hmac
import hashlib
import base64
import json
from typing import Dict, Any, Optional
from urllib.parse import quote
class NotificationSender:
"""通知发送器,支持钉钉、飞书、邮件。"""
# 内置消息模板
TEMPLATES = {
"ticket_created": {
"title": "新工单待审批",
"body": "**申请人**: {applicant}\n**目标实例**: {instance_name} ({db_type})\n**目标库**: {target_database}\n**工单标题**: {title}\n**SQL摘要**: {sql_preview}",
},
"ticket_approved": {
"title": "工单已审批通过",
"body": "**工单标题**: {title}\n**审批人**: {approver}\n**目标实例**: {instance_name} ({db_type})\n**状态**: 已批准,可执行",
},
"ticket_rejected": {
"title": "工单已被拒绝",
"body": "**工单标题**: {title}\n**审批人**: {approver}\n**拒绝原因**: {comment}\n**目标实例**: {instance_name} ({db_type})",
},
"ticket_executed": {
"title": "工单执行完成",
"body": "**工单标题**: {title}\n**执行人**: {operator}\n**目标实例**: {instance_name} ({db_type})\n**目标库**: {target_database}\n**执行结果**: 成功\n**耗时**: {duration}ms\n**影响行数**: {affected_rows}",
},
"ticket_failed": {
"title": "工单执行失败",
"body": "**工单标题**: {title}\n**执行人**: {operator}\n**目标实例**: {instance_name} ({db_type})\n**目标库**: {target_database}\n**错误信息**: {error}",
},
"ticket_rollback": {
"title": "工单已回滚",
"body": "**回滚人**: {operator}\n**工单ID**: {ticket_id}",
},
"backup_completed": {
"title": "备份完成",
"body": "**数据库**: {database}\n**文件大小**: {size}\n**耗时**: {duration}s",
},
"backup_failed": {
"title": "备份失败",
"body": "**错误信息**: {error}\n请检查数据库连接和磁盘空间",
},
"schema_drift_detected": {
"title": "检测到 Schema 漂移",
"body": "**实例**: {instance_name}\n**数据库**: {database}\n**发现 {drift_count} 处变更**:\n{drift_details}",
},
}
# 飞书卡片颜色映射
FEISHU_COLORS = {
"ticket_created": "turquoise",
"ticket_approved": "blue",
"ticket_rejected": "red",
"ticket_executed": "green",
"ticket_failed": "red",
"ticket_rollback": "orange",
"backup_completed": "green",
"backup_failed": "red",
"schema_drift_detected": "orange",
}
# 钉钉消息颜色标记
DINGTALK_TAGS = {
"ticket_created": "🆕",
"ticket_approved": "✅",
"ticket_rejected": "❌",
"ticket_executed": "✅",
"ticket_failed": "❌",
"ticket_rollback": "⚠️",
"backup_completed": "✅",
"backup_failed": "❌",
"schema_drift_detected": "⚠️",
}
async def send(self, channel_type: str, config: Dict[str, Any],
event_type: str, data: Dict[str, Any]) -> Dict[str, Any]:
"""发送通知。
Args:
channel_type: 渠道类型 (dingtalk / feishu / email)
config: 渠道配置
event_type: 事件类型
data: 模板数据
"""
try:
if channel_type == "dingtalk":
return await self._send_dingtalk(config, event_type, data)
elif channel_type in ("webhook_feishu", "feishu"):
return await self._send_feishu(config, event_type, data)
elif channel_type == "email":
return await self._send_email(config, event_type, data)
else:
return {"success": False, "message": f"不支持的通知类型: {channel_type}"}
except Exception as e:
return {"success": False, "message": f"发送失败: {str(e)}"}
async def _send_dingtalk(self, config: Dict[str, Any], event_type: str,
data: Dict[str, Any]) -> Dict[str, Any]:
"""发送钉钉消息。"""
import httpx
webhook_url = config.get("webhook_url", "")
secret = config.get("secret", "")
if not webhook_url:
return {"success": False, "message": "缺少 webhook_url 配置"}
# 生成签名
if secret:
timestamp = str(int(time.time() * 1000))
sign = self._generate_dingtalk_sign(timestamp, secret)
if "?" in webhook_url:
webhook_url = f"{webhook_url}×tamp={timestamp}&sign={sign}"
else:
webhook_url = f"{webhook_url}?timestamp={timestamp}&sign={sign}"
template = self.TEMPLATES.get(event_type, self.TEMPLATES["ticket_created"])
title = template["title"].format(**data)
body = template["body"].format(**data)
tag = self.DINGTALK_TAGS.get(event_type, "📌")
# 钉钉 markdown 单个 \n 视为空格,需要 \n\n 才能换行
body = body.replace("\n", "\n\n")
payload = {
"msgtype": "markdown",
"markdown": {
"title": f"{tag} {title}",
"text": f"### {tag} {title}\n\n{body}\n\n---\n*来自 SQL 审核平台*",
}
}
async with httpx.AsyncClient(timeout=10) as client:
response = await client.post(webhook_url, json=payload)
response.raise_for_status()
result = response.json()
if result.get("errcode") == 0:
return {"success": True, "message": "钉钉消息发送成功"}
return {"success": False, "message": f"钉钉返回错误: {result}"}
async def _send_feishu(self, config: Dict[str, Any], event_type: str,
data: Dict[str, Any]) -> Dict[str, Any]:
"""发送飞书消息 — 使用 Interactive Card。"""
import httpx
webhook_url = config.get("webhook_url", "")
if not webhook_url:
return {"success": False, "message": "缺少 webhook_url 配置"}
template = self.TEMPLATES.get(event_type, self.TEMPLATES["ticket_created"])
title = template["title"].format(**data)
body = template["body"].format(**data)
color = self.FEISHU_COLORS.get(event_type, "blue")
payload = {
"msg_type": "interactive",
"card": {
"header": {
"title": {"tag": "plain_text", "content": title},
"template": color,
},
"elements": [
{"tag": "div", "text": {"tag": "lark_md", "content": body}},
{"tag": "note", "elements": [{"tag": "plain_text", "content": "来自 SQL 审核平台"}]},
]
}
}
async with httpx.AsyncClient(timeout=10) as client:
response = await client.post(webhook_url, json=payload)
response.raise_for_status()
return {"success": True, "message": "飞书消息发送成功"}
async def _send_email(self, config: Dict[str, Any], event_type: str,
data: Dict[str, Any]) -> Dict[str, Any]:
"""发送邮件通知。"""
try:
import aiosmtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
except ImportError:
return {"success": False, "message": "缺少 aiosmtplib 依赖,请安装: pip install aiosmtplib"}
smtp_host = config.get("smtp_host", "")
smtp_port = config.get("smtp_port", 587)
smtp_user = config.get("smtp_user", "")
smtp_password = config.get("smtp_password", "")
from_addr = config.get("from_addr", smtp_user)
to_addr = data.get("recipient_email", "")
use_tls = config.get("use_tls", True)
if not all([smtp_host, smtp_user, smtp_password, to_addr]):
missing = [k for k, v in {
"smtp_host": smtp_host, "smtp_user": smtp_user,
"smtp_password": smtp_password, "recipient_email": to_addr,
}.items() if not v]
return {"success": False, "message": f"缺少邮件配置: {', '.join(missing)}"}
template = self.TEMPLATES.get(event_type, self.TEMPLATES["ticket_created"])
subject = template["title"].format(**data)
body_text = template["body"].format(**data).replace("\n", "
")
html_body = f"""
本邮件由 SQL 审核平台自动发送
""" msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = from_addr msg["To"] = to_addr msg.attach(MIMEText(html_body, "html", "utf-8")) await aiosmtplib.send( msg, hostname=smtp_host, port=smtp_port, username=smtp_user, password=smtp_password, start_tls=use_tls, ) return {"success": True, "message": f"邮件已发送至 {to_addr}"} def _generate_dingtalk_sign(self, timestamp: str, secret: str) -> str: """生成钉钉 webhook 签名。""" string_to_sign = f"{timestamp}\n{secret}" hmac_code = hmac.new( secret.encode("utf-8"), string_to_sign.encode("utf-8"), digestmod=hashlib.sha256, ).digest() return quote(base64.b64encode(hmac_code).decode("utf-8")) async def notify_ticket_event(self, ticket_id: int, event_type: str, channels: list, ticket_data: Dict[str, Any]) -> list: """发送工单事件通知到所有启用的渠道。""" results = [] for channel in channels: try: config = json.loads(channel.config) if isinstance(channel.config, str) else channel.config result = await self.send( channel.channel_type, config, event_type, ticket_data ) results.append({"channel": channel.name, **result}) except Exception as e: results.append({"channel": getattr(channel, "name", "unknown"), "success": False, "message": str(e)}) return results