konjacpotato
f93fda8efa
All checks were successful
Gitea Actions Demo / host-commands (push) Successful in 0s
71 lines
2.8 KiB
Python
71 lines
2.8 KiB
Python
import smtplib
|
||
from email.mime.text import MIMEText
|
||
from email.header import Header
|
||
from email.utils import formataddr
|
||
from utils import logger
|
||
from config import settings
|
||
|
||
|
||
class MailSender:
|
||
"""邮件发送工具，从环境变量读取SMTP配置"""
|
||
def execute(self, to_addrs, subject, body, from_addr=None, body_type='plain'):
|
||
"""
|
||
发送邮件
|
||
:param to_addrs: 收件人，可以是单个邮箱字符串或邮箱列表
|
||
:param subject: 邮件主题
|
||
:param body: 邮件正文
|
||
:param from_addr: 发件人地址，若为None则使用 SMTP_USER
|
||
:param body_type: 正文类型，'plain' 或 'html'
|
||
:return: True 表示发送成功
|
||
:raises ValueError: 配置缺失或发送失败时抛出异常
|
||
"""
|
||
# ----- 1. 获取SMTP配置 -----
|
||
smtp_host = settings.SMTP_HOST
|
||
smtp_port = settings.SMTP_PORT
|
||
username = settings.SMTP_USER
|
||
password = settings.SMTP_PASSWORD
|
||
use_tls = settings.SMTP_USE_TLS
|
||
from_name = settings.SMTP_FROM # 发件人显示名称，可选
|
||
|
||
# 必要配置校验
|
||
if not all([smtp_host, smtp_port, username, password]):
|
||
missing = []
|
||
if not smtp_host: missing.append(self.ENV_SMTP_HOST)
|
||
if not smtp_port: missing.append(self.ENV_SMTP_PORT)
|
||
if not username: missing.append(self.ENV_SMTP_USER)
|
||
if not password: missing.append(self.ENV_SMTP_PASSWORD)
|
||
raise ValueError(f"缺少必要的SMTP环境变量: {', '.join(missing)}")
|
||
|
||
# ----- 2. 构建邮件对象 -----
|
||
# 处理收件人格式
|
||
if isinstance(to_addrs, str):
|
||
to_addrs = [to_addrs]
|
||
to_str = ','.join(to_addrs)
|
||
|
||
# 确定发件人地址
|
||
if from_addr is None:
|
||
from_addr = username
|
||
|
||
# 创建邮件正文
|
||
msg = MIMEText(body, body_type, 'utf-8')
|
||
msg['Subject'] = Header(subject, 'utf-8')
|
||
# 发件人：可包含显示名称
|
||
if from_name:
|
||
msg['From'] = formataddr((Header(from_name, 'utf-8').encode(), from_addr))
|
||
else:
|
||
msg['From'] = from_addr
|
||
msg['To'] = to_str
|
||
|
||
# ----- 3. 发送邮件 -----
|
||
try:
|
||
logger.info(f"准备发送邮件，收件人: {to_str}, 主题: {subject}")
|
||
with smtplib.SMTP(smtp_host, int(smtp_port)) as server:
|
||
if use_tls:
|
||
server.starttls()
|
||
server.login(username, password)
|
||
server.send_message(msg)
|
||
logger.info(f"邮件发送成功，收件人: {to_str}")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"邮件发送失败: {e}")
|
||
raise ValueError(f"邮件发送失败: {e}") |