日常办公中我们经常需要处理大量邮件附件,手动下载、分类、归档不仅耗时还容易出错。通过Python结合imaplib协议可以实现邮件附件的自动获取与分类保存,大幅降低重复工作量。

准备工作
首先需要确保邮箱开启IMAP服务,不同邮箱的开启方式略有差异,一般在邮箱设置的账户安全选项中可以找到。同时需要记录邮箱的IMAP服务器地址和端口,比如常见邮箱的配置如下:
| 邮箱类型 | IMAP服务器地址 | 端口 |
|---|---|---|
| QQ邮箱 | imap.qq.com | 993 |
| 163邮箱 | imap.163.com | 993 |
| Gmail | imap.gmail.com | 993 |
另外需要安装Python内置的imaplib、email库,这两个库无需额外安装,Python环境自带。如果需要处理特定格式附件可能还需要额外安装对应库,比如处理PDF可以安装PyPDF2。
连接邮箱服务器
使用imaplib建立SSL连接,登录邮箱账户,示例代码如下:
import imaplib
import email
from email.header import decode_header
import os
# 邮箱配置信息
EMAIL_USER = "your_email@ipipp.com"
EMAIL_PASSWORD = "your_email_password" # QQ邮箱等需要使用授权码,不是登录密码
IMAP_SERVER = "imap.ipipp.com"
IMAP_PORT = 993
# 建立连接并登录
def connect_mailbox():
try:
# 创建SSL连接
mail = imaplib.IMAP4_SSL(IMAP_SERVER, IMAP_PORT)
# 登录邮箱
mail.login(EMAIL_USER, EMAIL_PASSWORD)
print("邮箱登录成功")
return mail
except Exception as e:
print(f"邮箱连接失败:{e}")
return None
获取邮件附件
登录成功后选择收件箱,遍历邮件提取附件内容,需要注意处理邮件的编码问题,避免中文文件名乱码。
def get_attachments(mail):
# 选择收件箱,只读模式
mail.select("INBOX", readonly=False)
# 搜索未读邮件,也可以根据需求修改搜索条件
status, messages = mail.search(None, "UNSEEN")
if status != "OK":
print("搜索邮件失败")
return []
email_ids = messages[0].split()
attachments = []
for email_id in email_ids:
# 获取邮件内容
status, msg_data = mail.fetch(email_id, "(RFC822)")
if status != "OK":
continue
# 解析邮件
raw_email = msg_data[0][1]
msg = email.message_from_bytes(raw_email)
# 获取发件人信息
from_addr = ""
from_header = msg.get("From")
if from_header:
from_parts = decode_header(from_header)
for part in from_parts:
if isinstance(part[0], bytes):
from_addr += part[0].decode(part[1] if part[1] else "utf-8")
else:
from_addr += part[0]
# 遍历邮件所有部分,提取附件
for part in msg.walk():
content_disposition = part.get("Content-Disposition", "")
if "attachment" in content_disposition or "filename" in content_disposition:
# 获取附件文件名
filename_header = part.get_filename()
if filename_header:
# 解码文件名
filename_parts = decode_header(filename_header)
filename = ""
for f_part in filename_parts:
if isinstance(f_part[0], bytes):
filename += f_part[0].decode(f_part[1] if f_part[1] else "utf-8")
else:
filename += f_part[0]
# 获取附件内容
file_data = part.get_payload(decode=True)
if file_data:
attachments.append({
"filename": filename,
"data": file_data,
"from_addr": from_addr,
"email_id": email_id.decode()
})
return attachments
自动化分类保存附件
根据附件的文件后缀、发件人等规则,将附件保存到不同的本地目录,示例分类规则为:按文件类型分类,同时发件人单独建立子目录。
def classify_and_save(attachments, save_root_dir):
# 确保根目录存在
if not os.path.exists(save_root_dir):
os.makedirs(save_root_dir)
for att in attachments:
filename = att["filename"]
file_data = att["data"]
from_addr = att["from_addr"]
# 提取文件后缀
file_ext = os.path.splitext(filename)[1].lower()
# 按文件类型分类目录
type_dir_map = {
".pdf": "PDF文件",
".doc": "Word文件",
".docx": "Word文件",
".xls": "Excel文件",
".xlsx": "Excel文件",
".jpg": "图片文件",
".png": "图片文件",
".txt": "文本文件"
}
type_dir = type_dir_map.get(file_ext, "其他文件")
# 提取发件人邮箱前缀作为子目录名
from_user = from_addr.split("@")[0] if "@" in from_addr else "未知发件人"
# 清理目录名中的特殊字符
from_user = "".join(c for c in from_user if c not in r'/:*?"<>|')
# 构建保存路径
save_dir = os.path.join(save_root_dir, type_dir, from_user)
if not os.path.exists(save_dir):
os.makedirs(save_dir)
# 保存文件
save_path = os.path.join(save_dir, filename)
with open(save_path, "wb") as f:
f.write(file_data)
print(f"附件保存成功:{save_path}")
完整执行流程
将上面的函数组合起来,实现完整的邮件附件获取、分类、保存流程,还可以添加标记已读、异常处理等逻辑。
def main():
# 本地保存根目录
SAVE_ROOT_DIR = "./邮件附件归档"
# 连接邮箱
mail = connect_mailbox()
if not mail:
return
try:
# 获取附件
attachments = get_attachments(mail)
if not attachments:
print("未找到符合条件的附件")
return
print(f"共找到{len(attachments)}个附件")
# 分类保存
classify_and_save(attachments, SAVE_ROOT_DIR)
# 可选:标记邮件为已读
# mail.select("INBOX")
# for att in attachments:
# mail.store(att["email_id"], "+FLAGS", "\Seen")
# 关闭连接
mail.close()
mail.logout()
print("邮箱连接已关闭")
except Exception as e:
print(f"处理过程出错:{e}")
if mail:
mail.close()
mail.logout()
if __name__ == "__main__":
main()
常见问题处理
- 登录失败:如果是QQ邮箱、163邮箱等,需要使用邮箱的授权码而不是登录密码,授权码在邮箱设置中生成。
- 附件文件名乱码:使用
email.header.decode_header方法解码,适配不同编码格式的邮件头。 - 大附件处理:如果附件体积过大,建议分块读取,避免内存占用过高,也可以添加附件大小过滤规则,跳过超大附件。
- 连接超时:可以添加重试机制,当连接失败时自动重试几次,提升脚本的稳定性。