在 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()
通過這些擴展實現,可以構建出適應不同場景的完整郵件解決方案,從簡單的通知郵件到復雜的企業級郵件服務。關鍵是根據實際需求選擇合適的技術方案,并注意處理各種邊界情況和異常狀態。