chore: AI并行workers 5→10

- _ai_processor 改用 ThreadPoolExecutor(5 workers) 并行调用AI
- _tg_worker 每轮排空所有待发TG消息(原来每轮只取1条)
- fetch_unseen_emails 单封FETCH失败不再丢弃整个批次，跳过继续

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 处理线程已启动")
+    logger.info("AI 处理线程已启动 (并行 workers=10)")
-    while _running:
+    max_workers = 10
-        try:
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
-            acct_idx, mail = _email_queue.get(timeout=1)
+        pending_futures = {}
-        except queue_module.Empty:
+        while _running:
-            continue
+            while len(pending_futures) < max_workers:
-        try:
+                try:
-            info = ai_process(cfg, acct_idx, mail)
+                    acct_idx, mail = _email_queue.get_nowait()
-            if info:
+                except queue_module.Empty:
-                _tg_queue.put(info)
+                    break
-        except Exception as e:
+                future = executor.submit(ai_process, cfg, acct_idx, mail)
-            logger.error(f"AI 处理邮件失败: {e}", exc_info=True)
+                pending_futures[future] = (acct_idx, mail)
-        finally:
+
-            with _pending_lock:
+            completed = [f for f in pending_futures if f.done()]
-                _pending_uids.discard(mail.uid)
+            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:
+        while _running:
-            info = _tg_queue.get_nowait()
+            try:
-            tg_send_and_mark(cfg, info)
+                info = _tg_queue.get_nowait()
-        except queue_module.Empty:
+                tg_send_and_mark(cfg, info)
-            pass
+            except queue_module.Empty:
-        except Exception as e:
+                break
-            logger.error(f"TG 发送失败: {e}", exc_info=True)
+            except Exception as e:
+                logger.error(f"TG 发送失败: {e}", exc_info=True)
 
         if _running:
             time.sleep(1)

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")
+        try:
-        if msg_data[0] is None:
+            _, msg_data = conn.uid("FETCH", uid, "RFC822")
-            logger.warning("UID %s FETCH 返回空，跳过", uid)
+            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))