Python查询Gmail收件箱邮件报错及结果限制问题

问题描述 #

使用以下Python代码实现Gmail的发送与搜索功能：

import smtplib
import imaplib
import email
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import pandas as pd


class EmailClient:
    def __init__(self, email, password, smtp_server='smtp.gmail.com', smtp_port=587, imap_server="imap.gmail.com"):
        self.email = email
        self.password = password
        self.smtp_server = smtplib.SMTP(smtp_server, smtp_port)
        self.imap_server = imaplib.IMAP4_SSL(imap_server)

    def send_email(self, to_addr, subject, body):
        msg = MIMEMultipart()
        msg['From'] = self.email
        msg['To'] = to_addr
        msg['Subject'] = subject
        msg.attach(MIMEText(body, 'plain'))

        self.smtp_server.starttls()
        self.smtp_server.login(self.email, self.password)
        self.smtp_server.send_message(msg)
        self.smtp_server.quit()

    def search_email(self, mailbox="INBOX", subject=None, to=None, from_=None, since_date=None, until_date=None,
                     since_emailid=None):

        self.imap_server.login(self.email, self.password)

        self.imap_server.select(mailbox)

        query_parts = []
        if subject is not None:
            query_parts.append(f'(SUBJECT "{subject}")')
        if to is not None:
            query_parts.append(f'(TO "{to}")')
        if from_ is not None:
            query_parts.append(f'(FROM "{from_}")')
        if since_date is not None:
            since_date_str = since_date.strftime("%d-%b-%Y")
            query_parts.append(f'(SINCE "{since_date_str}")')
        if until_date is not None:
            until_date_str = until_date.strftime("%d-%b-%Y")
            query_parts.append(f'(BEFORE "{until_date_str}")')
        if since_emailid is not None:
            query_parts.append(f'(UID {since_emailid}:*)')

        query = ' '.join(query_parts)

        ret = []

        resp, items = self.imap_server.uid('search', None, query)
        items = items[0].split()
        for emailid in items[::-1]:
            resp, data = self.imap_server.uid('fetch', emailid, "(BODY[HEADER.FIELDS (SUBJECT TO FROM DATE)])")
            try:
                raw_email = data[0][1].decode("utf-8")
            except UnicodeDecodeError:
                ValueError(f"Could not decode email with id {emailid}")
            email_message = email.message_from_string(raw_email)

            email_line = {}
            email_line['id'] = emailid
            email_line["to"] = email_message['To']
            email_line["from"] = email_message['From']
            email_line["subject"] = str(email_message['Subject'])
            email_line["created_at"] = email_message['Date']
            resp, email_data = self.imap_server.uid('fetch', emailid, '(BODY[TEXT])')
            email_line["body"] = email_data[0][1].decode('utf-8')

            ret.append(email_line)

        self.imap_server.logout()

        return pd.DataFrame(ret)

执行以下调用时：

email_client = EmailClient('<email>',  '<app_password>')
email_client.search_email()

期望返回收件箱所有邮件，却触发错误：

imaplib.IMAP4.error: UID command error: BAD [b'Could not parse command']

同时需要实现类似SQL LIMIT的功能，限制返回结果数量。

错误原因及修复 #

问题出在空查询字符串：当调用search_email()时没有传入任何筛选参数，query_parts是空列表，query就变成了空字符串。IMAP的UID SEARCH命令不接受空查询，因此触发解析错误。

修复方法：给查询设置默认值，当没有任何筛选条件时，传入ALL作为查询参数，表示匹配所有邮件。修改search_email方法中的查询拼接部分：

query = ' '.join(query_parts) if query_parts else 'ALL'

这样当没有筛选条件时，IMAP会执行UID SEARCH ALL，正确返回所有邮件。

另外还有两个潜在问题需要注意：

• send_email方法中，每次发送邮件都会重新连接SMTP服务器并退出，建议把starttls()和login()移到__init__方法中，或者添加连接复用逻辑，避免重复建立连接的开销。
• 解析邮件内容时的UnicodeDecodeError处理不完整，当前只是抛出ValueError但没有捕获，会导致程序中断，建议添加try-except块捕获并记录错误，跳过无法解码的邮件。

限制返回结果数量（类似SQL LIMIT） #

IMAP协议本身没有直接的LIMIT语法，但可以通过以下两种方式实现：

方法1：截取搜索结果 #

在获取到所有匹配的邮件UID后，只取前N个（或后N个）进行处理。修改search_email方法，添加limit参数：

def search_email(self, mailbox="INBOX", subject=None, to=None, from_=None, since_date=None, until_date=None,
                 since_emailid=None, limit=None):
    # ... 原有代码 ...

    resp, items = self.imap_server.uid('search', None, query)
    items = items[0].split()
    
    # 应用limit限制
    if limit is not None and len(items) > limit:
        # 按时间倒序取最新的limit条，items默认是升序，反转后取前limit个
        items = items[::-1][:limit][::-1]  # 保持UID升序，或直接取反转后的前limit个根据需求调整
    
    for emailid in items[::-1]:
        # ... 原有处理逻辑 ...

调用时传入limit参数即可：

email_client.search_email(limit=10)  # 返回最新10封邮件

方法2：结合IMAP的SORT和FETCH命令（更高效） #

如果只需要最新的N封邮件，可以先按时间排序，再取前N个，避免获取所有UID后再截取，适合邮件数量较多的场景：

def search_email(self, mailbox="INBOX", subject=None, to=None, from_=None, since_date=None, until_date=None,
                 since_emailid=None, limit=None):
    self.imap_server.login(self.email, self.password)
    self.imap_server.select(mailbox)

    query_parts = []
    # ... 原有查询拼接逻辑 ...
    query = ' '.join(query_parts) if query_parts else 'ALL'

    ret = []
    if limit is not None:
        # 按日期降序排序，取前limit个UID
        resp, items = self.imap_server.uid('sort', '(REVERSE DATE)', 'UTF-8', query)
        items = items[0].split()[:limit]
    else:
        resp, items = self.imap_server.uid('search', None, query)
        items = items[0].split()
    
    # ... 后续邮件处理逻辑 ...

这种方式直接在IMAP服务器端完成排序和截取，减少数据传输量，效率更高。

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