下载地址:https://www.pan38.com/share.php?code=JCnzE 提取密码:7789
这个邮件群发工具包含完整的模块化设计,主要功能包括:
使用前需要先在config.py中配置您的QQ邮箱和授权码,然后通过命令行参数运行程序。
# QQ邮箱SMTP服务器配置
SMTP_SERVER = "smtp.qq.com"
SMTP_PORT = 465
SSL_ENABLED = True
# 发件人账号配置
SENDER_EMAIL = "your_email@qq.com"
SENDER_PASSWORD = "your_authorization_code" # 注意是授权码不是QQ密码
# 邮件内容默认配置
DEFAULT_SUBJECT = "来自Python邮件群发系统的问候"
DEFAULT_CONTENT = "这是一封通过Python自动发送的测试邮件"
smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from email.header import Header
import os
from config import *
class EmailSender:
def __init__(self):
self.server = None
self.connect_server()
def connect_server(self):
try:
if SSL_ENABLED:
self.server = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT)
else:
self.server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT)
self.server.login(SENDER_EMAIL, SENDER_PASSWORD)
print("SMTP服务器连接成功")
except Exception as e:
print(f"连接SMTP服务器失败: {str(e)}")
raise
def send_email(self, to_email, subject, content,
content_type="plain", attachments=None):
if attachments is None:
attachments = []
msg = MIMEMultipart()
msg['From'] = SENDER_EMAIL
msg['To'] = to_email
msg['Subject'] = Header(subject, 'utf-8')
# 添加邮件正文
if content_type == "html":
msg.attach(MIMEText(content, 'html', 'utf-8'))
else:
msg.attach(MIMEText(content, 'plain', 'utf-8'))
# 添加附件
for attachment in attachments:
if os.path.exists(attachment):
with open(attachment, "rb") as f:
part = MIMEApplication(f.read())
part.add_header(
'Content-Disposition',
'attachment',
filename=os.path.basename(attachment)
)
msg.attach(part)
try:
self.server.sendmail(SENDER_EMAIL, to_email, msg.as_string())
print(f"邮件发送成功: {to_email}")
return True
except Exception as e:
print(f"发送邮件到 {to_email} 失败: {str(e)}")
return False
def close(self):
if self.server:
self.server.quit()
print("SMTP连接已关闭")
csv
import time
from typing import List
def read_recipients_from_csv(csv_file: str) -> List[str]:
"""从CSV文件读取收件人列表"""
recipients = []
try:
with open(csv_file, mode='r', encoding='utf-8') as file:
reader = csv.reader(file)
for row in reader:
if row and row[0].strip():
recipients.append(row[0].strip())
print(f"成功从 {csv_file} 读取 {len(recipients)} 个收件人")
except Exception as e:
print(f"读取CSV文件失败: {str(e)}")
return recipients
def generate_email_content(template: str, variables: dict) -> str:
"""使用模板和变量生成邮件内容"""
content = template
for key, value in variables.items():
content = content.replace(f"{{{{{key}}}}}", str(value))
return content
def batch_send_with_delay(sender, recipients: List[str],
subject: str, content: str,
delay_seconds: int = 5,
max_retries: int = 3,
attachments: List[str] = None):
"""批量发送邮件,带有延迟和重试机制"""
if attachments is None:
attachments = []
success_count = 0
fail_count = 0
for i, recipient in enumerate(recipients, 1):
print(f"\n正在处理第 {i}/{len(recipients)} 封邮件: {recipient}")
for attempt in range(1, max_retries + 1):
try:
if sender.send_email(recipient, subject, content,
attachments=attachments):
success_count += 1
break
else:
if attempt < max_retries:
print(f"第 {attempt} 次尝试失败,等待重试...")
time.sleep(2)
else:
fail_count += 1
print(f"达到最大重试次数 {max_retries},放弃发送")
except Exception as e:
print(f"发送过程中出现异常: {str(e)}")
fail_count += 1
break
if i < len(recipients):
print(f"等待 {delay_seconds} 秒后发送下一封...")
time.sleep(delay_seconds)
print(f"\n批量发送完成: 成功 {success_count} 封,失败 {fail_count} 封")
return success_count, fail_count
email_sender import EmailSender
from utils import read_recipients_from_csv, batch_send_with_delay
import argparse
import os
def main():
parser = argparse.ArgumentParser(description="QQ邮件群发工具")
parser.add_argument("-l", "--list", required=True,
help="包含收件人列表的CSV文件路径")
parser.add_argument("-s", "--subject",
help="邮件主题,默认为配置中的默认主题")
parser.add_argument("-c", "--content",
help="邮件正文内容或HTML文件路径")
parser.add_argument("-a", "--attachments", nargs="+",
help="附件文件路径,多个附件用空格分隔")
parser.add_argument("-d", "--delay", type=int, default=5,
help="每封邮件发送间隔(秒),默认5秒")
parser.add_argument("--html", action="store_true",
help="内容是否为HTML格式")
args = parser.parse_args()
# 读取收件人列表
recipients = read_recipients_from_csv(args.list)
if not recipients:
print("错误: 未找到有效的收件人地址")
return
# 设置邮件主题
subject = args.subject if args.subject else DEFAULT_SUBJECT
# 读取邮件内容
content = ""
if args.content:
if os.path.exists(args.content) and args.html:
with open(args.content, "r", encoding="utf-8") as f:
content = f.read()
else:
content = args.content
else:
content = DEFAULT_CONTENT
# 初始化邮件发送器
sender = EmailSender()
try:
# 批量发送邮件
batch_send_with_delay(
sender=sender,
recipients=recipients,
subject=subject,
content=content,
delay_seconds=args.delay,
attachments=args.attachments
)
finally:
sender.close()
if __name__ == "__main__":
main()
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。