2

我正在尝试使用一个 python 脚本来不断监控 gmail 帐户中的新电子邮件。使用IMAPClient，它打开两个 imap 连接，一个处于空闲模式。每当收到新消息时，处于空闲模式的连接会告诉另一个连接它应该获取新邮件。非空闲连接将获取邮件，执行一些处理，然后存档电子邮件。

当我在短时间内收到许多电子邮件时，问题就出现了，一分钟内超过几封。在这种情况下，我得到一个 AssertionError。下面是重现错误的最小示例。除了 imap 连接之外，它还打开一个 smtp 连接，以便它可以将电子邮件发送给它自己。在发送 5-7 封电子邮件后，它通常会在某个时间点因 AssertionError 而失败。AssertionError 来自对idle_check.

关于运行代码的一些简短评论。它不使用 OAuth，因此必须将 gmail 设置为允许不太安全的应用程序。必须设置脚本底部的“用户名”和“密码”字段。该脚本还将归档当前在收件箱中的所有电子邮件，因此不应在主电子邮件帐户上运行。

#!/usr/bin/env python3

import smtplib
import imapclient
import email
import threading
import time

class Server(object):
    def __init__(self,username,password):
        self.username = username
        self.password = password

    def start(self):
        self.stop_running = threading.Event()
        self.has_mail = threading.Event()

        threading.Thread(target=self._idle).start()
        threading.Thread(target=self._poll).start()

        print('Listening for messages now')

    def _idle(self):
        imap_idle = self.imap_connect()

        while not self.stop_running.is_set():
            imap_idle.idle()

            for i in range(600):
                try:
                    if imap_idle.idle_check(1):
                        self.has_mail.set()
                except AssertionError as e:
                    self.stop_running.set()
                    raise

            imap_idle.idle_done()
            imap_idle.noop()

    def _poll(self):
        imap_poll = self.imap_connect()
        self.process_unread(imap_poll)

        while True:
            if self.has_mail.wait(1):
                self.has_mail.clear()
                self.process_unread(imap_poll)
            if self.stop_running.is_set():
                return

    def imap_connect(self):
        imap = imapclient.IMAPClient('imap.gmail.com',use_uid=True,ssl=True)
        imap.login(self.username,self.password)
        imap.select_folder('INBOX')
        return imap

    def process_unread(self, imap):
        imap.select_folder('INBOX')
        messages = imap.search()
        if messages:
            imap.copy(messages,'[Gmail]/All Mail')
            imap.delete_messages(messages)

    def smtp_connect(self):
        smtp = smtplib.SMTP('smtp.gmail.com',587)
        smtp.ehlo()
        smtp.starttls()
        smtp.ehlo()
        smtp.login(self.username,self.password)
        return smtp

    def send(self,recipient,subject='',body=''):
        headers = ['from: ' + self.username,
                   'subject: ' + subject,
                   'to: ' + recipient,
                   'mime-version: 1.0',
                   'content-type: text/html']
        headers = '\r\n'.join(headers)
        self.smtp_connect().sendmail(self.username,recipient,headers+'\r\n\r\n'+body)



def main():
    username = 'username@gmail.com'
    password = 'password'

    s = Server(username, password)
    s.start()

    for i in range(8):
        if s.stop_running.is_set():
            break

        print('Sending message',i)
        s.send(username,
               'Test Message'.format(i),
               'Body {}'.format(i))
        time.sleep(15)


if __name__=='__main__':
    main()

给出的错误信息如下。错误时的text变量有时是b'XISTS'有时b' FLAGS (\\Seen))'。

Exception in thread Thread-1:                                                                                                                                                          
Traceback (most recent call last):                                                                                                                                                     
  File "/usr/lib/python3.4/threading.py", line 920, in _bootstrap_inner                                                                                                                
    self.run()                                                                                                                                                                         
  File "/usr/lib/python3.4/threading.py", line 868, in run                                                                                                                             
    self._target(*self._args, **self._kwargs)                                                                                                                                          
  File "./emailer.py", line 31, in _idle                                                                                                                                               
    if imap_idle.idle_check(1):                                                                                                                                                        
  File "/path/to/the/venv/lib/python3.4/site-packages/imapclient/imapclient.py", line 519, in idle_check                                                                               
    resps.append(_parse_untagged_response(line))                                                                                                                                       
  File "/path/to/the/venv/lib/python3.4/site-packages/imapclient/imapclient.py", line 1093, in _parse_untagged_response                                                                
    assert text.startswith(b'* ')                                                                                                                                                      
AssertionError

我使用 IMAPClient 0.13 使用 Python 3.4.0 运行它。这是在一个干净的 virtualenv 中运行的，没有安装其他库。任何帮助将不胜感激。

4

0 回答 0