Compare commits

...

2 Commits

View File

@@ -30,6 +30,7 @@ logger = logging.getLogger("main")
_running = True _running = True
_pending_lock = threading.Lock() _pending_lock = threading.Lock()
_pending_uids: set[bytes] = set() _pending_uids: set[bytes] = set()
_processing_uids: set[bytes] = set() # 正在处理含等待TG发送的邮件UID
# Queue 1: raw emails → AI processor # Queue 1: raw emails → AI processor
_email_queue: queue_module.Queue = queue_module.Queue() _email_queue: queue_module.Queue = queue_module.Queue()
@@ -51,7 +52,8 @@ def _email_poller(cfg):
if not _running: if not _running:
return return
with _pending_lock: with _pending_lock:
if mail.uid not in _pending_uids: if mail.uid not in _processing_uids:
_processing_uids.add(mail.uid)
_pending_uids.add(mail.uid) _pending_uids.add(mail.uid)
_email_queue.put((acct_idx, mail)) _email_queue.put((acct_idx, mail))
except Exception as e: except Exception as e:
@@ -90,6 +92,7 @@ def _ai_processor(cfg):
# AI 失败也要移除 pending否则永远卡住 # AI 失败也要移除 pending否则永远卡住
with _pending_lock: with _pending_lock:
_pending_uids.discard(mail.uid) _pending_uids.discard(mail.uid)
_processing_uids.discard(mail.uid)
time.sleep(0.1 if pending_futures else 0.5) time.sleep(0.1 if pending_futures else 0.5)
@@ -110,6 +113,7 @@ def _tg_worker(cfg):
# 发送并标记已读成功后才移除 pending # 发送并标记已读成功后才移除 pending
with _pending_lock: with _pending_lock:
_pending_uids.discard(info["uid"]) _pending_uids.discard(info["uid"])
_processing_uids.discard(info["uid"])
except queue_module.Empty: except queue_module.Empty:
break break
except Exception as e: except Exception as e:
@@ -118,6 +122,7 @@ def _tg_worker(cfg):
try: try:
with _pending_lock: with _pending_lock:
_pending_uids.discard(info["uid"]) _pending_uids.discard(info["uid"])
_processing_uids.discard(info["uid"])
except Exception: except Exception:
pass pass