如何编写批量下载邮件附件

wen 实用脚本 2026-08-01 198

本文目录导读：

如何编写批量下载邮件附件

使用邮件客户端（最简单）
使用IMAP协议（推荐）
使用Poplib（POP3协议）
Gmail API（功能最强大）
邮件转发设置
最佳实践建议

我来介绍几种批量下载邮件附件的方法,从简单到复杂：

使用邮件客户端（最简单）

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__)

选择哪种方法取决于你的需求：