Compare commits
2 Commits
479b855390
...
ae140df3fb
| Author | SHA1 | Date | |
|---|---|---|---|
| ae140df3fb | |||
| e9398526e6 |
@@ -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)
|
||||||
|
|||||||
Reference in New Issue
Block a user