112 lines
4.3 KiB
Python
112 lines
4.3 KiB
Python
"""邮件发送工具。"""
|
||
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
|