From d70b817fd47c668eeaad5a1c78050496d738821e Mon Sep 17 00:00:00 2001 From: Zichao Lin Date: Mon, 6 Jul 2026 20:12:24 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20AI=E5=B9=B6=E8=A1=8C=E5=A4=84=E7=90=86?= =?UTF-8?q?=20+=20=E4=BF=AE=E5=A4=8D=E9=82=AE=E4=BB=B6=E4=B8=A2=E5=A4=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _ai_processor 改用 ThreadPoolExecutor(5 workers) 并行调用AI - _tg_worker 每轮排空所有待发TG消息(原来每轮只取1条) - fetch_unseen_emails 单封FETCH失败不再丢弃整个批次，跳过继续 --- main.py | 58 ++++++++++++++++++++++++++++----------------- src/email_client.py | 38 ++++++++++++++++------------- 2 files changed, 57 insertions(+), 39 deletions(-) diff --git a/main.py b/main.py index 9eb3d90..d490855 100644 --- a/main.py +++ b/main.py @@ -5,6 +5,7 @@ import signal import sys import threading import time +from concurrent.futures import ThreadPoolExecutor from src.config import load_config from src.summarizer import poll_accounts, ai_process, tg_send_and_mark from src.tg_bot import poll_telegram @@ -64,21 +65,33 @@ def _email_poller(cfg): def _ai_processor(cfg): - logger.info("AI 处理线程已启动") - while _running: - try: - acct_idx, mail = _email_queue.get(timeout=1) - except queue_module.Empty: - continue - try: - info = ai_process(cfg, acct_idx, mail) - if info: - _tg_queue.put(info) - except Exception as e: - logger.error(f"AI 处理邮件失败: {e}", exc_info=True) - finally: - with _pending_lock: - _pending_uids.discard(mail.uid) + logger.info("AI 处理线程已启动 (并行 workers=5)") + max_workers = 5 + with ThreadPoolExecutor(max_workers=max_workers) as executor: + pending_futures = {} + while _running: + while len(pending_futures) < max_workers: + try: + acct_idx, mail = _email_queue.get_nowait() + except queue_module.Empty: + break + future = executor.submit(ai_process, cfg, acct_idx, mail) + pending_futures[future] = (acct_idx, mail) + + completed = [f for f in pending_futures if f.done()] + for future in completed: + acct_idx, mail = pending_futures.pop(future) + try: + info = future.result() + if info: + _tg_queue.put(info) + except Exception as e: + logger.error(f"AI 处理邮件失败: {e}", exc_info=True) + finally: + with _pending_lock: + _pending_uids.discard(mail.uid) + + time.sleep(0.1 if pending_futures else 0.5) def _tg_worker(cfg): @@ -90,13 +103,14 @@ def _tg_worker(cfg): except Exception as e: logger.error(f"TG 轮询错误: {e}", exc_info=True) - try: - info = _tg_queue.get_nowait() - tg_send_and_mark(cfg, info) - except queue_module.Empty: - pass - except Exception as e: - logger.error(f"TG 发送失败: {e}", exc_info=True) + while _running: + try: + info = _tg_queue.get_nowait() + tg_send_and_mark(cfg, info) + except queue_module.Empty: + break + except Exception as e: + logger.error(f"TG 发送失败: {e}", exc_info=True) if _running: time.sleep(1) diff --git a/src/email_client.py b/src/email_client.py index 4ec63e5..e76f5d5 100644 --- a/src/email_client.py +++ b/src/email_client.py @@ -115,24 +115,28 @@ def fetch_unseen_emails(account: EmailAccount) -> list[Email]: emails = [] for uid in uids: - _, msg_data = conn.uid("FETCH", uid, "RFC822") - if msg_data[0] is None: - logger.warning("UID %s FETCH 返回空，跳过", uid) + 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) + + 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)) + except Exception as e: + logger.error("UID %s FETCH 失败，跳过: %s", uid, e) continue - raw_email = msg_data[0][1] - msg = email.message_from_bytes(raw_email) - - 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)) conn.logout() logger.info("共获取 %d 封新邮件", len(emails))