From 1be3aadb08ede0aa2e07f87bc434522e3eb990d5 Mon Sep 17 00:00:00 2001 From: Zichao Lin Date: Thu, 2 Jul 2026 20:48:05 +0800 Subject: [PATCH] refactor: separate email poll, email process, TG poll into threads - main.py spawns 3 daemon threads: email poller, email processor, TG poller - Each thread runs independently so retries don't block other operations - _pending_uids set + lock prevents duplicate queueing across poll cycles - summarizer.py split into poll_accounts() (generator) and process_email() - Graceful shutdown via shared _running flag --- main.py | 77 +++++++++++++++++++++++++++++++++++++++-------- src/summarizer.py | 49 ++++++++++++------------------ 2 files changed, 83 insertions(+), 43 deletions(-) diff --git a/main.py b/main.py index 3aa4702..58e2ed9 100644 --- a/main.py +++ b/main.py @@ -1,9 +1,11 @@ import logging +import queue import signal import sys +import threading import time from src.config import load_config -from src.summarizer import process_all +from src.summarizer import poll_accounts, process_email from src.tg_bot import poll_telegram logging.basicConfig( @@ -14,6 +16,8 @@ logging.basicConfig( logger = logging.getLogger("main") _running = True +_pending_lock = threading.Lock() +_pending_uids: set[bytes] = set() def _signal_handler(signum, frame): @@ -22,6 +26,51 @@ def _signal_handler(signum, frame): _running = False +def _email_poller(cfg, processing_queue): + logger.info("邮件轮询线程已启动") + while _running: + for acct_idx, mail in poll_accounts(cfg): + if not _running: + return + with _pending_lock: + if mail.uid not in _pending_uids: + _pending_uids.add(mail.uid) + processing_queue.put((acct_idx, mail)) + if _running: + for _ in range(cfg.polling.interval_seconds): + if not _running: + return + time.sleep(1) + + +def _email_processor(cfg, processing_queue): + logger.info("邮件处理线程已启动") + while _running: + try: + acct_idx, mail = processing_queue.get(timeout=1) + except queue.Empty: + continue + try: + process_email(cfg, acct_idx, mail) + except Exception as e: + logger.error(f"处理邮件失败: {e}", exc_info=True) + finally: + with _pending_lock: + _pending_uids.discard(mail.uid) + + +def _tg_poller(cfg): + logger.info("TG 轮询线程已启动") + last_update_id = 0 + while _running: + try: + last_update_id = poll_telegram(cfg.telegram, cfg, last_update_id) + except Exception as e: + logger.error(f"TG 轮询错误: {e}", exc_info=True) + if _running: + time.sleep(1) + + def main(): global _running signal.signal(signal.SIGINT, _signal_handler) @@ -31,23 +80,25 @@ def main(): cfg = load_config(cfg_path) logger.info(f"AI邮件摘要机器人已启动，轮询间隔: {cfg.polling.interval_seconds}s") - last_check = 0.0 - last_update_id = 0 + processing_queue: queue.Queue = queue.Queue() - while _running: - try: - now = time.time() - if now - last_check >= cfg.polling.interval_seconds: - process_all(cfg) - last_check = now + threads = [ + threading.Thread(target=_email_poller, args=(cfg, processing_queue), daemon=True), + threading.Thread(target=_email_processor, args=(cfg, processing_queue), daemon=True), + threading.Thread(target=_tg_poller, args=(cfg,), daemon=True), + ] - last_update_id = poll_telegram(cfg.telegram, cfg, last_update_id) - except Exception as e: - logger.error(f"主循环出错: {e}", exc_info=True) + for t in threads: + t.start() - if _running: + try: + while _running: time.sleep(1) + except KeyboardInterrupt: + _running = False + for t in threads: + t.join(timeout=5) logger.info("机器人已停止") diff --git a/src/summarizer.py b/src/summarizer.py index facf669..9789f00 100644 --- a/src/summarizer.py +++ b/src/summarizer.py @@ -1,43 +1,32 @@ import logging +from typing import Generator from src.config import Config -from src.email_client import fetch_unseen_emails, mark_as_seen +from src.email_client import fetch_unseen_emails, mark_as_seen, Email from src.ai_client import summarize_email from src.tg_bot import send_summary, format_summary logger = logging.getLogger(__name__) -def process_all(cfg: Config): +def poll_accounts(cfg: Config) -> Generator[tuple[int, Email], None, None]: for idx, acct in enumerate(cfg.email_accounts): try: - _process_account(cfg, acct, idx) + logger.info(f"检查邮箱: {acct.username}") + emails = fetch_unseen_emails(acct) + if emails: + logger.info(f" 发现 {len(emails)} 封新邮件") + for mail in emails: + yield idx, mail except Exception as e: - logger.error(f"处理邮箱 {acct.username} 时出错: {e}", exc_info=True) + logger.error(f"轮询 {acct.username} 失败: {e}", exc_info=True) -def _process_account(cfg: Config, acct, acct_idx: int): - logger.info(f"检查邮箱: {acct.username}") - emails = fetch_unseen_emails(acct) - - if not emails: - logger.info(f" 没有新邮件") - return - - logger.info(f" 发现 {len(emails)} 封新邮件") - seen_uids = [] - - for mail in emails: - try: - logger.info(f" 正在摘要: {mail.subject}") - summary = summarize_email(cfg.ai, acct.username, mail.subject, mail.sender, mail.body) - summary["recipient"] = acct.username - text = format_summary(summary) - send_summary(cfg.telegram, cfg.telegram.chat_id, text, summary, mail.body, acct_idx) - seen_uids.append(mail.uid) - logger.info(f" 已发送到 Telegram") - except Exception as e: - logger.error(f" 处理邮件 '{mail.subject}' 失败: {e}", exc_info=True) - seen_uids.append(mail.uid) - - if seen_uids: - mark_as_seen(acct, seen_uids) +def process_email(cfg: Config, acct_idx: int, mail: Email): + acct = cfg.email_accounts[acct_idx] + logger.info(f" 正在摘要: {mail.subject}") + summary = summarize_email(cfg.ai, acct.username, mail.subject, mail.sender, mail.body) + summary["recipient"] = acct.username + text = format_summary(summary) + send_summary(cfg.telegram, cfg.telegram.chat_id, text, summary, mail.body, acct_idx) + mark_as_seen(acct, [mail.uid]) + logger.info(f" 已处理: {mail.subject}")