Compare commits

..

2 Commits

View File

@@ -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)
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)