Python中如何实现邮件实时收取功能？

IMAP POP3 都没有推送，只能定时查询
Exchange ActiveSync 了解下

import imaplib
import email
from email.header import decode_header
import time

class EmailMonitor:
    def __init__(self, username, password, imap_server='imap.gmail.com'):
        self.username = username
        self.password = password
        self.imap_server = imap_server
        self.mail = None
        self.last_uid = 0  # 记录最后处理的邮件UID
        
    def connect(self):
        """建立IMAP连接"""
        self.mail = imaplib.IMAP4_SSL(self.imap_server)
        self.mail.login(self.username, self.password)
        self.mail.select('INBOX')
        
    def get_new_emails(self):
        """获取新邮件"""
        # 搜索所有未读邮件
        status, messages = self.mail.search(None, 'UNSEEN')
        if status != 'OK':
            return []
            
        email_ids = messages[0].split()
        new_emails = []
        
        for e_id in email_ids:
            # 获取邮件内容
            status, msg_data = self.mail.fetch(e_id, '(RFC822)')
            if status != 'OK':
                continue
                
            # 解析邮件
            raw_email = msg_data[0][1]
            msg = email.message_from_bytes(raw_email)
            
            # 解码邮件主题
            subject, encoding = decode_header(msg['Subject'])[0]
            if isinstance(subject, bytes):
                subject = subject.decode(encoding if encoding else 'utf-8')
                
            # 获取发件人
            from_ = msg.get('From')
            
            # 获取邮件正文
            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()
            
            new_emails.append({
                'id': e_id.decode(),
                'subject': subject,
                'from': from_,
                'body': body[:200]  # 只取前200字符
            })
            
        return new_emails
    
    def mark_as_read(self, email_id):
        """标记邮件为已读"""
        self.mail.store(email_id, '+FLAGS', '\\Seen')
    
    def monitor_loop(self, interval=30):
        """监控循环"""
        print(f"开始监控邮箱: {self.username}")
        print("按 Ctrl+C 停止监控\n")
        
        try:
            while True:
                try:
                    new_emails = self.get_new_emails()
                    
                    if new_emails:
                        print(f"发现 {len(new_emails)} 封新邮件:")
                        for email_data in new_emails:
                            print(f"\n主题: {email_data['subject']}")
                            print(f"发件人: {email_data['from']}")
                            print(f"内容预览: {email_data['body']}")
                            print("-" * 50)
                            
                            # 标记为已读
                            self.mark_as_read(email_data['id'])
                    
                    time.sleep(interval)
                    
                except Exception as e:
                    print(f"错误: {e}")
                    # 尝试重新连接
                    time.sleep(5)
                    self.connect()
                    
        except KeyboardInterrupt:
            print("\n停止监控")
        finally:
            self.disconnect()
    
    def disconnect(self):
        """断开连接"""
        if self.mail:
            self.mail.close()
            self.mail.logout()

# 使用示例
if __name__ == "__main__":
    # 配置你的邮箱信息
    monitor = EmailMonitor(
        username='your_email@gmail.com',
        password='your_app_password'  # 使用应用专用密码
    )
    
    # 连接邮箱
    monitor.connect()
    
    # 开始监控，每30秒检查一次
    monitor.monitor_loop(interval=30)

核心要点：