fix: update email_client and smtp_client for new config structure - email_client: use account.imap.server/port/ssl/starttls instead of flat fields - smtp_client: use s.ssl/starttls instead of s.use_ssl/use_starttls - tg_bot: fix _do_send_reply to use find_user_by_chat_id for email_accounts

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

.gitignore

@@ -1,4 +1,5 @@
 .venv/
 __pycache__/
 *.pyc
-config.yaml
+config.json
+data/

MEMORY.md

@@ -0,0 +1,6 @@
+# AI Mail Bot - 记忆
+
+## 固定规则
+
+- **Telegram parse_mode 永远为 MarkdownV2**，不传 None。需要发纯文本时，用 Markdown 转义包装，而不是去掉 parse_mode。
+- **所有代码改动必须完整测试**，除非用户明确说不需要测试。没有测试的代码不算完成。

config.example.json

@@ -0,0 +1,32 @@
+{
+  "bot_token": "your_bot_token_here",
+  "polling_interval": 30,
+  "users": [
+    {
+      "chat_id": "your_telegram_chat_id",
+      "ai": {
+        "api_key": "your_deepseek_api_key",
+        "model": "deepseek-v4-flash",
+        "base_url": "https://api.deepseek.com"
+      },
+      "email_accounts": [
+        {
+          "username": "you@126.com",
+          "password": "your_imap_password",
+          "imap": {
+            "server": "imap.126.com",
+            "port": 993,
+            "ssl": true,
+            "starttls": false
+          },
+          "smtp": {
+            "server": "smtp.126.com",
+            "port": 465,
+            "ssl": true,
+            "starttls": false
+          }
+        }
+      ]
+    }
+  ]
+}

config.yaml.example

@@ -1,22 +0,0 @@
-email_accounts:
-  - imap_server: "imap.gmail.com"
-    imap_port: 993
-    username: "your_email@gmail.com"
-    password: "your_app_password"
-
-  - imap_server: "imap.qq.com"
-    imap_port: 993
-    username: "your_email@qq.com"
-    password: "your_authorization_code"
-
-ai:
-  api_key: "sk-your_deepseek_api_key"
-  model: "deepseek-chat"
-  base_url: "https://api.deepseek.com"
-
-telegram:
-  bot_token: "1234567890:ABCdefGHIjklmNOPqrstUVwxyz"
-  chat_id: "123456789"
-
-polling:
-  interval_seconds: 60

main.py

@@ -1,24 +1,140 @@
 import logging
+import logging.handlers
+import queue as queue_module
 import signal
 import sys
+import threading
 import time
-from src.config import load_config
-from src.summarizer import process_all
+from concurrent.futures import ThreadPoolExecutor
+from src.config import load_config, Config, UserConfig
+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.database import init_db, is_uid_processed, mark_fetched, mark_processing, mark_sent, mark_failed
 
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+# Thread-safe logging
+_log_queue = queue_module.Queue(-1)
+_queue_handler = logging.handlers.QueueHandler(_log_queue)
+_console_handler = logging.StreamHandler()
+_console_handler.setFormatter(logging.Formatter(
+    "%(asctime)s [%(levelname)s] [%(threadName)s] %(name)s: %(message)s",
     datefmt="%Y-%m-%d %H:%M:%S",
-)
+))
+_log_listener = logging.handlers.QueueListener(_log_queue, _console_handler)
+
+root = logging.getLogger()
+root.addHandler(_queue_handler)
+root.setLevel(logging.INFO)
+_log_listener.start()
+
 logger = logging.getLogger("main")
 
 _running = True
+_shutdown_event = threading.Event()
+
+# Queue 1: raw emails → AI processor (user, acct_idx, mail, tg_msg_id)
+_email_queue: queue_module.Queue = queue_module.Queue()
+# Queue 2: AI summaries → TG sender (user, info_dict)
+_tg_queue: queue_module.Queue = queue_module.Queue()
 
 
 def _signal_handler(signum, frame):
     global _running
     logger.info("收到退出信号，正在停止...")
     _running = False
+    _shutdown_event.set()
+
+
+def _email_poller(cfg: Config):
+    logger.info("邮件轮询线程已启动")
+    while _running:
+        try:
+            for user in cfg.users:
+                if not _running:
+                    return
+                for acct_idx, mail in poll_accounts(user):
+                    if not _running:
+                        return
+                    uid_str = mail.uid.decode("utf-8", errors="replace") if isinstance(mail.uid, bytes) else str(mail.uid)
+                    if not is_uid_processed(uid_str):
+                        mark_fetched(uid_str, mail.account_email, mail.folder,
+                                     mail.subject, mail.sender)
+                        try:
+                            tg_msg_id = send_thinking(cfg.bot_token, user.chat_id,
+                                f"📥 *获取邮件中…*\n{_mail_label(mail)}")
+                            logger.info("  [%s] 获取邮件中消息已发送: msg_id=%d", user.chat_id, tg_msg_id)
+                        except Exception as e:
+                            logger.error("  [%s] 获取邮件中消息发送失败: %s", user.chat_id, e)
+                            tg_msg_id = 0
+                        _email_queue.put((user, acct_idx, mail, tg_msg_id))
+        except Exception as e:
+            logger.error(f"邮件轮询线程异常: {e}", exc_info=True)
+            time.sleep(5)
+        if _running:
+            _shutdown_event.wait(cfg.polling_interval)
+
+
+def _ai_processor(cfg: Config):
+    logger.info("AI 处理线程已启动 (并行 workers=10)")
+    max_workers = 10
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        pending_futures = {}
+        while _running:
+            while len(pending_futures) < max_workers:
+                try:
+                    user, acct_idx, mail, tg_msg_id = _email_queue.get_nowait()
+                except queue_module.Empty:
+                    break
+                uid_str = mail.uid.decode("utf-8", errors="replace") if isinstance(mail.uid, bytes) else str(mail.uid)
+                mark_processing(uid_str)
+                future = executor.submit(ai_process, cfg, user, acct_idx, mail, tg_msg_id)
+                pending_futures[future] = (user, acct_idx, mail)
+
+            completed = [f for f in pending_futures if f.done()]
+            for future in completed:
+                user, acct_idx, mail = pending_futures.pop(future)
+                try:
+                    info = future.result()
+                    if info:
+                        _tg_queue.put((user, info))
+                except Exception as e:
+                    logger.error(f"AI 处理邮件失败: {e}", exc_info=True)
+                    uid_str = mail.uid.decode("utf-8", errors="replace") if isinstance(mail.uid, bytes) else str(mail.uid)
+                    mark_failed(uid_str)
+
+            time.sleep(0.1 if pending_futures else 0.5)
+
+
+def _tg_poller(cfg: Config):
+    logger.info("TG 轮询线程已启动（按钮回调专用）")
+    last_update_id = 0
+    while _running:
+        try:
+            last_update_id = poll_telegram(cfg.bot_token, cfg, last_update_id)
+        except Exception as e:
+            logger.error(f"TG 轮询错误: {e}", exc_info=True)
+        if _running:
+            _shutdown_event.wait(0.05)
+
+
+def _tg_sender(cfg: Config):
+    logger.info("TG 发送线程已启动（邮件摘要专用）")
+    while _running:
+        try:
+            user, info = _tg_queue.get_nowait()
+            tg_send_and_mark(cfg, user, 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():
@@ -26,22 +142,40 @@ def main():
     signal.signal(signal.SIGINT, _signal_handler)
     signal.signal(signal.SIGTERM, _signal_handler)
 
-    cfg_path = sys.argv[1] if len(sys.argv) > 1 else "config.yaml"
+    cfg_path = sys.argv[1] if len(sys.argv) > 1 else "config.json"
     cfg = load_config(cfg_path)
-    logger.info(f"AI邮件摘要机器人已启动，轮询间隔: {cfg.polling.interval_seconds}s")
+    init_db()
 
-    while _running:
+    user_count = len(cfg.users)
+    acct_count = sum(len(u.email_accounts) for u in cfg.users)
+    logger.info(f"AI邮件摘要机器人已启动: {user_count} 用户, {acct_count} 邮箱, 轮询间隔: {cfg.polling_interval}s")
+
+    threads = [
+        threading.Thread(target=_email_poller, args=(cfg,), daemon=True),
+        threading.Thread(target=_ai_processor, 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:
+        t.start()
+
+    logger.info("所有线程已启动，进入主循环")
+    tick = 0
     try:
-            process_all(cfg)
-        except Exception as e:
-            logger.error(f"轮询出错: {e}", exc_info=True)
-
-        if _running:
-            for _ in range(cfg.polling.interval_seconds):
-                if not _running:
-                    break
-                time.sleep(1)
+        while _running:
+            tick += 1
+            if tick % 30 == 0:
+                logger.info("队列状态: email_queue=%d tg_queue=%d",
+                            _email_queue.qsize(), _tg_queue.qsize())
+            _shutdown_event.wait(1)
+    except KeyboardInterrupt:
+        _running = False
+        _shutdown_event.set()
 
+    for t in threads:
+        t.join(timeout=5)
+    _log_listener.stop()
     logger.info("机器人已停止")
 
 

requirements.txt

@@ -0,0 +1,2 @@
+PyYAML>=6.0
+requests>=2.28.0

src/ai_client.py

@@ -1,30 +1,48 @@
 import json
+import logging
+import time
 from typing import Any
 import requests
 from src.config import AIConfig
 from src.retry import retry
 
-SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以 JSON 格式返回结构化摘要。
+logger = logging.getLogger(__name__)
+
+SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以 JSON 格式返回结构化摘要。所有输出（subject、summary 等）必须使用简体中文。
 
 返回格式必须严格遵循以下 JSON schema：
 {
   "subject": "你概括的简短标题（10字以内）",
   "recipient": "收件邮箱地址",
   "sender": "发件人",
+  "category": "normal/promotion/notification",
+  "plain_text": "邮件全文的纯文本版本，去除HTML标签，保留所有文字内容，保持原始段落结构，不删减不修改任何内容",
   "summary": "50字以内的核心摘要",
-  "priority": "high/medium/low",
-  "action_required": true/false,
-  "action_items": ["待办事项1", "待办事项2"],
-  "key_points": ["关键要点1", "关键要点2"]
+  "verification_code": "",
+  "app_name": "",
+  "links": [{"text": "按钮文字", "url": "https://..."}],
+  "can_reply": true/false
 }
 
-其中 subject 字段不能照抄原邮件主题，必须是你总结生成的简短标题。
+其中：
+- subject 不能照抄原邮件主题，必须是你总结生成的简短标题
+- category: 邮件类型。推广邮件（营销、促销、广告、 newsletter 推广等）标记为 "promotion"；验证码邮件标记为 "notification"；普通邮件标记为 "normal"
+- verification_code: 仅当 category 为 "notification" 且邮件包含验证码时填写，提取完整的验证码字符串（纯数字、字母数字混合等均可，如「123456」「abc-789」「701383」）。非验证码邮件留空字符串
+- app_name: 仅当 category 为 "notification" 且邮件包含验证码时填写，填写发送验证码的应用/平台名称（如「GitHub」「微信」「Kraken」）。非验证码邮件留空字符串
+- links: 仅提取以下两类链接：1）邮件明确要求用户点击才能完成某操作的链接（如确认邮箱、重置密码、查看账单、审批请求、查看通知详情等）；2）退订/取消订阅链接。不要提取推广性质的链接（如下载App、注册领优惠、查看更多商品等营销链接）、追踪像素、邮件签名中的社交链接。每条包含简短描述文字和完整 URL。最多 3 条。没有符合条件的链接时为空数组
+- 退订链接提取规则：推广邮件必须提取退订链接。退订链接通常在邮件最底部，包含以下关键词之一：unsubscribe、退订、取消订阅、opt out、manage preferences、email preferences。即使链接被包裹在 HTML 标签中，也要提取 href 属性里的完整 URL。如果邮件底部同时有多个退订相关链接，只提取最主要的那个
+- plain_text: 将邮件全文转换为可读的纯文本版本。去除所有HTML标签、CSS样式、脚本代码，保留完整的文字内容和原始段落结构。不删减、不修改、不总结任何内容，完整保留原文所有信息
+- summary 要简洁，把需要知道的信息浓缩成一句话，不需要分条列出
+- 当 category 为 "promotion" 时，summary 直接写成一句话：「[发送方名称] 在推广 [推广的产品/服务/活动]」，例如「腾讯云在推广双十一云服务器折扣」。此时 can_reply 为 false
+- can_reply 为 false 表示无需回复或不适合建议回复
 只返回 JSON，不要包含任何其他文字。"""
 
 
 @retry()
-def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str, body: str) -> dict[str, Any]:
-    content = f"收件人: {recipient}\n发件人: {sender}\n主题: {subject}\n正文:\n{body[:4000]}"
+def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str, body: str, account: str = "") -> dict[str, Any]:
+    body_preview = body[:80].replace("\n", " ")
+    logger.info("AI 摘要请求: sender=%s subject=%s body_preview=%s", sender, subject, body_preview)
+    content = f"收件人: {recipient}\n发件人: {sender}\n主题: {subject}\n账号: {account}\n正文:\n{body}"
 
     payload = {
         "model": ai_cfg.model,
@@ -35,18 +53,118 @@ def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str,
         "response_format": {"type": "json_object"},
     }
 
+    return _call_deepseek(ai_cfg, payload)
+
+
+PROMO_DETAIL_PROMPT = """你是一个邮件分析助手。用户想了解更多关于这封推广邮件的详情。
+请根据邮件原文，输出 3-5 条关键信息，每条一句话，简洁明了。包括：优惠力度、适用条件、截止时间、核心卖点等用户做决策需要知道的信息。所有输出使用简体中文。
+只返回一个 JSON: {"details": ["信息1", "信息2", ...]}
+只返回 JSON，不要包含任何其他文字。"""
+
+
+@retry()
+def expand_promo_detail(ai_cfg: AIConfig, sender: str, subject: str, body: str) -> list[str]:
+    logger.info("AI 推广详情请求: sender=%s subject=%s", sender, subject)
+    content = f"发件人: {sender}\n主题: {subject}\n正文:\n{body}"
+    payload = {
+        "model": ai_cfg.model,
+        "messages": [
+            {"role": "system", "content": PROMO_DETAIL_PROMPT},
+            {"role": "user", "content": content},
+        ],
+        "response_format": {"type": "json_object"},
+    }
+    result = _call_deepseek(ai_cfg, payload)
+    return result.get("details", [])
+
+
+MORE_REPLIES_PROMPT_SHORT = """你是一个邮件回复助手。根据以下邮件内容生成3个完全不同的简洁回复。
+
+规则：
+- 简洁模式：直接说要点，不需要称呼、落款、邮件格式。像微信/短信一样简短。
+- 必须使用与邮件相同的语言回复（英文邮件用英文，中文邮件用中文）。
+- 不要添加任何占位符如 [Your Name]。
+
+以下是一些已经生成过的回复，请生成全新的回复，不要与已有回复重复。
+
+返回 JSON:
+{
+  "suggestions": [
+    {"text": "回复内容"},
+    {"text": "回复内容"},
+    {"text": "回复内容"}
+  ]
+}
+每条回复控制在 50 字以内。
+只返回 JSON，不要包含任何其他文字。"""
+
+MORE_REPLIES_PROMPT_FULL = """你是一个邮件回复助手。根据以下邮件内容生成3个不同的正式邮件回复。
+
+规则：
+- 完整模式：生成完整的邮件格式，包括称呼（如 Dear xxx,）和正文，结尾只写问候语逗号，不写落款，用户会自己编辑添加。
+- 必须使用与邮件相同的语言回复（英文邮件用英文，中文邮件用中文）。
+- 语气正式专业，适合商务邮件。
+
+以下是一些已经生成过的回复，请生成全新的回复，不要与已有回复重复。
+
+返回 JSON:
+{
+  "suggestions": [
+    {"text": "回复内容"},
+    {"text": "回复内容"},
+    {"text": "回复内容"}
+  ]
+}
+每条回复控制在 300 字以内。
+只返回 JSON，不要包含任何其他文字。"""
+
+
+@retry()
+def generate_more_replies(ai_cfg: AIConfig, sender: str, subject: str, body: str,
+                          previous_suggestions: list[dict], user_hint: str = "",
+                          style: str = "short") -> list[dict]:
+    logger.info("AI 换批请求: sender=%s subject=%s user_hint=%s prev=%d 条 style=%s",
+                sender, subject, user_hint or "(无)", len(previous_suggestions), style)
+    content = f"发件人: {sender}\n主题: {subject}\n正文:\n{body}\n\n已生成过的回复:\n"
+    for i, s in enumerate(previous_suggestions, 1):
+        content += f"{i}. {s['text']}\n"
+    if user_hint:
+        content += f"\n用户要求: {user_hint}"
+
+    prompt = MORE_REPLIES_PROMPT_FULL if style == "full" else MORE_REPLIES_PROMPT_SHORT
+    payload = {
+        "model": ai_cfg.model,
+        "messages": [
+            {"role": "system", "content": prompt},
+            {"role": "user", "content": content},
+        ],
+        "response_format": {"type": "json_object"},
+    }
+    result = _call_deepseek(ai_cfg, payload)
+    return result.get("suggestions", [])
+
+
+def _call_deepseek(ai_cfg: AIConfig, payload: dict) -> dict:
+    model = payload.get("model", ai_cfg.model)
+    logger.info("DeepSeek API 请求: model=%s", model)
+    t0 = time.time()
     headers = {
         "Authorization": f"Bearer {ai_cfg.api_key}",
         "Content-Type": "application/json",
     }
-
     resp = requests.post(
         f"{ai_cfg.base_url.rstrip('/')}/chat/completions",
         headers=headers,
         json=payload,
-        timeout=60,
+        timeout=120,
     )
+    elapsed = time.time() - t0
     resp.raise_for_status()
     result = resp.json()
+    usage = result.get("usage", {})
+    prompt_tokens = usage.get("prompt_tokens", "?")
+    completion_tokens = usage.get("completion_tokens", "?")
     raw = result["choices"][0]["message"]["content"]
+    logger.info("DeepSeek API 响应: %.2fs, token=%d+%d, 输出长度=%d",
+                elapsed, prompt_tokens, completion_tokens, len(raw))
     return json.loads(raw)

src/config.py

@@ -1,14 +1,30 @@
 from dataclasses import dataclass, field
-from typing import List
-import yaml
+from typing import List, Optional
+import json
+
+
+@dataclass
+class ImapConfig:
+    server: str
+    port: int = 993
+    ssl: bool = True
+    starttls: bool = False
+
+
+@dataclass
+class SmtpConfig:
+    server: str
+    port: int = 465
+    ssl: bool = True
+    starttls: bool = False
 
 
 @dataclass
 class EmailAccount:
-    imap_server: str
-    imap_port: int
     username: str
     password: str
+    imap: ImapConfig
+    smtp: SmtpConfig
 
 
 @dataclass
@@ -19,36 +35,50 @@ class AIConfig:
 
 
 @dataclass
-class TelegramConfig:
-    bot_token: str
+class UserConfig:
     chat_id: str
-
-
-@dataclass
-class PollingConfig:
-    interval_seconds: int
+    ai: AIConfig
+    email_accounts: List[EmailAccount] = field(default_factory=list)
 
 
 @dataclass
 class Config:
-    email_accounts: List[EmailAccount]
-    ai: AIConfig
-    telegram: TelegramConfig
-    polling: PollingConfig
+    bot_token: str
+    polling_interval: int
+    users: List[UserConfig] = field(default_factory=list)
 
 
 def load_config(path: str) -> Config:
     with open(path, "r", encoding="utf-8") as f:
-        raw = yaml.safe_load(f)
+        raw = json.load(f)
 
-    accounts = [EmailAccount(**a) for a in raw["email_accounts"]]
-    ai = AIConfig(**raw["ai"])
-    tg = TelegramConfig(**raw["telegram"])
-    polling = PollingConfig(**raw["polling"])
+    users = []
+    for u in raw.get("users", []):
+        email_accounts = []
+        for a in u.get("email_accounts", []):
+            email_accounts.append(EmailAccount(
+                username=a["username"],
+                password=a["password"],
+                imap=ImapConfig(**a.get("imap", {})),
+                smtp=SmtpConfig(**a.get("smtp", {})),
+            ))
+        users.append(UserConfig(
+            chat_id=str(u["chat_id"]),
+            ai=AIConfig(**u["ai"]),
+            email_accounts=email_accounts,
+        ))
 
     return Config(
-        email_accounts=accounts,
-        ai=ai,
-        telegram=tg,
-        polling=polling,
+        bot_token=raw["bot_token"],
+        polling_interval=raw.get("polling_interval", 30),
+        users=users,
     )
+
+
+def find_user_by_chat_id(cfg: Config, chat_id: int | str) -> Optional[UserConfig]:
+    """根据 chat_id 查找用户"""
+    cid = str(chat_id)
+    for user in cfg.users:
+        if user.chat_id == cid:
+            return user
+    return None

src/database.py

@@ -0,0 +1,177 @@
+import json
+import logging
+import sqlite3
+import threading
+from contextlib import contextmanager
+from datetime import datetime
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+DB_PATH = Path(__file__).parent.parent / "data" / "mail_bot.db"
+_local = threading.local()
+
+
+def init_db():
+    """初始化数据库表，启动时清空邮件处理记录"""
+    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
+    conn = _get_conn()
+    conn.executescript("""
+        CREATE TABLE IF NOT EXISTS processed_emails (
+            uid TEXT PRIMARY KEY,
+            account TEXT NOT NULL,
+            folder TEXT NOT NULL DEFAULT 'INBOX',
+            subject TEXT,
+            sender TEXT,
+            status TEXT NOT NULL DEFAULT 'fetched',
+            created_at TEXT NOT NULL,
+            sent_at TEXT
+        );
+
+        CREATE TABLE IF NOT EXISTS email_contexts (
+            msg_id INTEGER PRIMARY KEY,
+            data TEXT NOT NULL,
+            created_at TEXT NOT NULL
+        );
+
+        CREATE TABLE IF NOT EXISTS conversations (
+            chat_id INTEGER PRIMARY KEY,
+            data TEXT NOT NULL,
+            updated_at TEXT NOT NULL
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_processed_status ON processed_emails(status);
+        CREATE INDEX IF NOT EXISTS idx_processed_account ON processed_emails(account);
+    """)
+    conn.execute("DELETE FROM processed_emails")
+    conn.execute("DELETE FROM email_contexts")
+    conn.execute("DELETE FROM conversations")
+    conn.commit()
+    logger.info("数据库初始化完成: %s", DB_PATH)
+
+
+def _get_conn() -> sqlite3.Connection:
+    if not hasattr(_local, "conn") or _local.conn is None:
+        DB_PATH.parent.mkdir(parents=True, exist_ok=True)
+        _local.conn = sqlite3.connect(str(DB_PATH), timeout=10)
+        _local.conn.execute("PRAGMA journal_mode=WAL")
+        _local.conn.execute("PRAGMA busy_timeout=5000")
+        _local.conn.row_factory = sqlite3.Row
+    return _local.conn
+
+
+@contextmanager
+def get_db():
+    conn = _get_conn()
+    try:
+        yield conn
+        conn.commit()
+    except Exception:
+        conn.rollback()
+        raise
+
+
+# ── processed_emails ────────────────────────────────────
+
+def is_uid_processed(uid: str) -> bool:
+    with get_db() as conn:
+        row = conn.execute(
+            "SELECT 1 FROM processed_emails WHERE uid=? AND status != 'failed'",
+            (uid,)
+        ).fetchone()
+        return row is not None
+
+
+def is_uids_processed(uids: list[str]) -> set[str]:
+    """批量检查，返回已处理的 uid 集合"""
+    if not uids:
+        return set()
+    with get_db() as conn:
+        placeholders = ",".join("?" * len(uids))
+        rows = conn.execute(
+            f"SELECT uid FROM processed_emails WHERE uid IN ({placeholders}) AND status != 'failed'",
+            uids
+        ).fetchall()
+        return {row["uid"] for row in rows}
+
+
+def mark_fetched(uid: str, account: str, folder: str = "INBOX",
+                 subject: str = "", sender: str = ""):
+    with get_db() as conn:
+        conn.execute(
+            """INSERT OR IGNORE INTO processed_emails
+               (uid, account, folder, subject, sender, status, created_at)
+               VALUES (?, ?, ?, ?, ?, 'fetched', ?)""",
+            (uid, account, folder, subject, sender, datetime.now().isoformat())
+        )
+
+
+def mark_processing(uid: str):
+    with get_db() as conn:
+        conn.execute(
+            "UPDATE processed_emails SET status='processing' WHERE uid=?",
+            (uid,)
+        )
+
+
+def mark_sent(uid: str):
+    with get_db() as conn:
+        conn.execute(
+            "UPDATE processed_emails SET status='sent', sent_at=? WHERE uid=?",
+            (datetime.now().isoformat(), uid)
+        )
+
+
+def mark_failed(uid: str):
+    with get_db() as conn:
+        conn.execute(
+            "UPDATE processed_emails SET status='failed' WHERE uid=?",
+            (uid,)
+        )
+
+
+# ── email_contexts ──────────────────────────────────────
+
+def save_email_context(msg_id: int, data: dict):
+    with get_db() as conn:
+        conn.execute(
+            """INSERT OR REPLACE INTO email_contexts (msg_id, data, created_at)
+               VALUES (?, ?, ?)""",
+            (msg_id, json.dumps(data, ensure_ascii=False), datetime.now().isoformat())
+        )
+
+
+def load_email_context(msg_id: int) -> dict | None:
+    with get_db() as conn:
+        row = conn.execute(
+            "SELECT data FROM email_contexts WHERE msg_id=?", (msg_id,)
+        ).fetchone()
+        if row:
+            return json.loads(row["data"])
+        return None
+
+
+# ── conversations ───────────────────────────────────────
+
+def save_conversation(chat_id: int, data: dict):
+    with get_db() as conn:
+        conn.execute(
+            """INSERT OR REPLACE INTO conversations (chat_id, data, updated_at)
+               VALUES (?, ?, ?)""",
+            (chat_id, json.dumps(data, ensure_ascii=False), datetime.now().isoformat())
+        )
+
+
+def load_conversation(chat_id: int) -> dict | None:
+    with get_db() as conn:
+        row = conn.execute(
+            "SELECT data FROM conversations WHERE chat_id=?", (chat_id,)
+        ).fetchone()
+        if row:
+            return json.loads(row["data"])
+        return None
+
+
+def delete_conversation(chat_id: int):
+    with get_db() as conn:
+        conn.execute("DELETE FROM conversations WHERE chat_id=?", (chat_id,))

src/email_client.py

@@ -1,19 +1,29 @@
 import imaplib
 import email
+import logging
+import re
 from email.header import decode_header
-from email.utils import parsedate_to_datetime
+from email.utils import parsedate_to_datetime, parseaddr
 from typing import Optional
 from src.config import EmailAccount
 from src.retry import retry
 
+logger = logging.getLogger(__name__)
+
 
 class Email:
-    def __init__(self, uid: bytes, subject: str, sender: str, body: str, date: str):
+    def __init__(self, uid: bytes, subject: str, sender: str, recipient: str,
+                 body: str, date: str, reply_to: str = "", account_email: str = "",
+                 folder: str = "INBOX"):
         self.uid = uid
         self.subject = subject
         self.sender = sender
+        self.recipient = recipient
         self.body = body
         self.date = date
+        self.reply_to = reply_to
+        self.account_email = account_email
+        self.folder = folder
 
 
 def _decode_str(s: str) -> str:
@@ -30,6 +40,18 @@ def _decode_str(s: str) -> str:
     return "".join(result)
 
 
+def _decode_address_header(header_value: str) -> str:
+    if not header_value:
+        return ""
+    decoded = _decode_str(header_value)
+    name, addr = parseaddr(decoded)
+    if not addr:
+        return name
+    if name:
+        return f"{name} <{addr}>"
+    return addr
+
+
 def _get_text_from_msg(msg) -> str:
     if msg.is_multipart():
         for part in msg.walk():
@@ -68,46 +90,126 @@ def _select_mailbox(conn, mailbox: str = "INBOX"):
         raise RuntimeError(f"无法选择邮箱 {mailbox}: {data}")
 
 
-def _login_and_prepare(account: EmailAccount):
-    conn = imaplib.IMAP4_SSL(account.imap_server, account.imap_port)
+def _login_and_prepare(account: EmailAccount, timeout: int = 30, select_inbox: bool = True):
+    imap = account.imap
+    logger.info("连接 %s:%d (超时%ds)", imap.server, imap.port, timeout)
+    if imap.ssl:
+        conn = imaplib.IMAP4_SSL(imap.server, imap.port, timeout=timeout)
+    else:
+        conn = imaplib.IMAP4(imap.server, imap.port, timeout=timeout)
+        if imap.starttls:
+            conn.starttls()
     conn.login(account.username, account.password)
+    logger.info("登录成功: %s", account.username)
     if _check_provider(account.username, _PROVIDERS_NEED_ID):
+        logger.info("发送 ID 命令 (126/163 兼容)")
         _send_id_command(conn)
+    if select_inbox:
         _select_mailbox(conn)
+        logger.info("已选择 INBOX")
     return conn
 
 
-@retry()
-def fetch_unseen_emails(account: EmailAccount) -> list[Email]:
-    conn = _login_and_prepare(account)
+def fetch_unseen_emails(account: EmailAccount, skip_uids: set[bytes] | None = None) -> list[Email]:
+    conn = _login_and_prepare(account, select_inbox=False)
+    if skip_uids is None:
+        skip_uids = set()
 
+    # 列出所有文件夹，跳过 NoSelect 文件夹
+    _, folder_data = conn.list()
+    folders = []
+    if folder_data:
+        for item in folder_data:
+            if item is None:
+                continue
+            line = item.decode("utf-8", errors="replace")
+            # 跳过 \NoSelect 文件夹（父文件夹，不能直接 SELECT）
+            if "\\NoSelect" in line:
+                continue
+            m = re.search(r'"([^"]*)"\s*$', line)
+            if m:
+                folders.append(m.group(1))
+    if not folders:
+        folders = ["INBOX"]
+    logger.info("扫描文件夹: %s", ", ".join(folders))
+
+    emails = []
+    skipped = 0
+    for folder in folders:
+        try:
+            status, _ = conn.select(folder, readonly=True)
+            if status != "OK":
+                logger.debug("无法选择文件夹: %s", folder)
+                continue
             _, data = conn.uid("SEARCH", None, "UNSEEN")
             uids = data[0].split() if data[0] else []
-    emails = []
+            if not uids:
+                continue
 
-    for uid in uids:
+            # 先过滤掉正在处理的 UID，避免下载不需要的邮件
+            new_uids = [uid for uid in uids if uid not in skip_uids]
+            skipped += len(uids) - len(new_uids)
+            if not new_uids:
+                continue
+            logger.info("  %s: %d 封未读 (%d 封跳过)", folder, len(new_uids), len(uids) - len(new_uids))
+
+            for uid in new_uids:
+                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_str(msg.get("From", ""))
+                    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)
 
-        emails.append(Email(uid=uid, subject=subject, sender=sender, body=body, date=date_str))
+                    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,
+                                        folder=folder))
 
+                    # FETCH 可能被服务器自动标已读，立即标回未读
+                    conn.uid("STORE", uid, "-FLAGS", "\\Seen")
+                except Exception as e:
+                    logger.error("    UID %s FETCH 失败，跳过: %s", uid, e)
+                    continue
+        except Exception as e:
+            logger.warning("扫描文件夹 %s 失败: %s", folder, e)
+            continue
+
+    try:
         conn.logout()
+    except Exception:
+        pass
+    logger.info("共获取 %d 封新邮件", len(emails))
     return emails
 
 
 @retry()
-def mark_as_seen(account: EmailAccount, uids: list[bytes]):
-    if not uids:
+def mark_as_seen(account: EmailAccount, uids_with_folder: list[tuple[bytes, str]]):
+    if not uids_with_folder:
         return
-    conn = _login_and_prepare(account)
+    logger.info("标记 %d 封邮件为已读", len(uids_with_folder))
+    conn = _login_and_prepare(account, select_inbox=False)
+    # 按文件夹分组
+    from collections import defaultdict
+    by_folder: dict[str, list[bytes]] = defaultdict(list)
+    for uid, folder in uids_with_folder:
+        by_folder[folder].append(uid)
+    for folder, uids in by_folder.items():
+        try:
+            conn.select(folder, readonly=False)
             for uid in uids:
                 conn.uid("STORE", uid, "+FLAGS", "\\Seen")
+            logger.info("  %s: 标记 %d 封", folder, len(uids))
+        except Exception as e:
+            logger.error("  标记 %s 失败: %s", folder, e)
     conn.logout()
+    logger.info("标记完成")

src/retry.py

@@ -1,10 +1,11 @@
 import logging
+import time
 from functools import wraps
 
 logger = logging.getLogger(__name__)
 
 
-def retry(max_tries=10):
+def retry(max_tries=10, base_delay=1, max_delay=60):
     def decorator(func):
         @wraps(func)
         def wrapper(*args, **kwargs):
@@ -15,10 +16,12 @@ def retry(max_tries=10):
                 except Exception as e:
                     last_exc = e
                     if attempt < max_tries:
+                        delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
                         logger.warning(
-                            "%s 失败(第%d次)，正在重试: %s",
-                            func.__name__, attempt, e,
+                            "%s 失败(第%d次)，%d秒后重试: %s",
+                            func.__name__, attempt, delay, e,
                         )
+                        time.sleep(delay)
                     else:
                         logger.error(
                             "%s 失败 %d 次，放弃: %s",

src/smtp_client.py

@@ -0,0 +1,67 @@
+import smtplib
+import logging
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+from email.mime.base import MIMEBase
+from email import encoders
+from pathlib import Path
+from src.config import EmailAccount
+from src.retry import retry
+
+logger = logging.getLogger(__name__)
+
+
+@retry()
+def send_reply(account: EmailAccount, to_addr: str, subject: str, body: str,
+               attachments: list[tuple[str, str]] | None = None):
+    """发送回复邮件，支持附件。
+
+    Args:
+        attachments: [(本地文件路径, mime_type), ...] 列表
+    """
+    if not account.smtp:
+        raise RuntimeError(f"账号 {account.username} 未配置 SMTP")
+
+    s = account.smtp
+    logger.info("SMTP 连接: %s:%d (ssl=%s)", s.server, s.port, s.ssl)
+    if s.ssl:
+        conn = smtplib.SMTP_SSL(s.server, s.port, timeout=15)
+    else:
+        conn = smtplib.SMTP(s.server, s.port, timeout=15)
+        if s.starttls:
+            conn.starttls()
+
+    conn.login(account.username, account.password)
+    logger.info("SMTP 登录成功")
+
+    reply_subject = subject if subject.lower().startswith("re:") else f"Re: {subject}"
+    logger.info("SMTP 发送: to=%s subject=%s body=%d 字符 attachments=%d",
+                to_addr, reply_subject, len(body), len(attachments or []))
+
+    if attachments:
+        msg = MIMEMultipart()
+        msg.attach(MIMEText(body, "plain", "utf-8"))
+        for filepath, mime_type in attachments:
+            p = Path(filepath)
+            if not p.exists():
+                logger.warning("  附件不存在: %s", filepath)
+                continue
+            maintype, subtype = mime_type.split("/", 1)
+            with open(p, "rb") as f:
+                part = MIMEBase(maintype, subtype)
+                part.set_payload(f.read())
+            encoders.encode_base64(part)
+            part.add_header("Content-Disposition", "attachment", filename=p.name)
+            msg.attach(part)
+            logger.info("  添加附件: %s (%s, %d bytes)", p.name, mime_type, p.stat().st_size)
+    else:
+        msg = MIMEMultipart("alternative")
+        msg.attach(MIMEText(body, "plain", "utf-8"))
+
+    msg["From"] = account.username
+    msg["To"] = to_addr
+    msg["Subject"] = reply_subject
+
+    conn.sendmail(account.username, [to_addr], msg.as_string())
+    conn.quit()
+    logger.info("SMTP 发送完成")

src/summarizer.py

@@ -1,43 +1,111 @@
 import logging
-from src.config import Config
-from src.email_client import fetch_unseen_emails, mark_as_seen
+import re
+import threading
+from typing import Generator, Optional
+from src.config import Config, UserConfig
+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_message, format_summary
+from src.tg_bot import send_summary, send_thinking, edit_message, format_summary
 
 logger = logging.getLogger(__name__)
 
+_MD_SPECIAL = re.compile(r"([_*\[\]()~`>#+\-=|{}.!\\])")
 
-def process_all(cfg: Config):
-    for acct in cfg.email_accounts:
+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(user: UserConfig, skip_uids: set[bytes] | None = None) -> Generator[tuple[int, Email], None, None]:
+    """轮询指定用户的所有邮箱"""
+    for idx, acct in enumerate(user.email_accounts):
         try:
-            _process_account(cfg, acct)
-        except Exception as e:
-            logger.error(f"处理邮箱 {acct.username} 时出错: {e}", exc_info=True)
-
-
-def _process_account(cfg: Config, acct):
-    logger.info(f"检查邮箱: {acct.username}")
-    emails = fetch_unseen_emails(acct)
-
-    if not emails:
-        logger.info(f"  没有新邮件")
-        return
-
-    logger.info(f"  发现 {len(emails)} 封新邮件")
-    seen_uids = []
-
+            logger.info("检查邮箱: %s (user=%s)", acct.username, user.chat_id)
+            emails = fetch_unseen_emails(acct, skip_uids=skip_uids)
+            if emails:
+                logger.info("  发现 %d 封新邮件", len(emails))
             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_message(cfg.telegram, text)
-            seen_uids.append(mail.uid)
-            logger.info(f"  已发送到 Telegram")
+                yield idx, mail
         except Exception as e:
-            logger.error(f"  处理邮件 '{mail.subject}' 失败: {e}", exc_info=True)
-            seen_uids.append(mail.uid)
+            logger.error("轮询 %s 失败: %s", acct.username, e, exc_info=True)
 
-    if seen_uids:
-        mark_as_seen(acct, seen_uids)
+
+def ai_process(cfg: Config, user: UserConfig, acct_idx: int, mail: Email, tg_msg_id: int = 0) -> Optional[dict]:
+    logger.info("  [%s] 正在摘要: %s", user.chat_id, mail.subject)
+    label = _mail_label(mail)
+    # 阶段1：显示等待处理
+    if tg_msg_id:
+        edit_message(cfg.bot_token, user.chat_id, tg_msg_id, f"⏳ *等待处理…*\n{label}")
+    else:
+        tg_msg_id = send_thinking(cfg.bot_token, user.chat_id, f"⏳ *等待处理…*\n{label}")
+
+    # 阶段2：AI 处理，更新为思考中
+    edit_message(cfg.bot_token, user.chat_id, tg_msg_id, f"🤖 *AI思考中…*\n{label}")
+    summary = summarize_email(user.ai, mail.recipient, mail.subject, mail.sender, mail.body, mail.account_email)
+    summary["recipient"] = mail.recipient
+    if "@" not in summary.get("sender", ""):
+        summary["sender"] = mail.sender
+    text = format_summary(summary, mail.account_email)
+    # 阶段3：编辑为最终结果
+    return {
+        "text": text,
+        "data": summary,
+        "original_body": mail.body,
+        "original_sender": mail.sender,
+        "original_recipient": mail.recipient,
+        "original_reply_to": mail.reply_to,
+        "acct_idx": acct_idx,
+        "account_email": mail.account_email,
+        "uid": mail.uid,
+        "folder": mail.folder,
+        "thinking_msg_id": tg_msg_id if tg_msg_id else 0,
+    }
+
+
+def tg_send_and_mark(cfg: Config, user: UserConfig, info: dict):
+    # 找到用户对应的邮箱账户
+    acct = user.email_accounts[info["acct_idx"]]
+    thinking_msg_id = info.get("thinking_msg_id")
+    if thinking_msg_id:
+        from src.tg_bot import _summary_keyboard
+        from src.database import save_email_context
+        can_reply = info["data"].get("can_reply", True)
+        is_promotion = info["data"].get("category") == "promotion"
+        links = info["data"].get("links", [])
+        edit_message(cfg.bot_token, user.chat_id, thinking_msg_id,
+                     info["text"], _summary_keyboard(can_reply, is_promotion, links))
+        save_email_context(thinking_msg_id, {
+            "summary_text": info["text"],
+            "summary_data": info["data"],
+            "original_body": info["original_body"],
+            "plain_text": info["data"].get("plain_text", ""),
+            "original_sender": info.get("original_sender", ""),
+            "original_recipient": info.get("original_recipient", ""),
+            "original_reply_to": info.get("original_reply_to", ""),
+            "account_email": info.get("account_email", ""),
+            "account_idx": info["acct_idx"],
+            "sender": info["data"].get("sender", ""),
+            "subject": info["data"].get("subject", ""),
+            "can_reply": can_reply,
+        })
+    else:
+        send_summary(cfg.bot_token, user.chat_id,
+                     info["text"], info["data"],
+                     info["original_body"], info["acct_idx"],
+                     original_sender=info.get("original_sender", ""),
+                     original_recipient=info.get("original_recipient", ""),
+                     original_reply_to=info.get("original_reply_to", ""),
+                     account_email=info.get("account_email", ""))
+    # 标记已读放到后台线程
+    def _bg_mark():
+        try:
+            mark_as_seen(acct, [(info["uid"], info.get("folder", "INBOX"))])
+        except Exception:
+            pass
+    threading.Thread(target=_bg_mark, daemon=True).start()
+    logger.info("  [%s] TG 发送成功，标记已读已提交后台", user.chat_id)

src/tg_bot.py

@@ -1,58 +1,689 @@
+import json
+import logging
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from email.utils import parseaddr
 import requests
-from src.config import TelegramConfig
+from src.config import Config, UserConfig, find_user_by_chat_id
+from src.ai_client import generate_more_replies, expand_promo_detail
+from src.smtp_client import send_reply
 from src.retry import retry
+from src.database import save_email_context, load_email_context, save_conversation, load_conversation, delete_conversation
+
+logger = logging.getLogger(__name__)
+
+_ai_pool = ThreadPoolExecutor(max_workers=3, thread_name_prefix="ai_regen")
+
+
+# ── Public API ──────────────────────────────────────────
+
+@retry()
+def send_summary(bot_token: str, chat_id: str, summary_text: str,
+                 summary_data: dict, original_body: str, account_idx: int,
+                 original_sender: str = "", original_recipient: str = "",
+                 original_reply_to: str = "", account_email: str = "") -> int:
+    can_reply = summary_data.get("can_reply", True)
+    is_promotion = summary_data.get("category") == "promotion"
+    links = summary_data.get("links", [])
+    result = _tg_req(bot_token, "sendMessage", {
+        "chat_id": chat_id,
+        "text": summary_text,
+        "parse_mode": "MarkdownV2",
+        "reply_markup": _summary_keyboard(can_reply, is_promotion, links),
+    })
+    msg_id = result["result"]["message_id"]
+    save_email_context(msg_id, {
+        "summary_text": summary_text,
+        "summary_data": summary_data,
+        "original_body": original_body,
+        "original_sender": original_sender or summary_data.get("sender", ""),
+        "original_recipient": original_recipient or summary_data.get("recipient", ""),
+        "original_reply_to": original_reply_to,
+        "account_idx": account_idx,
+        "account_email": account_email,
+        "sender": summary_data.get("sender", ""),
+        "subject": summary_data.get("subject", ""),
+        "can_reply": can_reply,
+    })
+    return msg_id
 
 
 @retry()
-def send_message(tg_cfg: TelegramConfig, text: str):
-    url = f"https://api.telegram.org/bot{tg_cfg.bot_token}/sendMessage"
-    payload = {
-        "chat_id": tg_cfg.chat_id,
+def send_text(bot_token: str, chat_id: str, text: str,
+              reply_markup: dict = None) -> int:
+    payload = {"chat_id": chat_id, "text": text}
+    if reply_markup:
+        payload["reply_markup"] = reply_markup
+    result = _tg_req(bot_token, "sendMessage", payload)
+    return result["result"]["message_id"]
+
+
+def send_thinking(bot_token: str, chat_id: str, text: str = "💭 思考中…") -> int:
+    result = _tg_req(bot_token, "sendMessage", {
+        "chat_id": chat_id,
         "text": text,
-        "parse_mode": "MarkdownV2",
+    })
+    return result["result"]["message_id"]
+
+
+def edit_message(bot_token: str, chat_id: str, msg_id: int,
+                 text: str, reply_markup: dict = None):
+    payload = {
+        "chat_id": chat_id, "message_id": msg_id,
+        "text": text, "parse_mode": "MarkdownV2",
     }
-    resp = requests.post(url, json=payload, timeout=30)
+    if reply_markup:
+        payload["reply_markup"] = reply_markup
+    _tg_req(bot_token, "editMessageText", payload)
+
+
+def delete_message(bot_token: str, chat_id: str, msg_id: int):
+    try:
+        _tg_req(bot_token, "deleteMessage", {
+            "chat_id": chat_id, "message_id": msg_id,
+        })
+    except Exception:
+        pass
+
+
+def poll_telegram(bot_token: str, cfg: Config, last_update_id: int) -> int:
+    try:
+        resp = requests.get(
+            f"https://api.telegram.org/bot{bot_token}/getUpdates",
+            params={"offset": last_update_id, "timeout": 1},
+            timeout=10,
+        )
         resp.raise_for_status()
+        data = resp.json()
+        if not data.get("ok"):
+            return last_update_id
+        for update in data.get("result", []):
+            uid = update["update_id"]
+            try:
+                _handle_update(bot_token, cfg, update)
+            except Exception as e:
+                logger.warning("处理 update %d 失败: %s", uid, e)
+            last_update_id = uid + 1
+    except Exception as e:
+        logger.warning("Telegram polling error: %s", e)
+    return last_update_id
 
 
-_priority_icon = {"high": "🔴", "medium": "🟡", "low": "🟢"}
+# ── Formatting ──────────────────────────────────────────
+
+def format_summary(data: dict, account_email: str = "") -> str:
+    category = data.get("category", "normal")
+    if category == "promotion":
+        return _format_promotion(data, account_email)
+    if category == "notification" and data.get("verification_code"):
+        return _format_verification(data, account_email)
+    return _format_normal(data, account_email)
 
 
-def format_summary(data: dict) -> str:
-    priority = data.get("priority", "medium")
-    icon = _priority_icon.get(priority, "⚪")
-
+def _format_promotion(data: dict, account_email: str = "") -> str:
+    summary = data.get("summary", "")
     lines = [
-        f"*📧 {_escape(data.get('subject', '邮件摘要'))}*",
+        f"📢 *推广*",
         f"━━━━━━━━━━━━━━━━━━",
+        f"*收件账户:* {_escape(account_email)}",
+        "",
+        f"{_escape(summary)}",
+    ]
+    return "\n".join(lines)
+
+
+def _format_verification(data: dict, account_email: str = "") -> str:
+    code = data.get("verification_code", "")
+    app = data.get("app_name", "")
+    lines = [
+        f"🔑 *验证码*",
+        f"━━━━━━━━━━━━━━━━━━",
+        f"*收件账户:* {_escape(account_email)}",
+        f"*应用:* {_escape(app)}",
+        f"*验证码:* `{_escape(code)}`",
+    ]
+    return "\n".join(lines)
+
+
+def _format_normal(data: dict, account_email: str = "") -> str:
+    lines = [
+        f"📧 *{_escape(data.get('subject', '邮件摘要'))}*",
+        f"━━━━━━━━━━━━━━━━━━",
+        f"*收件账户:* {_escape(account_email)}",
         f"*收件人:* {_escape(data.get('recipient', '未知'))}",
         f"*发件人:* {_escape(data.get('sender', '未知'))}",
-        f"*优先级:* {icon} {priority.upper()}",
         "",
         _escape(data.get("summary", "")),
-        "",
     ]
-
-    if data.get("action_required"):
-        lines.append(f"*📌 需要处理:* 是")
-        items = data.get("action_items", [])
-        if items:
-            lines.append(f"*待办事项:*")
-            for item in items:
-                lines.append(f"  • {_escape(item)}")
-        lines.append("")
-
-    points = data.get("key_points", [])
-    if points:
-        lines.append(f"*关键要点:*")
-        for p in points:
-            lines.append(f"  • {_escape(p)}")
-
     return "\n".join(lines)
 
 
 def _escape(text: str) -> str:
-    special = "_*[]()~`>#+-=|{}.!"
-    for ch in special:
+    for ch in "_*[]()~`>#+-=|{}.!":
         text = text.replace(ch, f"\\{ch}")
     return text
+
+
+# ── Inline keyboards ────────────────────────────────────
+
+CALLBACK_VIEW_ORIG = "v"
+CALLBACK_VIEW_SUMM = "s"
+CALLBACK_REPLY = "r"
+CALLBACK_AI_REPLY_FULL = "af"
+CALLBACK_AI_REPLY_SHORT = "as"
+CALLBACK_SELECT_REPLY = "sr"
+CALLBACK_SEND_REPLY = "ss"  # 直接发送（不需要编辑）
+CALLBACK_REGEN = "rg"
+CALLBACK_HINT = "h"
+CALLBACK_CANCEL = "c"
+CALLBACK_PROMO_DETAIL = "pd"
+
+
+def _summary_keyboard(can_reply: bool = True, is_promotion: bool = False, links: list = None) -> dict:
+    if is_promotion:
+        return {"inline_keyboard": [
+            [{"text": "告诉我更多", "callback_data": CALLBACK_PROMO_DETAIL}],
+            [{"text": "查看原文", "callback_data": CALLBACK_VIEW_ORIG}],
+        ]}
+    kb = [
+        [
+            {"text": "查看原文", "callback_data": CALLBACK_VIEW_ORIG},
+            {"text": "回复", "callback_data": CALLBACK_REPLY},
+        ],
+    ]
+    if can_reply:
+        kb.append([
+            {"text": "AI完整回复", "callback_data": CALLBACK_AI_REPLY_FULL},
+            {"text": "AI简洁回复", "callback_data": CALLBACK_AI_REPLY_SHORT},
+        ])
+    for link in (links or []):
+        url = link.get("url", "")
+        text = link.get("text", "打开链接")[:30]
+        # 校验 URL：合法 http/https，无多重协议头，长度合理
+        if url.startswith("http") and url.count("://") == 1 and len(url) < 200:
+            kb.append([{"text": f"🔗 {text}", "url": url}])
+    return {"inline_keyboard": kb}
+
+
+def _orig_keyboard(can_reply: bool = True) -> dict:
+    kb = [
+        [
+            {"text": "返回摘要", "callback_data": CALLBACK_VIEW_SUMM},
+            {"text": "回复", "callback_data": CALLBACK_REPLY},
+        ],
+    ]
+    if can_reply:
+        kb.append([
+            {"text": "AI完整回复", "callback_data": CALLBACK_AI_REPLY_FULL},
+            {"text": "AI简洁回复", "callback_data": CALLBACK_AI_REPLY_SHORT},
+        ])
+    return {"inline_keyboard": kb}
+
+
+def _cancel_keyboard() -> dict:
+    return {
+        "inline_keyboard": [
+            [{"text": "取消回复", "callback_data": CALLBACK_CANCEL}],
+        ]
+    }
+
+
+def _ai_reply_keyboard(msg_id: int, suggestions: list) -> dict:
+    kb = []
+    for i, s in enumerate(suggestions):
+        send_data = f"{CALLBACK_SEND_REPLY}|{msg_id}|{i}"
+        edit_data = f"{CALLBACK_SELECT_REPLY}|{msg_id}|{i}"
+        kb.append([
+            {"text": f"回复{i+1} 直接发送", "callback_data": send_data},
+            {"text": f"回复{i+1} 编辑", "callback_data": edit_data},
+        ])
+    kb.append([
+        {"text": "换一批", "callback_data": f"{CALLBACK_REGEN}|{msg_id}"},
+        {"text": "我想说：", "callback_data": f"{CALLBACK_HINT}|{msg_id}"},
+    ])
+    kb.append([{"text": "取消", "callback_data": CALLBACK_CANCEL}])
+    return {"inline_keyboard": kb}
+
+
+# ── Telegram API ────────────────────────────────────────
+
+def _tg_req(bot_token: str, method: str, payload: dict) -> dict:
+    url = f"https://api.telegram.org/bot{bot_token}/{method}"
+    logger.info("Telegram API %s: %s", method, {k: (v if k != "reply_markup" else "...") for k, v in payload.items()})
+    r = requests.post(url, json=payload, timeout=30)
+    if not r.ok:
+        logger.error("Telegram API %s 失败: %s %s", method, r.status_code, r.text[:500])
+    r.raise_for_status()
+    result = r.json()
+    if not result.get("ok"):
+        raise RuntimeError(f"Telegram API error: {result}")
+    return result
+
+
+# ── Update handling ─────────────────────────────────────
+
+def _do_regen(bot_token: str, cfg: Config, chat_id: int, msg_id: int, ctx: dict, conv: dict):
+    """在 AI 线程池中执行换一批，不阻塞按钮轮询"""
+    user = find_user_by_chat_id(cfg, chat_id)
+    if not user:
+        return
+    style = conv.get("reply_style", "short")
+    try:
+        new_suggestions = generate_more_replies(
+            user.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
+            conv.get("all_suggestions", []), style=style,
+        )
+    except Exception as e:
+        send_text(bot_token, str(chat_id), f"生成失败: {e}")
+        return
+    if new_suggestions:
+        conv["ai_suggestions"] = new_suggestions
+        conv["all_suggestions"].extend(new_suggestions)
+        save_conversation(chat_id, conv)
+        logger.info("[ai_regen] 新生成 %d 条回复 (style=%s)", len(new_suggestions), style)
+        _show_ai_suggestions(bot_token, chat_id, msg_id, new_suggestions)
+
+
+def _do_ai_hint(bot_token: str, cfg: Config, chat_id: int, conv: dict, ctx: dict, hint: str):
+    """在 AI 线程池中根据用户提示生成回复"""
+    user = find_user_by_chat_id(cfg, chat_id)
+    if not user:
+        return
+    style = conv.get("reply_style", "short")
+    try:
+        new_suggestions = generate_more_replies(
+            user.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
+            conv.get("all_suggestions", []), user_hint=hint, style=style,
+        )
+    except Exception as e:
+        send_text(bot_token, str(chat_id), f"生成失败: {e}")
+        delete_conversation(chat_id)
+        return
+    if new_suggestions:
+        conv["ai_suggestions"] = new_suggestions
+        conv["all_suggestions"].extend(new_suggestions)
+        conv["state"] = "ai_reply_selection"
+        conv.pop("prompt_msg_id", None)
+        save_conversation(chat_id, conv)
+        logger.info("[ai_hint] 根据提示生成 %d 条新回复", len(new_suggestions))
+        _show_ai_suggestions(bot_token, chat_id, conv["summary_msg_id"], new_suggestions)
+
+
+def _handle_update(bot_token: str, cfg: Config, update: dict):
+    if "callback_query" in update:
+        _handle_callback(bot_token, cfg, update["callback_query"])
+    elif "message" in update:
+        _handle_message(bot_token, cfg, update["message"])
+
+
+def _handle_callback(bot_token: str, cfg: Config, cb: dict):
+    chat_id = cb["message"]["chat"]["id"]
+    msg_id = cb["message"]["message_id"]
+    data = cb["data"]
+    cb_id = cb["id"]
+
+    try:
+        _tg_req(bot_token, "answerCallbackQuery", {"callback_query_id": cb_id})
+    except Exception:
+        pass
+
+    parts = data.split("|")
+    action = parts[0]
+
+    ctx = load_email_context(msg_id)
+    subject = ctx["subject"] if ctx else "?"
+    logger.info("TG 回调: action=%s msg_id=%d subject=%s", action, msg_id, subject)
+
+    if action == CALLBACK_VIEW_ORIG and ctx:
+        can_reply = ctx.get("can_reply", True)
+        text = f"*📄 原文 \\- {_escape(ctx['subject'])}*\n\n{_escape(ctx.get('plain_text', ctx['original_body']))}"
+        _tg_req(bot_token, "editMessageText", {
+            "chat_id": chat_id, "message_id": msg_id,
+            "text": text, "parse_mode": "MarkdownV2",
+            "reply_markup": _orig_keyboard(can_reply),
+        })
+        logger.info("  切换为原文视图")
+
+    elif action == CALLBACK_VIEW_SUMM and ctx:
+        can_reply = ctx.get("can_reply", True)
+        is_promotion = ctx.get("summary_data", {}).get("category") == "promotion"
+        links = ctx.get("summary_data", {}).get("links", [])
+        _tg_req(bot_token, "editMessageText", {
+            "chat_id": chat_id, "message_id": msg_id,
+            "text": ctx["summary_text"], "parse_mode": "MarkdownV2",
+            "reply_markup": _summary_keyboard(can_reply, is_promotion, links),
+        })
+        logger.info("  切换回摘要视图")
+
+    elif action == CALLBACK_REPLY and ctx:
+        prompt_msg_id = send_text(
+            bot_token, str(chat_id), "请输入你的回复内容：",
+            reply_markup=_cancel_keyboard(),
+        )
+        save_conversation(chat_id, {
+            "state": "awaiting_reply", "summary_msg_id": msg_id,
+            "prompt_msg_id": prompt_msg_id,
+        })
+        logger.info("  进入回复流程, prompt_msg_id=%d", prompt_msg_id)
+
+    elif action in (CALLBACK_AI_REPLY_FULL, CALLBACK_AI_REPLY_SHORT) and ctx:
+        style = "full" if action == CALLBACK_AI_REPLY_FULL else "short"
+        # 总是用对应风格重新生成，不使用摘要时的通用缓存
+        logger.info("  重新生成 AI 回复 (style=%s)", style)
+        _tg_req(bot_token, "editMessageText", {
+            "chat_id": chat_id, "message_id": msg_id,
+            "text": "🤖 *AI思考中…*", "parse_mode": "MarkdownV2",
+        })
+        conv = load_conversation(chat_id) or {}
+        conv["state"] = "ai_reply_selection"
+        conv["summary_msg_id"] = msg_id
+        conv["reply_style"] = style
+        conv["all_suggestions"] = []
+        save_conversation(chat_id, conv)
+        _ai_pool.submit(_do_regen, bot_token, cfg, chat_id, msg_id, ctx, conv)
+
+    elif action == CALLBACK_SEND_REPLY and ctx:
+        idx = int(parts[2])
+        conv = load_conversation(chat_id)
+        if not conv or conv.get("summary_msg_id") != msg_id:
+            return
+        suggestions = conv.get("ai_suggestions", [])
+        if idx >= len(suggestions):
+            return
+        sel_text = suggestions[idx]["text"]
+        logger.info("  直接发送回复 #%d: %s", idx, sel_text[:60])
+        _do_send_reply(bot_token, cfg, chat_id, msg_id, ctx, sel_text)
+        send_text(bot_token, str(chat_id), f"✅ 回复已发送")
+        delete_conversation(chat_id)
+        can_reply = ctx.get("can_reply", True)
+        is_promotion = ctx.get("summary_data", {}).get("category") == "promotion"
+        links = ctx.get("summary_data", {}).get("links", [])
+        _tg_req(bot_token, "editMessageText", {
+            "chat_id": chat_id, "message_id": msg_id,
+            "text": ctx["summary_text"], "parse_mode": "MarkdownV2",
+            "reply_markup": _summary_keyboard(can_reply, is_promotion, links),
+        })
+
+    elif action == CALLBACK_SELECT_REPLY and ctx:
+        idx = int(parts[2])
+        conv = load_conversation(chat_id)
+        if not conv or conv.get("summary_msg_id") != msg_id:
+            return
+        suggestions = conv.get("ai_suggestions", [])
+        if idx >= len(suggestions):
+            return
+        sel = suggestions[idx]
+        sel_text = sel["text"]
+        logger.info("  选择回复 #%d: %s", idx, sel_text[:60])
+
+        # 用代码块显示，点击可复制（MarkdownV2）
+        escaped = sel_text.replace("\\", "\\\\").replace("`", "\\`")
+        _tg_req(bot_token, "sendMessage", {
+            "chat_id": str(chat_id),
+            "text": f"*建议回复（点击代码块复制）:*\n`{escaped}`",
+            "parse_mode": "MarkdownV2",
+        })
+
+        # 进入 awaiting_reply 状态，用户可编辑后发送
+        prompt_msg_id = send_text(
+            bot_token, str(chat_id),
+            "请编辑后发送（或直接回复附件）：",
+            reply_markup=_cancel_keyboard(),
+        )
+        conv["state"] = "awaiting_reply"
+        conv["summary_msg_id"] = msg_id
+        conv["prompt_msg_id"] = prompt_msg_id
+        conv["draft_text"] = sel_text
+        save_conversation(chat_id, conv)
+        logger.info("  建议已展示，等待用户编辑发送")
+
+    elif action == CALLBACK_REGEN and ctx:
+        conv = load_conversation(chat_id)
+        if not conv or conv.get("summary_msg_id") != msg_id:
+            return
+        logger.info("  换一批 (已有 %d 条历史)", len(conv.get("all_suggestions", [])))
+        # AI 处理中
+        _tg_req(bot_token, "editMessageText", {
+            "chat_id": chat_id, "message_id": msg_id,
+            "text": "🤖 *AI思考中…*", "parse_mode": "MarkdownV2",
+        })
+        # 丢到 AI 线程池处理，不阻塞按钮轮询
+        _ai_pool.submit(_do_regen, bot_token, cfg, chat_id, msg_id, ctx, conv)
+
+    elif action == CALLBACK_HINT and ctx:
+        prompt_msg_id = send_text(
+            bot_token, str(chat_id), "请输入你对回复的提示/要求：",
+            reply_markup=_cancel_keyboard(),
+        )
+        conv = load_conversation(chat_id)
+        if conv and conv.get("summary_msg_id") == msg_id:
+            conv["state"] = "awaiting_ai_hint"
+            conv["prompt_msg_id"] = prompt_msg_id
+            save_conversation(chat_id, conv)
+            logger.info("  等待用户输入 AI 提示")
+
+    elif action == CALLBACK_CANCEL:
+        delete_conversation(chat_id)
+        # 清除回复建议缓存，重新选择时会重新生成
+        if ctx:
+            ctx["summary_data"]["reply_suggestions"] = []
+            save_email_context(msg_id, ctx)
+            can_reply = ctx.get("can_reply", True)
+            is_promotion = ctx.get("summary_data", {}).get("category") == "promotion"
+            links = ctx.get("summary_data", {}).get("links", [])
+            _tg_req(bot_token, "editMessageText", {
+                "chat_id": chat_id, "message_id": msg_id,
+                "text": ctx["summary_text"], "parse_mode": "MarkdownV2",
+                "reply_markup": _summary_keyboard(can_reply, is_promotion, links),
+            })
+            logger.info("  取消, 恢复摘要视图")
+        else:
+            delete_message(bot_token, str(chat_id), msg_id)
+            logger.info("  取消, 删除提示消息")
+
+    elif action == CALLBACK_PROMO_DETAIL and ctx:
+        _tg_req(bot_token, "editMessageText", {
+            "chat_id": chat_id, "message_id": msg_id,
+            "text": "🤖 *AI思考中…*", "parse_mode": "MarkdownV2",
+        })
+        try:
+            details = expand_promo_detail(
+                cfg.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
+            )
+        except Exception as e:
+            send_text(bot_token, str(chat_id), f"获取详情失败: {e}")
+            return
+        summary = ctx["summary_data"].get("summary", "")
+        detail_lines = "\n".join(f"\\- {_escape(d)}" for d in details) if details else "暂无更多详情"
+        expanded = (
+            f"📢 *推广*\n"
+            f"━━━━━━━━━━━━━━━━━━\n"
+            f"*收件账户:* {_escape(ctx.get('account_email', ''))}\n"
+            f"\n{_escape(summary)}\n"
+            f"\n*📋 详情:*\n{detail_lines}"
+        )
+        _tg_req(bot_token, "editMessageText", {
+            "chat_id": chat_id, "message_id": msg_id,
+            "text": expanded, "parse_mode": "MarkdownV2",
+            "reply_markup": {"inline_keyboard": [
+                [{"text": "收起", "callback_data": CALLBACK_VIEW_SUMM}],
+                [{"text": "查看原文", "callback_data": CALLBACK_VIEW_ORIG}],
+            ]},
+        })
+        logger.info("  显示推广详情")
+
+
+def _handle_message(bot_token: str, cfg: Config, msg: dict):
+    chat_id = msg["chat"]["id"]
+    text = msg.get("text") or msg.get("caption") or ""
+    text = text.strip()
+    conv = load_conversation(chat_id)
+
+    if not conv:
+        return
+
+    user_msg_id = msg["message_id"]
+    chat_id_str = str(chat_id)
+    prompt_msg_id = conv.get("prompt_msg_id")
+    state = conv.get("state")
+    media = _extract_media(msg)
+
+    logger.info("TG 消息: state=%s text=%s media=%d", state, text[:60] if text else "(无)", len(media))
+
+    if state == "awaiting_reply":
+        ctx = load_email_context(conv["summary_msg_id"])
+        if not ctx:
+            delete_conversation(chat_id)
+            delete_message(bot_token, chat_id_str, user_msg_id)
+            if prompt_msg_id:
+                delete_message(bot_token, chat_id_str, prompt_msg_id)
+            logger.info("  上下文已丢失，清理")
+            return
+
+        delete_message(bot_token, chat_id_str, user_msg_id)
+        if prompt_msg_id:
+            delete_message(bot_token, chat_id_str, prompt_msg_id)
+
+        # 合并累积的附件
+        all_attachments = conv.get("pending_attachments", [])
+        all_attachments.extend(media)
+
+        _do_send_reply(bot_token, cfg, chat_id, conv["summary_msg_id"], ctx, text,
+                       attachments=all_attachments or None)
+        attach_info = f" (+{len(all_attachments)}个附件)" if all_attachments else ""
+        send_text(bot_token, chat_id_str, f"✅ 回复已发送{attach_info}")
+        delete_conversation(chat_id)
+        logger.info("  回复发送完成")
+
+    elif state == "awaiting_ai_hint":
+        ctx = load_email_context(conv["summary_msg_id"])
+        if not ctx:
+            delete_conversation(chat_id)
+            delete_message(bot_token, chat_id_str, user_msg_id)
+            if prompt_msg_id:
+                delete_message(bot_token, chat_id_str, prompt_msg_id)
+            return
+
+        delete_message(bot_token, chat_id_str, user_msg_id)
+        if prompt_msg_id:
+            delete_message(bot_token, chat_id_str, prompt_msg_id)
+
+        # 更新消息为 AI 思考中
+        _tg_req(bot_token, "editMessageText", {
+            "chat_id": chat_id, "message_id": conv["summary_msg_id"],
+            "text": "🤖 *AI思考中…*", "parse_mode": "MarkdownV2",
+        })
+        # 丢到 AI 线程池处理
+        _ai_pool.submit(_do_ai_hint, bot_token, cfg, chat_id, conv, ctx, text)
+
+
+def _show_ai_suggestions(bot_token: str, chat_id: int, msg_id: int,
+                         suggestions: list):
+    text = "*AI 建议回复，请选择：*\n\n"
+    for i, s in enumerate(suggestions, 1):
+        text += f"{i}\\. {_escape(s['text'])}\n"
+    _tg_req(bot_token, "editMessageText", {
+        "chat_id": chat_id, "message_id": msg_id,
+        "text": text, "parse_mode": "MarkdownV2",
+        "reply_markup": _ai_reply_keyboard(msg_id, suggestions),
+    })
+
+
+def _download_tg_file(bot_token: str, file_id: str, filename: str) -> str:
+    """从 Telegram 下载文件到本地 media/ 目录，返回本地路径"""
+    import os
+    from pathlib import Path
+    media_dir = Path("media")
+    media_dir.mkdir(exist_ok=True)
+    resp = requests.get(
+        f"https://api.telegram.org/bot{bot_token}/getFile",
+        params={"file_id": file_id}, timeout=30,
+    )
+    resp.raise_for_status()
+    file_path = resp.json()["result"]["file_path"]
+    dl = requests.get(
+        f"https://api.telegram.org/file/bot{bot_token}/{file_path}",
+        timeout=60,
+    )
+    dl.raise_for_status()
+    dest = media_dir / filename
+    dest.write_bytes(dl.content)
+    logger.info("  下载附件: %s (%d bytes)", filename, len(dl.content))
+    return str(dest)
+
+
+def _extract_media(msg: dict) -> list[dict]:
+    """从 Telegram 消息中提取附件信息列表 [{file_id, filename, mime_type}]"""
+    attachments = []
+    # 照片取最大尺寸
+    if "photo" in msg:
+        photo = max(msg["photo"], key=lambda p: p.get("file_size", 0))
+        ext = ".jpg"
+        attachments.append({
+            "file_id": photo["file_id"],
+            "filename": f"photo_{photo['file_id'][:8]}{ext}",
+            "mime_type": "image/jpeg",
+        })
+    if "video" in msg:
+        v = msg["video"]
+        attachments.append({
+            "file_id": v["file_id"],
+            "filename": v.get("file_name", f"video_{v['file_id'][:8]}.mp4"),
+            "mime_type": v.get("mime_type", "video/mp4"),
+        })
+    if "document" in msg:
+        d = msg["document"]
+        attachments.append({
+            "file_id": d["file_id"],
+            "filename": d.get("file_name", f"doc_{d['file_id'][:8]}"),
+            "mime_type": d.get("mime_type", "application/octet-stream"),
+        })
+    if "voice" in msg:
+        vc = msg["voice"]
+        attachments.append({
+            "file_id": vc["file_id"],
+            "filename": f"voice_{vc['file_id'][:8]}.ogg",
+            "mime_type": vc.get("mime_type", "audio/ogg"),
+        })
+    if "sticker" in msg:
+        s = msg["sticker"]
+        if not s.get("is_animated"):
+            attachments.append({
+                "file_id": s["file_id"],
+                "filename": f"sticker_{s['file_id'][:8]}.webp",
+                "mime_type": "image/webp",
+            })
+    return attachments
+
+
+def _do_send_reply(bot_token: str, cfg: Config,
+                   chat_id: int, msg_id: int, ctx: dict, reply_text: str,
+                   attachments: list[dict] | None = None):
+    user = find_user_by_chat_id(cfg, chat_id)
+    if not user:
+        send_text(bot_token, str(chat_id), "❌ 用户未找到。")
+        return
+    acct = user.email_accounts[ctx["account_idx"]]
+    if not acct.smtp:
+        send_text(bot_token, str(chat_id), "❌ 该邮箱未配置 SMTP，无法发送回复。")
+        return
+    to_addr = parseaddr(ctx.get("original_reply_to", ""))[1]
+    if not to_addr:
+        to_addr = parseaddr(ctx["original_sender"])[1]
+    if not to_addr:
+        to_addr = parseaddr(ctx["sender"])[1]
+    if not to_addr:
+        to_addr = ctx["sender"]
+    subject = ctx["subject"]
+
+    # 下载 Telegram 附件到本地
+    local_files = []
+    if attachments:
+        for att in attachments:
+            try:
+                path = _download_tg_file(bot_token, att["file_id"], att["filename"])
+                local_files.append((path, att["mime_type"]))
+            except Exception as e:
+                logger.error("  附件下载失败: %s - %s", att["filename"], e)
+
+    send_reply(acct, to_addr, subject, reply_text, attachments=local_files or None)