Compare commits

...

2 Commits

View File

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