smtplib与email:邮件自动发送

Python 办公自动化专题 · 用Python实现邮件发送全自动化

专题:Python 自动化办公系统学习

关键词:Python, 自动化办公, smtplib, email, 邮件发送, SMTP, 邮件自动化, Python办公

一、邮件协议基础

电子邮件系统是互联网最经典的应用之一,其背后依赖一套完整的协议体系。要理解Python的smtplib和email库,首先需要掌握三个核心协议:SMTP、POP3和IMAP。SMTP(Simple Mail Transfer Protocol,简单邮件传输协议)负责邮件的发送和中继,默认使用25号端口,加密版本使用465(SSL)或587(TLS)端口。当我们调用smtplib发送邮件时,本质上就是通过SMTP协议与邮件服务器通信,将邮件内容传递给服务器,再由服务器转发到收件人的邮箱服务商。

POP3(Post Office Protocol version 3)和IMAP(Internet Message Access Protocol)则是收件协议,用于从邮件服务器接收邮件。POP3将邮件下载到本地后通常删除服务器上的副本,适合单设备使用的场景;IMAP则保持邮件在服务器上,支持多设备同步,适合移动办公。在实际的自动化办公场景中,我们通常只用SMTP发送邮件,但如果需要监控邮件回复或自动处理收到的邮件,则需要结合IMAP或POP3协议。

邮件在传输过程中,其内容格式遵循MIME(Multipurpose Internet Mail Extensions,多用途互联网邮件扩展)标准。MIME定义了邮件内容的类型、编码方式和多部分结构。简单地说,一个MIME邮件可以包含多个部分,每个部分可以是不同的内容类型——纯文本、HTML、图片、附件等。Python的email库提供了MIME相关模块(如MIMEText、MIMEMultipart、MIMEBase、MIMEImage等),让我们可以方便地构建符合MIME标准的邮件对象。理解MIME的分层结构对于构建复杂邮件至关重要:外层是一个multipart容器,内部嵌套多个独立的MIME部分,每个部分有各自的Content-Type和Content-Transfer-Encoding头信息。

核心要点:SMTP负责发,POP3/IMAP负责收。MIME标准定义了邮件内容的组织方式。Python的smtplib对标SMTP协议,email库对标MIME标准,二者配合即可实现任意类型的邮件发送。

邮件从发送到接收的完整流程可以概括为以下几步:用户客户端通过SMTP将邮件提交到发件人的邮件服务器;发件服务器查询收件人域名的MX(Mail Exchange)记录,确定收件服务器的地址;发件服务器与收件服务器建立SMTP连接,传输邮件内容;收件服务器将邮件存入收件人的邮箱;收件人通过POP3或IMAP协议从服务器收取邮件。在Python中,smtplib帮我们封装了与SMTP服务器通信的所有细节,我们只需要提供服务器地址、端口、账户信息即可。

1.1 SMTP协议详解

SMTP使用命令-响应的工作模式。客户端发送命令,服务器返回状态码和响应信息。常见的命令包括HELO/EHLO(握手)、MAIL FROM(发件人)、RCPT TO(收件人)、DATA(邮件内容)、QUIT(退出连接)。smtplib在底层自动处理这些命令,但理解其原理有助于排错——比如收到"550"状态码通常表示邮件被拒,收到"535"表示认证失败。

# SMTP协议底层命令示例(理解原理) # 实际连接时smtplib会自动发送这些命令 HELO my-pc EHLO my-pc AUTH LOGIN MAIL FROM:<sender@example.com> RCPT TO:<recipient@example.com> DATA Subject: Test Email From: sender@example.com To: recipient@example.com Hello, this is a test email. . QUIT

1.2 MIME格式详解

MIME标准扩展了邮件内容格式,使其可以包含非ASCII字符、附件和多媒体内容。MIME邮件通过Content-Type头部声明内容类型,常见的类型有text/plain(纯文本)、text/html(HTML)、multipart/mixed(混合内容,如含附件)、multipart/alternative(替代内容,如同时提供文本和HTML版本)、image/jpeg(图片附件)、application/pdf(PDF附件)等。Content-Transfer-Encoding头部指定内容的传输编码方式,常见的编码有base64(二进制数据)、quoted-printable(主要含ASCII的可读文本)和7bit/8bit(纯ASCII或扩展ASCII文本)。

# MIME邮件结构示意图 # multipart/mixed (外层) # | # +-- multipart/alternative (邮件正文) # | | # | +-- text/plain (纯文本版本) # | +-- text/html (HTML版本) # | # +-- image/jpeg (内嵌图片,通过CID引用) # +-- application/pdf (附件) # +-- application/vnd.openxmlformats (Excel附件)
# 查看MIME邮件头结构 from email import message_from_string raw_email = """From: sender@example.com To: recipient@example.com Subject: MIME示例 MIME-Version: 1.0 Content-Type: text/plain; charset="utf-8" 这是邮件正文""" msg = message_from_string(raw_email) print("From:", msg["From"]) print("To:", msg["To"]) print("Subject:", msg["Subject"]) print("Content-Type:", msg.get_content_type()) print("Payload:", msg.get_payload())

二、smtplib核心

smtplib是Python标准库中用于发送邮件的核心模块,它封装了SMTP协议的客户端实现,提供了一套简洁而强大的API。使用smtplib发送邮件的基本流程包括:创建SMTP连接、登录认证、发送邮件、关闭连接。smtplib提供了两种连接方式:普通SMTP(使用SMTP类,默认端口25)和SSL加密SMTP(使用SMTP_SSL类,默认端口465)。此外,还可以使用starttls()方法在普通连接基础上升级为TLS加密连接(端口587)。

在实际开发中,推荐优先使用SSL或TLS加密方式,避免邮件内容和登录密码在网络上明文传输。绝大多数现代邮件服务商都已禁用25端口的明文SMTP。对于主流邮箱服务,QQ邮箱支持465(SSL)和587(TLS),163邮箱同样支持465和587,Gmail使用587(TLS),企业邮箱则根据服务商配置各有不同。

smtplib的debug级别设置对于排查发送失败问题非常有帮助。通过设置set_debuglevel(1),可以打印出客户端与服务器之间的所有通信细节,包括发送的命令和服务器的响应码。常见的错误码:220表示服务就绪,235表示认证成功,250表示请求完成,354表示开始输入邮件内容,550表示邮件被拒(可能是被反垃圾策略拦截)。

2.1 基础连接与发送

# smtplib基础使用——SSL加密方式 import smtplib from email.mime.text import MIMEText # 创建SSL连接(推荐) with smtplib.SMTP_SSL("smtp.qq.com", 465, timeout=30) as server: # 登录认证 server.login("your_email@qq.com", "授权码") # 构建邮件 msg = MIMEText("这是邮件正文内容", "plain", "utf-8") msg["From"] = "your_email@qq.com" msg["To"] = "recipient@example.com" msg["Subject"] = "测试邮件" # 发送 server.send_message(msg) print("邮件发送成功!")

2.2 TLS加密与调试模式

# 使用TLS加密 + 调试模式 import smtplib from email.mime.text import MIMEText # 创建普通SMTP连接 server = smtplib.SMTP("smtp.qq.com", 587, timeout=30) # 开启调试模式:打印与服务器通信的所有细节 server.set_debuglevel(1) # 开始TLS加密 server.starttls() # 登录 server.login("your_email@qq.com", "授权码") # 构建并发送 msg = MIMEText("调试模式下的测试邮件", "plain", "utf-8") msg["From"] = "your_email@qq.com" msg["To"] = "recipient@example.com" msg["Subject"] = "TLS测试" server.send_message(msg) server.quit()

2.3 异常处理

# 完善的异常处理 import smtplib from email.mime.text import MIMEText import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def send_email(sender, password, recipient, subject, body): try: with smtplib.SMTP_SSL("smtp.qq.com", 465, timeout=30) as server: server.login(sender, password) msg = MIMEText(body, "plain", "utf-8") msg["From"] = sender msg["To"] = recipient msg["Subject"] = subject server.send_message(msg) logger.info(f"邮件发送成功 -> {recipient}") return True except smtplib.SMTPAuthenticationError: logger.error("认证失败:请检查账号和授权码") except smtplib.SMTPRecipientsRefused: logger.error("收件人被拒:请检查收件人地址") except smtplib.SMTPServerDisconnected: logger.error("服务器连接断开") except smtplib.SMTPException as e: logger.error(f"SMTP错误:{e}") except Exception as e: logger.error(f"未知错误:{e}") return False

三、纯文本邮件

纯文本邮件是最基础也最兼容的邮件形式,几乎所有的邮件客户端都能正确显示。在Python中,构建纯文本邮件主要使用email.mime.text模块中的MIMEText类。MIMEText构造函数接受三个参数:邮件正文内容、内容类型("plain"表示纯文本)和字符编码(通常使用"utf-8")。邮件头的设置则通过对MIMEText对象的字典式操作完成:设置From(发件人地址及可选名称)、To(收件人地址)、Subject(邮件主题)等标准邮件头字段。

在实际应用中,邮件头设置有一些重要的细节需要注意。首先,如果希望显示发件人名称而非纯邮箱地址,可以使用RFC 2047格式,例如"=?utf-8?B?5Y+R5L6L5Lq6?= ",或者更简单地在地址前加名称并使用Header对象。其次,当需要发送给多个收件人时,To字段可以包含多个以逗号分隔的地址,但注意send_message()方法的第二个参数(实际收件人列表)必须与To字段一致,否则服务器会报错。Cc(抄送)和Bcc(密送)是常用的附加收件人类型,Cc的收件人会看到彼此,Bcc的收件人则互相不可见。

编码处理是纯文本邮件中容易出错的环节。当邮件主题或发件人名称包含中文时,必须进行Base64或Quoted-Printable编码。Python的email.header.Header类可以自动处理这种编码转换。此外,正文中的特殊字符(如超过ASCII范围的Unicode字符)也需要正确处理,设置charset="utf-8"后MIMEText会自动完成编码。

3.1 基础纯文本邮件

# 基础纯文本邮件 from email.mime.text import MIMEText from email.header import Header import smtplib # 构建邮件正文 body = """您好, 这是一封自动化测试邮件。 本邮件由Python程序自动发送,请勿回复。 祝工作愉快! 自动化办公系统""" # 创建MIMEText对象 msg = MIMEText(body, "plain", "utf-8") # 设置邮件头——含中文名称 msg["From"] = "自动化系统 " msg["To"] = "user@example.com" msg["Subject"] = Header("系统测试邮件", "utf-8").encode() # 发送 with smtplib.SMTP_SSL("smtp.qq.com", 465) as server: server.login("sender@qq.com", "授权码") server.send_message(msg)

3.2 多收件人 + Cc + Bcc

# 多个收件人、抄送、密送 from email.mime.text import MIMEText from email.header import Header import smtplib msg = MIMEText("这是一封群发测试邮件", "plain", "utf-8") msg["From"] = "通知中心 " msg["To"] = "user1@example.com, user2@example.com" msg["Cc"] = "manager@example.com" msg["Subject"] = Header("重要通知", "utf-8").encode() # 全部收件人列表(包括To、Cc、Bcc) all_recipients = [ "user1@example.com", "user2@example.com", "manager@example.com", "bcc_user@example.com" # 密送 ] with smtplib.SMTP_SSL("smtp.qq.com", 465) as server: server.login("sender@qq.com", "授权码") server.send_message(msg, all_recipients)

3.3 使用Header正确处理中文编码

# 使用Header类处理中文 from email.mime.text import MIMEText from email.header import Header from email.utils import formataddr import smtplib def build_message(sender_name, sender_addr, subject, body): msg = MIMEText(body, "plain", "utf-8") # formataddr自动处理名称编码 msg["From"] = formataddr((Header(sender_name, "utf-8").encode(), sender_addr)) msg["To"] = formataddr((Header("尊敬的客户", "utf-8").encode(), "client@example.com")) msg["Subject"] = Header(subject, "utf-8").encode() # 添加日期头 from email.utils import formatdate msg["Date"] = formatdate(localtime=True) # 添加消息ID from email.utils import make_msgid msg["Message-ID"] = make_msgid() return msg msg = build_message( "客户服务部", "service@example.com", "关于账户安全的通知", "尊敬的客户,您的账户安全状态正常。" )

四、HTML邮件

HTML邮件允许发送格式丰富的邮件内容,包括字体样式、颜色、布局、图片和链接等。在Python中,构建HTML邮件同样使用MIMEText类,只需将第二个参数改为"html"即可。HTML邮件的核心优势是可以制作精美的邮件模板,适合营销邮件、通知公告、数据报表等场景。然而,HTML邮件也面临兼容性问题——不同的邮件客户端(Outlook、Gmail、QQ邮箱、网易邮箱等)对CSS的支持程度差异很大。

编写HTML邮件时,需要遵循一些重要原则:使用内联样式而非外部或嵌入样式表,因为大多数邮件客户端会移除<style>标签;使用表格布局而非div布局,因为Outlook对div的支持很差;避免使用JavaScript,所有邮件客户端都会禁用脚本;图片需要使用绝对URL(https链接)而非相对路径;邮件宽度建议控制在600-700像素以内,确保在移动设备上也能正常显示。此外,HTML邮件通常需要同时提供纯文本版本作为降级方案,使用MIMEMultipart的alternative类型来实现——邮件客户端会优先显示HTML版本,如果客户端不支持HTML则回退到纯文本版本。

在实际项目中,推荐使用邮件模板引擎来管理HTML邮件内容。可以将HTML邮件模板保存为独立文件,使用Python的string.Template或Jinja2等模板引擎进行变量替换。这种方式的优点是将邮件内容和业务逻辑分离,设计和维护更加方便。当需要发送复杂的数据报表邮件时,还可以在Python中动态生成HTML表格、图表甚至使用pandas的to_html()方法直接将DataFrame转换为HTML表格嵌入邮件。

4.1 基础HTML邮件

# 基础HTML邮件 from email.mime.text import MIMEText from email.header import Header import smtplib html_content = """<html> <body style="font-family: 'Microsoft YaHei', Arial, sans-serif; padding: 20px;"> <h2 style="color: #2e7d32;">系统通知</h2> <p>您好,</p> <p>以下是本日系统运行摘要:</p> <table border="1" cellpadding="8" cellspacing="0" style="border-collapse: collapse; width: 100%;"> <tr style="background: #2e7d32; color: #fff;"> <th>指标</th> <th>数值</th> <th>状态</th> </tr> <tr> <td>CPU使用率</td> <td>45%</td> <td style="color: green;">正常</td> </tr> <tr> <td>内存使用率</td> <td>62%</td> <td style="color: green;">正常</td> </tr> <tr> <td>磁盘使用率</td> <td>78%</td> <td style="color: orange;">注意</td> </tr> </table> <p style="color: #888; font-size: 12px;">本邮件由系统自动发送</p> </body> </html>""" msg = MIMEText(html_content, "html", "utf-8") msg["From"] = "监控系统 " msg["To"] = "admin@example.com" msg["Subject"] = Header("系统运行日报", "utf-8").encode() with smtplib.SMTP_SSL("smtp.qq.com", 465) as server: server.login("sender@qq.com", "授权码") server.send_message(msg)

4.2 纯文本+HTML 双版本邮件(最佳实践)

# 同时提供纯文本和HTML版本 from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.header import Header import smtplib # 创建multipart/alternative容器 msg = MIMEMultipart("alternative") msg["From"] = "通知中心 " msg["To"] = "user@example.com" msg["Subject"] = Header("月度数据报告", "utf-8").encode() # 纯文本版本 text_content = """月度数据报告 总用户数:12,580 新增用户:1,230 活跃用户:8,456 收入:¥1,250,000""" # HTML版本(更丰富) html_content = """<html> <body style="font-family: Arial, sans-serif;"> <h2>月度数据报告</h2> <table> <tr><td>总用户数</td><td>12,580</td></tr> <tr><td>新增用户</td><td>1,230</td></tr> <tr><td>活跃用户</td><td>8,456</td></tr> <tr><td>收入</td><td>¥1,250,000</td></tr> </table> </body></html>""" # 先附加纯文本(优先级低),后附加HTML(优先级高) msg.attach(MIMEText(text_content, "plain", "utf-8")) msg.attach(MIMEText(html_content, "html", "utf-8")) with smtplib.SMTP_SSL("smtp.qq.com", 465) as server: server.login("sender@qq.com", "授权码") server.send_message(msg)

4.3 使用模板引擎管理HTML邮件

# 使用string.Template管理邮件模板 from string import Template from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import smtplib # 模板文件(可保存为独立文件) html_template = Template("""<html> <body> <h2>${title}</h2> <p>${greeting}</p> <table border="1" cellpadding="6" style="border-collapse:collapse;"> <tr><th>项目</th><th>数值</th></tr> ${table_rows} </table> <p>${footer}</p> </body> </html>""") # 动态生成表格行 data = {"用户数": 12580, "订单数": 3421, "收入(元)": 1250000} rows = "".join(f"<tr><td>{k}</td><td>{v}</td></tr>" for k, v in data.items()) # 填充模板 html = html_template.substitute( title="日度运营数据", greeting="您好,以下是今日数据简报:", table_rows=rows, footer="本邮件由系统自动生成" ) msg = MIMEText(html, "html", "utf-8") msg["From"] = "数据组 " msg["To"] = "team@example.com" msg["Subject"] = "日度运营数据"

五、附件发送

在自动化办公中,邮件附件是最常用的功能之一——发送Excel报表、PDF文档、图片压缩包等是日常工作的常见需求。Python的email库通过MIMEMultipart和MIMEBase类支持任意类型的附件添加。构建带附件的邮件时,需要将邮件主体设置为multipart/mixed类型,然后将邮件正文和各个附件作为独立的MIME部件添加到邮件容器中。每个附件部件需要设置正确的Content-Type(对应文件类型)和Content-Disposition(指定文件名和处理方式)。

附件发送的核心原理是:将文件以二进制模式读取,使用base64编码后嵌入邮件中。对于常见文件类型,email库提供了便捷的辅助类:MIMEApplication适用于任意二进制数据(文档、PDF、压缩包等),MIMEImage适用于图片,MIMEAudio适用于音频文件。如果使用通用的MIMEBase,则需要手动设置Content-Type和Content-Transfer-Encoding头部。无论使用哪种方式,都需要通过set_payload()设置文件内容,并通过add_header()设置Content-Disposition中的filename参数。

内嵌图片(Inline Image)是附件的一种特殊形式。与普通附件不同,内嵌图片在邮件正文中被引用显示,而不是作为独立附件出现在附件列表中。实现内嵌图片需要使用multipart/related类型,图片部分通过Content-ID(CID)标识,HTML正文中通过<img src="cid:image_id">引用。同一个邮件可以同时包含内嵌图片和外部附件,此时邮件结构为multipart/mixed外层容器包含multipart/related(正文+内嵌图片)和独立的附件MIME部件。这种嵌套结构在构建时需要特别注意层级顺序。

5.1 发送带PDF附件的邮件

# 发送带PDF附件的邮件 from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email import encoders import smtplib def send_with_attachment(sender, password, recipient, subject, body, file_path): # 创建multipart/mixed容器 msg = MIMEMultipart("mixed") msg["From"] = sender msg["To"] = recipient msg["Subject"] = subject # 添加邮件正文 msg.attach(MIMEText(body, "plain", "utf-8")) # 添加附件 with open(file_path, "rb") as f: attachment = MIMEBase("application", "octet-stream") attachment.set_payload(f.read()) encoders.encode_base64(attachment) attachment.add_header( "Content-Disposition", "attachment", filename=("utf-8", "", file_path.split("/")[-1]) ) msg.attach(attachment) with smtplib.SMTP_SSL("smtp.qq.com", 465) as server: server.login(sender, password) server.send_message(msg) # 调用 send_with_attachment( "sender@qq.com", "授权码", "manager@company.com", "月度财务报表", "请查收本月度财务报表。", "月度财务报表_2026年4月.pdf" )

5.2 内嵌图片到邮件正文

# HTML邮件中内嵌图片(CID方式) from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.image import MIMEImage import smtplib msg = MIMEMultipart("related") # related类型用于内嵌资源 msg["From"] = "designer@example.com" msg["To"] = "client@example.com" msg["Subject"] = "设计方案展示" # HTML正文,通过cid引用图片 html = """<html> <body> <h2>设计方案</h2> <p>以下是首页设计效果图:</p> <p><img src="cid:homepage_design" width="600"></p> <p>如有疑问请随时联系。</p> </body> </html>""" msg.attach(MIMEText(html, "html", "utf-8")) # 添加内嵌图片 with open("design_homepage.png", "rb") as f: img = MIMEImage(f.read()) img.add_header("Content-ID", "<homepage_design>") img.add_header("Content-Disposition", "inline", filename="design_homepage.png") msg.attach(img) with smtplib.SMTP_SSL("smtp.qq.com", 465) as server: server.login("sender@qq.com", "授权码") server.send_message(msg)

5.3 多附件(Excel+图片+压缩包)+ 正文

# 同时发送多个不同类型附件 from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email import encoders import smtplib import os def send_multiple_attachments(sender, password, recipient, subject, body, files): msg = MIMEMultipart("mixed") msg["From"] = sender msg["To"] = recipient msg["Subject"] = subject msg.attach(MIMEText(body, "plain", "utf-8")) for file_path in files: if not os.path.exists(file_path): print(f"文件不存在:{file_path}") continue # 根据扩展名设置Content-Type ext = os.path.splitext(file_path)[1].lower() content_type_map = { ".pdf": ("application", "pdf"), ".xlsx": ("application", "vnd.openxmlformats-officedocument.spreadsheetml.sheet"), ".xls": ("application", "vnd.ms-excel"), ".docx": ("application", "vnd.openxmlformats-officedocument.wordprocessingml.document"), ".zip": ("application", "zip"), ".rar": ("application", "x-rar-compressed"), ".png": ("image", "png"), ".jpg": ("image", "jpeg"), ".jpeg": ("image", "jpeg"), } maintype, subtype = content_type_map.get(ext, ("application", "octet-stream")) with open(file_path, "rb") as f: part = MIMEBase(maintype, subtype) part.set_payload(f.read()) encoders.encode_base64(part) part.add_header( "Content-Disposition", "attachment", filename=("utf-8", "", os.path.basename(file_path)) ) msg.attach(part) with smtplib.SMTP_SSL("smtp.qq.com", 465) as server: server.login(sender, password) server.send_message(msg) # 一次发送多个附件 send_multiple_attachments( "sender@qq.com", "授权码", "team@company.com", "项目交付材料", "您好,以下是本次项目的交付材料,请查收。", ["需求文档.pdf", "数据报表.xlsx", "设计素材.zip"] )

六、邮箱配置

配置SMTP邮件服务是Python邮件发送的首要步骤。不同的邮件服务商有不同的SMTP服务器地址、端口和认证方式。绝大多数个人邮箱需要使用"授权码"而非登录密码进行SMTP认证,这是出于安全考虑——授权码是专门用于第三方客户端登录的独立密码,可以单独禁用而不影响网页端登录。获取授权码通常需要在邮箱设置中开启SMTP服务并生成专用授权码,以QQ邮箱为例,需要在设置-账户-POP3/SMTP服务中开启并生成授权码。

在实际开发中,不应将邮箱账号和授权码硬编码在代码中,而应通过环境变量、配置文件或密钥管理服务来管理。常见的做法是使用.env文件(配合python-dotenv库加载)、config.ini配置文件(使用Python内置的configparser模块解析)或直接在环境变量中设置。对于企业级应用,推荐使用密钥管理服务(如AWS Secrets Manager、HashiCorp Vault)来管理邮箱凭证。此外,还可以封装一个邮件配置管理类,根据不同的运行环境(开发、测试、生产)加载不同的邮箱配置。

各主流邮箱的SMTP配置参数是开发者必须掌握的技能。QQ邮箱是国内使用最广泛的邮箱服务之一,其SMTP配置为smtp.qq.com,SSL端口465,TLS端口587。163邮箱的SMTP服务器为smtp.163.com。Gmail的SMTP为smtp.gmail.com,需要注意Gmail可能需要启用"允许不够安全的应用"选项或使用应用专用密码。企业邮箱的配置取决于企业的邮件服务提供商,常见的有腾讯企业邮(smtp.exmail.qq.com)、阿里企业邮(smtp.mxhichina.com)等。新浪邮箱(smtp.sina.com.cn)等国内邮箱也各有其配置参数。

6.1 配置文件管理

# config.ini 配置文件示例 # [mail] # host = smtp.qq.com # port = 465 # use_ssl = true # username = your_email@qq.com # password = your_authorization_code # sender_name = 自动化系统 # 从配置文件读取邮箱配置 import configparser from pathlib import Path def load_mail_config(config_path="config.ini"): config = configparser.ConfigParser() config.read(config_path, encoding="utf-8") conf = {} if "mail" in config: conf["host"] = config["mail"].get("host", "smtp.qq.com") conf["port"] = config["mail"].getint("port", 465) conf["use_ssl"] = config["mail"].getboolean("use_ssl", True) conf["username"] = config["mail"]["username"] conf["password"] = config["mail"]["password"] conf["sender_name"] = config["mail"].get("sender_name", "") return conf mail_conf = load_mail_config()

6.2 主流邮箱SMTP配置一览

# 主流邮箱SMTP配置字典 MAIL_CONFIGS = { "qq": { "host": "smtp.qq.com", "ssl_port": 465, "tls_port": 587, "note": "需要开启POP3/SMTP服务,使用授权码" }, "163": { "host": "smtp.163.com", "ssl_port": 465, "tls_port": 587, "note": "需要开启SMTP服务,使用授权码" }, "gmail": { "host": "smtp.gmail.com", "ssl_port": 465, "tls_port": 587, "note": "需要启用应用专用密码或OAuth2认证" }, "outlook": { "host": "smtp.office365.com", "ssl_port": 587, "tls_port": 587, "note": "使用TLS加密,端口587" }, "qq_exmail": { "host": "smtp.exmail.qq.com", "ssl_port": 465, "tls_port": 587, "note": "腾讯企业邮箱" }, "aliyun_exmail": { "host": "smtp.mxhichina.com", "ssl_port": 465, "tls_port": 80, "note": "阿里企业邮箱" }, "sina": { "host": "smtp.sina.com.cn", "ssl_port": 465, "tls_port": 587, "note": "新浪邮箱" }, "126": { "host": "smtp.126.com", "ssl_port": 465, "tls_port": 587, "note": "126邮箱" } } def create_smtp_connection(mail_type="qq"): config = MAIL_CONFIGS[mail_type] if config.get("ssl_port"): return smtplib.SMTP_SSL(config["host"], config["ssl_port"], timeout=30) return smtplib.SMTP(config["host"], config["tls_port"], timeout=30)

6.3 使用环境变量管理凭证(推荐)

# 使用环境变量管理邮箱凭证(最安全的做法) import os import smtplib from email.mime.text import MIMEText # 设置环境变量(在.bashrc或系统环境变量中设置) # export MAIL_USERNAME="your_email@qq.com" # export MAIL_PASSWORD="your_authorization_code" # export SMTP_HOST="smtp.qq.com" # export SMTP_PORT="465" def get_mail_config_from_env(): return { "host": os.environ.get("SMTP_HOST", "smtp.qq.com"), "port": int(os.environ.get("SMTP_PORT", 465)), "username": os.environ["MAIL_USERNAME"], "password": os.environ["MAIL_PASSWORD"], } def send_with_env_config(recipient, subject, body): conf = get_mail_config_from_env() with smtplib.SMTP_SSL(conf["host"], conf["port"]) as server: server.login(conf["username"], conf["password"]) msg = MIMEText(body, "plain", "utf-8") msg["From"] = conf["username"] msg["To"] = recipient msg["Subject"] = subject server.send_message(msg)

七、批量发送

批量发送邮件是自动化办公中的高频需求——例如向所有客户发送活动通知、向各部门发送周报、向全体员工发送公告等。Python的smtplib天然支持批量发送,但需要考虑一些工程化的问题:发送间隔控制以避免触发反垃圾策略、失败重试机制保证送达率、日志记录用于审计和排查、进度显示改善用户体验。批量发送的核心是在循环中逐封发送,每次调用send_message()发送一封邮件到单个或多个收件人。

在批量发送时,一个重要的设计决策是:使用"单封发送"还是"批量发送"?单封发送是指每封邮件只包含一个收件人地址,循环调用send_message()多次。这种方式的好处是可以在邮件中使用收件人个性化信息(如"尊敬的张三先生"),并且即使部分邮件发送失败也不影响其他邮件。批量发送是指一封邮件同时发送给多个收件人(To字段包含多个地址),由服务器分发。这种方式效率更高(一次SMTP连接即可),但无法个性化内容,而且如果服务器拒绝则所有收件人都收不到邮件。通常来说,对于个性化营销邮件使用单封发送,对于统一通知使用批量发送。

发送间隔控制是防止账号被标记为垃圾发送者的关键策略。大多数邮件服务商对发送频率有隐形限制——例如QQ邮箱限制每小时不超过100封,超出后可能会暂时封禁SMTP服务。合理的做法是每发送一封后sleep(0.5-2)秒,避免短时间内大量发送。对于超大批量发送(上千封),建议使用专门的邮件发送服务(如SendGrid、Mailgun、阿里云邮件推送等),而非直接使用个人邮箱SMTP。此外,发送失败重试也是必要的——对于临时性错误(如网络超时),可以重试2-3次;对于永久性错误(如账号被禁),则应停止发送并报警。

7.1 从列表批量发送

# 从收件人列表批量发送(单封发送,支持个性化) import smtplib import time import logging from email.mime.text import MIMEText from email.header import Header logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def batch_send_individual(recipients, subject_template, body_template): """ 批量发送个性化邮件 recipients: [{"name": "张三", "email": "zhangsan@example.com"}, ...] """ config = { "host": "smtp.qq.com", "port": 465, "user": "sender@qq.com", "password": "授权码", "sender_name": "客户服务部" } success = 0 failed = 0 for i, recipient in enumerate(recipients): try: # 个性化内容 subject = subject_template.format(name=recipient["name"]) body = body_template.format(name=recipient["name"]) msg = MIMEText(body, "plain", "utf-8") msg["From"] = f'{config["sender_name"]} <{config["user"]}>' msg["To"] = recipient["email"] msg["Subject"] = Header(subject, "utf-8").encode() with smtplib.SMTP_SSL(config["host"], config["port"]) as server: server.login(config["user"], config["password"]) server.send_message(msg) success += 1 logger.info(f"[{i+1}/{len(recipients)}] 已发送 -> {recipient['name']} ({recipient['email']})") # 发送间隔控制 if i < len(recipients) - 1: time.sleep(1) except Exception as e: failed += 1 logger.error(f"发送失败 -> {recipient['email']}: {e}") logger.info(f"发送完成:成功 {success},失败 {failed}") return success, failed # 示例收件人列表 recipients = [ {"name": "张三", "email": "zhangsan@example.com"}, {"name": "李四", "email": "lisi@example.com"}, {"name": "王五", "email": "wangwu@example.com"}, ] batch_send_individual( recipients, "尊敬的{name},您好", "尊敬的{name},感谢您对我们产品的支持..." )

7.2 带失败重试的批量发送

# 带重试机制的批量发送 import smtplib import time import logging from email.mime.text import MIMEText from email.header import Header logger = logging.getLogger(__name__) def send_with_retry(config, recipient, subject, body, max_retries=3): """带重试的邮件发送""" for attempt in range(1, max_retries + 1): try: with smtplib.SMTP_SSL(config["host"], config["port"], timeout=30) as server: server.login(config["user"], config["password"]) msg = MIMEText(body, "plain", "utf-8") msg["From"] = config["user"] msg["To"] = recipient msg["Subject"] = Header(subject, "utf-8").encode() server.send_message(msg) return True except smtplib.SMTPServerDisconnected: logger.warning(f"连接断开,第{attempt}次重试 -> {recipient}") time.sleep(2 ** attempt) # 指数退避 except smtplib.SMTPResponseException as e: if e.smtp_code == 550: # 永久性错误,不重试 logger.error(f"邮件被拒(550)-> {recipient}") return False logger.warning(f"SMTP错误,第{attempt}次重试 -> {recipient}") time.sleep(2 ** attempt) except Exception as e: logger.warning(f"未知错误,第{attempt}次重试 -> {recipient}: {e}") time.sleep(2 ** attempt) return False def batch_send_with_progress(config, recipients, subject, body): """批量发送带进度显示""" total = len(recipients) success = 0 failed = 0 print(f"开始批量发送,共 {total} 封邮件") print("=" * 40) for i, recipient in enumerate(recipients, 1): result = send_with_retry(config, recipient, subject, body) if result: success += 1 status = "OK" else: failed += 1 status = "FAIL" progress = i / total * 100 print(f"[{i}/{total}] {progress:.1f}% | {status} | {recipient}") print("=" * 40) print(f"发送完成:成功 {success},失败 {failed}") return success, failed

7.3 从Excel读取收件人列表批量发送

# 从Excel文件读取收件人列表并批量发送 import pandas as pd import smtplib import time from email.mime.text import MIMEText from email.header import Header def batch_send_from_excel(excel_path, sheet_name=0, subject="通知"): # 从Excel读取收件人列表 df = pd.read_excel(excel_path, sheet_name=sheet_name) required_cols = {"name", "email"} if not required_cols.issubset(df.columns): raise ValueError(f"Excel必须包含列:{required_cols}") config = { "host": "smtp.qq.com", "port": 465, "user": "sender@qq.com", "password": "授权码" } success = 0 for _, row in df.iterrows(): try: body = f"尊敬的{row['name']},您好!\n\n请查收附件中的最新资料。" msg = MIMEText(body, "plain", "utf-8") msg["From"] = config["user"] msg["To"] = row["email"] msg["Subject"] = Header(subject, "utf-8").encode() with smtplib.SMTP_SSL(config["host"], config["port"]) as server: server.login(config["user"], config["password"]) server.send_message(msg) success += 1 time.sleep(1) # 间隔控制 except Exception as e: print(f"发送失败 {row['email']}: {e}") print(f"批量发送完成:成功 {success}/{len(df)}") # batch_send_from_excel("客户列表.xlsx")

八、安全与反垃圾

当你开始使用Python批量发送邮件时,很快会发现一个现实问题:大量邮件被收件人邮箱系统拦截或投入垃圾箱。这是因为邮件服务商部署了复杂的反垃圾策略来保护用户。要确保邮件正常送达收件箱,需要从技术和管理两个层面进行规范。技术层面,需要正确配置域名的SPF、DKIM和DMARC记录,这三项是邮件服务商验证发件人身份的核心机制。管理层面,需要注意发送频率控制、内容合规度、退订机制等。

SPF(Sender Policy Framework)是一种域名所有权验证机制。通过在域名的DNS记录中添加SPF TXT记录,声明哪些IP地址被授权使用该域名发送邮件。例如,如果你使用QQ邮箱发送邮件,需要在你的域名DNS中添加类似"v=spf1 include:spf.mail.qq.com ~all"的记录。DKIM(DomainKeys Identified Mail)使用公钥加密签名技术,在邮件头中嵌入数字签名,收件服务器通过查询发件域名的DNS公钥来验证邮件是否被篡改。DMARC(Domain-based Message Authentication, Reporting and Conformance)则基于SPF和DKIM的验证结果,定义收件服务器如何处理验证失败的邮件(放行、隔离或拒收),并向发件人发送认证报告。

除了技术层面的域名认证,日常发送行为的管理同样重要。发送频率控制是避免被标记为垃圾邮件的关键——不建议在短时间内发送大量邮件,每小时发送量控制在邮箱服务商限制以内(如QQ邮箱建议不超过100封/小时)。邮件内容应避免敏感词汇(如"免费""中奖""点击这里"等营销陷阱词汇),使用合规的语言和格式。营销类邮件必须在正文中包含退订链接,这是中国《电子邮件营销服务规范》的明确要求。退订链接可以指向一个简单的页面,收件人点击后更新邮件订阅状态,此后不再向该地址发送邮件。图片和文本的比例也需要控制,纯图片邮件容易被判断为垃圾邮件。

8.1 SPF/DKIM/DMARC 记录示例

# DNS TXT记录配置示例(在域名管理面板中设置) # SPF记录 # 记录类型: TXT # 主机记录: @ (或留空,代表根域名) # 记录值: v=spf1 include:spf.mail.qq.com include:spf.example.com ~all # DKIM记录(需要邮件服务商提供的公钥) # 记录类型: TXT # 主机记录: default._domainkey # 记录值: v=DKIM1; k=rsa; p=MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQC... # DMARC记录 # 记录类型: TXT # 主机记录: _dmarc # 记录值: v=DMARC1; p=quarantine; rua=mailto:dmarc@example.com; pct=100 # p策略可选值: none(不处理) / quarantine(标记垃圾) / reject(拒收)

8.2 发送频率与合规控制

# 发送速率限制器 import time import logging from collections import deque from datetime import datetime, timedelta logger = logging.getLogger(__name__) class RateLimiter: """发送频率限制器""" def __init__(self, max_per_hour=80, max_per_day=500): self.max_per_hour = max_per_hour self.max_per_day = max_per_day self.sent_times = deque() # 发送时间戳队列 def can_send(self): now = datetime.now() hour_ago = now - timedelta(hours=1) day_ago = now - timedelta(days=1) # 清理过期记录 while self.sent_times and self.sent_times[0] < hour_ago: self.sent_times.popleft() # 检查限制 hourly_count = sum(1 for t in self.sent_times if t > hour_ago) daily_count = sum(1 for t in self.sent_times if t > day_ago) if hourly_count >= self.max_per_hour: logger.warning(f"达到小时发送上限 ({self.max_per_hour})") return False if daily_count >= self.max_per_day: logger.warning(f"达到日发送上限 ({self.max_per_day})") return False return True def record_send(self): self.sent_times.append(datetime.now()) def wait_if_needed(self): """如需等待则阻塞""" while not self.can_send(): wait_seconds = 60 logger.info(f"达到频率限制,等待 {wait_seconds} 秒") time.sleep(wait_seconds) # 使用示例 limiter = RateLimiter(max_per_hour=80) for recipient in recipients: limiter.wait_if_needed() # 发送邮件... limiter.record_send()

8.3 退订链接与内容合规

# 生成带退订链接的HTML邮件 from string import Template import hashlib def build_compliant_email(user_name, user_email, base_url="https://example.com"): """构建合规的营销邮件(含退订链接)""" # 生成退订Token(用邮箱+盐值哈希) token = hashlib.sha256( f"{user_email}:your_salt_value".encode() ).hexdigest()[:16] unsubscribe_url = f"{base_url}/unsubscribe?email={user_email}&token={token}" html_template = Template(""" <html> <body style="font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto;"> <div style="background: #f8f8f8; padding: 30px;"> <h2 style="color: #333;">${greeting}</h2> <p>感谢您订阅我们的产品更新通知。</p> <p>本月我们发布了以下新功能:</p> <ul> <li>新功能A - 提升操作效率</li> <li>新功能B - 优化用户体验</li> </ul> <p>如有任何问题,请随时与我们联系。</p> <hr style="border: none; border-top: 1px solid #ddd; margin: 20px 0;"> <p style="font-size: 12px; color: #999; text-align: center;"> 如果您不希望继续接收此类邮件,请 <a href="${unsubscribe_url}" style="color: #999;">点击此处退订</a> </p> </div> </body> </html> """) return html_template.substitute( greeting=f"尊敬的{user_name},", unsubscribe_url=unsubscribe_url )

九、实战案例

理论学习最终要服务于实际应用。本章通过三个完整的实战案例,展示smtplib和email库在实际办公场景中的综合运用。每个案例都是可以直接运行的完整代码,涵盖了从数据准备、邮件构建到发送的完整流程。这些案例代表了邮件自动化的三种典型应用场景:周期性报告、实时告警通知和大规模批量通知。

第一个案例是自动发送日报/周报。在数据驱动的团队中,定期发送运营数据报告是一项重复性高、容易出错的工作。通过Python脚本,可以从数据库或API获取数据,生成HTML格式的报表邮件,定时发送给团队成员。配合操作系统的定时任务(Linux的crontab或Windows的任务计划程序),可以实现完全自动化的报表发送系统。代码示例展示了如何从数据库查询当日数据、生成带有数据可视化图表的HTML邮件,并发送给指定的收件人列表。

第二个案例是系统告警邮件。在运维监控场景中,当服务器指标异常、服务宕机或错误率飙升时,需要立即通知相关人员。基于Python的告警邮件系统可以根据告警级别(Info/Warning/Critical)决定发送策略:Info级别只发送给值班人员,Warning级别发送给整个技术团队,Critical级别则直接发送给管理层并附加大容量日志附件。代码示例展示了如何构建告警邮件,包括错误日志附件、告警级别标识、以及防止告警风暴的去重机制。

第三个案例是大规模批量发送通知。在实际业务中,可能需要向数百甚至数千用户发送活动通知、账单提醒或重要公告。案例三展示了完整的批量发送方案,包括从数据库读取收件人列表、个性化内容生成、发送间隔控制、失败重试、以及发送报告生成。此外,还引入了并发发送技术(使用ThreadPoolExecutor)来提升发送效率,同时控制并发度避免被封禁。

9.1 实战:自动发送日报/周报

# 自动发送日报——完整实战 import smtplib import time import logging from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.header import Header from datetime import datetime, date import pandas as pd logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class DailyReportSender: """日报自动发送器""" def __init__(self, smtp_config): self.config = smtp_config def fetch_today_data(self): """模拟从数据库获取今日数据""" # 实际项目中替换为数据库查询 return { "date": date.today().strftime("%Y-%m-%d"), "new_users": 328, "active_users": 12560, "orders": 892, "revenue": 156800.50, "avg_response_time": "1.2s", "error_rate": "0.03%", } def build_html_report(self, data): """构建HTML格式日报""" html = f"""<html> <body style="font-family: 'Microsoft YaHei', Arial, sans-serif; padding: 20px;"> <h2 style="color: #2e7d32;">运营日报 - {data["date"]}</h2> <table style="width: 100%; border-collapse: collapse; margin: 20px 0;"> <tr style="background: #2e7d32; color: white;"> <th style="padding: 10px;">指标</th> <th style="padding: 10px;">今日数据</th> <th style="padding: 10px;">环比</th> <th style="padding: 10px;">状态</th> </tr> <tr> <td>新增用户</td> <td>{data["new_users"]}</td> <td style="color: green;">+8.2%</td> <td style="color: green;">良好</td> </tr> <tr> <td>活跃用户</td> <td>{data["active_users"]}</td> <td style="color: green;">+3.5%</td> <td style="color: green;">良好</td> </tr> <tr> <td>订单数</td> <td>{data["orders"]}</td> <td>-1.2%</td> <td style="color: orange;">持平</td> </tr> <tr> <td>收入(元)</td> <td>{data["revenue"]:,.2f}</td> <td style="color: green;">+5.1%</td> <td style="color: green;">良好</td> </tr> <tr> <td>平均响应</td> <td>{data["avg_response_time"]}</td> <td>-0.1s</td> <td style="color: green;">正常</td> </tr> <tr> <td>错误率</td> <td>{data["error_rate"]}</td> <td>-0.01%</td> <td style="color: green;">正常</td> </tr> </table> <h3>关键发现</h3> <ul> <li>新增用户连续3日增长,可能与最近的上线活动有关</li> <li>订单数略有下滑,建议关注转化率</li> <li>系统性能指标均正常,无异常告警</li> </ul> <p style="color: #999; font-size: 12px; margin-top: 30px;"> 本日报由系统自动生成 | {datetime.now().strftime("%Y-%m-%d %H:%M")} </p> </body> </html>""" return html def send_report(self, recipients): """发送日报""" data = self.fetch_today_data() html = self.build_html_report(data) msg = MIMEMultipart("alternative") msg["From"] = f'数据报表 <{self.config["user"]}>' msg["To"] = ", ".join(recipients) msg["Subject"] = Header(f"运营日报 - {data['date']}", "utf-8").encode() msg.attach(MIMEText(html, "html", "utf-8")) with smtplib.SMTP_SSL(self.config["host"], self.config["port"]) as server: server.login(self.config["user"], self.config["password"]) server.send_message(msg) logger.info(f"日报已发送给 {len(recipients)} 人") # 使用 sender = DailyReportSender({ "host": "smtp.qq.com", "port": 465, "user": "report@qq.com", "password": "授权码" }) sender.send_report(["manager@company.com", "team@company.com"])

9.2 实战:系统告警邮件

# 系统告警邮件发送 import smtplib import logging from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email import encoders from email.header import Header from datetime import datetime import socket logger = logging.getLogger(__name__) class AlertMailer: """系统告警邮件发送器""" ALERT_LEVELS = { "INFO": {"color": "#2196F3", "label": "信息"}, "WARNING": {"color": "#FF9800", "label": "警告"}, "CRITICAL": {"color": "#F44336", "label": "严重"}, } def __init__(self, smtp_config): self.config = smtp_config self.hostname = socket.gethostname() # 去重缓存:避免告警风暴 self._dedup_cache = {} def should_send(self, alert_key, min_interval=300): """相同告警在min_interval秒内不重复发送""" now = datetime.now().timestamp() if alert_key in self._dedup_cache: elapsed = now - self._dedup_cache[alert_key] if elapsed < min_interval: logger.info(f"告警去重:{alert_key} 距上次发送仅 {elapsed:.0f}s") return False self._dedup_cache[alert_key] = now return True def send_alert(self, alert_name, message, level="INFO", log_file=None): """发送告警邮件""" if not self.should_send(alert_name): return level_info = self.ALERT_LEVELS.get(level, self.ALERT_LEVELS["INFO"]) html = f"""<html> <body style="font-family: Arial, sans-serif; padding: 20px;"> <div style="border-left: 4px solid {level_info['color']}; padding: 15px; background: #f5f5f5;"> <h2 style="color: {level_info['color']}; margin: 0;"> [{level_info['label']}] {alert_name} </h2> <p style="margin-top: 10px;">{message}</p> </div> <table style="margin-top: 15px;"> <tr><td>服务器:</td><td>{self.hostname}</td></tr> <tr><td>时间:</td><td>{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}</td></tr> </table> </body></html>""" msg = MIMEMultipart("mixed") msg["From"] = f'监控系统 <{self.config["user"]}>' msg["To"] = self._get_recipients(level) msg["Subject"] = Header(f"[{level_info['label']}] {alert_name}", "utf-8").encode() msg.attach(MIMEText(html, "html", "utf-8")) # 附加错误日志 if log_file: try: with open(log_file, "rb") as f: part = MIMEBase("text", "plain") part.set_payload(f.read(5 * 1024 * 1024)) # 限制5MB encoders.encode_base64(part) part.add_header("Content-Disposition", "attachment", filename="error.log") msg.attach(part) except Exception as e: logger.error(f"日志附件失败:{e}") with smtplib.SMTP_SSL(self.config["host"], self.config["port"]) as server: server.login(self.config["user"], self.config["password"]) server.send_message(msg) logger.info(f"告警邮件已发送:[{level}] {alert_name}") def _get_recipients(self, level): """根据告警级别决定收件人""" if level == "CRITICAL": return "cto@company.com, ops-leader@company.com" elif level == "WARNING": return "ops-team@company.com" else: return "oncall@company.com" # 使用示例 alerter = AlertMailer({ "host": "smtp.qq.com", "port": 465, "user": "alerts@qq.com", "password": "授权码" }) # 发送严重告警 alerter.send_alert( "磁盘空间不足", "服务器 /data 分区使用率已达 95%,请立即处理!", level="CRITICAL", log_file="/var/log/disk_alert.log" )

9.3 实战:批量发送通知(含并发)

# 批量发送通知——高并发版本 import smtplib import time import logging from concurrent.futures import ThreadPoolExecutor, as_completed from email.mime.text import MIMEText from email.header import Header from datetime import datetime logger = logging.getLogger(__name__) class BulkNotificationSender: """批量通知发送器""" def __init__(self, smtp_config, max_workers=5): self.config = smtp_config self.max_workers = max_workers self.sent_count = 0 self.fail_count = 0 def _send_single(self, recipient): """发送单封邮件(供线程池调用)""" try: name = recipient.get("name", recipient["email"]) body = f"""尊敬的{name},您好! 我们非常高兴地通知您,系统已完成最新版本更新。 主要更新内容: 1. 性能优化 - 页面加载速度提升50% 2. 新增功能 - 数据导出支持Excel格式 3. Bug修复 - 修复了搜索功能异常问题 如您在使用过程中遇到任何问题,请通过以下方式联系我们: - 客服邮箱:support@example.com - 客服电话:400-xxx-xxxx 感谢您的支持! {datetime.now().strftime("%Y年%m月%d日")}""" msg = MIMEText(body, "plain", "utf-8") msg["From"] = f'通知中心 <{self.config["user"]}>' msg["To"] = recipient["email"] msg["Subject"] = Header("系统版本更新通知", "utf-8").encode() with smtplib.SMTP_SSL( self.config["host"], self.config["port"], timeout=30 ) as server: server.login(self.config["user"], self.config["password"]) server.send_message(msg) return (True, recipient["email"], None) except Exception as e: return (False, recipient["email"], str(e)) def send_bulk(self, recipients): """并发批量发送""" total = len(recipients) logger.info(f"开始批量发送 {total} 封通知") start_time = time.time() # 使用线程池并发发送 with ThreadPoolExecutor(max_workers=self.max_workers) as executor: futures = { executor.submit(self._send_single, r): r for r in recipients } completed = 0 for future in as_completed(futures): success, email, error = future.result() completed += 1 if success: self.sent_count += 1 else: self.fail_count += 1 logger.warning(f"发送失败 [{completed}/{total}]: {email} - {error}") # 每隔10%输出一次进度 if completed % max(1, total // 10) == 0: pct = completed / total * 100 elapsed = time.time() - start_time eta = (elapsed / completed) * (total - completed) logger.info( f"进度: {pct:.0f}% ({completed}/{total}) | " f"已用: {elapsed:.0f}s | 预计剩余: {eta:.0f}s | " f"成功: {self.sent_count} | 失败: {self.fail_count}" ) elapsed = time.time() - start_time logger.info( f"批量发送完成!" f"总计: {total} | 成功: {self.sent_count} | " f"失败: {self.fail_count} | 耗时: {elapsed:.1f}s" ) return self.sent_count, self.fail_count # 生成大量收件人(模拟) recipients = [ {"name": f"用户{i}", "email": f"user{i}@example.com"} for i in range(1, 101) # 100个收件人 ] sender = BulkNotificationSender( {"host": "smtp.qq.com", "port": 465, "user": "notice@qq.com", "password": "授权码"}, max_workers=5 ) # sender.send_bulk(recipients)

核心要点总结:

1. 协议基础:SMTP负责发送,POP3/IMAP负责接收,MIME标准定义邮件结构。Python的smtplib封装SMTP客户端,email库封装MIME构造。

2. 发送流程:创建连接(SSL/TLS)→ 登录认证(使用授权码而非密码)→ 构建邮件(MIMEText/MIMEMultipart)→ 发送(send_message)→ 关闭连接。

3. 邮件类型:纯文本(MIMEText "plain")、HTML邮件(MIMEText "html")、带附件邮件(MIMEMultipart "mixed" + MIMEBase)、内嵌图片(MIMEMultipart "related" + CID引用)、双版本(MIMEMultipart "alternative" 提供plain+html)。

4. 邮箱配置:不同服务商SMTP地址和端口不同,推荐SSL加密(端口465)或TLS加密(端口587),使用环境变量或配置文件管理凭证,切勿硬编码。

5. 批量策略:单封发送支持个性化但效率较低,批量发送效率高但无法个性化。必须控制发送频率(建议间隔1秒+,每小时不超过80封),配置失败重试和指数退避。

6. 安全合规:配置SPF/DKIM/DMARC DNS记录提高送达率,邮件正文含退订链接,控制图片/文本比例,避免敏感营销词汇。

7. 实战应用:日报/周报自动发送、系统告警通知(支持级别判断和去重)、大规模批量通知(线程池并发发送+进度显示)。综合使用smtplib、email、pandas、concurrent.futures等库。