在 Python3 中，使用内置的 smtplib 库和 email 模块发送邮件是一个常见的需求。以下是更详细的实现指南，包含各种场景的解决方案和技术细节：
一、发送纯文本邮件的完整实现
- 准备工作：
- 确保已开通 SMTP 服务（各邮箱开启方式不同）
- 获取 SMTP 授权码（非登录密码）
- 确认服务器地址和端口（常见配置见下表）
| 邮箱服务 | SMTP服务器 | SSL端口 | TLS端口 |
|---|---|---|---|
| QQ邮箱 | smtp.qq.com | 465 | 587 |
| 163邮箱 | smtp.163.com | 465 | 994 |
| Gmail | smtp.gmail.com | 465 | 587 |
- 核心代码详解：
import smtplib
import ssl
from email.mime.text import MIMEText
from email.header import Header
from email.utils import formataddr, formatdate, make_msgid
from datetime import datetime# 增强的邮件配置
config = {"smtp_server": "smtp.examples.com","smtp_port": 465,"sender_email": "your_email@example.com","sender_name": "系统管理员", # 发件人显示名称"password": "your_smtp_password","receivers": [{"email": "user1@example.com", "name": "张经理"},{"email": "user2@example.com", "name": "李主管"}]
}# 构建邮件内容（支持多行模板）
email_template = """尊敬的{recipient_name}：这是来自{system_name}的系统通知邮件。当前时间：{current_time}
系统状态：正常运行
最近事件：
{events}请及时处理相关事务。
"""events = "\n".join(["1. 用户登录异常(3次)","2. 数据库备份完成","3. 新版本发布通知"
])# 填充模板内容
email_body = email_template.format(recipient_name="各位", # 群发时的通用称呼system_name="OA系统",current_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),events=events
)# 创建MIME对象
msg = MIMEText(email_body, "plain", "utf-8")# 设置邮件头（规范格式）
msg["From"] = formataddr((config["sender_name"], config["sender_email"]))
msg["To"] = ", ".join([formataddr((r["name"], r["email"])) for r in config["receivers"]]
)
msg["Subject"] = Header("【重要】系统状态通知", "utf-8")# 高级邮件头设置
msg["Date"] = formatdate(localtime=True)
msg["Message-ID"] = make_msgid()
msg["X-Priority"] = "1" # 邮件优先级(1-5, 1最高)
msg["X-Mailer"] = "Python SMTP" # 邮件客户端标识# 安全发送流程
context = ssl.create_default_context()
try:with smtplib.SMTP_SSL(config["smtp_server"],config["smtp_port"],context=context,timeout=10 # 设置超时时间) as server:server.login(config["sender_email"], config["password"])# 实际发送时区分密送和抄送to_addresses = [r["email"] for r in config["receivers"]]server.sendmail(config["sender_email"],to_addresses,msg.as_string())print(f"成功发送邮件至 {len(to_addresses)} 位收件人")
except smtplib.SMTPException as e:print(f"邮件发送失败，SMTP错误: {str(e)}")
except Exception as e:print(f"发生未知错误: {str(e)}")
二、HTML邮件的专业实现
- 高级功能支持：
- 响应式设计（适应移动端）
- 嵌入式CSS和JavaScript
- 动态内容渲染
- 邮件跟踪（通过嵌入图片）
- 完整示例：
from email.mime.multipart import MIMEMultipart# 创建多部分邮件
msg = MIMEMultipart("alternative")
msg["From"] = formataddr(("市场部", "marketing@company.com"))
msg["To"] = "customer@example.com"# 纯文本备用内容
text_part = MIMEText("这是纯文本备用内容", "plain", "utf-8")# HTML主要内容
html_content = """
<!DOCTYPE html>
<html>
<head><meta http-equiv="Content-Type" content="text/html; charset=utf-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>产品推广邮件</title><style type="text/css">/* 响应式设计 */body { font-family: 'Helvetica Neue', Arial, sans-serif; margin: 0; padding: 0; }.container { max-width: 600px; margin: 0 auto; padding: 20px; }.header { background-color: #3498db; color: white; padding: 20px; text-align: center; }.content { padding: 20px; line-height: 1.6; }.button { display: inline-block; background: #2ecc71; color: white; padding: 10px 20px; text-decoration: none; border-radius: 5px; }.footer { color: #7f8c8d; font-size: 12px; text-align: center; padding: 20px; }@media screen and (max-width: 480px) {.container { width: 100% !important; }}</style>
</head>
<body><div class="container"><div class="header"><h1>新产品发布</h1></div><div class="content"><p>尊敬的客户：</p><p>我们很高兴向您介绍我们的最新产品...</p><p><a href="https://example.com/product" class="button">立即查看</a></p><!-- 产品特性表格 --><table border="0" cellpadding="0" cellspacing="0" width="100%"><tr><td width="50%" valign="top"><h3>核心功能</h3><ul><li>高性能处理</li><li>简单易用</li></ul></td><td width="50%" valign="top"><h3>技术优势</h3><ul><li>先进算法</li><li>稳定可靠</li></ul></td></tr></table><!-- 邮件跟踪像素 --><img src="https://example.com/track?email=customer@example.com" width="1" height="1"></div><div class="footer"><p>© 2023 公司名称. 保留所有权利.</p><p><a href="%unsubscribe_url%">退订邮件</a></p></div></div>
</body>
</html>
"""html_part = MIMEText(html_content, "html", "utf-8")# 添加邮件部分
msg.attach(text_part) # 注意顺序：先添加纯文本版本
msg.attach(html_part)# 发送逻辑同上...
三、带附件邮件的专业实现
邮件附件的处理是邮件系统开发中的关键环节，需要注意以下技术要点：
自动检测MIME类型
- 通过文件扩展名识别（如.jpg对应image/jpeg）
- 使用文件头信息检测（如PDF文件的%PDF标识）
- 可集成第三方库如Apache Tika进行精确识别
- 示例：检测到.txt文件自动设置Content-Type为text/plain
支持大文件分块处理
- 采用BASE64编码时注意76字符换行规则
- 实施断点续传机制
- 使用HTTP/1.1的chunked传输编码
- 典型场景：10MB以上的视频文件传输
处理中文文件名
- 必须进行RFC 2231编码（如=?UTF-8?B?5Lit5paH?=）
- 文件名长度限制为75个字符
- 避免使用非ASCII字符的扩展名
- 示例处理流程：UTF-8编码 → BASE64编码 → MIME头格式化
添加多个附件
- 每个附件作为独立的MIME part
- 使用multipart/mixed作为顶层Content-Type
- 注意附件顺序对邮件客户端显示的影响
- 典型实现：附件1（合同.pdf）+附件2（报价单.xlsx）
其他注意事项
- 设置正确的Content-Disposition（attachment/inline）
- 处理Windows/Linux路径差异
- 添加附件描述信息（Content-Description头字段）
- 安全考虑：病毒扫描和文件类型限制
- 完整示例：
import os
import mimetypes
from email.mime.base import MIMEBase
from email import encodersdef add_attachment(msg, filepath):"""专业添加附件方法"""if not os.path.exists(filepath):raise FileNotFoundError(f"附件文件不存在: {filepath}")# 猜测MIME类型ctype, encoding = mimetypes.guess_type(filepath)if ctype is None or encoding is not None:ctype = "application/octet-stream"maintype, subtype = ctype.split("/", 1)with open(filepath, "rb") as fp:part = MIMEBase(maintype, subtype)part.set_payload(fp.read())# 编码和设置头信息encoders.encode_base64(part)# 处理中文文件名filename = os.path.basename(filepath)part.add_header("Content-Disposition","attachment",filename=Header(filename, "utf-8").encode())msg.attach(part)# 创建带附件的邮件
msg = MIMEMultipart()
msg["Subject"] = "季度报告和数据分析"# 添加正文
msg.attach(MIMEText("请查收附件中的季度报告", "plain"))# 添加多个附件
attachments = ["/reports/Q3_Report.pdf","/data/sales_data.xlsx","/images/performance_chart.png"
]for attachment in attachments:try:add_attachment(msg, attachment)print(f"已添加附件: {os.path.basename(attachment)}")except Exception as e:print(f"添加附件失败: {str(e)}")# 发送逻辑...
四、企业级最佳实践
连接池管理增强版
from smtplib import SMTP_SSL
from queue import Queue
import threadingclass SMTPConnectionPool:def __init__(self, host, port, username, password, pool_size=5):self.host = hostself.port = portself.username = usernameself.password = passwordself.pool = Queue(pool_size)self.lock = threading.Lock()# 初始化连接池for _ in range(pool_size):conn = SMTP_SSL(host, port)conn.login(username, password)self.pool.put(conn)def get_connection(self):return self.pool.get()def release_connection(self, conn):self.pool.put(conn)def send_email(self, msg, recipients):conn = Nonetry:conn = self.get_connection()conn.sendmail(self.username, recipients, msg.as_string())finally:if conn:self.release_connection(conn)# 使用示例
pool = SMTPConnectionPool(host="smtp.example.com",port=465,username="user@example.com",password="password",pool_size=10
)# 在多线程环境中使用
def worker(email_list):for email in email_list:msg = build_email_message(email)pool.send_email(msg, [email["address"]])threads = []
for i in range(5):t = threading.Thread(target=worker, args=(email_chunks[i],))threads.append(t)t.start()for t in threads:t.join()
完整的邮件服务类
import logging
from logging.handlers import RotatingFileHandlerclass EmailService:def __init__(self, config):self.config = configself.logger = self._setup_logger()self.connection_pool = self._init_connection_pool()def _setup_logger(self):logger = logging.getLogger("EmailService")logger.setLevel(logging.INFO)handler = RotatingFileHandler("email_service.log",maxBytes=10*1024*1024, # 10MBbackupCount=5)formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")handler.setFormatter(formatter)logger.addHandler(handler)return loggerdef _init_connection_pool(self):return SMTPConnectionPool(host=self.config["smtp_host"],port=self.config["smtp_port"],username=self.config["smtp_user"],password=self.config["smtp_password"],pool_size=self.config.get("pool_size", 5))def send_email(self, template_name, recipient, context, attachments=None):"""发送模板邮件"""try:# 1. 加载模板template = self._load_template(template_name)# 2. 构建邮件内容msg = self._build_message(template, recipient, context)# 3. 添加附件if attachments:for attachment in attachments:self._add_attachment(msg, attachment)# 4. 发送邮件self.connection_pool.send_email(msg,[recipient["email"]])self.logger.info(f"邮件发送成功: {template_name} -> {recipient['email']}")return Trueexcept Exception as e:self.logger.error(f"邮件发送失败: {template_name} -> {recipient['email']}: {str(e)}",exc_info=True)return False# 其他辅助方法...def _load_template(self, name):"""加载邮件模板"""passdef _build_message(self, template, recipient, context):"""构建邮件消息"""passdef _add_attachment(self, msg, filepath):"""添加附件"""pass
监控与统计集成
from prometheus_client import CollectorRegistry, push_to_gateway
from datetime import datetimeclass EmailMetrics:def __init__(self):self.registry = CollectorRegistry()self.emails_sent = Counter("emails_sent_total","Total emails sent",["template"],registry=self.registry)self.send_time = Summary("email_send_time_seconds","Time spent sending emails",registry=self.registry)self.errors = Counter("email_errors_total","Total email sending errors",["type"],registry=self.registry)def record_success(self, template, duration):self.emails_sent.labels(template).inc()self.send_time.observe(duration)def record_error(self, error_type):self.errors.labels(error_type).inc()def push_metrics(self):push_to_gateway("metrics.example.com:9091",job="email_service",registry=self.registry)# 使用示例
metrics = EmailMetrics()@metrics.send_time.time()
def send_email_with_metrics(email):try:start_time = datetime.now()# 发送邮件逻辑...duration = (datetime.now() - start_time).total_seconds()metrics.record_success(email["template"], duration)return Trueexcept smtplib.SMTPException as e:metrics.record_error("smtp")raiseexcept Exception as e:metrics.record_error("other")raisefinally:metrics.push_metrics()
五、高级主题扩展
DKIM签名支持
import dkimdef add_dkim_signature(msg, domain, selector, private_key):"""添加DKIM签名"""headers = ["From", "To", "Subject"]sig = dkim.sign(message=msg.as_bytes(),selector=selector.encode(),domain=domain.encode(),privkey=private_key.encode(),include_headers=headers)msg["DKIM-Signature"] = sig[len("DKIM-Signature: "):].decode()return msg# 使用示例
private_key = """-----BEGIN RSA PRIVATE KEY-----
...
-----END RSA PRIVATE KEY-----"""msg = add_dkim_signature(msg,domain="example.com",selector="selector1",private_key=private_key
)
邮件队列系统集成
import redis
import json
import pickleclass EmailQueue:def __init__(self, redis_host="localhost", redis_port=6379):self.redis = redis.Redis(host=redis_host,port=redis_port,db=0,decode_responses=False)self.queue_name = "email_queue"def enqueue(self, email_task):"""将邮件任务加入队列"""serialized = pickle.dumps(email_task)self.redis.rpush(self.queue_name, serialized)def dequeue(self):"""从队列取出邮件任务"""serialized = self.redis.lpop(self.queue_name)if serialized:return pickle.loads(serialized)return Nonedef process_queue(self, worker_count=4):"""处理队列中的邮件"""def worker():while True:task = self.dequeue()if not task:breaktry:send_email(**task)except Exception as e:self._handle_failure(task, e)threads = []for _ in range(worker_count):t = threading.Thread(target=worker)threads.append(t)t.start()for t in threads:t.join()# 使用示例
queue = EmailQueue()# 生产者添加任务
queue.enqueue({"to": "user@example.com","subject": "测试邮件","body": "这是一封测试邮件","template": "welcome"
})# 消费者处理队列
queue.process_queue()
通过这些扩展实现，可以构建出适应不同场景的完整邮件解决方案，从简单的通知邮件到复杂的企业级邮件服务。关键是根据实际需求选择合适的技术方案，并注意处理各种边界情况和异常状态。
