第165集:邮件自动化

1. 邮件自动化概述

邮件自动化是指使用编程语言自动发送、接收、处理和管理电子邮件的技术。它可以帮助我们实现邮件的批量发送、自动回复、定时发送、邮件内容提取等功能,大大提高工作效率。

应用场景:

  • 批量发送通知邮件、营销邮件
  • 自动回复客户咨询邮件
  • 定时发送报告邮件
  • 自动提取邮件附件并处理
  • 监控特定邮件内容并触发相应操作
  • 邮件归档和分类管理

2. 核心库介绍

Python提供了多个用于邮件处理的标准库和第三方库:

2.1 smtplib - 发送邮件

smtplib是Python的标准库,用于通过SMTP(Simple Mail Transfer Protocol)协议发送邮件。

主要功能:

  • 连接到SMTP服务器
  • 身份验证
  • 发送文本邮件、HTML邮件
  • 发送带附件的邮件

2.2 email - 构建邮件内容

email是Python的标准库,用于构建和解析邮件内容,包括邮件头、正文、附件等。

主要组件:

  • MIMEText:用于创建纯文本或HTML格式的邮件正文
  • MIMEImage:用于添加图片附件
  • MIMEApplication:用于添加其他类型的附件
  • MIMEMultipart:用于组合多个MIME部分,构建复杂邮件

2.3 imaplib - 接收邮件

imaplib是Python的标准库,用于通过IMAP(Internet Message Access Protocol)协议接收邮件。

主要功能:

  • 连接到IMAP服务器
  • 浏览邮件文件夹
  • 搜索和获取邮件
  • 解析邮件内容和附件

2.4 第三方库

  • yagmail:简化SMTP邮件发送的第三方库
  • imapclient:更友好的IMAP客户端库
  • mailparser:用于解析邮件的第三方库

3. 发送邮件实战

3.1 发送简单文本邮件

import smtplib
from email.mime.text import MIMEText
from email.header import Header

# 邮件服务器配置
mail_host = "smtp.example.com"  # SMTP服务器地址
mail_user = "your_email@example.com"  # 用户名
mail_pass = "your_password"  # 密码或授权码

# 发件人和收件人
sender = "your_email@example.com"
receivers = ["recipient1@example.com", "recipient2@example.com"]  # 可添加多个收件人

# 邮件内容
subject = "Python邮件测试"
content = "这是一封使用Python smtplib发送的测试邮件。"

# 创建邮件对象
message = MIMEText(content, 'plain', 'utf-8')
message['From'] = Header("Python自动化脚本", 'utf-8')
message['To'] = Header("测试用户", 'utf-8')
message['Subject'] = Header(subject, 'utf-8')

# 发送邮件
try:
    smtpObj = smtplib.SMTP()
    smtpObj.connect(mail_host, 25)  # 25为SMTP端口号
    smtpObj.login(mail_user, mail_pass)
    smtpObj.sendmail(sender, receivers, message.as_string())
    print("邮件发送成功")
except smtplib.SMTPException as e:
    print(f"邮件发送失败: {e}")

3.2 发送HTML格式邮件

import smtplib
from email.mime.text import MIMEText
from email.header import Header

# 邮件内容(HTML格式)
html_content = """
<html>
<head>
    <title>Python邮件测试</title>
    <style>
        body {font-family: Arial, sans-serif;}
        .content {background-color: #f0f0f0; padding: 20px;}
        .highlight {color: #0066cc; font-weight: bold;}
    </style>
</head>
<body>
    <h1>Python邮件自动化测试</h1>
    <div class="content">
        <p>这是一封使用Python smtplib发送的<b>HTML格式邮件</b>。</p>
        <p>我们可以在邮件中添加:</p>
        <ul>
            <li>文本格式(如<span class="highlight">高亮文字</span>)</li>
            <li>列表和表格</li>
            <li>图片(内嵌或附件)</li>
            <li>超链接</li>
        </ul>
        <p>更多信息请访问:<a href="https://www.python.org">Python官方网站</a></p>
    </div>
</body>
</html>
"""

# 创建邮件对象
message = MIMEText(html_content, 'html', 'utf-8')
message['From'] = Header("Python自动化脚本", 'utf-8')
message['To'] = Header("测试用户", 'utf-8')
message['Subject'] = Header("Python HTML邮件测试", 'utf-8')

# 发送邮件(代码与之前相同)
try:
    smtpObj = smtplib.SMTP()
    smtpObj.connect(mail_host, 25)
    smtpObj.login(mail_user, mail_pass)
    smtpObj.sendmail(sender, receivers, message.as_string())
    print("HTML邮件发送成功")
except smtplib.SMTPException as e:
    print(f"HTML邮件发送失败: {e}")

3.3 发送带附件的邮件

import smtplib
import os
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from email.header import Header

# 创建带附件的邮件对象
message = MIMEMultipart()
message['From'] = Header("Python自动化脚本", 'utf-8')
message['To'] = Header("测试用户", 'utf-8')
message['Subject'] = Header("Python带附件邮件测试", 'utf-8')

# 添加邮件正文
text_content = "这是一封带附件的邮件,附件包含了测试文档和图片。"
message.attach(MIMEText(text_content, 'plain', 'utf-8'))

# 添加附件1:文档
file_path1 = "测试文档.pdf"
with open(file_path1, 'rb') as f:
    attach1 = MIMEApplication(f.read())
    attach1.add_header('Content-Disposition', 'attachment', filename=os.path.basename(file_path1))
    message.attach(attach1)

# 添加附件2:图片
file_path2 = "测试图片.jpg"
with open(file_path2, 'rb') as f:
    attach2 = MIMEApplication(f.read())
    attach2.add_header('Content-Disposition', 'attachment', filename=os.path.basename(file_path2))
    message.attach(attach2)

# 发送邮件
try:
    smtpObj = smtplib.SMTP()
    smtpObj.connect(mail_host, 25)
    smtpObj.login(mail_user, mail_pass)
    smtpObj.sendmail(sender, receivers, message.as_string())
    print("带附件邮件发送成功")
except smtplib.SMTPException as e:
    print(f"带附件邮件发送失败: {e}")

3.4 使用SSL/TLS加密发送邮件

import smtplib
from email.mime.text import MIMEText
from email.header import Header

# 邮件服务器配置(使用SSL/TLS)
mail_host = "smtp.example.com"
mail_user = "your_email@example.com"
mail_pass = "your_password"
smtp_port = 465  # SSL端口
# smtp_port = 587  # TLS端口

# 创建邮件对象
message = MIMEText("这是一封使用SSL加密的测试邮件", 'plain', 'utf-8')
message['From'] = Header("Python自动化脚本", 'utf-8')
message['To'] = Header("测试用户", 'utf-8')
message['Subject'] = Header("Python SSL邮件测试", 'utf-8')

# 发送邮件(SSL)
try:
    # 使用SSL连接
    smtpObj = smtplib.SMTP_SSL(mail_host, smtp_port)
    smtpObj.login(mail_user, mail_pass)
    smtpObj.sendmail(sender, receivers, message.as_string())
    print("SSL邮件发送成功")
except smtplib.SMTPException as e:
    print(f"SSL邮件发送失败: {e}")

# 发送邮件(TLS)
try:
    # 使用TLS连接
    smtpObj = smtplib.SMTP(mail_host, 587)
    smtpObj.starttls()  # 启动TLS加密
    smtpObj.login(mail_user, mail_pass)
    smtpObj.sendmail(sender, receivers, message.as_string())
    print("TLS邮件发送成功")
except smtplib.SMTPException as e:
    print(f"TLS邮件发送失败: {e}")

4. 接收邮件实战

4.1 连接到IMAP服务器并获取邮件

import imaplib
import email
from email.header import decode_header

# IMAP服务器配置
imap_host = "imap.example.com"
imap_user = "your_email@example.com"
imap_pass = "your_password"

# 连接到IMAP服务器
try:
    # 创建IMAP客户端
    imap = imaplib.IMAP4_SSL(imap_host)
    
    # 登录
    imap.login(imap_user, imap_pass)
    
    # 选择邮件文件夹
    status, messages = imap.select("INBOX")
    
    # 获取邮件数量
    messages = int(messages[0])
    print(f"收件箱共有 {messages} 封邮件")
    
    # 关闭连接
    imap.close()
    imap.logout()
    
except Exception as e:
    print(f"连接IMAP服务器失败: {e}")

4.2 读取和解析邮件内容

# 连接到IMAP服务器(代码同上)
imap = imaplib.IMAP4_SSL(imap_host)
imap.login(imap_user, imap_pass)
imap.select("INBOX")

# 获取最新的5封邮件
for i in range(messages, messages-5, -1):
    # 获取邮件
    res, msg = imap.fetch(str(i), "(RFC822)")
    
    for response in msg:
        if isinstance(response, tuple):
            # 解析邮件内容
            msg = email.message_from_bytes(response[1])
            
            # 解码邮件主题
            subject, encoding = decode_header(msg["Subject"])[0]
            if isinstance(subject, bytes):
                subject = subject.decode(encoding if encoding else "utf-8")
            
            # 解码发件人
            from_, encoding = decode_header(msg.get("From"))[0]
            if isinstance(from_, bytes):
                from_ = from_.decode(encoding if encoding else "utf-8")
            
            print(f"\n邮件 {i}: {subject}")
            print(f"发件人: {from_}")
            
            # 解析邮件正文
            if msg.is_multipart():
                # 多部分邮件
                for part in msg.walk():
                    # 跳过附件
                    if part.get_content_disposition() == "attachment":
                        continue
                    
                    # 获取正文内容
                    content_type = part.get_content_type()
                    content_disposition = str(part.get("Content-Disposition"))
                    
                    try:
                        body = part.get_payload(decode=True).decode()
                    except:
                        continue
                    
                    if content_type == "text/plain" and "attachment" not in content_disposition:
                        # 纯文本正文
                        print(f"正文(纯文本): {body[:100]}...")
                    elif content_type == "text/html" and "attachment" not in content_disposition:
                        # HTML正文
                        print(f"正文(HTML): {body[:100]}...")
            else:
                # 单部分邮件
                content_type = msg.get_content_type()
                body = msg.get_payload(decode=True).decode()
                if content_type == "text/plain":
                    print(f"正文: {body[:100]}...")

# 关闭连接
imap.close()
imap.logout()

4.3 下载邮件附件

import os

# 创建附件保存目录
attachments_dir = "email_attachments"
os.makedirs(attachments_dir, exist_ok=True)

# 连接到IMAP服务器(代码同上)

# 获取最新的一封邮件
res, msg = imap.fetch(str(messages), "(RFC822)")

for response in msg:
    if isinstance(response, tuple):
        msg = email.message_from_bytes(response[1])
        
        # 解析主题和发件人(代码同上)
        
        # 下载附件
        if msg.is_multipart():
            for part in msg.walk():
                content_disposition = str(part.get("Content-Disposition"))
                
                # 检查是否为附件
                if "attachment" in content_disposition:
                    # 获取附件文件名
                    filename = part.get_filename()
                    if filename:
                        # 解码文件名
                        filename, encoding = decode_header(filename)[0]
                        if isinstance(filename, bytes):
                            filename = filename.decode(encoding if encoding else "utf-8")
                        
                        # 保存附件
                        filepath = os.path.join(attachments_dir, filename)
                        with open(filepath, "wb") as f:
                            f.write(part.get_payload(decode=True))
                        print(f"附件已保存: {filepath}")

# 关闭连接
imap.close()
imap.logout()

5. 批量邮件发送

5.1 使用邮件列表批量发送

import smtplib
from email.mime.text import MIMEText
from email.header import Header

# 读取收件人列表
with open("recipients.txt", "r", encoding="utf-8") as f:
    recipients = [line.strip() for line in f if line.strip()]

# 邮件内容
subject = "重要通知:系统升级维护"
content = """
尊敬的用户:

您好!

我们计划于2024年1月1日凌晨2:00-4:00进行系统升级维护,期间系统将暂时不可用。

给您带来的不便,敬请谅解!

系统管理团队
"""

# 发送批量邮件
success_count = 0
error_count = 0

try:
    smtpObj = smtplib.SMTP_SSL(mail_host, 465)
    smtpObj.login(mail_user, mail_pass)
    
    for recipient in recipients:
        try:
            # 创建邮件对象
            message = MIMEText(content, 'plain', 'utf-8')
            message['From'] = Header("系统管理团队", 'utf-8')
            message['To'] = Header(recipient, 'utf-8')
            message['Subject'] = Header(subject, 'utf-8')
            
            # 发送邮件
            smtpObj.sendmail(sender, [recipient], message.as_string())
            print(f"成功发送给: {recipient}")
            success_count += 1
            
        except Exception as e:
            print(f"发送失败 {recipient}: {e}")
            error_count += 1
    
    print(f"\n批量发送完成:成功 {success_count} 封,失败 {error_count} 封")
    
except Exception as e:
    print(f"连接邮件服务器失败: {e}")
finally:
    if 'smtpObj' in locals():
        smtpObj.quit()

5.2 使用邮件模板

import smtplib
from email.mime.text import MIMEText
from email.header import Header

# 邮件模板
email_template = """
尊敬的 {username} 先生/女士:

您好!

感谢您对我们产品的支持。您的订单 {order_id} 已发货,预计将于 {delivery_date} 送达。

订单详情:
- 产品名称:{product_name}
- 数量:{quantity}
- 金额:{amount} 元

如有任何问题,请随时联系我们。

祝商祺!

客户服务团队
"""

# 订单数据
orders = [
    {"username": "张三", "order_id": "2024001", "delivery_date": "2024-01-05", "product_name": "Python编程教程", "quantity": 1, "amount": 99},
    {"username": "李四", "order_id": "2024002", "delivery_date": "2024-01-06", "product_name": "数据分析实战", "quantity": 2, "amount": 198},
    {"username": "王五", "order_id": "2024003", "delivery_date": "2024-01-07", "product_name": "机器学习入门", "quantity": 1, "amount": 129}
]

# 发送个性化邮件
try:
    smtpObj = smtplib.SMTP_SSL(mail_host, 465)
    smtpObj.login(mail_user, mail_pass)
    
    for order in orders:
        # 填充模板
        content = email_template.format(**order)
        subject = f"订单 {order['order_id']} 发货通知"
        
        # 创建邮件对象
        message = MIMEText(content, 'plain', 'utf-8')
        message['From'] = Header("客户服务团队", 'utf-8')
        message['To'] = Header(order['username'], 'utf-8')
        message['Subject'] = Header(subject, 'utf-8')
        
        # 获取收件人邮箱(假设订单数据中包含)
        recipient_email = f"{order['username']}@example.com"
        
        # 发送邮件
        smtpObj.sendmail(sender, [recipient_email], message.as_string())
        print(f"已发送订单通知给 {order['username']}")
    
except Exception as e:
    print(f"发送邮件失败: {e}")
finally:
    if 'smtpObj' in locals():
        smtpObj.quit()

6. 使用yagmail简化邮件发送

yagmail是一个简化SMTP邮件发送的第三方库,可以大幅减少代码量。

安装:

pip install yagmail

使用示例:

import yagmail

# 连接到SMTP服务器
yag = yagmail.SMTP(user='your_email@example.com', password='your_password', host='smtp.example.com', port=465, smtp_ssl=True)

# 发送简单邮件
subject = "yagmail测试邮件"
content = "这是使用yagmail发送的测试邮件"
yag.send(to='recipient@example.com', subject=subject, contents=content)

# 发送HTML邮件
html_content = "<h1>yagmail HTML邮件</h1><p>这是一封<b>HTML格式</b>的邮件</p>"
yag.send(to='recipient@example.com', subject="yagmail HTML测试", contents=html_content)

# 发送带附件的邮件
yag.send(
    to='recipient@example.com',
    subject="yagmail带附件测试",
    contents=["这是一封带附件的邮件", "测试文档.pdf", "测试图片.jpg"]
)

# 关闭连接
yag.close()

7. 邮件自动化最佳实践

7.1 安全考虑

  • 使用授权码而不是密码登录邮件服务器
  • 启用SSL/TLS加密保护邮件传输
  • 不要在代码中硬编码敏感信息(如密码、授权码)
  • 使用环境变量或配置文件存储敏感信息

7.2 性能优化

  • 批量发送邮件时使用连接池,避免频繁建立和关闭连接
  • 对大量收件人进行分组发送,避免单次发送过多邮件
  • 使用异步发送方式提高发送效率

7.3 合规性

  • 遵守邮件发送的法律法规(如CAN-SPAM Act)
  • 提供退订选项,尊重用户意愿
  • 不要发送垃圾邮件或未经请求的邮件

7.4 错误处理

  • 对邮件发送和接收过程进行异常处理
  • 记录发送日志,便于追踪问题
  • 实现重试机制,处理临时网络问题

7.5 测试与验证

  • 在发送正式邮件前进行充分测试
  • 验证收件人地址的有效性
  • 检查邮件内容和格式是否正确

8. 总结

邮件自动化是Python自动化办公的重要组成部分,通过smtplibemailimaplib等库,我们可以轻松实现邮件的发送、接收和处理。无论是批量发送通知邮件、自动回复客户咨询,还是定时发送报告,Python都能胜任。

关键要点:

  • 发送邮件:使用smtplibemail库构建和发送各种类型的邮件
  • 接收邮件:使用imaplib连接到邮件服务器,获取和解析邮件内容
  • 批量处理:利用邮件列表和模板实现个性化批量邮件发送
  • 安全合规:遵循安全最佳实践和法律法规

通过不断练习和实践,你可以将邮件自动化应用到更多实际工作场景中,大大提高工作效率!

« 上一篇 办公文档自动化 下一篇 » 系统监控脚本