""" 异步邮件发送工具 使用 aiosmtplib 实现异步 SMTP 邮件发送，支持： - 纯文本邮件 - HTML 邮件 - 批量发送（并发） - 自动 TLS/SSL """ import asyncio import logging from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from typing import Dict, List, Optional import aiosmtplib from app.config import settings logger = logging.getLogger(__name__) class EmailSender: """异步邮件发送器""" @staticmethod async def _get_smtp_config() -> dict: """ 通过 config_manager 三级获取 SMTP 配置（Redis → 数据库 → env） """ from app.config_manager import config_manager group_config = await config_manager.get_group("notify_email") return { "host": group_config.get("smtp_host") or None, "port": int(group_config.get("smtp_port") or settings.SMTP_PORT or 465), "user": group_config.get("smtp_user") or None, "password": group_config.get("smtp_password") or None, "use_tls": str(group_config.get("smtp_use_tls", "True")).lower() in ("true", "1", "yes"), "from_name": group_config.get("smtp_from_name") or settings.APP_NAME, "from_email": group_config.get("smtp_from_email") or group_config.get("smtp_user") or None, } @staticmethod def is_configured() -> bool: """检查 SMTP 是否已配置（仅检查 env，同步兼容）""" return bool(settings.SMTP_HOST and settings.SMTP_USER and settings.SMTP_PASSWORD) @staticmethod async def is_configured_async() -> bool: """检查 SMTP 是否已配置（通过 config_manager 三级获取）""" config = await EmailSender._get_smtp_config() return bool(config["host"] and config["user"] and config["password"]) @staticmethod def _build_from_address(config: dict) -> str: """根据配置构建发件人地址""" from_email = config["from_email"] or config["user"] from_name = config["from_name"] or settings.APP_NAME return f"{from_name} <{from_email}>" @staticmethod def _get_from_address() -> str: """获取发件人地址（同步兼容，仅读 env）""" from_email = settings.SMTP_FROM_EMAIL or settings.SMTP_USER from_name = settings.SMTP_FROM_NAME or settings.APP_NAME return f"{from_name} <{from_email}>" @staticmethod async def send( to_email: str, subject: str, content: str, html: bool = False, ) -> bool: """ 发送单封邮件 Args: to_email: 收件人邮箱 subject: 邮件主题 content: 邮件内容（纯文本或 HTML） html: 是否为 HTML 内容 Returns: 是否发送成功 """ config = await EmailSender._get_smtp_config() if not (config["host"] and config["user"] and config["password"]): logger.warning("SMTP 未配置，跳过邮件发送") return False try: msg = MIMEMultipart("alternative") msg["From"] = EmailSender._build_from_address(config) msg["To"] = to_email msg["Subject"] = subject if html: msg.attach(MIMEText(content, "html", "utf-8")) else: msg.attach(MIMEText(content, "plain", "utf-8")) # sender 必须使用 smtp_user，部分邮箱（如QQ）要求 MAIL FROM 与认证用户一致 sender = config["user"] if config["use_tls"]: await aiosmtplib.send( msg, sender=sender, hostname=config["host"], port=config["port"], username=config["user"], password=config["password"], use_tls=True, ) else: await aiosmtplib.send( msg, sender=sender, hostname=config["host"], port=config["port"], username=config["user"], password=config["password"], start_tls=True, ) logger.info(f"邮件发送成功: {to_email}") return True except Exception as e: logger.error(f"邮件发送失败 [{to_email}]: {e}") return False @staticmethod async def send_batch( to_emails: List[str], subject: str, content: str, html: bool = False, ) -> Dict[str, bool]: """ 批量发送邮件（并发） Args: to_emails: 收件人邮箱列表 subject: 邮件主题 content: 邮件内容 html: 是否为 HTML 内容 Returns: 每个邮箱的发送结果 """ config = await EmailSender._get_smtp_config() if not (config["host"] and config["user"] and config["password"]): logger.warning("SMTP 未配置，跳过批量邮件发送") return {email: False for email in to_emails} tasks = [ EmailSender.send(email, subject, content, html) for email in to_emails ] results = await asyncio.gather(*tasks, return_exceptions=True) return { email: (result is True) for email, result in zip(to_emails, results) } @staticmethod def build_notification_html(title: str, content: str, app_name: Optional[str] = None) -> str: """ 构建通知邮件的 HTML 模板 Args: title: 通知标题 content: 通知内容 app_name: 应用名称 Returns: HTML 字符串 """ app = app_name or settings.APP_NAME return f"""

{app}

{title}

{content}

This email was sent by {app}. Please do not reply directly.

"""