fix: login without selecting INBOX first so conn.list() returns all folders
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
import imaplib
|
||||
import email
|
||||
import logging
|
||||
import re
|
||||
from email.header import decode_header
|
||||
from email.utils import parsedate_to_datetime, parseaddr
|
||||
from typing import Optional
|
||||
@@ -89,7 +90,7 @@ def _select_mailbox(conn, mailbox: str = "INBOX"):
|
||||
raise RuntimeError(f"无法选择邮箱 {mailbox}: {data}")
|
||||
|
||||
|
||||
def _login_and_prepare(account: EmailAccount, timeout: int = 30):
|
||||
def _login_and_prepare(account: EmailAccount, timeout: int = 30, select_inbox: bool = True):
|
||||
logger.info("连接 %s:%d (超时%ds)", account.imap_server, account.imap_port, timeout)
|
||||
if account.use_ssl:
|
||||
conn = imaplib.IMAP4_SSL(account.imap_server, account.imap_port, timeout=timeout)
|
||||
@@ -102,13 +103,14 @@ def _login_and_prepare(account: EmailAccount, timeout: int = 30):
|
||||
if _check_provider(account.username, _PROVIDERS_NEED_ID):
|
||||
logger.info("发送 ID 命令 (126/163 兼容)")
|
||||
_send_id_command(conn)
|
||||
_select_mailbox(conn)
|
||||
logger.info("已选择 INBOX")
|
||||
if select_inbox:
|
||||
_select_mailbox(conn)
|
||||
logger.info("已选择 INBOX")
|
||||
return conn
|
||||
|
||||
|
||||
def fetch_unseen_emails(account: EmailAccount) -> list[Email]:
|
||||
conn = _login_and_prepare(account)
|
||||
conn = _login_and_prepare(account, select_inbox=False)
|
||||
|
||||
# 列出所有文件夹
|
||||
_, folder_data = conn.list()
|
||||
@@ -117,10 +119,12 @@ def fetch_unseen_emails(account: EmailAccount) -> list[Email]:
|
||||
for item in folder_data:
|
||||
if item is None:
|
||||
continue
|
||||
parts = item.decode("utf-8", errors="replace").split(' " ')
|
||||
if len(parts) >= 2:
|
||||
folder_name = parts[1].strip('"')
|
||||
folders.append(folder_name)
|
||||
line = item.decode("utf-8", errors="replace")
|
||||
# IMAP LIST 格式: (flags) "delimiter" "folder_name"
|
||||
# 用正则提取最后一个引号内的内容
|
||||
m = re.search(r'"([^"]*)"\s*$', line)
|
||||
if m:
|
||||
folders.append(m.group(1))
|
||||
if not folders:
|
||||
folders = ["INBOX"]
|
||||
logger.info("扫描文件夹: %s", ", ".join(folders))
|
||||
@@ -182,7 +186,7 @@ def mark_as_seen(account: EmailAccount, uids_with_folder: list[tuple[bytes, str]
|
||||
if not uids_with_folder:
|
||||
return
|
||||
logger.info("标记 %d 封邮件为已读", len(uids_with_folder))
|
||||
conn = _login_and_prepare(account)
|
||||
conn = _login_and_prepare(account, select_inbox=False)
|
||||
# 按文件夹分组
|
||||
from collections import defaultdict
|
||||
by_folder: dict[str, list[bytes]] = defaultdict(list)
|
||||
|
||||
Reference in New Issue
Block a user