313 lines
9.8 KiB
Python
313 lines
9.8 KiB
Python
"""
|
||
邮件发送模块
|
||
自动将生成的报告通过QQ邮箱发送给指定用户
|
||
"""
|
||
import smtplib
|
||
import os
|
||
from email.mime.text import MIMEText
|
||
from email.mime.multipart import MIMEMultipart
|
||
from email.mime.base import MIMEBase
|
||
from email import encoders
|
||
from datetime import datetime
|
||
from typing import List, Optional
|
||
|
||
|
||
class EmailSender:
|
||
"""邮件发送器"""
|
||
|
||
def __init__(self, sender_email: str, sender_password: str, smtp_server: str = "smtp.qq.com", smtp_port: int = 465):
|
||
"""
|
||
初始化邮件发送器
|
||
|
||
Args:
|
||
sender_email: 发件人邮箱（QQ邮箱）
|
||
sender_password: 发件人密码（QQ邮箱授权码）
|
||
smtp_server: SMTP服务器地址
|
||
smtp_port: SMTP服务器端口
|
||
"""
|
||
self.sender_email = sender_email
|
||
self.sender_password = sender_password
|
||
self.smtp_server = smtp_server
|
||
self.smtp_port = smtp_port
|
||
|
||
def send_report(self,
|
||
recipient_emails: List[str],
|
||
report_path: str,
|
||
subject: Optional[str] = None,
|
||
body: Optional[str] = None,
|
||
cc_emails: Optional[List[str]] = None) -> bool:
|
||
"""
|
||
发送报告邮件
|
||
|
||
Args:
|
||
recipient_emails: 收件人邮箱列表
|
||
report_path: 报告文件路径
|
||
subject: 邮件主题（可选）
|
||
body: 邮件正文（可选）
|
||
cc_emails: 抄送邮箱列表（可选）
|
||
|
||
Returns:
|
||
是否发送成功
|
||
"""
|
||
try:
|
||
# 创建邮件对象
|
||
message = MIMEMultipart()
|
||
message['From'] = self.sender_email
|
||
message['To'] = ', '.join(recipient_emails)
|
||
|
||
if cc_emails:
|
||
message['Cc'] = ', '.join(cc_emails)
|
||
|
||
# 设置邮件主题
|
||
if subject is None:
|
||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
subject = f"交通新闻分析报告 - {timestamp}"
|
||
message['Subject'] = subject
|
||
|
||
# 设置邮件正文
|
||
if body is None:
|
||
body = self._generate_default_body(report_path)
|
||
message.attach(MIMEText(body, 'plain', 'utf-8'))
|
||
|
||
# 添加附件
|
||
if os.path.exists(report_path):
|
||
self._attach_file(message, report_path)
|
||
else:
|
||
print(f"[警告] 报告文件不存在: {report_path}")
|
||
return False
|
||
|
||
# 发送邮件
|
||
print(f"\n[邮件] 正在发送邮件...")
|
||
print(f" 发件人: {self.sender_email}")
|
||
print(f" 收件人: {', '.join(recipient_emails)}")
|
||
if cc_emails:
|
||
print(f" 抄送: {', '.join(cc_emails)}")
|
||
print(f" 主题: {subject}")
|
||
|
||
# 连接SMTP服务器并发送
|
||
with smtplib.SMTP_SSL(self.smtp_server, self.smtp_port) as server:
|
||
server.login(self.sender_email, self.sender_password)
|
||
|
||
# 合并收件人和抄送人
|
||
all_recipients = recipient_emails.copy()
|
||
if cc_emails:
|
||
all_recipients.extend(cc_emails)
|
||
|
||
server.send_message(message)
|
||
|
||
print(f"[成功] 邮件发送成功!")
|
||
return True
|
||
|
||
except smtplib.SMTPAuthenticationError:
|
||
print(f"[错误] 邮箱认证失败，请检查邮箱和授权码是否正确")
|
||
print(f"提示: QQ邮箱需要使用授权码，不是QQ密码")
|
||
return False
|
||
except smtplib.SMTPException as e:
|
||
print(f"[错误] 邮件发送失败: {str(e)}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"[错误] 发送邮件时出错: {str(e)}")
|
||
return False
|
||
|
||
def _attach_file(self, message: MIMEMultipart, file_path: str) -> None:
|
||
"""
|
||
添加附件到邮件
|
||
|
||
Args:
|
||
message: 邮件对象
|
||
file_path: 文件路径
|
||
"""
|
||
filename = os.path.basename(file_path)
|
||
|
||
with open(file_path, 'rb') as f:
|
||
part = MIMEBase('application', 'octet-stream')
|
||
part.set_payload(f.read())
|
||
|
||
encoders.encode_base64(part)
|
||
part.add_header('Content-Disposition', f'attachment; filename="{filename}"')
|
||
message.attach(part)
|
||
|
||
def _generate_default_body(self, report_path: str) -> str:
|
||
"""
|
||
生成默认邮件正文
|
||
|
||
Args:
|
||
report_path: 报告文件路径
|
||
|
||
Returns:
|
||
邮件正文
|
||
"""
|
||
timestamp = datetime.now().strftime("%Y年%m月%d日 %H:%M")
|
||
filename = os.path.basename(report_path)
|
||
|
||
body = f"""您好！
|
||
|
||
这是交通新闻自动报表系统生成的分析报告。
|
||
|
||
报告时间: {timestamp}
|
||
报告文件: {filename}
|
||
|
||
报告内容包括：
|
||
- 多源数据采集（赛文交通网、政府采购网等）
|
||
- 智能分析与总结
|
||
- 热点话题提取
|
||
- 趋势分析
|
||
|
||
请查收附件中的详细报告。
|
||
|
||
---
|
||
此邮件由交通新闻自动报表系统自动发送
|
||
如有问题，请联系系统管理员
|
||
"""
|
||
return body
|
||
|
||
def send_multiple_reports(self,
|
||
recipient_emails: List[str],
|
||
report_paths: List[str],
|
||
subject: Optional[str] = None,
|
||
body: Optional[str] = None) -> bool:
|
||
"""
|
||
发送多个报告文件
|
||
|
||
Args:
|
||
recipient_emails: 收件人邮箱列表
|
||
report_paths: 报告文件路径列表
|
||
subject: 邮件主题（可选）
|
||
body: 邮件正文（可选）
|
||
|
||
Returns:
|
||
是否发送成功
|
||
"""
|
||
try:
|
||
# 创建邮件对象
|
||
message = MIMEMultipart()
|
||
message['From'] = self.sender_email
|
||
message['To'] = ', '.join(recipient_emails)
|
||
|
||
# 设置邮件主题
|
||
if subject is None:
|
||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
subject = f"交通新闻分析报告（{len(report_paths)}份）- {timestamp}"
|
||
message['Subject'] = subject
|
||
|
||
# 设置邮件正文
|
||
if body is None:
|
||
body = self._generate_multiple_reports_body(report_paths)
|
||
message.attach(MIMEText(body, 'plain', 'utf-8'))
|
||
|
||
# 添加所有附件
|
||
for report_path in report_paths:
|
||
if os.path.exists(report_path):
|
||
self._attach_file(message, report_path)
|
||
else:
|
||
print(f"[警告] 报告文件不存在: {report_path}")
|
||
|
||
# 发送邮件
|
||
print(f"\n[邮件] 正在发送邮件...")
|
||
print(f" 发件人: {self.sender_email}")
|
||
print(f" 收件人: {', '.join(recipient_emails)}")
|
||
print(f" 主题: {subject}")
|
||
print(f" 附件数量: {len(report_paths)}")
|
||
|
||
with smtplib.SMTP_SSL(self.smtp_server, self.smtp_port) as server:
|
||
server.login(self.sender_email, self.sender_password)
|
||
server.send_message(message)
|
||
|
||
print(f"[成功] 邮件发送成功!")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"[错误] 发送邮件时出错: {str(e)}")
|
||
return False
|
||
|
||
def _generate_multiple_reports_body(self, report_paths: List[str]) -> str:
|
||
"""
|
||
生成多个报告的邮件正文
|
||
|
||
Args:
|
||
report_paths: 报告文件路径列表
|
||
|
||
Returns:
|
||
邮件正文
|
||
"""
|
||
timestamp = datetime.now().strftime("%Y年%m月%d日 %H:%M")
|
||
|
||
files_list = "\n".join([f" - {os.path.basename(path)}" for path in report_paths])
|
||
|
||
body = f"""您好！
|
||
|
||
这是交通新闻自动报表系统生成的分析报告。
|
||
|
||
报告时间: {timestamp}
|
||
报告数量: {len(report_paths)} 份
|
||
|
||
报告文件:
|
||
{files_list}
|
||
|
||
请查收附件中的详细报告。
|
||
|
||
---
|
||
此邮件由交通新闻自动报表系统自动发送
|
||
如有问题，请联系系统管理员
|
||
"""
|
||
return body
|
||
|
||
|
||
def main():
|
||
"""测试函数"""
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv()
|
||
|
||
# 从环境变量获取配置
|
||
sender_email = os.getenv("EMAIL_SENDER")
|
||
sender_password = os.getenv("EMAIL_PASSWORD")
|
||
recipient_email = os.getenv("EMAIL_RECIPIENT")
|
||
|
||
if not sender_email or not sender_password:
|
||
print("错误: 请在.env文件中配置邮箱信息")
|
||
print("\n请添加以下配置到.env文件:")
|
||
print("EMAIL_SENDER=你的QQ邮箱")
|
||
print("EMAIL_PASSWORD=你的QQ邮箱授权码")
|
||
print("EMAIL_RECIPIENT=收件人邮箱")
|
||
print("\n注意: EMAIL_PASSWORD是QQ邮箱的授权码，不是QQ密码")
|
||
print("获取授权码: QQ邮箱 > 设置 > 账户 > POP3/IMAP/SMTP/Exchange/CardDAV/CalDAV服务")
|
||
return
|
||
|
||
if not recipient_email:
|
||
print("警告: 未配置收件人邮箱，使用发件人邮箱作为收件人")
|
||
recipient_email = sender_email
|
||
|
||
# 创建邮件发送器
|
||
email_sender = EmailSender(sender_email, sender_password)
|
||
|
||
# 查找最新的报告文件
|
||
data_dir = os.getenv("DATA_DIR", "./data")
|
||
report_files = [f for f in os.listdir(data_dir) if f.startswith("report_") and (f.endswith(".md") or f.endswith(".txt"))]
|
||
|
||
if not report_files:
|
||
print("错误: 未找到报告文件，请先生成报告")
|
||
print("运行: python main.py --mode report")
|
||
return
|
||
|
||
# 获取最新的报告
|
||
report_files.sort(reverse=True)
|
||
latest_report = os.path.join(data_dir, report_files[0])
|
||
|
||
print(f"找到最新报告: {latest_report}")
|
||
|
||
# 发送邮件
|
||
success = email_sender.send_report(
|
||
recipient_emails=[recipient_email],
|
||
report_path=latest_report
|
||
)
|
||
|
||
if success:
|
||
print("\n测试成功！")
|
||
else:
|
||
print("\n测试失败，请检查配置")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|