Compare commits
3 Commits
87e4b0b01e
...
479b855390
| Author | SHA1 | Date | |
|---|---|---|---|
| 479b855390 | |||
| 46bc7c7057 | |||
| bbc43f8972 |
11
main.py
11
main.py
@@ -87,7 +87,7 @@ def _ai_processor(cfg):
|
||||
_tg_queue.put(info)
|
||||
except Exception as e:
|
||||
logger.error(f"AI 处理邮件失败: {e}", exc_info=True)
|
||||
finally:
|
||||
# AI 失败也要移除 pending,否则永远卡住
|
||||
with _pending_lock:
|
||||
_pending_uids.discard(mail.uid)
|
||||
|
||||
@@ -107,10 +107,19 @@ def _tg_worker(cfg):
|
||||
try:
|
||||
info = _tg_queue.get_nowait()
|
||||
tg_send_and_mark(cfg, info)
|
||||
# 发送并标记已读成功后才移除 pending
|
||||
with _pending_lock:
|
||||
_pending_uids.discard(info["uid"])
|
||||
except queue_module.Empty:
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"TG 发送失败: {e}", exc_info=True)
|
||||
# 发送失败也要移除 pending,避免永久阻塞
|
||||
try:
|
||||
with _pending_lock:
|
||||
_pending_uids.discard(info["uid"])
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if _running:
|
||||
time.sleep(1)
|
||||
|
||||
@@ -8,7 +8,7 @@ from src.retry import retry
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以 JSON 格式返回结构化摘要。
|
||||
SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以 JSON 格式返回结构化摘要。所有输出(subject、summary 等)必须使用简体中文,但 reply_suggestions(智能回复)请使用与邮件相同的语言,以便用户直接回复。
|
||||
|
||||
返回格式必须严格遵循以下 JSON schema:
|
||||
{
|
||||
@@ -62,7 +62,7 @@ def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str,
|
||||
|
||||
|
||||
PROMO_DETAIL_PROMPT = """你是一个邮件分析助手。用户想了解更多关于这封推广邮件的详情。
|
||||
请根据邮件原文,输出 3-5 条关键信息,每条一句话,简洁明了。包括:优惠力度、适用条件、截止时间、核心卖点等用户做决策需要知道的信息。
|
||||
请根据邮件原文,输出 3-5 条关键信息,每条一句话,简洁明了。包括:优惠力度、适用条件、截止时间、核心卖点等用户做决策需要知道的信息。所有输出使用简体中文。
|
||||
只返回一个 JSON: {"details": ["信息1", "信息2", ...]}
|
||||
只返回 JSON,不要包含任何其他文字。"""
|
||||
|
||||
|
||||
@@ -12,7 +12,8 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
class Email:
|
||||
def __init__(self, uid: bytes, subject: str, sender: str, recipient: str,
|
||||
body: str, date: str, reply_to: str = "", account_email: str = ""):
|
||||
body: str, date: str, reply_to: str = "", account_email: str = "",
|
||||
folder: str = "INBOX"):
|
||||
self.uid = uid
|
||||
self.subject = subject
|
||||
self.sender = sender
|
||||
@@ -21,6 +22,7 @@ class Email:
|
||||
self.date = date
|
||||
self.reply_to = reply_to
|
||||
self.account_email = account_email
|
||||
self.folder = folder
|
||||
|
||||
|
||||
def _decode_str(s: str) -> str:
|
||||
@@ -108,36 +110,63 @@ def _login_and_prepare(account: EmailAccount, timeout: int = 30):
|
||||
def fetch_unseen_emails(account: EmailAccount) -> list[Email]:
|
||||
conn = _login_and_prepare(account)
|
||||
|
||||
_, data = conn.uid("SEARCH", None, "UNSEEN")
|
||||
uids = data[0].split() if data[0] else []
|
||||
logger.info("UNSEEN 数量: %d", len(uids))
|
||||
emails = []
|
||||
|
||||
for uid in uids:
|
||||
try:
|
||||
_, msg_data = conn.uid("FETCH", uid, "RFC822")
|
||||
if msg_data[0] is None:
|
||||
logger.warning("UID %s FETCH 返回空,跳过", uid)
|
||||
# 列出所有文件夹
|
||||
_, folder_data = conn.list()
|
||||
folders = []
|
||||
if folder_data:
|
||||
for item in folder_data:
|
||||
if item is None:
|
||||
continue
|
||||
raw_email = msg_data[0][1]
|
||||
msg = email.message_from_bytes(raw_email)
|
||||
parts = item.decode("utf-8", errors="replace").split(' " ')
|
||||
if len(parts) >= 2:
|
||||
folder_name = parts[1].strip('"')
|
||||
folders.append(folder_name)
|
||||
if not folders:
|
||||
folders = ["INBOX"]
|
||||
logger.info("扫描文件夹: %s", ", ".join(folders))
|
||||
|
||||
subject = _decode_str(msg.get("Subject", ""))
|
||||
sender = _decode_address_header(msg.get("From", ""))
|
||||
recipient = _decode_address_header(msg.get("To", ""))
|
||||
reply_to = _decode_address_header(msg.get("Reply-To", ""))
|
||||
date_str = msg.get("Date", "")
|
||||
body = _get_text_from_msg(msg)
|
||||
body_len = len(body)
|
||||
emails = []
|
||||
for folder in folders:
|
||||
try:
|
||||
status, _ = conn.select(folder, readonly=True)
|
||||
if status != "OK":
|
||||
logger.warning("无法选择文件夹: %s", folder)
|
||||
continue
|
||||
_, data = conn.uid("SEARCH", None, "UNSEEN")
|
||||
uids = data[0].split() if data[0] else []
|
||||
if not uids:
|
||||
continue
|
||||
logger.info(" %s: %d 封未读", folder, len(uids))
|
||||
|
||||
logger.info(" 邮件: [%s] from=%s to=%s reply-to=%s (%d 字符)", subject, sender, recipient, reply_to, body_len)
|
||||
emails.append(Email(uid=uid, subject=subject, sender=sender, recipient=recipient,
|
||||
body=body, date=date_str, reply_to=reply_to, account_email=account.username))
|
||||
for uid in uids:
|
||||
try:
|
||||
_, msg_data = conn.uid("FETCH", uid, "RFC822")
|
||||
if msg_data[0] is None:
|
||||
logger.warning("UID %s FETCH 返回空,跳过", uid)
|
||||
continue
|
||||
raw_email = msg_data[0][1]
|
||||
msg = email.message_from_bytes(raw_email)
|
||||
|
||||
# FETCH 可能被服务器自动标已读,立即标回未读
|
||||
conn.uid("STORE", uid, "-FLAGS", "\\Seen")
|
||||
subject = _decode_str(msg.get("Subject", ""))
|
||||
sender = _decode_address_header(msg.get("From", ""))
|
||||
recipient = _decode_address_header(msg.get("To", ""))
|
||||
reply_to = _decode_address_header(msg.get("Reply-To", ""))
|
||||
date_str = msg.get("Date", "")
|
||||
body = _get_text_from_msg(msg)
|
||||
body_len = len(body)
|
||||
|
||||
logger.info(" 邮件: [%s] from=%s to=%s reply-to=%s (%d 字符)", subject, sender, recipient, reply_to, body_len)
|
||||
emails.append(Email(uid=uid, subject=subject, sender=sender, recipient=recipient,
|
||||
body=body, date=date_str, reply_to=reply_to, account_email=account.username,
|
||||
folder=folder))
|
||||
|
||||
# FETCH 可能被服务器自动标已读,立即标回未读
|
||||
conn.uid("STORE", uid, "-FLAGS", "\\Seen")
|
||||
except Exception as e:
|
||||
logger.error(" UID %s FETCH 失败,跳过: %s", uid, e)
|
||||
continue
|
||||
except Exception as e:
|
||||
logger.error("UID %s FETCH 失败,跳过: %s", uid, e)
|
||||
logger.warning("扫描文件夹 %s 失败: %s", folder, e)
|
||||
continue
|
||||
|
||||
try:
|
||||
@@ -149,12 +178,23 @@ def fetch_unseen_emails(account: EmailAccount) -> list[Email]:
|
||||
|
||||
|
||||
@retry()
|
||||
def mark_as_seen(account: EmailAccount, uids: list[bytes]):
|
||||
if not uids:
|
||||
def mark_as_seen(account: EmailAccount, uids_with_folder: list[tuple[bytes, str]]):
|
||||
if not uids_with_folder:
|
||||
return
|
||||
logger.info("标记 %d 封邮件为已读", len(uids))
|
||||
logger.info("标记 %d 封邮件为已读", len(uids_with_folder))
|
||||
conn = _login_and_prepare(account)
|
||||
for uid in uids:
|
||||
conn.uid("STORE", uid, "+FLAGS", "\\Seen")
|
||||
# 按文件夹分组
|
||||
from collections import defaultdict
|
||||
by_folder: dict[str, list[bytes]] = defaultdict(list)
|
||||
for uid, folder in uids_with_folder:
|
||||
by_folder[folder].append(uid)
|
||||
for folder, uids in by_folder.items():
|
||||
try:
|
||||
conn.select(folder, readonly=True)
|
||||
for uid in uids:
|
||||
conn.uid("STORE", uid, "+FLAGS", "\\Seen")
|
||||
logger.info(" %s: 标记 %d 封", folder, len(uids))
|
||||
except Exception as e:
|
||||
logger.error(" 标记 %s 失败: %s", folder, e)
|
||||
conn.logout()
|
||||
logger.info("标记完成")
|
||||
|
||||
@@ -42,6 +42,7 @@ def ai_process(cfg: Config, acct_idx: int, mail: Email) -> Optional[dict]:
|
||||
"acct_idx": acct_idx,
|
||||
"account_email": mail.account_email,
|
||||
"uid": mail.uid,
|
||||
"folder": mail.folder,
|
||||
"thinking_msg_id": thinking_msg_id,
|
||||
}
|
||||
|
||||
@@ -81,5 +82,5 @@ def tg_send_and_mark(cfg: Config, info: dict):
|
||||
original_reply_to=info.get("original_reply_to", ""),
|
||||
account_email=info.get("account_email", ""))
|
||||
# 只有发送成功才标记已读
|
||||
mark_as_seen(acct, [info["uid"]])
|
||||
mark_as_seen(acct, [(info["uid"], info.get("folder", "INBOX"))])
|
||||
logger.info(f" TG 发送成功,已标记已读")
|
||||
|
||||
Reference in New Issue
Block a user