第四十八篇：电子邮件——SMTP 发送与 POP3 收取
前言
Python 内置的 smtplib 和 email 模块让邮件收发变得非常简单。无论是自动发送通知、批量发送邮件，还是从邮箱拉取消息进行分析，这些功能都能帮你实现。本文将系统讲解 SMTP 发信和 POP3/IMAP 收信的完整用法。
SMTP —— 发送邮件
基本概念
┌──────────┐ SMTP ┌──────────┐
│ 发件人 │ ──→ 发件人 SMTP 服务器 → │ 收件人 │
│ Email │ │ Mailbox │
└──────────┘ └──────────┘
↓ ↑
┌───────────────┐
│ POP3 / IMAP │ ← 客户端拉取
└───────────────┘
简单文本邮件
import smtplib
from email.mime.text import MIMEText
from email.header import Header
def send_simple_email(to_addr, subject, body):
"""发送纯文本邮件"""
# 配置邮箱信息（以 QQ 邮箱为例）
smtp_server = "smtp.qq.com"
smtp_port = 465
sender_email = "your_email@qq.com"
sender_password = "your_auth_code" # 不是密码！是授权码
# 构建邮件
message = MIMEText(body, 'plain', 'utf-8')
message['From'] = Header("我的发信机器人", 'utf-8')
message['To'] = Header(to_addr, 'utf-8')
message['Subject'] = Header(subject, 'utf-8')
try:
# 使用 SSL 连接（推荐端口 465）
server = smtplib.SMTP_SSL(smtp_server, smtp_port)
server.login(sender_email, sender_password)
server.sendmail(sender_email, [to_addr], message.as_string())
server.quit()
print(f"✓ 邮件已发送至 {to_addr}")
return True
except smtplib.SMTPAuthenticationError:
print("❌ 认证失败：请检查用户名和授权码")
return False
except smtplib.SMTPConnectError:
print("❌ 连接失败：请检查网络和防火墙设置")
return False
except Exception as e:
print(f"❌ 发送失败: {e}")
return False
# send_simple_email(
# "recipient@example.com",
# "测试邮件",
# "这是一封由 Python 自动发送的测试邮件。\n\n你好世界！"
# )
HTML 格式邮件
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
def send_html_email(to_addr, subject, html_content):
"""发送 HTML 格式邮件"""
smtp_config = {
'server': 'smtp.qq.com',
'port': 465,
'user': 'your_email@qq.com',
'password': 'your_auth_code',
}
message = MIMEMultipart('alternative')
message['From'] = Header("HTML 邮件机器人", 'utf-8')
message['To'] = Header(to_addr, 'utf-8')
message['Subject'] = Header(subject, 'utf-8')
# 纯文本备份（兼容不支持 HTML 的客户端）
text_part = MIMEText("您正在查看纯文本版本。请使用支持 HTML 的客户端查看。", 'plain', 'utf-8')
# HTML 主体
html_part = MIMEText(html_content, 'html', 'utf-8')
# 附加两种格式
message.attach(text_part)
message.attach(html_part)
try:
server = smtplib.SMTP_SSL(smtp_config['server'], smtp_config['port'])
server.login(smtp_config['user'], smtp_config['password'])
server.sendmail(smtp_config['user'], [to_addr], message.as_string())
server.quit()
print("✓ HTML 邮件已发送")
return True
except Exception as e:
print(f"❌ 发送失败: {e}")
return False
# HTML 内容示例
html = """
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>Python 自动化邮件报告</title>
<style>
body {{ font-family: Arial, sans-serif; max-width: 700px; margin: 0 auto; padding: 20px; }}
h1 {{ color: #654ACB; border-bottom: 2px solid #654ACB; padding-bottom: 10px; }}
table {{ border-collapse: collapse; width: 100%; margin: 20px 0; }}
th, td {{ border: 1px solid #ddd; padding: 12px; text-align: left; }}
th {{ background-color: #654ACB; color: white; }}
tr:nth-child(even) {{ background-color: #f9f9f9; }}
.badge {{ display: inline-block; padding: 4px 12px; border-radius: 12px;
color: white; font-size: 12px; }}
.success {{ background-color: #28a745; }}
.warning {{ background-color: #ffc107; color: black; }}
.danger {{ background-color: #dc3545; }}
</style>
</head>
<body>
<h1>📊 日报报表</h1>
<p>您好，以下是今日系统运行报告。</p>
<h2>关键指标</h2>
<table>
<tr><th>指标</th><th>值</th><th>状态</th></tr>
<tr><td>CPU 使用率</td><td>45%</td><td><span class="badge success">正常</span></td></tr>
<tr><td>内存使用率</td><td>78%</td><td><span class="badge warning">偏高</span></td></tr>
<tr><td>磁盘空间</td><td>12%</td><td><span class="badge danger">异常</span></td></tr>
</table>
<p style="color: gray; font-size: 12px;">此邮件由 Python 脚本自动生成，请勿回复。</p>
</body>
</html>
"""
# send_html_email("recipient@example.com", "日报报表", html)
带附件的邮件
import smtplib
import os
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email import encoders
from email.header import Header
from datetime import datetime
def send_file_email(to_addr, subject, body, attachments=None):
"""发送带附件的邮件"""
if attachments is None:
attachments = []
message = MIMEMultipart()
message['From'] = Header("附件邮件发送器", 'utf-8')
message['To'] = Header(to_addr, 'utf-8')
message['Subject'] = Header(subject, 'utf-8')
message.attach(MIMEText(body, 'plain', 'utf-8'))
# 添加附件
for filepath in attachments:
if not os.path.exists(filepath):
print(f"警告: 文件不存在: {filepath}")
continue
filename = os.path.basename(filepath)
with open(filepath, '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)
try:
smtp_config = {
'server': 'smtp.qq.com',
'port': 465,
'user': 'your_email@qq.com',
'password': 'your_auth_code',
}
server = smtplib.SMTP_SSL(smtp_config['server'], smtp_config['port'])
server.login(smtp_config['user'], smtp_config['password'])
server.sendmail(smtp_config['user'], [to_addr], message.as_string())
server.quit()
print(f"✓ 邮件已发送至 {to_addr}（含 {len(attachments)} 个附件）")
return True
except Exception as e:
print(f"❌ 发送失败: {e}")
return False
# send_file_email(
# "recipient@example.com",
# "项目报告",
# "请查收附件中的季度报告。",
# ["report.pdf", "data.xlsx"]
# )
批量邮件发送器
import smtplib
from email.mime.text import MIMEText
from email.header import Header
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
class BatchMailSender:
"""批量邮件发送器"""
def __init__(self, config):
self.config = config
self.results = {'success': [], 'failed': []}
def _get_smtp(self):
"""获取新的 SMTP 连接"""
server = smtplib.SMTP_SSL(
self.config['server'],
self.config['port']
)
server.login(self.config['user'], self.config['password'])
return server
def send_one(self, recipient, subject, body):
"""发送单封邮件"""
message = MIMEText(body, 'plain', 'utf-8')
message['From'] = Header(self.config.get('from_name', '发信机器人'), 'utf-8')
message['To'] = Header(recipient, 'utf-8')
message['Subject'] = Header(subject, 'utf-8')
try:
server = self._get_smtp()
server.sendmail(
self.config['user'],
[recipient],
message.as_string()
)
server.quit()
return {'recipient': recipient, 'status': 'success'}
except Exception as e:
return {'recipient': recipient, 'status': 'failed', 'error': str(e)}
def send_batch(self, recipients, subject, body_template, max_workers=5):
"""批量发送个性化邮件"""
total = len(recipients)
start_time = time.time()
print(f"开始发送批量邮件: {total} 封")
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {}
for recipient_info in recipients:
name = recipient_info['name']
address = recipient_info['address']
# 个性化模板
personalized_body = body_template.format(
name=name,
date=time.strftime("%Y-%m-%d %H:%M"),
)
future = executor.submit(
self.send_one, address, subject, personalized_body
)
futures[future] = address
for i, future in enumerate(as_completed(futures), 1):
result = future.result()
status_icon = "✓" if result['status'] == 'success' else "✗"
print(f"\r[{i}/{total}] {status_icon} {result['recipient']}", end='')
if result['status'] == 'success':
self.results['success'].append(result)
else:
self.results['failed'].append(result)
elapsed = time.time() - start_time
print(f"\n{'='*50}")
print(f"发送完成: {len(self.results['success'])} 成功, "
f"{len(self.results['failed'])} 失败 (耗时 {elapsed:.1f}s)")
return self.results
# 使用示例
config = {
'server': 'smtp.qq.com',
'port': 465,
'user': 'your_email@qq.com',
'password': 'your_auth_code',
'from_name': 'HR 部门',
}
sender = BatchMailSender(config)
recipients = [
{"name": "Alice", "address": "alice@example.com"},
{"name": "Bob", "address": "bob@example.com"},
{"name": "Charlie", "address": "charlie@example.com"},
]
subject = "入职通知"
body_template = """尊敬的 {name}：
欢迎您加入我们的团队！
您的入职日期为 {date}，请准时参加入职培训。
如有任何问题，欢迎随时联系 HR 部门。
此致
敬礼
"""
# sender.send_batch(recipients, subject, body_template)
POP3/IMAP —— 接收邮件
POP3 基础用法
import poplib
from email.parser import Parser
from email.header import decode_header
def fetch_emails_pop3(host='pop3.qq.com', port=995, user='', password=''):
"""通过 POP3 收取邮件"""
try:
# 连接服务器
server = poplib.POP3_SSL(host, port)
server.user(user)
server.pass_(password)
# 获取邮件统计
num_messages, mailbox_size = server.stat()
print(f"收件箱共有 {num_messages} 封信件，占用 {mailbox_size} 字节")
# 收取最新 5 封邮件
for i in range(min(5, num_messages)):
response, lines, octets = server.retr(i + 1)
msg_data = b'\r\n'.join(lines).decode('utf-8')
# 解析邮件
msg = Parser().parsestr(msg_data)
subject = decode_header(msg.get('Subject'))[0][0]
if isinstance(subject, bytes):
subject = subject.decode('utf-8')
from_addr = msg.get('From', '')
date = msg.get('Date', '')
body = extract_body(msg)
print(f"\n--- 邮件 #{i+1} ---")
print(f"来自: {from_addr}")
print(f"主题: {subject}")
print(f"日期: {date[:25]}")
print(f"正文预览: {body[:100]}...")
server.quit()
return True
except poplib.error_proto as e:
print(f"POP3 协议错误: {e}")
return False
except Exception as e:
print(f"收信失败: {e}")
return False
def extract_body(msg):
"""提取邮件正文"""
if msg.is_multipart():
for part in msg.walk():
content_type = part.get_content_type()
if content_type == 'text/plain':
payload = part.get_payload(decode=True)
charset = part.get_content_charset() or 'utf-8'
return payload.decode(charset, errors='ignore')
else:
payload = msg.get_payload(decode=True)
charset = msg.get_content_charset() or 'utf-8'
if payload:
return payload.decode(charset, errors='ignore')
return ""
# fetch_emails_pop3(
# host='pop3.qq.com',
# port=995,
# user='your_email@qq.com',
# password='your_auth_code'
# )
IMAP 进阶用法
import imaplib
import email
from email.header import decode_header
def fetch_emails_imap(host='imap.qq.com', port=993, user='', password=''):
"""通过 IMAP 收取和管理邮件"""
try:
server = imaplib.IMAP4_SSL(host, port)
server.login(user, password)
# 选择收件箱
server.select('INBOX')
# 搜索所有未读邮件
status, messages = server.search(None, 'UNSEEN')
mail_ids = messages[0].split()
print(f"找到 {len(mail_ids)} 封未读邮件")
# 标记已读并提取
for mid in mail_ids[-5:]: # 只看最新 5 封
status, data = server.fetch(mid, '(RFC822)')
raw_email = data[0][1]
msg = email.message_from_bytes(raw_email)
# 解码主题
decoded_subject = decode_header(msg['Subject'])[0]
subject = decoded_subject[0]
if isinstance(subject, bytes):
charset = decoded_subject[1] or 'utf-8'
subject = subject.decode(charset, errors='ignore')
# 提取正文
body = ""
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == 'text/plain':
body = part.get_payload(decode=True).decode(
part.get_content_charset() or 'utf-8',
errors='ignore'
)
break
elif msg.get_content_type() == 'text/plain':
body = msg.get_payload(decode=True).decode(
msg.get_content_charset() or 'utf-8',
errors='ignore'
)
# 标记已读
server.store(mid, '+FLAGS', '\\Seen')
print(f"\n{subject[:50]}")
print(f" 正文: {body[:80]}..." if body else " [无正文]")
server.close()
server.logout()
return True
except imaplib.error as e:
print(f"IMAP 错误: {e}")
return False
except Exception as e:
print(f"收信失败: {e}")
return False
# fetch_emails_imap(
# host='imap.qq.com',
# port=993,
# user='your_email@qq.com',
# password='your_auth_code'
# )
邮件分析工具
import imaplib
import email
from collections import Counter
import re
def analyze_inbox(user, password):
"""分析收件箱结构"""
connections = imaplib.IMAP4_SSL('imap.qq.com', 993)
connections.login(user, password)
connections.select('INBOX')
# 搜索所有邮件
status, data = connections.search(None, 'ALL')
all_ids = data[0].split()
total_count = len(all_ids)
print(f"=== 收件箱分析报告 ===")
print(f"总邮件数: {total_count}")
sender_counter = Counter()
subject_keywords = Counter()
for mid in all_ids:
status, msg_data = connections.fetch(mid, '(BODY.PEEK[HEADER])')
if status != 'OK':
continue
raw_headers = msg_data[0][1]
msg = email.message_from_bytes(raw_headers)
# 统计发件人
from_addr = msg.get('From', '')
if from_addr:
match = re.search(r'<([^>]+)>', from_addr)
domain = match.group(1).split('@')[-1] if match else 'unknown'
sender_counter[domain] += 1
# 关键词分析
subject = msg.get('Subject', '').lower()
for keyword in ['报告', '发票', '通知', '确认', '提醒', '活动']:
if keyword in subject:
subject_keywords[keyword] += 1
print(f"\n按域名分类:")
for domain, count in sender_counter.most_common(10):
bar = "█" * (count // max(total_count // 100, 1))
print(f" {domain:>20s}: {count:4d} {bar}")
print(f"\n关键词分布:")
for keyword, count in subject_keywords.most_common():
print(f" {keyword:>10s}: {count:4d}")
connections.close()
connections.logout()
print("\n分析完成!")
# analyze_inbox('your_email@qq.com', 'your_auth_code')
实际应用场景
自动监控告警邮件
import smtplib
import psutil
import socket
from email.mime.text import MIMEText
from email.header import Header
from datetime import datetime
class SystemMonitorMailer:
"""系统监控邮件告警"""
def __init__(self, smtp_config, alert_thresholds=None):
self.smtp_config = smtp_config
self.alert_thresholds = alert_thresholds or {
'cpu_percent': 85,
'memory_percent': 90,
'disk_percent': 95,
}
def check_system_health(self):
"""检测系统健康状态"""
checks = []
# CPU 使用率
cpu_usage = psutil.cpu_percent(interval=1)
checks.append(('CPU 使用率', cpu_usage, self.alert_thresholds['cpu_percent']))
# 内存使用率
mem = psutil.virtual_memory()
checks.append(('内存使用率', mem.percent, self.alert_thresholds['memory_percent']))
# 磁盘使用率
disk = psutil.disk_usage('/')
checks.append(('磁盘使用率', disk.percent, self.alert_thresholds['disk_percent']))
# 网络连接数
net_connections = psutil.net_connections(kind='inet')
checks.append(('活跃连接', len(net_connections), 1000))
return checks
def send_alert(self, hostname, checks):
"""发送告警邮件"""
# 收集异常项
alerts = [c for c in checks if c[1] > c[2]]
if not alerts:
print("✓ 系统各项指标正常，无需告警")
return
html_body = f"""
<html>
<head><style>
body {{ font-family: monospace; background: #f5f5f5; padding: 20px; }}
table {{ border-collapse: collapse; width: 100%; background: white; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background: #dc3545; color: white; }}
.normal {{ color: green; }}
.warning {{ color: orange; }}
.critical {{ color: red; font-weight: bold; }}
</style></head>
<body>
<h1>⚠️ 系统健康告警</h1>
<p>检测到以下异常指标：</p>
<table>
<tr><th>指标</th><th>当前值</th><th>阈值</th><th>状态</th></tr>
"""
for name, value, threshold in alerts:
cls = 'critical' if value > threshold * 1.2 else 'warning'
html_body += f"""
<tr>
<td>{name}</td>
<td class="{cls}">{value}%</td>
<td>{threshold}%</td>
<td class="{cls}">异常</td>
</tr>
"""
html_body += f"""
</table>
<p style="color:gray;margin-top:20px">主机: {hostname}<br>时间: {datetime.now()}</p>
</body></html>
"""
message = MIMEText(html_body, 'html', 'utf-8')
message['From'] = Header("系统监控告警", 'utf-8')
message['To'] = Header(self.smtp_config['to'], 'utf-8')
message['Subject'] = Header(f"[告警] {hostname} 系统异常", 'utf-8')
try:
server = smtplib.SMTP_SSL(
self.smtp_config['server'],
self.smtp_config['port']
)
server.login(self.smtp_config['user'], self.smtp_config['password'])
server.sendmail(self.smtp_config['user'],
[self.smtp_config['to']],
message.as_string())
server.quit()
print(f"✓ 告警邮件已发送至 {self.smtp_config['to']}")
except Exception as e:
print(f"❌ 告警发送失败: {e}")
# 使用示例
monitor = SystemMonitorMailer(
smtp_config={
'server': 'smtp.qq.com',
'port': 465,
'user': 'your_email@qq.com',
'password': 'your_auth_code',
'to': 'admin@example.com',
}
)
# checks = monitor.check_system_health()
# monitor.send_alert(socket.gethostname(), checks)
邮件签名档生成器
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from PIL import Image, ImageDraw, ImageFont
import io
import os
def generate_signature_image(name, title, phone, company, output_file="signature.png"):
"""用 Pillow 生成电子签名图片"""
width, height = 400, 120
img = Image.new('RGBA', (width, height), (255, 255, 255, 0))
draw = ImageDraw.Draw(img)
# 绘制分隔线
draw.line([(10, 10), (390, 10)], fill=(101, 74, 203), width=2)
try:
font_main = ImageFont.truetype("simhei.ttf", 20)
font_sub = ImageFont.truetype("simhei.ttf", 14)
except:
font_main = ImageFont.load_default()
font_sub = font_main
# 姓名
draw.text((20, 20), name, fill=(101, 74, 203), font=font_main)
# 职位
draw.text((20, 48), title, fill=(100, 100, 100), font=font_sub)
# 公司
draw.text((20, 68), company, fill=(140, 140, 140), font=font_sub)
# 电话
draw.text((20, 88), f"Tel: {phone}", fill=(100, 100, 100), font=font_sub)
img.save(output_file, 'PNG')
return output_file
def send_with_signature(to_addr, subject, body_html, signature_image_path=None):
"""发送带签名的邮件"""
message = MIMEMultipart('related')
message['From'] = Header("技术支持部", 'utf-8')
message['To'] = Header(to_addr, 'utf-8')
message['Subject'] = Header(subject, 'utf-8')
# HTML 主体（包含签名图片引用）
alt_part = MIMEMultipart('alternative')
html_with_sig = body_html
if signature_image_path and os.path.exists(signature_image_path):
html_with_sig += f'<img src="cid:signature_img"><br>'
alt_part.attach(MIMEText(html_with_sig, 'html', 'utf-8'))
message.attach(alt_part)
# 内嵌签名图片
if signature_image_path and os.path.exists(signature_image_path):
with open(signature_image_path, 'rb') as f:
img_data = f.read()
from email.mime.image import MIMEImage
sig_img = MIMEImage(img_data)
sig_img.add_header('Content-ID', '<signature_img>')
sig_img.add_header('Content-Disposition', 'inline')
message.attach(sig_img)
try:
server = smtplib.SMTP_SSL('smtp.qq.com', 465)
server.login('your_email@qq.com', 'your_auth_code')
server.sendmail('your_email@qq.com', [to_addr], message.as_string())
server.quit()
print(f"✓ 带签名邮件已发送至 {to_addr}")
except Exception as e:
print(f"❌ 发送失败: {e}")
# # 生成签名图片
# sig_file = generate_signature_image(
# name="张伟",
# title="高级工程师",
# phone="138-0000-0000",
# company="某科技有限公司"
# )
#
# # 发送带签名的邮件
# send_with_signature(
# "recipient@example.com",
# "项目进度汇报",
# "<p>您好，项目进展顺利...</p>",
# sig_file
# )
SMTP 常用配置速查表
| 服务商 | SMTP 服务器 | 端口(SSL) | 端口(STARTTLS) |
|---|---|---|---|
| QQ 邮箱 | smtp.qq.com | 465 | 587 |
| 163 邮箱 | smtp.163.com | 465 | 587 |
| Gmail | smtp.gmail.com | 465 | 587 |
| Outlook | smtp-mail.outlook.com | 587 | 587 |
| Yahoo | smtp.mail.yahoo.com | 465 | 587 |
| Foxmail | smtp.foxmail.com | 465 | 587 |
POP3/IMAP 对比
| 特性 | POP3 | IMAP |
|---|---|---|
| 工作方式 | 下载后删除服务器副本 | 同步浏览服务器内容 |
| 多设备支持 | ❌ 仅本地 | ✅ 完美支持 |
| 离线访问 | ✅ 下载后可离线 | ⚠️ 需额外配置 |
| 文件夹管理 | ❌ 通常只有 inbox | ✅ 支持标签/文件夹 |
| 带宽消耗 | 较低（一次性下载） | 较高（实时同步） |
| 适用场景 | 单设备、低带宽 | 多设备、企业邮箱 |
最佳实践总结
✅ 推荐做法
- 使用授权码而非密码 — 各大邮箱都推荐使用独立授权码
- SSL/TLS 加密传输 — 始终使用 SMTP_SSL 或 STARTTLS
- 限制发送频率 — 避免被判定为垃圾邮件（每分钟不超过 20 封）
- 设置合理超时 —
settimeout(30)防止永久阻塞 - IMAP 优先于 POP3 — 现代应用都应采用 IMAP
❌ 避免做法
import smtplib
# ❌ 明文传输（不安全）
server = smtplib.SMTP("smtp.qq.com", 25) # 端口 25 不加密!
# ✅ 使用 SSL 加密
server = smtplib.SMTP_SSL("smtp.qq.com", 465)
# ❌ 硬编码密码
password = "my_password_123" # 代码仓库中泄露风险!
# ✅ 使用环境变量或配置文件
import os
password = os.environ.get('EMAIL_PASSWORD')
# ❌ 大量连续发送（可能被当作 spam）
for i in range(10000):
send_email(f"user{i}@example.com", ...) # 触发反垃圾规则!
# ✅ 控制发送速率
import time
for ...:
send_email(...)
time.sleep(3) # 间隔 3 秒
总结
本文系统讲解了 Python 邮件处理的核心技能：
- SMTP 发信 — 纯文本/HTML/附件邮件、批量发送、并发控制
- POP3 收信 — 基础邮件读取与解析、编码处理
- IMAP 管理 — 未读标记、文件夹操作、头部预览
- 实战应用 — 系统监控告警、邮件签名图生成、收件箱分析
- 安全实践 — 授权码替代密码、SSL 加密、防垃圾策略
本文是 Python 教程系列的第四十八篇。