baodan/api/insurance/utils/email.py

"""邮件发送工具。"""
import smtplib
import socket
from email.message import EmailMessage
from html import escape

from insurance.config import get_config


class EmailSendError(Exception):
    """邮件发送失败，携带可展示给用户的错误信息。"""

    def __init__(self, user_message: str):
        super().__init__(user_message)
        self.user_message = user_message


def _get_mail_config(key: str, default=None):
    value = get_config(key, "")
    if value != "":
        return value
    try:
        from configs import dify_config
        value = getattr(dify_config, key, None)
        return default if value is None else value
    except Exception:
        return default


def _as_bool(value) -> bool:
    if isinstance(value, bool):
        return value
    return str(value or "").strip().lower() in ("true", "1", "yes", "on")


def _send_with_dify_mail(to_email: str, subject: str, content: str) -> bool:
    """优先复用 Dify 已初始化的邮件客户端，兼容 smtp/resend/sendgrid。"""
    try:
        from extensions.ext_mail import mail
    except Exception:
        return False

    if not getattr(mail, "is_inited", lambda: False)():
        return False

    html = "<pre style=\"font-family: sans-serif; white-space: pre-wrap;\">" + escape(content) + "</pre>"
    try:
        mail.send(to=to_email, subject=subject, html=html)
        return True
    except Exception as exc:
        raise EmailSendError("Dify 邮件服务发送失败，请检查邮件服务配置") from exc


def send_email(to_email: str, subject: str, content: str) -> bool:
    """通过 Dify 风格的 SMTP 配置发送纯文本邮件。"""
    if _send_with_dify_mail(to_email, subject, content):
        return True

    mail_type = str(_get_mail_config("MAIL_TYPE", "") or "").strip().lower()
    host = _get_mail_config("SMTP_SERVER", "")
    if mail_type and mail_type != "smtp":
        raise EmailSendError("当前邮件服务未初始化，请检查 Dify 邮件服务配置")
    if not mail_type and not host:
        raise EmailSendError("邮件服务未配置，请联系管理员")

    port = int(_get_mail_config("SMTP_PORT", "587"))
    username = _get_mail_config("SMTP_USERNAME", "")
    password = _get_mail_config("SMTP_PASSWORD", "")
    default_send_from = _get_mail_config("MAIL_DEFAULT_SEND_FROM", "")
    from_email = default_send_from

    # Dify 语义：SMTP_USE_TLS=true 且 SMTP_OPPORTUNISTIC_TLS=false 表示 SMTP_SSL；
    # SMTP_USE_TLS=true 且 SMTP_OPPORTUNISTIC_TLS=true 表示 STARTTLS。
    use_tls = _as_bool(_get_mail_config("SMTP_USE_TLS", "false"))
    opportunistic_tls = _as_bool(_get_mail_config("SMTP_OPPORTUNISTIC_TLS", "false"))
    use_starttls = use_tls and opportunistic_tls
    use_ssl = use_tls and not use_starttls

    placeholders = {"smtp.example.com", "smtp.gmail.com", "your_smtp_password", "no-reply@example.com"}
    if not host or not from_email:
        raise EmailSendError("邮件服务未配置，请联系管理员")
    if host in placeholders or password in placeholders or any(value in from_email for value in placeholders):
        raise EmailSendError("邮件服务仍是示例配置，请联系管理员")

    message = EmailMessage()
    message["Subject"] = subject
    message["From"] = from_email
    message["To"] = to_email
    message.set_content(content)

    try:
        if use_ssl:
            with smtplib.SMTP_SSL(host, port, timeout=10) as server:
                if username and password:
                    server.login(username, password)
                server.send_message(message)
        else:
            with smtplib.SMTP(host, port, timeout=10) as server:
                if use_starttls:
                    server.starttls()
                if username and password:
                    server.login(username, password)
                server.send_message(message)
    except smtplib.SMTPAuthenticationError as exc:
        raise EmailSendError("SMTP 认证失败，请检查邮箱账号或授权码") from exc
    except (smtplib.SMTPConnectError, socket.timeout, TimeoutError, OSError) as exc:
        raise EmailSendError("SMTP 连接失败，请检查服务器地址、端口和网络") from exc
    except smtplib.SMTPException as exc:
        raise EmailSendError("SMTP 发送失败，请检查邮箱服务配置") from exc

    return True