Python使用imaplib和email库实现自动化邮件处理教程

import imaplib          # imap协议客户端
import email           # 邮件解析
from email.header import decode_header  # 解码邮件头
import re              # 正则表达式
import chardet         # 字符编码检测
import os
import pandas as pd    # 数据处理
from datetime import datetime

# 设置邮箱信息
username = "your_email@139.com"
password = "your_password" # 不是密码，是密钥，对应的右键系统设置中获取
imap_url = "imap.139.com"

# 连接imap服务器
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

def search_emails(mail, folder="inbox", criteria='unseen'):
    """
    搜索邮件
    :param mail: imap连接对象
    :param folder: 邮箱文件夹，默认收件箱
    :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

def parse_email(mail, message_id):
    """解析邮件内容"""
    # 获取邮件数据
    status, data = mail.fetch(message_id, '(rfc822)')
    email_body = data[0][1]
    
    # 解析邮件
    email_message = email.message_from_bytes(email_body)
    
    # 解析发件人
    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
    }

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

def process_excel_attachment(filepath):
    """处理excel文件并生成报告"""
    # 读取excel数据
    data = pd.read_excel(filepath, sheet_name='数据表')
    
    # 数据处理（示例：筛选特定条件）
    filtered_data = data[data['状态'] == '待处理']
    
    # 数据分析
    summary = {
        '总记录数': len(data),
        '待处理数': len(filtered_data),
        '处理率': f"{(1 - len(filtered_data)/len(data))*100:.2f}%"
    }
    
    # 生成报告
    report_path = filepath.replace('.xlsx', '_报告.txt')
    with open(report_path, 'w', encoding='utf-8') as f:
        f.write(f"数据分析报告\n")
        f.write(f"生成时间: {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():
    """主函数：完整的邮件处理流程"""
    # 创建附件保存目录
    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"发件人: {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"生成报告: {report_path}")
            
            # 标记邮件为已读
            mail.store(msg_id, '+flags', '\\seen')
            print(f"邮件已标记为已读")
    
    except exception as e:
        print(f"处理邮件时出错: {e}")
    
    finally:
        # 关闭连接
        mail.close()
        mail.logout()
        print("\n邮件处理完成，连接已关闭")

if __name__ == "__main__":
    main()

import os
from dotenv import load_dotenv

load_dotenv()
username = os.getenv('email_username')
password = os.getenv('email_password')

def safe_decode(data, encoding=none):
    """安全解码函数"""
    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

# 标记邮件
mail.store(msg_id, '+flags', '\\flagged')  # 标记为重要
mail.store(msg_id, '+flags', '\\deleted')  # 标记为删除

# 移动邮件到其他文件夹
mail.copy(msg_id, 'processed')  # 复制到已处理文件夹
mail.store(msg_id, '+flags', '\\deleted')  # 标记原邮件为删除

import schedule
import time

def job():
    print("开始处理邮件...")
    main()

# 每小时执行一次
schedule.every(1).hours.do(job)

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