"""
|
邮件发送服务
|
"""
|
|
import smtplib
|
from email.mime.text import MIMEText
|
from email.mime.multipart import MIMEMultipart
|
from typing import List, Optional
|
from loguru import logger
|
from config import settings
|
|
|
class EmailService:
|
"""邮件发送服务"""
|
|
def __init__(self):
|
self.smtp_server = settings.email_smtp_server
|
self.smtp_port = settings.email_smtp_port
|
self.username = settings.email_smtp_username
|
self.password = settings.email_smtp_password
|
self.from_email = settings.email_from_email
|
|
def send_email(self, to_emails: List[str], subject: str, content: str) -> bool:
|
"""
|
发送邮件
|
|
Args:
|
to_emails: 收件人邮箱列表
|
subject: 邮件主题
|
content: 邮件内容
|
|
Returns:
|
发送成功返回True，失败返回False
|
"""
|
if not settings.email_enabled:
|
logger.info("邮件发送功能已禁用")
|
return True
|
|
if not to_emails:
|
logger.warning("收件人邮箱列表为空")
|
return False
|
|
try:
|
# 创建邮件对象
|
msg = MIMEMultipart()
|
msg["From"] = self.from_email
|
msg["To"] = ", ".join(to_emails)
|
msg["Subject"] = subject
|
|
# 添加邮件内容
|
msg.attach(MIMEText(content, "plain", "utf-8"))
|
|
# 连接SMTP服务器并发送邮件
|
server = None
|
email_sent = False
|
|
try:
|
server = smtplib.SMTP(self.smtp_server, self.smtp_port)
|
server.starttls() # 启用TLS加密
|
server.login(self.username, self.password)
|
|
# 发送邮件
|
text = msg.as_string()
|
server.sendmail(self.from_email, to_emails, text)
|
email_sent = True
|
|
logger.info(f"邮件发送成功: to={to_emails}, subject={subject}")
|
|
finally:
|
# 安全关闭连接，忽略关闭时的异常
|
if server:
|
try:
|
server.quit()
|
except Exception:
|
# 忽略关闭连接时的异常，因为邮件可能已经发送成功
|
pass
|
|
return email_sent
|
|
except smtplib.SMTPAuthenticationError as e:
|
logger.error(f"邮件发送认证失败: error={str(e)}")
|
return False
|
except smtplib.SMTPException as e:
|
logger.error(f"邮件发送SMTP错误: error={str(e)}")
|
return False
|
except Exception as e:
|
logger.error(f"邮件发送异常: error={str(e)}")
|
return False
|
|
def send_notification(self, subject: str, content: str) -> bool:
|
"""
|
发送通知邮件到配置的收件人列表
|
|
Args:
|
subject: 邮件主题
|
content: 邮件内容
|
|
Returns:
|
发送成功返回True，失败返回False
|
"""
|
return self.send_email(settings.email_to_emails, subject, content)
|
|
def test_connection(self) -> bool:
|
"""
|
测试邮件服务器连接
|
|
Returns:
|
连接成功返回True，失败返回False
|
"""
|
if not settings.email_enabled:
|
logger.info("邮件发送功能已禁用")
|
return True
|
|
try:
|
server = None
|
connection_success = False
|
|
try:
|
server = smtplib.SMTP(self.smtp_server, self.smtp_port)
|
server.starttls()
|
server.login(self.username, self.password)
|
connection_success = True
|
|
logger.info("邮件服务器连接测试成功")
|
|
finally:
|
# 安全关闭连接，忽略关闭时的异常
|
if server:
|
try:
|
server.quit()
|
except Exception:
|
# 忽略关闭连接时的异常
|
pass
|
|
return connection_success
|
|
except smtplib.SMTPAuthenticationError as e:
|
logger.error(f"邮件服务器认证失败: error={str(e)}")
|
return False
|
except smtplib.SMTPException as e:
|
logger.error(f"邮件服务器连接SMTP错误: error={str(e)}")
|
return False
|
except Exception as e:
|
logger.error(f"邮件服务器连接异常: error={str(e)}")
|
return False
|
|
|
# 全局邮件服务实例
|
email_service = EmailService()
|
