#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
定时检查新邮件并自动分类脚本
支持 IMAP 邮箱，根据发件人/主题关键词移动邮件到指定文件夹
"""
import imaplib
import email
from email.header import decode_header
import time
import logging
import os
import sys
from datetime import datetime, timedelta
from typing import List, Dict, Tuple, Optional
# ================== 配置区 ==================
# 邮箱配置（请修改为您的邮箱信息）
EMAIL_CONFIG = {
"imap_server": "imap.gmail.com", # imap.gmail.com / imap.qq.com
"port": 993,
"use_ssl": True,
"username": "your_email@gmail.com", # 邮箱地址
"password": "your_app_password" # 建议使用应用专用密码
}
# 检查间隔（秒）
CHECK_INTERVAL = 300 # 5分钟
# 分类规则：规则列表，按顺序匹配，匹配第一个则移动
# folder: 目标文件夹（IMAP 文件夹名，如 INBOX.分类名 或 直接使用中文）
# 如果目标文件夹不存在，脚本会自动创建
CLASSIFY_RULES = [
{
"name": "工作邮件",
"folder": "工作",
"conditions": {
"from_keywords": ["@company.com", "boss@", "hr@"], # 发件人包含这些词
"subject_keywords": ["会议", "报告", "项目", "deadline", "urgent"]
}
},
{
"name": "订阅/通知",
"folder": "订阅",
"conditions": {
"from_keywords": ["noreply@", "notification@", "weekly@", "newsletter@"],
"subject_keywords": ["订阅", "周报", "月报", "newsletter", "unsubscribe"]
}
},
{
"name": "个人邮件",
"folder": "个人",
"conditions": {
"from_keywords": ["@gmail.com", "@qq.com", "friend@", "family@"],
"subject_keywords": ["聚会", "旅行", "照片", "hello"]
}
},
# 可以添加更多规则
]
# 日志配置
LOG_FILE = "email_classifier.log"
# ==========================================
# 全局变量
logger = logging.getLogger(__name__)
processed_uid_set = set() # 已处理邮件的 UID 集合，防止重复处理
def setup_logging():
"""配置日志输出到文件和控制台"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(LOG_FILE, encoding='utf-8'),
logging.StreamHandler(sys.stdout)
]
)
def decode_mime_string(s: str) -> str:
"""解码 MIME 编码的字符串（如 =?UTF-8?B?xxx?=）"""
if not s:
return ""
try:
decoded_parts = decode_header(s)
result = []
for part, charset in decoded_parts:
if isinstance(part, bytes):
if charset:
result.append(part.decode(charset, errors='replace'))
else:
result.append(part.decode('utf-8', errors='replace'))
else:
result.append(str(part))
return ''.join(result)
except Exception:
return str(s)
def get_text_from_email(msg) -> str:
"""从邮件对象中提取纯文本内容（用于调试/日志）"""
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain":
try:
return part.get_payload(decode=True).decode('utf-8', errors='replace')
except:
continue
else:
try:
return msg.get_payload(decode=True).decode('utf-8', errors='replace')
except:
return ""
return ""
def check_conditions(msg, rules: dict) -> bool:
"""
检查邮件是否匹配规则
规则说明：
- from_keywords: 发件人地址包含任一关键词（大小写不敏感）
- subject_keywords: 主题包含任一关键词（大小写不敏感）
当同一类条件（如发件人）有多词时，满足任一即可；
不同条件之间是 AND 关系（发件人 AND 主题）。
"""
# 获取发件人
from_addr = ""
from_header = msg.get("From", "")
if from_header:
from_addr = decode_mime_string(from_header).lower()
# 获取主题
subject = ""
subject_header = msg.get("Subject", "")
if subject_header:
subject = decode_mime_string(subject_header).lower()
conditions = rules.get("conditions", {})
from_kw = conditions.get("from_keywords", [])
subject_kw = conditions.get("subject_keywords", [])
# 如果没有任何条件，则不匹配
if not from_kw and not subject_kw:
return False
# 检查发件人关键词
from_match = True
if from_kw:
from_match = any(kw.lower() in from_addr for kw in from_kw)
# 检查主题关键词
subject_match = True
if subject_kw:
subject_match = any(kw.lower() in subject for kw in subject_kw)
# 两种条件同时满足（AND）
return from_match and subject_match
def classify_and_move(mail: imaplib.IMAP4_SSL, email_uid: bytes) -> Optional[str]:
"""
对单封邮件执行分类并移动
返回移动到的文件夹名称，如果未分类返回 None
"""
# 获取邮件数据
typ, data = mail.uid('fetch', email_uid, '(BODY.PEEK[HEADER])')
if typ != 'OK':
logger.warning(f"无法获取邮件 UID {email_uid} 的头部")
return None
msg = email.message_from_bytes(data[0][1])
subject = decode_mime_string(msg.get("Subject", ""))
from_addr = decode_mime_string(msg.get("From", ""))
logger.info(f"检查邮件: 来自 [{from_addr}] 主题 [{subject}]")
# 遍历规则，匹配第一个即停止
for rule in CLASSIFY_RULES:
if check_conditions(msg, rule):
target_folder = rule["folder"]
logger.info(f" -> 归类为 [{rule['name']}]，移动到文件夹 [{target_folder}]")
# 创建目标文件夹（如果不存在）
try:
# 尝试选择目标文件夹（判断是否存在）
status, _ = mail.select(f'"{target_folder}"')
if status != 'OK':
# 创建文件夹
status, _ = mail.create(f'"{target_folder}"')
if status == 'OK':
logger.info(f" 创建目标文件夹 [{target_folder}] 成功")
else:
logger.error(f" 创建目标文件夹 [{target_folder}] 失败")
return None
except Exception as e:
logger.error(f" 操作文件夹时出错: {e}")
return None
# 移动邮件：复制到目标文件夹，然后删除原邮件
try:
# 复制到目标文件夹
typ, _ = mail.uid('copy', email_uid, target_folder)
if typ != 'OK':
logger.error(f" 复制邮件到 [{target_folder}] 失败")
return None
# 标记原邮件为已删除
typ, _ = mail.uid('store', email_uid, '+FLAGS', '\\Deleted')
if typ != 'OK':
logger.warning(f" 标记删除原邮件失败，可能已移动")
else:
# 立即清除（expunge）当前文件夹中的已删除邮件
# 注意：expunge 会影响所有标记删除的邮件，而非仅此一封
# 但我们后续会调用 expunge，这里可以省略
pass
logger.info(f" 成功移动到 [{target_folder}]")
return target_folder
except Exception as e:
logger.error(f" 移动邮件时出错: {e}")
return None
logger.info(f" -> 未匹配任何规则，留在收件箱")
return None
def process_new_emails(mail: imaplib.IMAP4_SSL):
"""检查并处理新邮件"""
global processed_uid_set
try:
# 选择收件箱（只读模式先检查，但移动需要写入，所以直接选择可写）
status, messages = mail.select('INBOX', readonly=False)
if status != 'OK':
logger.error("无法选择 INBOX")
return
# 搜索今天收到的邮件（更高效，也可改为 'UNSEEN' 但会标记已读）
# 使用 SINCE 获取最近一天的邮件，避免无限扫描旧邮件
today = datetime.now().strftime("%d-%b-%Y")
typ, data = mail.uid('search', None, f'(SINCE {today})')
if typ != 'OK':
logger.error("搜索邮件失败")
return
uids = data[0].split()
if not uids:
logger.info("没有新邮件")
return
# 去重：跳过已处理的 UID
new_uids = [uid for uid in uids if uid not in processed_uid_set]
if not new_uids:
logger.info(f"所有邮件均已处理过（收件箱共 {len(uids)} 封，已处理 {len(processed_uid_set)} 封）")
return
logger.info(f"发现 {len(new_uids)} 封待处理邮件")
# 处理每一封新邮件
moved_count = 0
for uid in new_uids:
result = classify_and_move(mail, uid)
if result:
moved_count += 1
# 记录已处理（无论是否移动）
processed_uid_set.add(uid)
# 清理已删除的邮件（expunge），必须在同一个 select 后执行
if moved_count > 0:
mail.expunge()
logger.info(f"已完成清理，共移动 {moved_count} 封邮件")
# 限制 processed_uid_set 大小，避免内存无限增长（保留最近 5000 个）
if len(processed_uid_set) > 5000:
# 简单删除较早的一半
old_uids = list(processed_uid_set)[:2500]
for uid in old_uids:
processed_uid_set.discard(uid)
logger.info(f"清理 UID 缓存，当前大小 {len(processed_uid_set)}")
except Exception as e:
logger.error(f"处理邮件时发生异常: {e}", exc_info=True)
def main_loop():
"""主循环：定时连接并检查"""
logger.info("=== 邮件分类脚本启动 ===")
logger.info(f"IMAP 服务器: {EMAIL_CONFIG['imap_server']}")
logger.info(f"检查间隔: {CHECK_INTERVAL} 秒")
logger.info(f"分类规则数: {len(CLASSIFY_RULES)}")
logger.info("等待首次执行...")
# 连接邮箱
mail = None
while True:
try:
if mail is None:
logger.info("正在连接邮箱...")
if EMAIL_CONFIG["use_ssl"]:
mail = imaplib.IMAP4_SSL(EMAIL_CONFIG["imap_server"], EMAIL_CONFIG["port"])
else:
mail = imaplib.IMAP4(EMAIL_CONFIG["imap_server"], EMAIL_CONFIG["port"])
mail.login(EMAIL_CONFIG["username"], EMAIL_CONFIG["password"])
logger.info("邮箱连接成功")
# 执行一次检查
process_new_emails(mail)
# 等待下一次执行
logger.info(f"等待 {CHECK_INTERVAL} 秒后再次检查...")
time.sleep(CHECK_INTERVAL)
except imaplib.IMAP4.abort as e:
logger.error(f"IMAP 连接中断: {e}，将尝试重新连接...")
mail = None
time.sleep(30) # 等待30秒后重连
except KeyboardInterrupt:
logger.info("用户中断，退出程序")
if mail:
try:
mail.logout()
except:
pass
sys.exit(0)
except Exception as e:
logger.error(f"主循环异常: {e}", exc_info=True)
logger.info("等待 60 秒后重试...")
time.sleep(60)
if __name__ == "__main__":
setup_logging()
main_loop()
使用说明
安装依赖
脚本仅使用 Python 标准库（imaplib、email 等），无需额外安装，确保 Python 版本 ≥ 3.6 即可。
修改配置
- 邮箱配置：修改
EMAIL_CONFIG中的服务器地址、端口、用户名和密码（建议使用应用专用密码）。 - 检查间隔：调整
CHECK_INTERVAL变量（单位：秒），默认 300 秒（5分钟）。 - 分类规则：编辑
CLASSIFY_RULES列表，每个规则包含：name：规则描述（仅用于日志）folder：目标文件夹（IMAP 文件夹名，支持中文）conditions：匹配条件，支持from_keywords（发件人关键词）和subject_keywords（主题关键词）
运行脚本
python3 email_classifier.py
脚本会在后台持续运行,每 5 分钟检查一次收件箱，日志将输出到控制台和 email_classifier.log 文件。
注意事项
- 密码安全：建议为邮箱开启 IMAP 并生成“应用专用密码”，避免使用主密码。
- 文件夹自动创建：如果目标文件夹不存在，脚本会自动创建。
- 已处理缓存：脚本会记录已处理的邮件 UID，避免重复移动，重启脚本会清空缓存，但已移动的邮件不会重复处理（因为已不在收件箱）。
- 停止脚本：按
Ctrl+C可安全退出。
