ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

Python邮件群发工具实战:从SMTP协议到批量任务调度

Python邮件群发工具实战:从SMTP协议到批量任务调度 简介这是一款基于Python开发的永久免费、无广告的邮件群发工具面向需要批量发送SMTP邮件的运营人员、开发者与小团队。它支持添加多个发信邮箱、导入TXT文件或粘贴邮箱列表、查看与管理收件人并可为单个发信邮箱设置发送次数限制兼顾效率与送达稳定性。压缩包共1008个文件约19MB包含核心Python源码、pyd编译模块以及运行所需的dll、tcl等依赖资源另有少量gif、html界面素材整体结构完整既可直接运行也可作为二次开发参考。已有121人学习下载。虽然开发中原本设想的QQ群成员邮箱获取功能因平台限制未上线但核心邮件群发功能已可稳定使用。读者可借此快速搭建自己的邮件群发系统也能从中学习SMTP交互、多邮箱轮换、队列管理等Python实战技巧。1. 用 Python 做邮件群发工具先避开“发不出去”的坑如果你被同事叫去“写个脚本把这几百封邮件发一下”第一反应多半是打开邮箱联系人一封封复制粘贴。这种手工操作在 50 封以内勉强能忍超过 100 封之后重复劳动带来的不仅是时间损耗还有手滑发错收件人、漏掉附件、不小心把密送写进收件人列表之类的低级事故。用 Python 做邮件群发工具本质上是把 SMTP 协议、邮件编码规则、批量任务调度这三件事封装到一个可重复执行的脚本里再用 zip 打包分发到任意一台装了 Python 的机器上直接运行。这里先给出一个反直觉的结论Python 发邮件最大的障碍从来不是语法而是“退信”和“进垃圾箱”。真正的工程难点在于 SMTP 服务器的连接策略、收件人地址的合法性与去重、以及发送频率的控制。本文先讲 smtplib 和 email 库的最小可用实现,再讲批量发送的企业级落地套路最后给出 zip 分发的目录组织和接收端的一键运行方案。无论你是准备给内部员工发通知还是给客户做定期的营销邮件这套流程都能直接抄作业。2. 构建可复用的 Python 邮件发送核心模块2.1 SMTP 发送协议与 smtplib 的对应关系选对发送服务器是关键邮件群发的底层协议是 SMTPSimple Mail Transfer Protocol。Python 标准库中的smtplib负责与邮件服务器对话email库负责构造符合 RFC 5322 格式的邮件体。二者的分工是smtplib 管传输email 管内容。拆开理解就能避免后续排错时两头乱抓。实际开发中最常见的错误是直接用自己的企业邮箱域名作为 SMTP 服务器地址却没有确认该服务器是否开放了 SMTP 提交端口。目前主流邮箱服务商对第三方发信都有严格限制常见配置如下服务商SMTP 服务器SSL 端口认证方式限制腾讯企业邮smtp.exmail.qq.com465授权码单账号每日约 200 封网易企业邮smtp.qiye.163.com465授权码单账号每日约 500 封Gmailsmtp.gmail.com465应用专用密码单账号每日约 500 封阿里云企业邮smtp.qiye.aliyun.com465企业邮箱密码单账号每日约 200 封选择 SMTP 服务器时要看两个指标一是单账号日限额二是连接超时参数。群发 1000 封邮件不能只靠一个账号硬发否则触发频率限制后会被临时封禁。常见的做法是准备多个账号在代码里做轮询分发。2.2 最小可用脚本从纯文本到 MIME 格式的完整代码先写一个不依赖任何第三方库的最小发送函数它能验证 Python 环境、SMTP 配置、网络连通性这三件事是否都正常后续所有批量逻辑都在这层外壳上扩展。import smtplib from email.mime.text import MIMEText from email.header import Header from email.utils import formataddr def send_text_mail(smtp_host, smtp_port, sender, auth_code, receivers, subject, content): # 构造 MIMEText 对象subtype 为 plain 表示纯文本 msg MIMEText(content, plain, utf-8) msg[From] formataddr((str(Header(发件人, utf-8)), sender)) msg[To] ,.join(receivers) msg[Subject] Header(subject, utf-8) # 建立 SSL 连接并登录发送后显式退出 with smtplib.SMTP_SSL(smtp_host, smtp_port, timeout10) as server: server.login(sender, auth_code) server.sendmail(sender, receivers, msg.as_string())这段代码里最值得玩味的是formataddr和Header的组合。它负责把发件人姓名和邮箱地址编码成符合邮件头的格式避免中文发件人姓名在部分客户端显示为乱码。MIMEText的utf-8参数指定了邮件体的字符集如果漏掉这一步发往海外邮箱的邮件可能在标题处出现?UTF-8?B?...?的原始编码串。server.sendmail的第二个参数必须传列表格式单个字符串会造成部分邮件服务器解析异常。2.3 HTML 模板与附件构造不能只发纯文本实际群发场景中纯文本邮件的打开率远低于 HTML 邮件。发活动通知、周报汇总时需要嵌入图片、列表、按钮等排版元素。email.mime.multipart支持将纯文本和 HTML 版本同时放入同一封邮件这被称为 multipart/alternative 结构兼容性最好的做法是同时提供两者。from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email import encoders def build_html_mail(receiver, sender, subject, html_content, attachment_pathNone): # alternative 结构让不支持 HTML 的客户端自动显示纯文本版本 msg MIMEMultipart(alternative) msg[From] sender msg[To] receiver msg[Subject] Header(subject, utf-8) # 纯文本兜底版本 text_part MIMEText(您的邮件客户端不支持 HTML请切换到支持 HTML 的客户端查看。, plain, utf-8) html_part MIMEText(html_content, html, utf-8) msg.attach(text_part) msg.attach(html_part) if attachment_path: with open(attachment_path, rb) as f: part MIMEBase(application, octet-stream) part.set_payload(f.read()) encoders.encode_base64(part) part.add_header(Content-Disposition, attachment, filenameattachment_path.split(/)[-1]) msg.attach(part) return msg.as_string()这里MIMEMultipart(alternative)是容易被忽视的性能细节。如果只把 HTML 部分 attach 进MIMEMultipart()默认是 mixed 类型部分严格的垃圾邮件过滤网关会直接判定为“缺纯文本版本”导致进垃圾箱概率上升。附件读取用rb二进制模式再用encode_base64转码这一步不做的话视频、PDF 等二进制附件在邮件客户端会损坏。展开来讲MIMEBase的octet-stream是通用二进制类型对 xlsx、pdf、zip 等文件无需另设 MIME 类型反而能绕过部分附件类型过滤规则。3. 邮件群发的批量任务调度并发、失败重试与频率控制3.1 收件人清单的去重与邮箱合法性校验先过滤再发送批量群发和单发的最本质区别在于收件人数据质量。真实业务场景里的收件人列表来源各不相同Excel 导出、CRM 系统拉取、网页表单收集这些数据里混着空值、重复项、格式错误的地址。如果把这些脏数据直接喂给发送循环轻则浪费时间重试重则触发服务商的地址错误率上限。先做一次静态过滤再进入发送队列。下面给出一个不依赖第三方库的校验函数import re from email.utils import parseaddr def clean_receiver_list(raw_list): seen set() cleaned [] email_re re.compile(r^[\w.-][\w-](\.[\w-])$) for item in raw_list: addr parseaddr(item)[1].strip().lower() if not addr or not email_re.match(addr): print(f[跳过] 非法地址: {item}) continue if addr in seen: print(f[跳过] 重复地址: {addr}) continue seen.add(addr) cleaned.append(addr) return cleanedparseaddr负责把张三 zhangsanexample.com这种带显示名的字符串拆出真实邮箱地址lower()统一成小写是因为邮箱地址的域名部分不区分大小写。正则里的\w默认匹配下划线和数字但注意它不匹配中文字符因此有国际化域名IDN需求的要单独处理。这一步过滤后还可以根据业务需要做域名白名单只允许发给yourcompany.com后缀的内部邮件避免员工把公司通知误发到外部。3.2 发送速率控制在服务商限制之内防止账号被锁邮箱服务商的日限额是硬性指标但更隐蔽的是“瞬时速率限制”。单账号一分钟内发送超过 15~30 封服务商大概率返回550 4.7.0或454这类临时错误码。把每个收件人的发送间隔拉长是不错的办法但多账号轮询的效率更高。import time import threading from queue import Queue SEND_INTERVAL 2 # 单账号每次发送后的等待秒数 MAX_RETRY 3 # 单封邮件最大重试次数 retry_log [] def worker(account, task_queue): # account 结构: {sender:..., auth:..., smtp_host:..., smtp_port:...} smtp smtplib.SMTP_SSL(account[smtp_host], account[smtp_port], timeout10) smtp.login(account[sender], account[auth]) while not task_queue.empty(): try: receiver, subject, html task_queue.get_nowait() send_with_retry(smtp, account[sender], receiver, subject, html) except smtplib.SMTPServerDisconnected: print(f{account[sender]} 连接断开重新连接) smtp smtplib.SMTP_SSL(account[smtp_host], account[smtp_port], timeout10) smtp.login(account[sender], account[auth]) finally: task_queue.task_done() time.sleep(SEND_INTERVAL) smtp.quit() def send_with_retry(smtp, sender, receiver, subject, html): for attempt in range(MAX_RETRY): try: msg build_html_mail(receiver, sender, subject, html) smtp.sendmail(sender, [receiver], msg) return except smtplib.SMTPRecipientsRefused: # 收件人不存在这类硬错误重试无意义 retry_log.append((receiver, RecipientsRefused)) return except smtplib.SMTPSenderRefused: print(f发件人被拒绝: {sender}请检查账号状态) raise except smtplib.SMTPResponseException as e: if e.smtp_code in (451, 452, 550, 554): # 限流或临时不可达等指数时间再试 time.sleep(2 ** attempt) else: retry_log.append((receiver, e.smtp_code)) return这段代码里有两个设计意图值得展开说。第一task_queue.empty()只是粗略判断多线程下容易出现队列已空但还有 worker 在取数据的竞态严格做法是用哨兵值标记任务结束但群发场景下轻微重复可以接受空队列退出反而更简单。第二重试策略区分了“收件人拒绝”和“服务商限流”两类异常前者是百重试不变的硬失败后者需要拉长间隔。2 ** attempt是典型的指数退避三次重试分别等待 1 秒、2 秒、4 秒不会给服务商造成二次压力。3.3 多账号轮询与失败日志落盘群发任务的可观测性生产环境的群发工具必须留下完整的发送轨迹否则收到退信时无法定位是哪一批次、哪个账号、哪个收件人出了问题。常见做法是用csv模块把每次发送的结果追加到日志文件同时维护一个 SQLite 数据库存储任务状态。轻量场景下 CSV 足够字段设计如下import csv from datetime import datetime def log_result(result_file, receiver, status, detail, sender): file_exists os.path.isfile(result_file) with open(result_file, a, newline, encodingutf-8) as f: writer csv.writer(f) if not file_exists: writer.writerow([time, sender, receiver, status, detail]) writer.writerow([ datetime.now().isoformat(), sender, receiver, status, detail ])这类日志文件的作用不只是事后追溯。把状态字段设计成success / refused / timeout / invalid四种枚举后可以用一条 awk 命令统计各状态的分布判断某个邮箱账号是否触发了限流阈值。我一般会在脚本里加一个--summary参数任务结束后自动打印成功率这样发完 500 封邮件不用人工翻日志。4. 把工具打包成 zip 分发到任意机器运行4.1 目录结构设计与配置文件模板开发调试完毕的邮件群发工具最终要以 zip 压缩包形式交付给同事或部署到服务器。这里的关键不是打包动作本身而是目录结构要自洽接收方解压后不需要理解源码逻辑直接改配置就能跑。标准目录结构如下mail_sender/ ├── config/ │ ├── accounts.json # SMTP 账号池 │ └── task.json # 本次发送任务配置 ├── data/ │ ├── receivers.txt # 收件人列表每行一个 │ └── template.html # 邮件正文模板 ├── logs/ # 运行日志与发送结果 ├── attachments/ # 附件目录可留空 ├── main.py # 主入口 └── requirements.txt # 依赖清单标准库则无需本例完全使用 Python 标准库所以requirements.txt可以留空这极大地降低接收方的部署成本。accounts.json的设计要从多账号轮询的需求反推每个账号条目应包含发送间隔、退出策略等辅助字段而不只是邮箱和授权码。{ accounts: [ { sender: noreplyexample.com, auth_code: your-auth-code, smtp_host: smtp.exmail.qq.com, smtp_port: 465, send_interval: 2.5, max_sends_per_minute: 20 } ], global: { retry_times: 3, dry_run: false } }将配置抽离成 JSON 而不是硬编码在源码里是为了让非 Python 开发者也能安全修改。auth_code字段是唯一不需要手写的敏感信息直接扫码从服务商后台获取即可。4.2 压缩打包命令与解压后的依赖安装打包过程用 zip 命令配合-x参数排除缓存与日志文件确保压缩包体积最小cd ~/workspace zip -r mail_sender.zip mail_sender \ -x mail_sender/__pycache__/* \ -x mail_sender/logs/* \ -x mail_sender/.DS_Store接收端拿到mail_sender.zip后第一步不是急着运行而是确认目标机器的 Python 版本。当前主流 Linux 发行版的系统自带 Python 版本通常在 3.8 到 3.12 之间脚本中用到的smtplib、email、csv、queue全部是标准库不会出现版本兼容问题。建议运行环境检测命令python3 --version python3 main.py --check--check参数是我习惯加上的启动选项它只做三件事验证 SMTP 用户名密码能否登录、测试网络到 smtp 服务器是否连通、以及解析任务配置的 JSON 语法。写死在代码里会比在 readme 里写“请先检查网络”有效得多。4.3 用户配置的三处必改点与常见误用接收方拿到工具后第一时间会改的配置集中在三处accounts.json里的 SMTP 授权码、task.json里的邮件主题和发件人别名、receivers.txt的收件人清单。这三处如果改错功能直接不可用。配置位置必改内容常见错误accounts.jsonauth_code填授权码误填邮箱登录密码导致SMTPAuthenticationErrortask.jsonsender_name填公司显示名留空显示为邮箱前缀task.jsonsubject填邮件标题用中文时忘记配置 UTF-8收件端乱码receivers.txt的解析规则要特意在代码里写明“每行一个邮箱分号/逗号隔开的均视为非法”这样能过滤掉从 Excel 复制粘贴的残留空格。同时保留#开头的注释行允许用户暂时屏蔽某些收件人。这些规则的实现比预想的简单就是读文件后逐行strip然后跳过#开头行再交给clean_receiver_list做合法性校验。5. 单账号发送速度的极限压测与 SMTP 会话诊断当群发任务跑完日志显示全部成功后还有一项容易被忽略的验证工作——检查是否有邮件被服务商静默吞掉。SMTP 协议的设计决定了发件方只能确认“服务商接收了这封邮件”无法确认“收件人真的看到了”。所以最后一章聚焦于从协议层验证发送效果、定位被限流的会话特征。SMTP 交互过程可以通过在 smtplib 的SMTP_SSL和sendmail之间插入set_debuglevel(1)打印完整的报文。下面这段代码会输出客户端和服务器的每一行命令import smtplib smtp smtplib.SMTP_SSL(smtp.exmail.qq.com, 465, timeout10) smtp.set_debuglevel(1) smtp.login(noreplyexample.com, auth-code) smtp.sendmail( noreplyexample.com, [testgmail.com], msg.as_string() )运行后会看到类似输出send: ehlo [127.0.0.1]\r\n reply: b250-smtp.exmail.qq.com\r\n reply: b250-AUTH LOGIN PLAIN\r\n send: auth plain ...\r\n reply: b235 Authentication successful\r\n send: mail FROM:noreplyexample.com\r\n reply: b250 OK\r\n send: rcpt TO:testgmail.com\r\n reply: b250 OK\r\n send: data\r\n reply: b354 End data with CRLF.CRLF\r\n send: .... reply: b250 OK: queued as XYZ123\r\n如果rcpt TO返回的代码是550意味着服务商拒绝接收该收件人地址返回452表示资源不足通常是发送过快触发限流421表示服务不可用可能是 IP 被列入临时黑名单。在诊断脚本里对这三个码做显式判断归入不同的重试策略比笼统地捕获SMTPResponseException更精确。单账号的发送速度极限可以通过二分法压测得出初始设置send_interval2秒跑 50 封观察是否有452或451返回如果没有改为1秒再跑 50 封直到出现限流错误码后回退到上一档值。这个值在不同服务商间差异巨大不要照搬网上的默认值。实测中同一个企业邮账号A 网络下 1 秒间隔正常B 网络下 3 秒间隔仍有 452原因是服务商的限流基于 IP 而非单纯账号维度。最后提醒一个在群发工具里比功能更重要的参数异常重试时MAX_RETRY不要超过 3 次。超过 3 次仍然失败的地址要么是域名拼写错误要么是账号权限不足重试只是放大服务商侧的压力。把这类地址单独导出到failed.csv人工确认后再做第二轮比无限重试更可靠。本文还有配套的精品资源点击获取
返回列表