本文目录导读：
我来介绍几种批量下载邮件附件的方法,从简单到复杂：
使用邮件客户端（最简单）
Outlook客户端
# 使用Python的win32com库操作Outlook
import win32com.client
import os
from datetime import datetime, timedelta
def download_outlook_attachments():
outlook = win32com.client.Dispatch("Outlook.Application").GetNamespace("MAPI")
inbox = outlook.GetDefaultFolder(6) # 6 = Inbox
# 获取最近7天的邮件
date_filter = datetime.now() - timedelta(days=7)
for message in inbox.Items:
try:
if message.ReceivedTime >= date_filter:
# 检查是否有附件
if message.Attachments.Count > 0:
print(f"处理邮件: {message.Subject}")
# 保存附件
for attachment in message.Attachments:
save_path = os.path.join("C:\\attachments", attachment.FileName)
attachment.SaveAsFile(save_path)
print(f" 已保存: {attachment.FileName}")
except Exception as e:
print(f"处理邮件时出错: {e}")
# 运行
download_outlook_attachments()
使用IMAP协议（推荐）
Python + imaplib
import imaplib
import email
from email.header import decode_header
import os
import base64
class EmailAttachmentDownloader:
def __init__(self, imap_server, email_address, password):
self.imap_server = imap_server
self.email_address = email_address
self.password = password
self.connection = None
self.save_dir = "C:\\downloaded_attachments"
def connect(self):
"""连接到IMAP服务器"""
self.connection = imaplib.IMAP4_SSL(self.imap_server)
self.connection.login(self.email_address, self.password)
print("成功连接到邮件服务器")
def select_mailbox(self, mailbox="INBOX"):
"""选择邮箱文件夹"""
status, messages = self.connection.select(mailbox)
if status == 'OK':
print(f"已选择 {mailbox}，共有 {int(messages[0])} 封邮件")
else:
print(f"选择 {mailbox} 失败")
def decode_header(self, header_value):
"""解码邮件头信息"""
if isinstance(header_value, tuple):
return header_value[0]
try:
decoded = decode_header(header_value)
subject = ''
for part, encoding in decoded:
if isinstance(part, bytes):
subject += part.decode(encoding or 'utf-8', errors='ignore')
else:
subject += part
return subject
except:
return str(header_value)
def download_attachments_from_date(self, date_str):
"""按日期下载附件
date_str: 格式如 '2024-01-01'
"""
if not os.path.exists(self.save_dir):
os.makedirs(self.save_dir)
# 搜索该日期之后的所有邮件
status, messages = self.connection.search(None, f'SINCE {date_str}')
if status != 'OK':
print("搜索邮件失败")
return
email_ids = messages[0].split()
print(f"找到 {len(email_ids)} 封邮件")
for email_id in email_ids:
try:
# 获取邮件
status, msg_data = self.connection.fetch(email_id, '(RFC822)')
if status != 'OK':
continue
# 解析邮件
msg = email.message_from_bytes(msg_data[0][1])
subject = self.decode_header(msg['Subject'])
print(f"\n处理邮件: {subject}")
# 遍历邮件部分
for part in msg.walk():
# 检查是否有附件
if part.get_content_maintype() == 'multipart':
continue
filename = part.get_filename()
if filename:
# 解码文件名
filename = self.decode_header(filename)
print(f" 找到附件: {filename}")
# 处理文件名并发问题
if os.path.exists(os.path.join(self.save_dir, filename)):
filename = f"{os.path.splitext(filename)[0]}_{email_id.decode()}{os.path.splitext(filename)[1]}"
# 保存附件
filepath = os.path.join(self.save_dir, filename)
with open(filepath, 'wb') as f:
f.write(part.get_payload(decode=True))
print(f" 已保存到: {filepath}")
except Exception as e:
print(f"处理邮件 {email_id} 时出错: {e}")
def download_attachments_by_sender(self, sender_email):
"""按发件人下载附件"""
# 构建搜索条件
criteria = f'FROM "{sender_email}"'
status, messages = self.connection.search(None, criteria)
if status != 'OK':
print("搜索邮件失败")
return
email_ids = messages[0].split()
print(f"从 {sender_email} 找到 {len(email_ids)} 封邮件")
# 继续处理...（类似上面的方法）
self._process_emails(email_ids)
def _process_emails(self, email_ids):
"""处理邮件列表，提取附件"""
for email_id in email_ids:
status, msg_data = self.connection.fetch(email_id, '(RFC822)')
if status != 'OK':
continue
msg = email.message_from_bytes(msg_data[0][1])
subject = self.decode_header(msg['Subject'])
print(f"处理邮件: {subject}")
for part in msg.walk():
if part.get_content_maintype() == 'multipart':
continue
filename = part.get_filename()
if filename:
filename = self.decode_header(filename)
filepath = os.path.join(self.save_dir, filename)
with open(filepath, 'wb') as f:
f.write(part.get_payload(decode=True))
print(f" 已保存: {filename}")
def close(self):
"""关闭连接"""
if self.connection:
self.connection.close()
self.connection.logout()
print("已关闭连接")
# 使用示例
if __name__ == "__main__":
# 配置信息
IMAP_SERVER = "imap.gmail.com" # 如果使用Gmail
EMAIL = "your.email@gmail.com"
PASSWORD = "your_app_password" # 建议使用应用密码
# 创建下载器实例
downloader = EmailAttachmentDownloader(IMAP_SERVER, EMAIL, PASSWORD)
try:
# 连接并选择邮箱
downloader.connect()
downloader.select_mailbox("INBOX")
# 下载指定日期后的所有附件
downloader.download_attachments_from_date("01-Apr-2024")
# 或者按发件人下载
# downloader.download_attachments_by_sender("sender@example.com")
finally:
# 关闭连接
downloader.close()
使用Poplib（POP3协议）
import poplib
import email
from email.header import decode_header
import os
def download_pop3_attachments(server, username, password, max_emails=50):
"""使用POP3下载附件"""
# 连接POP3服务器
pop_conn = poplib.POP3_SSL(server)
pop_conn.user(username)
pop_conn.pass_(password)
# 获取邮件数量
num_messages = len(pop_conn.list()[1])
print(f"邮箱中共有 {num_messages} 封邮件")
# 创建保存目录
save_dir = "C:\\pop3_downloads"
if not os.path.exists(save_dir):
os.makedirs(save_dir)
# 只处理最近的邮件
messages_to_process = min(num_messages, max_emails)
for i in range(num_messages, num_messages - messages_to_process, -1):
try:
# 获取邮件
response, lines, octets = pop_conn.retr(i)
msg_content = b'\r\n'.join(lines)
# 解析邮件
msg = email.message_from_bytes(msg_content)
# 解码主题
subject = decode_header(msg['Subject'])[0][0]
if isinstance(subject, bytes):
subject = subject.decode(errors='ignore')
print(f"\n处理邮件: {subject}")
# 检查附件
for part in msg.walk():
if part.get_content_maintype() == 'multipart':
continue
filename = part.get_filename()
if filename:
# 解码文件名
decoded_header = decode_header(filename)
filename = decoded_header[0][0] if isinstance(decoded_header[0][0], str) else decoded_header[0][0].decode(errors='ignore')
# 保存附件
filepath = os.path.join(save_dir, filename)
with open(filepath, 'wb') as f:
f.write(part.get_payload(decode=True))
print(f" 已保存: {filename}")
except Exception as e:
print(f"处理邮件 {i} 时出错: {e}")
# 关闭连接
pop_conn.quit()
# 使用示例
# download_pop3_attachments("pop.gmail.com", "your.email@gmail.com", "your_password")
Gmail API（功能最强大）
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
import base64
import os
import json
# Gmail API 设置
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']
class GmailAttachmentDownloader:
def __init__(self, credentials_file):
self.credentials_file = credentials_file
self.service = None
self.creds = None
def authenticate(self):
"""认证Gmail API"""
# 如果token.json存在，使用已经保存的凭据
if os.path.exists('token.json'):
self.creds = Credentials.from_authorized_user_file('token.json', SCOPES)
# 如果没有有效的凭据，需要进行授权
if not self.creds or not self.creds.valid:
if self.creds and self.creds.expired and self.creds.refresh_token:
self.creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(
self.credentials_file, SCOPES)
self.creds = flow.run_local_server(port=0)
# 保存凭据
with open('token.json', 'w') as token:
token.write(self.creds.to_json())
self.service = build('gmail', 'v1', credentials=self.creds)
print("Gmail API连接成功")
def get_attachments_from_query(self, query):
"""按查询条件搜索邮件并下载附件"""
save_dir = "G:\\gmail_attachments"
if not os.path.exists(save_dir):
os.makedirs(save_dir)
# 搜索邮件
results = self.service.users().messages().list(
userId='me', q=query).execute()
messages = results.get('messages', [])
print(f"找到 {len(messages)} 封邮件")
for message in messages:
try:
# 获取邮件详情
msg = self.service.users().messages().get(
userId='me', id=message['id'], format='full').execute()
# 获取主题
subject = ''
for header in msg['payload']['headers']:
if header['name'] == 'Subject':
subject = header['value']
break
print(f"\n处理邮件: {subject}")
# 递归处理邮件部分
self._process_parts(msg['payload'], save_dir, msg['id'])
except Exception as e:
print(f"处理邮件出错: {e}")
def _process_parts(self, parts, save_dir, msg_id):
"""递归处理邮件内容"""
if 'parts' in parts:
# 处理子部分
for part in parts['parts']:
self._process_parts(part, save_dir, msg_id)
else:
# 检查是否有附件
filename = parts.get('filename')
if filename:
print(f" 找到附件: {filename}")
# 获取附件数据
attachment_id = parts['body'].get('attachmentId')
if attachment_id:
attachment = self.service.users().messages().attachments().get(
userId='me', messageId=msg_id, id=attachment_id).execute()
# 解码附件数据
file_data = base64.urlsafe_b64decode(attachment['data'])
# 保存文件
filepath = os.path.join(save_dir, filename)
with open(filepath, 'wb') as f:
f.write(file_data)
print(f" 已保存: {filepath}")
# 使用示例
if __name__ == "__main__":
downloader = GmailAttachmentDownloader("client_secret.json")
downloader.authenticate()
# 按条件搜索邮件
# query示例：
# "from:example@gmail.com has:attachment" - 特定发件人的附件
# "subject:财务报表 has:attachment" - 特定主题
# "after:2024/01/01 has:attachment" - 特定日期后
downloader.get_attachments_from_query("has:attachment from:example@gmail.com")
邮件转发设置
如果你不想写代码,也可以通过邮箱设置来自动处理：
Gmail设置
- 打开Gmail设置 → 转发和POP/IMAP
- 创建过滤器
- 设置转发规则
最佳实践建议
安全考虑
# 不要硬编码密码，使用环境变量
import os
EMAIL_PASSWORD = os.getenv('EMAIL_PASSWORD')
# 或者使用配置文件
import configparser
config = configparser.ConfigParser()
config.read('config.ini')
PASSWORD = config['email']['password']
错误处理
def safe_download(fn):
"""安全下载包装器"""
def wrapper(*args, **kwargs):
max_retries = 3
for attempt in range(max_retries):
try:
return fn(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
print(f"重试 {attempt + 1}/{max_retries}: {e}")
time.sleep(5)
return wrapper
日志记录
import logging
logging.basicConfig(
filename='attachment_downloader.log',
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
选择哪种方法取决于你的需求：
- 快速简单：使用邮件客户端
- 自动化：使用IMAP或POP3
- 高级功能：使用Gmail API
- 大规模处理：组合使用多种方法
需要更多具体帮助吗？我可以详细解释某个方法的实现。
