Compare commits

..

3 Commits

3 changed files with 47 additions and 28 deletions

52
main.py
View File

@@ -7,7 +7,7 @@ import threading
import time import time
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
from src.config import load_config from src.config import load_config
from src.summarizer import poll_accounts, ai_process, tg_send_and_mark from src.summarizer import poll_accounts, ai_process, tg_send_and_mark, _mail_label
from src.tg_bot import poll_telegram, send_thinking from src.tg_bot import poll_telegram, send_thinking
from src.database import init_db, is_uid_processed, mark_fetched, mark_processing, mark_sent, mark_failed from src.database import init_db, is_uid_processed, mark_fetched, mark_processing, mark_sent, mark_failed
@@ -56,7 +56,8 @@ def _email_poller(cfg):
mark_fetched(uid_str, mail.account_email, mail.folder, mark_fetched(uid_str, mail.account_email, mail.folder,
mail.subject, mail.sender) mail.subject, mail.sender)
try: try:
tg_msg_id = send_thinking(cfg.telegram, cfg.telegram.chat_id, "📥 获取邮件中…") tg_msg_id = send_thinking(cfg.telegram, cfg.telegram.chat_id,
f"📥 *获取邮件中…*\n{_mail_label(mail)}")
except Exception: except Exception:
tg_msg_id = 0 tg_msg_id = 0
_email_queue.put((acct_idx, mail, tg_msg_id)) _email_queue.put((acct_idx, mail, tg_msg_id))
@@ -98,33 +99,37 @@ def _ai_processor(cfg):
time.sleep(0.1 if pending_futures else 0.5) time.sleep(0.1 if pending_futures else 0.5)
def _tg_worker(cfg): def _tg_poller(cfg):
logger.info("TG 线程已启动") logger.info("TG 轮询线程已启动(按钮回调专用)")
last_update_id = 0 last_update_id = 0
while _running: while _running:
try: try:
last_update_id = poll_telegram(cfg.telegram, cfg, last_update_id) last_update_id = poll_telegram(cfg.telegram, cfg, last_update_id)
except Exception as e: except Exception as e:
logger.error(f"TG 轮询错误: {e}", exc_info=True) logger.error(f"TG 轮询错误: {e}", exc_info=True)
while _running:
try:
info = _tg_queue.get_nowait()
tg_send_and_mark(cfg, info)
uid_str = info["uid"].decode("utf-8", errors="replace") if isinstance(info["uid"], bytes) else str(info["uid"])
mark_sent(uid_str)
except queue_module.Empty:
break
except Exception as e:
logger.error(f"TG 发送失败: {e}", exc_info=True)
try:
uid_str = info["uid"].decode("utf-8", errors="replace") if isinstance(info["uid"], bytes) else str(info["uid"])
mark_failed(uid_str)
except Exception:
pass
if _running: if _running:
_shutdown_event.wait(1) _shutdown_event.wait(0.1)
def _tg_sender(cfg):
logger.info("TG 发送线程已启动(邮件摘要专用)")
while _running:
try:
info = _tg_queue.get_nowait()
tg_send_and_mark(cfg, info)
uid_str = info["uid"].decode("utf-8", errors="replace") if isinstance(info["uid"], bytes) else str(info["uid"])
mark_sent(uid_str)
except queue_module.Empty:
pass
except Exception as e:
logger.error(f"TG 发送失败: {e}", exc_info=True)
try:
uid_str = info["uid"].decode("utf-8", errors="replace") if isinstance(info["uid"], bytes) else str(info["uid"])
mark_failed(uid_str)
except Exception:
pass
if _running:
_shutdown_event.wait(0.2)
def main(): def main():
@@ -140,7 +145,8 @@ def main():
threads = [ threads = [
threading.Thread(target=_email_poller, args=(cfg,), daemon=True), threading.Thread(target=_email_poller, args=(cfg,), daemon=True),
threading.Thread(target=_ai_processor, args=(cfg,), daemon=True), threading.Thread(target=_ai_processor, args=(cfg,), daemon=True),
threading.Thread(target=_tg_worker, args=(cfg,), daemon=True), threading.Thread(target=_tg_poller, args=(cfg,), daemon=True),
threading.Thread(target=_tg_sender, args=(cfg,), daemon=True),
] ]
for t in threads: for t in threads:

View File

@@ -1,4 +1,5 @@
import logging import logging
import re
from typing import Generator, Optional from typing import Generator, Optional
from src.config import Config from src.config import Config
from src.email_client import fetch_unseen_emails, mark_as_seen, Email from src.email_client import fetch_unseen_emails, mark_as_seen, Email
@@ -7,6 +8,17 @@ from src.tg_bot import send_summary, send_thinking, edit_message, format_summary
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_MD_SPECIAL = re.compile(r"([_*\[\]()~`>#+\-=|{}.!\\])")
def _escape_md(text: str) -> str:
return _MD_SPECIAL.sub(r"\\\1", text)
def _mail_label(mail: Email) -> str:
sender_short = mail.sender.split("<")[0].strip() if "<" in mail.sender else mail.sender.split("@")[0]
subject_short = mail.subject[:30] + ("" if len(mail.subject) > 30 else "")
return f"{_escape_md(sender_short)} · {_escape_md(subject_short)}"
def poll_accounts(cfg: Config, skip_uids: set[bytes] | None = None) -> Generator[tuple[int, Email], None, None]: def poll_accounts(cfg: Config, skip_uids: set[bytes] | None = None) -> Generator[tuple[int, Email], None, None]:
for idx, acct in enumerate(cfg.email_accounts): for idx, acct in enumerate(cfg.email_accounts):
@@ -23,14 +35,15 @@ def poll_accounts(cfg: Config, skip_uids: set[bytes] | None = None) -> Generator
def ai_process(cfg: Config, acct_idx: int, mail: Email, tg_msg_id: int = 0) -> Optional[dict]: def ai_process(cfg: Config, acct_idx: int, mail: Email, tg_msg_id: int = 0) -> Optional[dict]:
logger.info(f" 正在摘要: {mail.subject}") logger.info(f" 正在摘要: {mail.subject}")
label = _mail_label(mail)
# 阶段1显示等待处理 # 阶段1显示等待处理
if tg_msg_id: if tg_msg_id:
edit_message(cfg.telegram, cfg.telegram.chat_id, tg_msg_id, "⏳ 等待处理…") edit_message(cfg.telegram, cfg.telegram.chat_id, tg_msg_id, f"*等待处理…*\n{label}")
else: else:
tg_msg_id = send_thinking(cfg.telegram, cfg.telegram.chat_id, "⏳ 等待处理…") tg_msg_id = send_thinking(cfg.telegram, cfg.telegram.chat_id, f"*等待处理…*\n{label}")
# 阶段2AI 处理,更新为思考中 # 阶段2AI 处理,更新为思考中
edit_message(cfg.telegram, cfg.telegram.chat_id, tg_msg_id, "🤖 AI思考中…") edit_message(cfg.telegram, cfg.telegram.chat_id, tg_msg_id, f"🤖 *AI思考中…*\n{label}")
summary = summarize_email(cfg.ai, mail.recipient, mail.subject, mail.sender, mail.body, mail.account_email) summary = summarize_email(cfg.ai, mail.recipient, mail.subject, mail.sender, mail.body, mail.account_email)
summary["recipient"] = mail.recipient summary["recipient"] = mail.recipient
if "@" not in summary.get("sender", ""): if "@" not in summary.get("sender", ""):

View File

@@ -86,8 +86,8 @@ def poll_telegram(tg_cfg: TelegramConfig, cfg: Config, last_update_id: int) -> i
try: try:
resp = requests.get( resp = requests.get(
f"https://api.telegram.org/bot{tg_cfg.bot_token}/getUpdates", f"https://api.telegram.org/bot{tg_cfg.bot_token}/getUpdates",
params={"offset": last_update_id, "timeout": 10}, params={"offset": last_update_id, "timeout": 2, "allowed_updates": ["callback_query", "message"]},
timeout=15, timeout=5,
) )
resp.raise_for_status() resp.raise_for_status()
data = resp.json() data = resp.json()