Python使用imaplib和email庫實(shí)現(xiàn)自動化郵件處理教程

更新時(shí)間：2025年08月01日 09:09:24 作者：老哥不老

在數(shù)字化辦公場景中,郵件自動化是提升工作效率的關(guān)鍵技能,下面這篇文章主要介紹了Python使用imaplib和email庫實(shí)現(xiàn)自動化郵件處理的相關(guān)資料,文中通過代碼介紹的非常詳細(xì),需要的朋友可以參考下

前言

在日常工作中，我們經(jīng)常需要自動化處理電子郵件，比如自動下載附件、解析郵件內(nèi)容、處理特定格式的數(shù)據(jù)等。本文將通過一個(gè)實(shí)際案例，詳細(xì)介紹如何使用Python的imaplib和email庫來實(shí)現(xiàn)郵件的自動化處理。

1. 環(huán)境準(zhǔn)備與庫介紹

首先，我們需要導(dǎo)入必要的庫：

import imaplib          # IMAP協(xié)議客戶端
import email           # 郵件解析
from email.header import decode_header  # 解碼郵件頭
import re              # 正則表達(dá)式
import chardet         # 字符編碼檢測
import os
import pandas as pd    # 數(shù)據(jù)處理
from datetime import datetime

主要庫說明：

imaplib: Python內(nèi)置庫，用于通過IMAP協(xié)議訪問郵件服務(wù)器
email: 用于解析郵件內(nèi)容和結(jié)構(gòu)
chardet: 自動檢測字符編碼，處理不同編碼的郵件內(nèi)容
pandas: 處理Excel等數(shù)據(jù)文件

2. IMAP郵件服務(wù)器連接

IMAP（Internet Message Access Protocol）是一種郵件獲取協(xié)議，相比POP3，它支持在線操作郵件，不會刪除服務(wù)器上的郵件。

# 設(shè)置郵箱信息
username = "your_email@139.com"
password = "your_password" # 不是密碼，是密鑰，對應(yīng)的右鍵系統(tǒng)設(shè)置中獲取
imap_url = "imap.139.com"

# 連接IMAP服務(wù)器
def connect_to_email():
    try:
        # 使用SSL加密連接
        mail = imaplib.IMAP4_SSL(imap_url)
        mail.login(username, password)
        return mail
    except Exception as e:
        print(f"連接失敗: {e}")
        return None

常見郵箱的IMAP服務(wù)器地址：

Gmail: imap.gmail.com
Outlook: outlook.office365.com
QQ郵箱: imap.qq.com
163郵箱: imap.163.com
139郵箱: imap.139.com

3. 郵件搜索與獲取

連接成功后，我們可以選擇郵箱文件夾并搜索郵件：

def search_emails(mail, folder="inbox", criteria='UNSEEN'):
    """
    搜索郵件
    :param mail: IMAP連接對象
    :param folder: 郵箱文件夾，默認(rèn)收件箱
    :param criteria: 搜索條件，UNSEEN表示未讀郵件
    """
    # 選擇郵箱文件夾
    mail.select(folder)
    
    # 搜索郵件
    status, messages = mail.search(None, criteria)
    
    if status != 'OK':
        print("搜索失敗")
        return []
    
    # 獲取郵件ID列表
    message_ids = messages[0].split()
    return message_ids

常用搜索條件：

'ALL': 所有郵件
'UNSEEN': 未讀郵件
'SEEN': 已讀郵件
'SUBJECT "關(guān)鍵詞"': 主題包含特定關(guān)鍵詞
'FROM "sender@email.com"': 來自特定發(fā)件人
'SINCE "01-Jan-2024"': 特定日期之后的郵件

4. 郵件內(nèi)容解析

獲取郵件后，需要解析郵件內(nèi)容：

def parse_email(mail, message_id):
    """解析郵件內(nèi)容"""
    # 獲取郵件數(shù)據(jù)
    status, data = mail.fetch(message_id, '(RFC822)')
    email_body = data[0][1]
    
    # 解析郵件
    email_message = email.message_from_bytes(email_body)
    
    # 解析發(fā)件人
    from_header = decode_header(email_message["From"])[0]
    if isinstance(from_header[0], bytes):
        sender = from_header[0].decode(from_header[1] or 'utf-8')
    else:
        sender = from_header[0]
    
    # 提取郵箱地址
    email_pattern = r'[\w\.-]+@[\w\.-]+'
    email_match = re.search(email_pattern, sender)
    sender_email = email_match.group() if email_match else sender
    
    # 解析主題
    subject_header = decode_header(email_message["Subject"])[0]
    if isinstance(subject_header[0], bytes):
        subject = subject_header[0].decode(subject_header[1] or 'utf-8')
    else:
        subject = subject_header[0]
    
    return {
        'sender': sender,
        'sender_email': sender_email,
        'subject': subject,
        'message': email_message
    }

5. 附件處理

處理郵件附件是郵件自動化的重要部分：

def process_attachments(email_message, save_dir):
    """處理郵件附件"""
    attachments = []
    
    for part in email_message.walk():
        # 跳過multipart容器
        if part.get_content_maintype() == 'multipart':
            continue
        
        # 獲取附件文件名
        filename = part.get_filename()
        if filename:
            # 解碼文件名
            filename_tuple = decode_header(filename)[0]
            if isinstance(filename_tuple[0], bytes):
                filename = filename_tuple[0].decode(filename_tuple[1] or 'utf-8')
            
            # 保存附件
            filepath = os.path.join(save_dir, filename)
            with open(filepath, 'wb') as f:
                f.write(part.get_payload(decode=True))
            
            attachments.append({
                'filename': filename,
                'filepath': filepath
            })
            print(f"已保存附件: {filename}")
    
    return attachments

6. 實(shí)戰(zhàn)案例：自動化處理Excel附件

下面是一個(gè)完整的實(shí)戰(zhàn)案例，展示如何自動處理包含Excel附件的郵件：

def process_excel_attachment(filepath):
    """處理Excel文件并生成報(bào)告"""
    # 讀取Excel數(shù)據(jù)
    data = pd.read_excel(filepath, sheet_name='數(shù)據(jù)表')
    
    # 數(shù)據(jù)處理（示例：篩選特定條件）
    filtered_data = data[data['狀態(tài)'] == '待處理']
    
    # 數(shù)據(jù)分析
    summary = {
        '總記錄數(shù)': len(data),
        '待處理數(shù)': len(filtered_data),
        '處理率': f"{(1 - len(filtered_data)/len(data))*100:.2f}%"
    }
    
    # 生成報(bào)告
    report_path = filepath.replace('.xlsx', '_報(bào)告.txt')
    with open(report_path, 'w', encoding='utf-8') as f:
        f.write(f"數(shù)據(jù)分析報(bào)告\n")
        f.write(f"生成時(shí)間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
        for key, value in summary.items():
            f.write(f"{key}: {value}\n")
    
    return report_path

def main():
    """主函數(shù)：完整的郵件處理流程"""
    # 創(chuàng)建附件保存目錄
    ATTACHMENT_DIR = "email_attachments"
    if not os.path.exists(ATTACHMENT_DIR):
        os.makedirs(ATTACHMENT_DIR)
    
    # 連接郵箱
    mail = connect_to_email()
    if not mail:
        return
    
    try:
        # 搜索未讀郵件
        message_ids = search_emails(mail, criteria='UNSEEN')
        
        if not message_ids:
            print("沒有未讀郵件")
            return
        
        # 處理每封郵件
        for msg_id in message_ids:
            # 解析郵件
            email_info = parse_email(mail, msg_id)
            print(f"\n處理郵件: {email_info['subject']}")
            print(f"發(fā)件人: {email_info['sender_email']}")
            
            # 處理附件
            attachments = process_attachments(
                email_info['message'], 
                ATTACHMENT_DIR
            )
            
            # 處理Excel附件
            for attachment in attachments:
                if attachment['filename'].endswith('.xlsx'):
                    report_path = process_excel_attachment(
                        attachment['filepath']
                    )
                    print(f"生成報(bào)告: {report_path}")
            
            # 標(biāo)記郵件為已讀
            mail.store(msg_id, '+FLAGS', '\\Seen')
            print(f"郵件已標(biāo)記為已讀")
    
    except Exception as e:
        print(f"處理郵件時(shí)出錯(cuò): {e}")
    
    finally:
        # 關(guān)閉連接
        mail.close()
        mail.logout()
        print("\n郵件處理完成，連接已關(guān)閉")

if __name__ == "__main__":
    main()

7. 最佳實(shí)踐與注意事項(xiàng)

7.1 安全性建議

密碼管理：不要在代碼中硬編碼密碼，使用環(huán)境變量或配置文件

import os
from dotenv import load_dotenv

load_dotenv()
username = os.getenv('EMAIL_USERNAME')
password = os.getenv('EMAIL_PASSWORD')

使用應(yīng)用專用密碼：很多郵箱服務(wù)（如Gmail）需要使用應(yīng)用專用密碼而非賬戶密碼

7.2 錯(cuò)誤處理

完善的錯(cuò)誤處理能讓程序更穩(wěn)定：

def safe_decode(data, encoding=None):
    """安全解碼函數(shù)"""
    if isinstance(data, bytes):
        if encoding:
            return data.decode(encoding, errors='ignore')
        else:
            # 自動檢測編碼
            detected = chardet.detect(data)
            return data.decode(detected['encoding'] or 'utf-8', errors='ignore')
    return data

7.3 性能優(yōu)化

批量處理：一次獲取多封郵件，避免頻繁連接
并發(fā)處理：使用多線程或異步處理大量郵件
緩存機(jī)制：對已處理的郵件ID進(jìn)行緩存，避免重復(fù)處理

7.4 郵件標(biāo)記和文件夾操作

# 標(biāo)記郵件
mail.store(msg_id, '+FLAGS', '\\Flagged')  # 標(biāo)記為重要
mail.store(msg_id, '+FLAGS', '\\Deleted')  # 標(biāo)記為刪除

# 移動郵件到其他文件夾
mail.copy(msg_id, 'Processed')  # 復(fù)制到已處理文件夾
mail.store(msg_id, '+FLAGS', '\\Deleted')  # 標(biāo)記原郵件為刪除

7.5 定時(shí)任務(wù)

使用Python的schedule庫或系統(tǒng)的cron來實(shí)現(xiàn)定時(shí)執(zhí)行：

import schedule
import time

def job():
    print("開始處理郵件...")
    main()

# 每小時(shí)執(zhí)行一次
schedule.every(1).hours.do(job)

while True:
    schedule.run_pending()
    time.sleep(60)