feat: include reply content in send confirmation message

- email_client: log IMAP connection, login, UNSEEN count, each email
  details (subject/sender/body size), mark-as-seen progress
- ai_client: log AI request params, timing, token usage, response size
- smtp_client: log SMTP connect, login, send details
- tg_bot: log all callback actions with subject context, message states
- main: periodic queue depth report (email_queue/tg_queue/pending_uids)

config.yaml.example

@@ -3,11 +3,19 @@ email_accounts:
     imap_port: 993
     username: "your_email@gmail.com"
     password: "your_app_password"
+    smtp:
+      server: "smtp.gmail.com"
+      port: 465
+      use_ssl: true
 
-  - imap_server: "imap.qq.com"
+  - imap_server: "imap.126.com"
     imap_port: 993
-    username: "your_email@qq.com"
+    username: "your_email@126.com"
-    password: "your_authorization_code"
+    password: "your_auth_code"
+    smtp:
+      server: "smtp.126.com"
+      port: 465
+      use_ssl: true
 
 ai:
   api_key: "sk-your_deepseek_api_key"

main.py

@@ -1,18 +1,39 @@
 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 src.summarizer import poll_accounts, ai_process, tg_send_and_mark
+from src.tg_bot import poll_telegram
 
-logging.basicConfig(
+# Thread-safe logging: all log records go through a single QueueListener thread
-    level=logging.INFO,
+_log_queue = queue_module.Queue(-1)
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+_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
+_pending_lock = threading.Lock()
+_pending_uids: set[bytes] = set()
+
+# Queue 1: raw emails → AI processor
+_email_queue: queue_module.Queue = queue_module.Queue()
+# Queue 2: AI summaries → TG sender
+_tg_queue: queue_module.Queue = queue_module.Queue()
 
 
 def _signal_handler(signum, frame):
@@ -21,6 +42,62 @@ def _signal_handler(signum, frame):
     _running = False
 
 
+def _email_poller(cfg):
+    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)
+                    _email_queue.put((acct_idx, mail))
+        if _running:
+            for _ in range(cfg.polling.interval_seconds):
+                if not _running:
+                    return
+                time.sleep(1)
+
+
+def _ai_processor(cfg):
+    logger.info("AI 处理线程已启动")
+    while _running:
+        try:
+            acct_idx, mail = _email_queue.get(timeout=1)
+        except queue_module.Empty:
+            continue
+        try:
+            info = ai_process(cfg, acct_idx, mail)
+            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)
+
+
+def _tg_worker(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)
+
+        try:
+            info = _tg_queue.get_nowait()
+            tg_send_and_mark(cfg, info)
+        except queue_module.Empty:
+            pass
+        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)
@@ -30,18 +107,30 @@ def main():
     cfg = load_config(cfg_path)
     logger.info(f"AI邮件摘要机器人已启动，轮询间隔: {cfg.polling.interval_seconds}s")
 
-    while _running:
+    threads = [
-        try:
+        threading.Thread(target=_email_poller, args=(cfg,), daemon=True),
-            process_all(cfg)
+        threading.Thread(target=_ai_processor, args=(cfg,), daemon=True),
-        except Exception as e:
+        threading.Thread(target=_tg_worker, args=(cfg,), daemon=True),
-            logger.error(f"轮询出错: {e}", exc_info=True)
+    ]
 
-        if _running:
+    for t in threads:
-            for _ in range(cfg.polling.interval_seconds):
+        t.start()
-                if not _running:
-                    break
-                time.sleep(1)
 
+    logger.info("所有线程已启动，进入主循环")
+    tick = 0
+    try:
+        while _running:
+            tick += 1
+            if tick % 30 == 0:
+                logger.info("队列状态: email_queue=%d tg_queue=%d pending_uids=%d",
+                            _email_queue.qsize(), _tg_queue.qsize(), len(_pending_uids))
+            time.sleep(1)
+    except KeyboardInterrupt:
+        _running = False
+
+    for t in threads:
+        t.join(timeout=5)
+    _log_listener.stop()
     logger.info("机器人已停止")
 
 

src/ai_client.py

@@ -1,9 +1,13 @@
 import json
+import logging
+import time
 from typing import Any
 import requests
 from src.config import AIConfig
 from src.retry import retry
 
+logger = logging.getLogger(__name__)
+
 SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以 JSON 格式返回结构化摘要。
 
 返回格式必须严格遵循以下 JSON schema：
@@ -15,15 +19,28 @@ SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以
   "priority": "high/medium/low",
   "action_required": true/false,
   "action_items": ["待办事项1", "待办事项2"],
-  "key_points": ["关键要点1", "关键要点2"]
+  "key_points": ["关键要点1", "关键要点2"],
+  "can_reply": true/false,
+  "reply_suggestions": [
+    {"text": "回复内容", "is_simple": true/false},
+    {"text": "回复内容", "is_simple": true/false},
+    {"text": "回复内容", "is_simple": true/false}
+  ]
 }
 
-其中 subject 字段不能照抄原邮件主题，必须是你总结生成的简短标题。
+其中：
+- subject 不能照抄原邮件主题，必须是你总结生成的简短标题
+- can_reply 为 false 表示无需回复或不适合建议回复（此时忽略 reply_suggestions）
+- is_simple 为 true 表示纯确认性回复（如"好的""收到"），用户选择后可直发
+- is_simple 为 false 表示涉及实质内容，需要用户确认再发送
+- 每条回复控制在 50 字以内
 只返回 JSON，不要包含任何其他文字。"""
 
 
 @retry()
 def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str, body: 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正文:\n{body[:4000]}"
 
     payload = {
@@ -35,18 +52,69 @@ def summarize_email(ai_cfg: AIConfig, recipient: str, subject: str, sender: str,
         "response_format": {"type": "json_object"},
     }
 
+    return _call_deepseek(ai_cfg, payload)
+
+
+MORE_REPLIES_PROMPT = """你是一个邮件回复助手。根据以下邮件内容生成3个完全不同的建议回复。
+
+以下是一些已经生成过的回复，请生成全新的回复，不要与已有回复重复。
+
+返回 JSON:
+{
+  "suggestions": [
+    {"text": "回复内容"},
+    {"text": "回复内容"},
+    {"text": "回复内容"}
+  ]
+}
+每条回复控制在 50 字以内。
+只返回 JSON，不要包含任何其他文字。"""
+
+
+@retry()
+def generate_more_replies(ai_cfg: AIConfig, sender: str, subject: str, body: str,
+                          previous_suggestions: list[dict], user_hint: str = "") -> list[dict]:
+    logger.info("AI 换批请求: sender=%s subject=%s user_hint=%s prev=%d 条",
+                sender, subject, user_hint or "(无)", len(previous_suggestions))
+    content = f"发件人: {sender}\n主题: {subject}\n正文:\n{body[:4000]}\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}"
+
+    payload = {
+        "model": ai_cfg.model,
+        "messages": [
+            {"role": "system", "content": MORE_REPLIES_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,
     )
+    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,23 @@
-from dataclasses import dataclass, field
+from dataclasses import dataclass
-from typing import List
+from typing import List, Optional
 import yaml
 
 
+@dataclass
+class SmtpConfig:
+    server: str
+    port: int
+    use_ssl: bool = True
+    use_starttls: bool = False
+
+
 @dataclass
 class EmailAccount:
     imap_server: str
     imap_port: int
     username: str
     password: str
+    smtp: Optional[SmtpConfig] = None
 
 
 @dataclass
@@ -41,7 +50,11 @@ def load_config(path: str) -> Config:
     with open(path, "r", encoding="utf-8") as f:
         raw = yaml.safe_load(f)
 
-    accounts = [EmailAccount(**a) for a in raw["email_accounts"]]
+    accounts = []
+    for a in raw["email_accounts"]:
+        if "smtp" in a and a["smtp"] is not None:
+            a["smtp"] = SmtpConfig(**a["smtp"])
+        accounts.append(EmailAccount(**a))
     ai = AIConfig(**raw["ai"])
     tg = TelegramConfig(**raw["telegram"])
     polling = PollingConfig(**raw["polling"])

src/email_client.py

@@ -1,11 +1,14 @@
 import imaplib
 import email
+import logging
 from email.header import decode_header
 from email.utils import parsedate_to_datetime
 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):
@@ -69,11 +72,15 @@ def _select_mailbox(conn, mailbox: str = "INBOX"):
 
 
 def _login_and_prepare(account: EmailAccount):
+    logger.info("连接 %s:%d", account.imap_server, account.imap_port)
     conn = imaplib.IMAP4_SSL(account.imap_server, account.imap_port)
     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)
     _select_mailbox(conn)
+    logger.info("已选择 INBOX")
     return conn
 
 
@@ -83,11 +90,13 @@ def fetch_unseen_emails(account: EmailAccount) -> list[Email]:
 
     _, data = conn.uid("SEARCH", None, "UNSEEN")
     uids = data[0].split() if data[0] else []
+    logger.info("UNSEEN 数量: %d", len(uids))
     emails = []
 
     for uid in uids:
         _, 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)
@@ -96,10 +105,13 @@ def fetch_unseen_emails(account: EmailAccount) -> list[Email]:
         sender = _decode_str(msg.get("From", ""))
         date_str = msg.get("Date", "")
         body = _get_text_from_msg(msg)
+        body_len = len(body)
 
+        logger.info("  邮件: [%s] %s <%s> (%d 字符)", subject, sender, date_str, body_len)
         emails.append(Email(uid=uid, subject=subject, sender=sender, body=body, date=date_str))
 
     conn.logout()
+    logger.info("共获取 %d 封新邮件", len(emails))
     return emails
 
 
@@ -107,7 +119,9 @@ def fetch_unseen_emails(account: EmailAccount) -> list[Email]:
 def mark_as_seen(account: EmailAccount, uids: list[bytes]):
     if not uids:
         return
+    logger.info("标记 %d 封邮件为已读", len(uids))
     conn = _login_and_prepare(account)
     for uid in uids:
         conn.uid("STORE", uid, "+FLAGS", "\\Seen")
     conn.logout()
+    logger.info("标记完成")

src/smtp_client.py

@@ -0,0 +1,39 @@
+import smtplib
+import logging
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+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):
+    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.use_ssl)
+    if s.use_ssl:
+        conn = smtplib.SMTP_SSL(s.server, s.port, timeout=15)
+    else:
+        conn = smtplib.SMTP(s.server, s.port, timeout=15)
+        if s.use_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 字符", to_addr, reply_subject, len(body))
+
+    msg = MIMEMultipart("alternative")
+    msg["From"] = account.username
+    msg["To"] = to_addr
+    msg["Subject"] = reply_subject
+    msg.attach(MIMEText(body, "plain", "utf-8"))
+
+    conn.sendmail(account.username, [to_addr], msg.as_string())
+    conn.quit()
+    logger.info("SMTP 发送完成")

src/summarizer.py

@@ -1,43 +1,47 @@
 import logging
+from typing import Generator, Optional
 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_message, format_summary
+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 acct in cfg.email_accounts:
+    for idx, acct in enumerate(cfg.email_accounts):
         try:
-            _process_account(cfg, acct)
+            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):
+def ai_process(cfg: Config, acct_idx: int, mail: Email) -> Optional[dict]:
-    logger.info(f"检查邮箱: {acct.username}")
+    acct = cfg.email_accounts[acct_idx]
-    emails = fetch_unseen_emails(acct)
+    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)
+    return {
+        "text": text,
+        "data": summary,
+        "original_body": mail.body,
+        "original_sender": mail.sender,
+        "acct_idx": acct_idx,
+        "uid": mail.uid,
+    }
 
-    if not emails:
-        logger.info(f"  没有新邮件")
-        return
 
-    logger.info(f"  发现 {len(emails)} 封新邮件")
+def tg_send_and_mark(cfg: Config, info: dict):
-    seen_uids = []
+    acct = cfg.email_accounts[info["acct_idx"]]
-
+    send_summary(cfg.telegram, cfg.telegram.chat_id,
-    for mail in emails:
+                 info["text"], info["data"],
-        try:
+                 info["original_body"], info["acct_idx"],
-            logger.info(f"  正在摘要: {mail.subject}")
+                 original_sender=info.get("original_sender", ""))
-            summary = summarize_email(cfg.ai, acct.username, mail.subject, mail.sender, mail.body)
+    mark_as_seen(acct, [info["uid"]])
-            summary["recipient"] = acct.username
+    logger.info(f"  已发送并标记已读")
-            text = format_summary(summary)
-            send_message(cfg.telegram, text)
-            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)

src/tg_bot.py

@@ -1,30 +1,99 @@
+import json
+import logging
+from email.utils import parseaddr
 import requests
-from src.config import TelegramConfig
+from src.config import TelegramConfig, Config
+from src.ai_client import generate_more_replies
+from src.smtp_client import send_reply
 from src.retry import retry
 
-
+logger = logging.getLogger(__name__)
-@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,
-        "text": text,
-        "parse_mode": "MarkdownV2",
-    }
-    resp = requests.post(url, json=payload, timeout=30)
-    resp.raise_for_status()
-
 
 _priority_icon = {"high": "🔴", "medium": "🟡", "low": "🟢"}
 
+# msg_id -> email context
+_email_contexts: dict[int, dict] = {}
+# chat_id -> conversation state
+_conversations: dict[int, dict] = {}
+
+
+# ── Public API ──────────────────────────────────────────
+
+@retry()
+def send_summary(tg_cfg: TelegramConfig, chat_id: str, summary_text: str,
+                 summary_data: dict, original_body: str, account_idx: int,
+                 original_sender: str = "") -> int:
+    can_reply = summary_data.get("can_reply", True)
+    result = _tg_req(tg_cfg, "sendMessage", {
+        "chat_id": chat_id,
+        "text": summary_text,
+        "parse_mode": "MarkdownV2",
+        "reply_markup": _summary_keyboard(can_reply),
+    })
+    msg_id = result["result"]["message_id"]
+    _email_contexts[msg_id] = {
+        "summary_text": summary_text,
+        "summary_data": summary_data,
+        "original_body": original_body[:10000],
+        "original_sender": original_sender or summary_data.get("sender", ""),
+        "account_idx": account_idx,
+        "sender": summary_data.get("sender", ""),
+        "subject": summary_data.get("subject", ""),
+        "can_reply": can_reply,
+    }
+    return msg_id
+
+
+@retry()
+def send_text(tg_cfg: TelegramConfig, 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(tg_cfg, "sendMessage", payload)
+    return result["result"]["message_id"]
+
+
+def delete_message(tg_cfg: TelegramConfig, chat_id: str, msg_id: int):
+    try:
+        _tg_req(tg_cfg, "deleteMessage", {
+            "chat_id": chat_id, "message_id": msg_id,
+        })
+    except Exception:
+        pass
+
+
+def poll_telegram(tg_cfg: TelegramConfig, cfg: Config, last_update_id: int) -> int:
+    try:
+        resp = requests.get(
+            f"https://api.telegram.org/bot{tg_cfg.bot_token}/getUpdates",
+            params={"offset": last_update_id, "timeout": 10},
+            timeout=15,
+        )
+        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(tg_cfg, 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
+
+
+# ── Formatting ──────────────────────────────────────────
 
 def format_summary(data: dict) -> str:
     priority = data.get("priority", "medium")
     icon = _priority_icon.get(priority, "⚪")
-
     lines = [
         f"*📧 {_escape(data.get('subject', '邮件摘要'))}*",
-        f"━━━━━━━━━━━━━━━━━━",
+        "━━━━━━━━━━━━━━━━━━",
         f"*收件人:* {_escape(data.get('recipient', '未知'))}",
         f"*发件人:* {_escape(data.get('sender', '未知'))}",
         f"*优先级:* {icon} {priority.upper()}",
@@ -32,27 +101,315 @@ def format_summary(data: dict) -> str:
         _escape(data.get("summary", "")),
         "",
     ]
-
     if data.get("action_required"):
-        lines.append(f"*📌 需要处理:* 是")
+        lines.append("*📌 需要处理:* 是")
-        items = data.get("action_items", [])
+        for item in data.get("action_items", []):
-        if items:
+            lines.append(f"  \\- {_escape(item)}")
-            lines.append(f"*待办事项:*")
-            for item in items:
-                lines.append(f"  • {_escape(item)}")
         lines.append("")
-
+    for p in data.get("key_points", []):
-    points = data.get("key_points", [])
+        lines.append(f"  \\- {_escape(p)}")
-    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 "_*[]()~`>#+-=|{}.!":
-    for ch in special:
         text = text.replace(ch, f"\\{ch}")
     return text
+
+
+# ── Inline keyboards ────────────────────────────────────
+
+CALLBACK_VIEW_ORIG = "v"
+CALLBACK_VIEW_SUMM = "s"
+CALLBACK_REPLY = "r"
+CALLBACK_AI_REPLY = "a"
+CALLBACK_SELECT_REPLY = "sr"
+CALLBACK_REGEN = "rg"
+CALLBACK_HINT = "h"
+CALLBACK_CANCEL = "c"
+
+
+def _summary_keyboard(can_reply: bool = True) -> dict:
+    kb = [
+        [
+            {"text": "查看原文", "callback_data": CALLBACK_VIEW_ORIG},
+            {"text": "回复", "callback_data": CALLBACK_REPLY},
+        ],
+    ]
+    if can_reply:
+        kb.append([{"text": "AI回复", "callback_data": CALLBACK_AI_REPLY}])
+    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}])
+    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):
+        data = f"{CALLBACK_SELECT_REPLY}|{msg_id}|{i}"
+        kb.append([{"text": s["text"][:30], "callback_data": 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(tg_cfg: TelegramConfig, method: str, payload: dict) -> dict:
+    url = f"https://api.telegram.org/bot{tg_cfg.bot_token}/{method}"
+    r = requests.post(url, json=payload, timeout=30)
+    r.raise_for_status()
+    result = r.json()
+    if not result.get("ok"):
+        raise RuntimeError(f"Telegram API error: {result}")
+    return result
+
+
+# ── Update handling ─────────────────────────────────────
+
+def _handle_update(tg_cfg: TelegramConfig, cfg: Config, update: dict):
+    if "callback_query" in update:
+        _handle_callback(tg_cfg, cfg, update["callback_query"])
+    elif "message" in update and "text" in update["message"]:
+        _handle_message(tg_cfg, cfg, update["message"])
+
+
+def _handle_callback(tg_cfg: TelegramConfig, 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(tg_cfg, "answerCallbackQuery", {"callback_query_id": cb_id})
+    except Exception:
+        pass
+
+    parts = data.split("|")
+    action = parts[0]
+
+    ctx = _email_contexts.get(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['original_body'][:3500])}"
+        _tg_req(tg_cfg, "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)
+        _tg_req(tg_cfg, "editMessageText", {
+            "chat_id": chat_id, "message_id": msg_id,
+            "text": ctx["summary_text"], "parse_mode": "MarkdownV2",
+            "reply_markup": _summary_keyboard(can_reply),
+        })
+        logger.info("  切换回摘要视图")
+
+    elif action == CALLBACK_REPLY and ctx:
+        prompt_msg_id = send_text(
+            tg_cfg, str(chat_id), "请输入你的回复内容：",
+            reply_markup=_cancel_keyboard(),
+        )
+        _conversations[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 == CALLBACK_AI_REPLY and ctx:
+        suggestions = ctx["summary_data"].get("reply_suggestions", [])
+        if not suggestions:
+            send_text(tg_cfg, str(chat_id), "此邮件无需回复。")
+            logger.info("  AI 判定无需回复，隐藏")
+            return
+
+        _conversations[chat_id] = {
+            "state": "ai_reply_selection",
+            "summary_msg_id": msg_id,
+            "ai_suggestions": suggestions,
+            "all_suggestions": list(suggestions),
+        }
+        logger.info("  显示 AI 回复建议 (%d 条)", len(suggestions))
+        _show_ai_suggestions(tg_cfg, chat_id, msg_id, suggestions)
+
+    elif action == CALLBACK_SELECT_REPLY and ctx:
+        idx = int(parts[2])
+        conv = _conversations.get(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]
+        logger.info("  选择回复 #%d: %s", idx, sel["text"][:60])
+        _do_send_reply(tg_cfg, cfg, chat_id, msg_id, ctx, sel["text"])
+        send_text(tg_cfg, str(chat_id), f"✅ 已发送: {sel['text']}")
+        _conversations.pop(chat_id, None)
+        can_reply = ctx.get("can_reply", True)
+        _tg_req(tg_cfg, "editMessageText", {
+            "chat_id": chat_id, "message_id": msg_id,
+            "text": ctx["summary_text"], "parse_mode": "MarkdownV2",
+            "reply_markup": _summary_keyboard(can_reply),
+        })
+
+    elif action == CALLBACK_REGEN and ctx:
+        conv = _conversations.get(chat_id)
+        if not conv or conv.get("summary_msg_id") != msg_id:
+            return
+        logger.info("  换一批 (已有 %d 条历史)", len(conv.get("all_suggestions", [])))
+        try:
+            new_suggestions = generate_more_replies(
+                cfg.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
+                conv.get("all_suggestions", []),
+            )
+        except Exception as e:
+            send_text(tg_cfg, str(chat_id), f"生成失败: {e}")
+            return
+        if new_suggestions:
+            conv["ai_suggestions"] = new_suggestions
+            conv["all_suggestions"].extend(new_suggestions)
+            logger.info("  新生成 %d 条回复", len(new_suggestions))
+            _show_ai_suggestions(tg_cfg, chat_id, msg_id, new_suggestions)
+
+    elif action == CALLBACK_HINT and ctx:
+        prompt_msg_id = send_text(
+            tg_cfg, str(chat_id), "请输入你对回复的提示/要求：",
+            reply_markup=_cancel_keyboard(),
+        )
+        conv = _conversations.get(chat_id)
+        if conv and conv.get("summary_msg_id") == msg_id:
+            conv["state"] = "awaiting_ai_hint"
+            conv["prompt_msg_id"] = prompt_msg_id
+            logger.info("  等待用户输入 AI 提示")
+
+    elif action == CALLBACK_CANCEL:
+        _conversations.pop(chat_id, None)
+        if ctx:
+            can_reply = ctx.get("can_reply", True)
+            _tg_req(tg_cfg, "editMessageText", {
+                "chat_id": chat_id, "message_id": msg_id,
+                "text": ctx["summary_text"], "parse_mode": "MarkdownV2",
+                "reply_markup": _summary_keyboard(can_reply),
+            })
+            logger.info("  取消, 恢复摘要视图")
+        else:
+            delete_message(tg_cfg, str(chat_id), msg_id)
+            logger.info("  取消, 删除提示消息")
+
+
+def _handle_message(tg_cfg: TelegramConfig, cfg: Config, msg: dict):
+    chat_id = msg["chat"]["id"]
+    text = msg["text"].strip()
+    conv = _conversations.get(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")
+
+    logger.info("TG 消息: state=%s text=%s", state, text[:60])
+
+    if state == "awaiting_reply":
+        ctx = _email_contexts.get(conv["summary_msg_id"])
+        if not ctx:
+            _conversations.pop(chat_id, None)
+            delete_message(tg_cfg, chat_id_str, user_msg_id)
+            if prompt_msg_id:
+                delete_message(tg_cfg, chat_id_str, prompt_msg_id)
+            logger.info("  上下文已丢失，清理")
+            return
+
+        delete_message(tg_cfg, chat_id_str, user_msg_id)
+        if prompt_msg_id:
+            delete_message(tg_cfg, chat_id_str, prompt_msg_id)
+        _do_send_reply(tg_cfg, cfg, chat_id, conv["summary_msg_id"], ctx, text)
+        send_text(tg_cfg, chat_id_str, f"✅ 回复已发送: {text}")
+        _conversations.pop(chat_id, None)
+        logger.info("  回复发送完成")
+
+    elif state == "awaiting_ai_hint":
+        ctx = _email_contexts.get(conv["summary_msg_id"])
+        if not ctx:
+            _conversations.pop(chat_id, None)
+            delete_message(tg_cfg, chat_id_str, user_msg_id)
+            if prompt_msg_id:
+                delete_message(tg_cfg, chat_id_str, prompt_msg_id)
+            return
+
+        delete_message(tg_cfg, chat_id_str, user_msg_id)
+        if prompt_msg_id:
+            delete_message(tg_cfg, chat_id_str, prompt_msg_id)
+        try:
+            new_suggestions = generate_more_replies(
+                cfg.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
+                conv.get("all_suggestions", []), user_hint=text,
+            )
+        except Exception as e:
+            send_text(tg_cfg, chat_id_str, f"生成失败: {e}")
+            _conversations.pop(chat_id, None)
+            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)
+            logger.info("  根据提示生成 %d 条新回复", len(new_suggestions))
+            _show_ai_suggestions(tg_cfg, chat_id, conv["summary_msg_id"], new_suggestions)
+
+
+def _show_ai_suggestions(tg_cfg: TelegramConfig, 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(tg_cfg, "editMessageText", {
+        "chat_id": chat_id, "message_id": msg_id,
+        "text": text, "parse_mode": "MarkdownV2",
+        "reply_markup": _ai_reply_keyboard(msg_id, suggestions),
+    })
+
+
+def _do_send_reply(tg_cfg: TelegramConfig, cfg: Config,
+                   chat_id: int, msg_id: int, ctx: dict, reply_text: str):
+    acct = cfg.email_accounts[ctx["account_idx"]]
+    if not acct.smtp:
+        send_text(tg_cfg, str(chat_id), "❌ 该邮箱未配置 SMTP，无法发送回复。")
+        return
+    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"]
+    send_reply(acct, to_addr, subject, reply_text)