定时检查新邮件并分类脚本

#!/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 可安全退出。