如何用Python的imap_tools无需重登持续获取最新邮件？

一、无需重新登录获取新邮件的修复方案 #

你的脚本无法获取新邮件的核心原因有两点：一是依赖「未读标记」的查询逻辑存在不确定性，二是IMAP服务器可能因连接闲置自动断开，导致后续查询使用失效连接。以下是针对性的修改方案：

关键优化点 #

• 跟踪邮件唯一UID：使用邮件的唯一标识（UID）替代未读状态作为查询依据，确保能精准获取所有新增邮件（UID在文件夹内是递增的）。
• 定期发送NOOP指令：向服务器发送空指令保持连接活跃，避免因闲置被断开。
• 异常处理后自动重连：确保连接失效时能重新建立并恢复状态。

修改后的脚本 #

from imap_tools import MailBox, AND
import time

# 配置信息
email_address = 'your_email@example.com'
password = 'yourpassword'
imap_server = 'imap.example.com'

def maintain_connection():
    """建立连接并获取当前最高邮件UID"""
    mailbox = MailBox(imap_server)
    mailbox.login(email_address, password, initial_folder='INBOX')
    # 获取初始最高UID，无邮件则设为0
    last_uid = max(msg.uid for msg in mailbox.fetch(limit=1, reverse=True)) if list(mailbox.fetch(limit=1)) else 0
    return mailbox, last_uid

def check_new_emails(mailbox, last_uid):
    """检查并处理新邮件，返回更新后的连接和UID"""
    try:
        # 发送NOOP保持连接活跃
        mailbox.noop()
        
        # 查询所有UID大于上次记录的邮件（即新增邮件）
        for msg in mailbox.fetch(AND(uid=f">{last_uid}")):
            print(f"新邮件：{msg.subject}")
            # 更新最新UID
            last_uid = msg.uid
            # 可选：标记邮件为已读
            # mailbox.flag(msg.uid, 'SEEN', True)
        
        return mailbox, last_uid
    except Exception as e:
        print(f"错误发生：{e}")
        print("尝试重新建立连接...")
        mailbox.logout()
        time.sleep(5)
        return maintain_connection()

# 初始化连接和UID
mailbox, last_uid = maintain_connection()

# 持续检查
while True:
    mailbox, last_uid = check_new_emails(mailbox, last_uid)
    print("等待下一次检查...")
    time.sleep(60)

二、更高效的实时邮件通知方案（IMAP IDLE） #

上述轮询方案仍存在延迟，更高效的方式是使用IMAP的IDLE命令：服务器会主动推送新邮件通知，无需定时查询，大幅减少资源消耗。

基于IDLE的脚本示例 #

from imap_tools import MailBox, AND
import time

# 配置信息
email_address = 'your_email@example.com'
password = 'yourpassword'
imap_server = 'imap.example.com'

def maintain_connection():
    mailbox = MailBox(imap_server)
    mailbox.login(email_address, password, initial_folder='INBOX')
    last_uid = max(msg.uid for msg in mailbox.fetch(limit=1, reverse=True)) if list(mailbox.fetch(limit=1)) else 0
    return mailbox, last_uid

def process_new_emails(mailbox, last_uid):
    """处理新增邮件并返回最新UID"""
    for msg in mailbox.fetch(AND(uid=f">{last_uid}")):
        print(f"实时收到新邮件：{msg.subject}")
        last_uid = msg.uid
        # 可选：标记为已读
        # mailbox.flag(msg.uid, 'SEEN', True)
    return last_uid

# 初始化
mailbox, last_uid = maintain_connection()

try:
    while True:
        print("进入IDLE模式，等待新邮件通知...")
        # 等待服务器通知，超时5分钟后发送NOOP保持连接
        mailbox.idle.wait(timeout=300)
        # 收到通知或超时后检查新邮件
        last_uid = process_new_emails(mailbox, last_uid)
        # 发送NOOP维持连接
        mailbox.noop()
except Exception as e:
    print(f"连接异常：{e}")
    mailbox.logout()
    time.sleep(5)
    # 可选：自动重启连接
    # mailbox, last_uid = maintain_connection()

方案优势 #

• 实时响应：新邮件到达时立即触发处理，无需等待轮询间隔。
• 低资源消耗：仅在有新邮件时执行查询，减少服务器和客户端的交互次数。
• 连接稳定性：通过超时后发送NOOP指令，避免连接被服务器闲置断开。

内容的提问来源于stack exchange，提问作者realsuspection