ESP8266上MicroPython通过IMAP服务器获取未读邮件异常问题

在ESP8266上使用MicroPython 1.22.2（带ussl的最新版本）连接Gmail IMAP服务器时，能成功连接并登录，但无法检测到未读邮件，邮件获取功能不符合预期。已确认服务器连接正常，测试过不同缓冲区大小，原代码如下：

import network
import usocket as socket
import ussl as ssl
import utime as time

SSID = "wifi"
WIFI_PASSWORD = "pass"

EMAIL = "ex@gmail.com"
PASSWORD = "APP password"

IMAP_SERVER = "imap.gmail.com"
IMAP_PORT = 993 

# Connect to Wi-Fi
def connect_wifi(ssid, password, timeout=30):
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    wlan.connect(ssid, password)
    start_time = time.time()
    while not wlan.isconnected():
        if time.time() - start_time > timeout:
            print("Failed to connect to Wi-Fi.")
            return False
        time.sleep(1)
    print("Connected to Wi-Fi:", wlan.ifconfig())
    return True

def connect_to_imap():
    try:
        addr = socket.getaddrinfo(IMAP_SERVER, IMAP_PORT)[0][-1]
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(10)
        s.connect(addr)
        s = ssl.wrap_socket(s)
        print("IMAP connected and SSL wrapped.")
        return s
    except Exception as e:
        print(f"IMAP connection error: {e}")
        return None

def send_command(s, command):
    print(f"Sending command: {command.decode()}")
    s.write(command + b'\r\n')
    response = b""
    while True:
        try:
            part = s.read(1024)
            if not part:
                break
            response += part
            if b'OK' in part or b'NO' in part or b'BAD' in part:
                break
        except OSError as e:
            print(f"Socket read error: {e}")
            break
    response_str = response.decode()
    print(f"Response: {response_str}")
    return response_str

def login_imap(s):
    login_cmd = f'A1 LOGIN "{EMAIL}" "{PASSWORD}"'.encode()
    response = send_command(s, login_cmd)
    return 'OK' in response

def fetch_unread_emails(s):
    print("Selecting INBOX...")
    send_command(s, b'A2 SELECT INBOX')
    
    print("Searching for unseen emails...")
    search_cmd = b'A3 SEARCH UNSEEN'
    search_response = send_command(s, search_cmd)
    
    parts = search_response.split()
    email_ids = [part for part in parts if part.isdigit()]
    
    if email_ids:
        print(f"Found email IDs: {email_ids}")
        for email_id in email_ids:
            fetch_cmd = f'A4 FETCH {email_id} BODY[TEXT]'.encode()
            email_response = send_command(s, fetch_cmd)
            print(f"Email ID {email_id} content: {email_response}")
    else:
        print("No new emails.")

def main():
    if connect_wifi(SSID, WIFI_PASSWORD):
        while True:
            s = connect_to_imap()
            if s:
                if login_imap(s):
                    fetch_unread_emails(s)
                s.close()
            time.sleep(15)

main()

问题排查与修复方案 #

原代码存在以下核心问题导致无法检测未读邮件：

1. 响应读取逻辑不完整：send_command函数仅读取到包含OK/NO/BAD的分段就停止，而IMAP SEARCH命令的响应可能分多段返回，前半段不含状态关键字，导致漏掉邮件ID。
2. 缺少命令执行校验：SELECT INBOX命令未校验执行结果，若会话异常后续SEARCH会无效。
3. 邮件ID解析逻辑粗糙：仅通过split提取数字，若响应不完整会无法正确识别邮件ID。

修复后的完整代码：

import network
import usocket as socket
import ussl as ssl
import utime as time

SSID = "wifi"
WIFI_PASSWORD = "pass"

EMAIL = "ex@gmail.com"
PASSWORD = "APP password"

IMAP_SERVER = "imap.gmail.com"
IMAP_PORT = 993 

def connect_wifi(ssid, password, timeout=30):
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    wlan.connect(ssid, password)
    start_time = time.time()
    while not wlan.isconnected():
        if time.time() - start_time > timeout:
            print("Failed to connect to Wi-Fi.")
            return False
        time.sleep(1)
    print("Connected to Wi-Fi:", wlan.ifconfig())
    return True

def connect_to_imap():
    try:
        addr = socket.getaddrinfo(IMAP_SERVER, IMAP_PORT)[0][-1]
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(10)
        s.connect(addr)
        s = ssl.wrap_socket(s)
        print("IMAP connected and SSL wrapped.")
        # 读取服务器初始问候，确认会话正常
        greeting = s.read(1024).decode()
        print(f"Server greeting: {greeting}")
        return s
    except Exception as e:
        print(f"IMAP connection error: {e}")
        return None

def send_command(s, tag, command):
    full_cmd = f"{tag} {command}".encode()
    print(f"Sending command: {full_cmd.decode()}")
    s.write(full_cmd + b'\r\n')
    response = b""
    while True:
        try:
            part = s.read(1024)
            if not part:
                break
            response += part
            # 等待包含当前命令标记的状态行，确保获取完整响应
            if f"{tag} OK".encode() in response or f"{tag} NO".encode() in response or f"{tag} BAD".encode() in response:
                break
        except OSError as e:
            print(f"Socket read error: {e}")
            break
    response_str = response.decode()
    print(f"Response: {response_str}")
    return response_str

def login_imap(s):
    tag = "A1"
    login_cmd = f'LOGIN "{EMAIL}" "{PASSWORD}"'
    response = send_command(s, tag, login_cmd)
    return f"{tag} OK" in response

def fetch_unread_emails(s):
    tag_counter = 2
    # 选择收件箱并校验结果
    select_tag = f"A{tag_counter}"
    tag_counter +=1
    select_response = send_command(s, select_tag, "SELECT INBOX")
    if f"{select_tag} OK" not in select_response:
        print("Failed to select INBOX")
        return
    
    # 搜索未读邮件
    search_tag = f"A{tag_counter}"
    tag_counter +=1
    search_response = send_command(s, search_tag, "SEARCH UNSEEN")
    
    # 精准提取邮件ID
    email_ids = []
    if f"{search_tag} OK" in search_response:
        # 定位SEARCH结果的有效区间
        result_part = search_response.split(f"{search_tag} SEARCH")[1].split(f"{search_tag} OK")[0].strip()
        email_ids = [part for part in result_part.split() if part.isdigit()]
    
    if email_ids:
        print(f"Found email IDs: {email_ids}")
        for email_id in email_ids:
            fetch_tag = f"A{tag_counter}"
            tag_counter +=1
            fetch_cmd = f'FETCH {email_id} BODY[TEXT]'
            email_response = send_command(s, fetch_tag, fetch_cmd)
            print(f"Email ID {email_id} content: {email_response}")
    else:
        print("No new emails.")

def main():
    if connect_wifi(SSID, WIFI_PASSWORD):
        while True:
            s = connect_to_imap()
            if s:
                if login_imap(s):
                    fetch_unread_emails(s)
                s.close()
            time.sleep(15)

main()

关键修改说明 #

• 完善响应读取：等待包含当前命令标记的状态行，确保获取完整的IMAP响应，避免漏掉SEARCH返回的邮件ID。
• 增加命令校验：执行SELECT INBOX后校验结果，确保后续操作在有效会话中进行。
• 精准解析邮件ID：定位SEARCH响应的有效结果区间，避免无效字符干扰ID提取。
• 动态生成命令标记：使用递增标记符合IMAP协议规范，避免潜在冲突。

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