from pydantic_settings import BaseSettings from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import smtplib from typing import Optional, List from datetime import datetime import logging logger = logging.getLogger(__name__) class EmailSettings(BaseSettings): """邮件配置""" SMTP_HOST: str = "localhost" SMTP_PORT: int = 25 SMTP_USER: Optional[str] = None SMTP_PASSWORD: Optional[str] = None SMTP_USE_TLS: bool = False SMTP_USE_SSL: bool = False SMTP_FROM: str = "ipam@localhost" class Config: env_prefix = "EMAIL_" class EmailService: """邮件服务""" def __init__(self, settings: EmailSettings): self.settings = settings def send_email(self, to: str, subject: str, html_content: str) -> bool: """ 发送HTML邮件 Args: to: 收件人邮箱 subject: 邮件主题 html_content: HTML内容 Returns: 是否发送成功 """ try: msg = MIMEMultipart('alternative') msg['From'] = self.settings.SMTP_FROM msg['To'] = to msg['Subject'] = subject html_part = MIMEText(html_content, 'html', 'utf-8') msg.attach(html_part) # 连接SMTP服务器 if self.settings.SMTP_USE_SSL: server = smtplib.SMTP_SSL( host=self.settings.SMTP_HOST, port=self.settings.SMTP_PORT, timeout=10 ) else: server = smtplib.SMTP( host=self.settings.SMTP_HOST, port=self.settings.SMTP_PORT, timeout=10 ) # 如果需要TLS加密 if self.settings.SMTP_USE_TLS and not self.settings.SMTP_USE_SSL: server.starttls() # 如果需要认证 if self.settings.SMTP_USER and self.settings.SMTP_PASSWORD: server.login(self.settings.SMTP_USER, self.settings.SMTP_PASSWORD) # 发送邮件 server.send_message(msg) server.quit() logger.info(f"邮件发送成功: {to} - {subject}") return True except Exception as e: logger.error(f"邮件发送失败: {str(e)}", exc_info=True) return False def send_alert_notification(self, to: str, alert_data: dict) -> bool: """发送告警通知邮件""" severity_colors = { 'critical': '#dc2626', 'error': '#ef4444', 'warning': '#f59e0b', 'info': '#3b82f6' } severity_texts = { 'critical': '严重', 'error': '错误', 'warning': '警告', 'info': '信息' } alert_type_texts = { 'ip_conflict': 'IP 地址冲突', 'unauthorized_access': '未授权设备接入', 'subnet_full': '网段容量不足', 'device_offline': '设备离线', 'new_device_detected': '新设备发现', 'mac_changed': 'MAC 地址变更' } severity = alert_data.get('severity', 'warning') color = severity_colors.get(severity, '#6b7280') severity_text = severity_texts.get(severity, severity) alert_type_text = alert_type_texts.get(alert_data.get('alert_type'), alert_data.get('alert_type')) html_content = f"""

告警类型: {alert_type_text}

告警标题: {alert_data.get('title', '-')}

详细信息: {alert_data.get('message', '-')}

相关IP: {alert_data.get('ip_address_str', '-')}

相关MAC: {alert_data.get('mac_address', '-')}

告警时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

""" subject = f"【{severity_text}】IPAM 告警 - {alert_data.get('title', '新告警')}" return self.send_email(to, subject, html_content) def send_scan_report(self, to: str, report_data: dict) -> bool: """发送扫描报告邮件""" html_content = f"""

扫描结果概览

{report_data.get('total_ips', 0)}

扫描IP总数

{report_data.get('online_ips', 0)}

在线设备

{report_data.get('with_mac', 0)}

发现MAC地址

{report_data.get('new_devices', 0)}

新发现设备

扫描网段: {report_data.get('network_cidr', '-')}

扫描时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

""" subject = f"IPAM 扫描报告 - {report_data.get('network_cidr', '')}" return self.send_email(to, subject, html_content) # 全局邮件服务实例 _email_service = None def get_email_service() -> EmailService: """获取邮件服务实例""" global _email_service if _email_service is None: settings = EmailSettings() _email_service = EmailService(settings) return _email_service def init_email_service(settings: EmailSettings): """初始化邮件服务""" global _email_service _email_service = EmailService(settings)

⚠️ IPAM 告警通知

📊 IPAM 网段扫描报告

扫描结果概览