Python使用imaplib与Scikit-learn打造智能邮件过滤器

pip install scikit-learn pandas numpy

import imaplib
import email
from email.header import decode_header

def fetch_emails(username, password, imap_server='imap.qq.com', limit=50):
    # 连接到服务器
    mail = imaplib.IMAP4_SSL(imap_server)
    mail.login(username, password)
    
    # 选择收件箱
    mail.select('INBOX')
    
    # 搜索所有邮件
    status, messages = mail.search(None, 'ALL')
    email_ids = messages[0].split()
    
    # 取最新的 limit 封邮件
    emails_data = []
    for e_id in email_ids[-limit:]:
        status, msg_data = mail.fetch(e_id, '(RFC822)')
        
        # 解析邮件内容
        for response_part in msg_data:
            if isinstance(response_part, tuple):
                msg = email.message_from_bytes(response_part[1])
                
                # 解码主题
                subject, encoding = decode_header(msg["Subject"])[0]
                if isinstance(subject, bytes):
                    subject = subject.decode(encoding if encoding else 'utf-8')
                
                # 提取发件人
                from_ = msg.get("From")
                
                # 提取正文 (简单处理，仅取第一个text/plain部分)
                body = ""
                if msg.is_multipart():
                    for part in msg.walk():
                        content_type = part.get_content_type()
                        content_disposition = str(part.get("Content-Disposition"))
                        if content_type == "text/plain" and "attachment" not in content_disposition:
                            body = part.get_payload(decode=True).decode()
                            break
                else:
                    body = msg.get_payload(decode=True).decode()
                
                emails_data.append({
                    "id": e_id,
                    "subject": subject,
                    "from": from_,
                    "body": body
                })
    mail.close()
    mail.logout()
    return emails_data

# 示例调用 (请替换为你自己的账号和授权码)
# emails = fetch_emails('your_email@qq.com', 'your_auth_code')
# print(f"获取了 {len(emails)} 封邮件")

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import make_pipeline
from sklearn.metrics import classification_report

# 1. 模拟训练数据 (实际中应从历史邮件提取并标注)
# 0: 重要/工作, 1: 垃圾/无关
data = [
    ("【报警】服务器CPU使用率超过90%", 0),
    ("项目周会通知 - 请准时参加", 0),
    ("恭喜你获得100元优惠券", 1),
    ("发票确认：请查收附件", 0),
    ("澳门首家赌场上线啦", 1),
    ("Python技术交流群邀请", 1),
    ("紧急：数据库连接失败", 0),
    ("Weekly Report - Team A", 0),
    ("代开发票，诚信合作", 1),
    ("系统维护通知", 0)
]

# 转换为DataFrame
df = pd.DataFrame(data, columns=['text', 'label'])

# 2. 构建管道 (Pipeline)
# TfidfVectorizer: 将文本转换为数值向量
# MultinomialNB: 朴素贝叶斯分类器
model = make_pipeline(TfidfVectorizer(), MultinomialNB())

# 3. 训练模型
# 这里的X是邮件内容，y是标签
X_train = df['text']
y_train = df['label']

model.fit(X_train, y_train)

print("模型训练完成！")

# 模拟新的邮件
new_emails = [
    "你好，这是本周的工作总结",
    "急急急！内部系统升级维护",
    "注册即送大礼包，点击领取",
    "财务部门请注意查收邮件"
]

predictions = model.predict(new_emails)

for email, pred in zip(new_emails, predictions):
    label = "重要" if pred == 0 else "垃圾/忽略"
    print(f"邮件: '{email}' -> 预测结果: {label}")

def auto_process_emails(username, password, model):
    # 1. 获取邮件
    emails = fetch_emails(username, password, limit=20)
    
    # 2. 连接服务器准备操作
    mail = imaplib.IMAP4_SSL('imap.qq.com')
    mail.login(username, password)
    mail.select('INBOX')
    
    for item in emails:
        # 提取特征 (这里简单用subject + body)
        text = item['subject'] + " " + item['body']
        
        # 3. 预测
        prediction = model.predict([text])[0]
        
        if prediction == 1: # 假设1是垃圾/低优先级
            print(f"正在处理低优先级邮件: {item['subject']}")
            
            # 获取邮件ID (注意：fetch_emails里的ID是bytes，需要转码查找)
            # 这里为了演示简化，实际需要根据RFC822头匹配或索引处理
            # 简单的IMAP操作示例：
            # 标记为已读
            # mail.store(item['id'], '+FLAGS', '\\Seen')
            # 移动到其他文件夹 (需要先创建该文件夹)
            # mail.copy(item['id'], 'ProcessedBox')
            
        else:
            print(f"发现重要邮件: {item['subject']} (保持原样)")

# 注意：在实际运行前，请确保你已经用足够多的数据训练了模型
# auto_process_emails('user', 'pass', model)