Compare commits

..

15 Commits

Author SHA1 Message Date
7a01b29655 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 2026-07-17 22:19:20 +08:00
c6180834bc chore: remove old YAML config files, keep config.example.json 2026-07-17 22:08:14 +08:00
f2b3793d4c feat: multi-user support - Config format: JSON with bot_token (global), users array - Each user has chat_id, AI config, email_accounts - Each email account has separate IMAP/SMTP with ssl/starttls - Config module: new dataclasses (UserConfig, ImapConfig, SmtpConfig) - find_user_by_chat_id() for callback dispatch - Main: email poller iterates all users, queues carry user object - Summarizer: accepts UserConfig, uses user-specific AI and chat_id - TG bot: all functions use bot_token string instead of TelegramConfig - Callback/message handlers look up user by chat_id - AI regen/hint threads use find_user_by_chat_id for AI config - Config example: config.example.json - Tests: 3/3 passing 2026-07-17 22:05:59 +08:00
9906f1675f chore: remove reply_suggestions from email summarization prompt AI reply suggestions are now generated on-demand when user clicks full/short reply buttons, using style-specific prompts. 2026-07-17 21:02:50 +08:00
8386ab3c89 fix: always regenerate AI replies when selecting style, ignore summary cache The cached reply_suggestions from summarization use a generic prompt, not matching full/short style. Now always triggers fresh generation. 2026-07-17 21:01:03 +08:00
1d70fe2491 fix: pass links/is_promotion to _summary_keyboard in cancel and send_reply handlers 2026-07-17 20:56:59 +08:00
bedf5b887c feat: 2-col reply buttons, direct send, language-matched prompts - Buttons: [回复N 直接发送] [回复N 编辑] in 2-column layout - CALLBACK_SEND_REPLY: send immediately without editing - Prompts enforce matching email language (no Chinese forced on English) - Full mode: greeting + comma only, no [Your Name] placeholder - Tests: 5/5 passing 2026-07-17 20:49:11 +08:00
6ce9a806ea fix: clear reply suggestions cache on cancel, regenerate when switching styles - Cancel now clears reply_suggestions from email_context - Selecting style with empty cache triggers fresh AI generation - Selecting style with cache uses cached suggestions - Tests: 3/3 passing 2026-07-17 20:33:09 +08:00
84f2428aeb fix: send AI suggestion as MarkdownV2 code block for copy-on-tap 2026-07-17 20:28:46 +08:00
51064e48aa feat: AI reply suggestions show as copyable code block, user edits before sending - Click suggestion -> show text in backtick code block (tap to copy) - Enter awaiting_reply state so user can edit/send like normal reply - No direct email sending on selection anymore - Tests: 3/3 passing 2026-07-17 20:23:59 +08:00
ab2bf7facb feat: AI reply modes - full (formal email) and short (concise) - Split 'AI回复' button into 'AI完整回复' and 'AI简洁回复' - Full mode: complete email format with greeting/signature, 300 char limit - Short mode: brief text message style, 50 char limit - Style persists in conversation state for regen/hint consistency - Tests: 6/6 passing 2026-07-17 20:14:48 +08:00
2a9be9896f fix: restore links when returning to summary view _summary_keyboard was called without links param in VIEW_SUMM handler, causing extracted links to disappear after viewing original text. 2026-07-17 20:10:18 +08:00
2f29e03cec fix: read Telegram caption field for media messages Telegram sends text in 'caption' field (not 'text') when message has media. Handle both cases: text or caption, with text taking priority. 2026-07-17 20:04:27 +08:00
b2e219ece5 fix: send reply immediately on media-only message, no accumulation - Remove accumulation logic: media-only now sends as reply directly - All 3 cases (text-only, media-only, text+media) send immediately - Tests: 4/4 passing 2026-07-17 19:53:38 +08:00
131e440d6b style: simplify media accumulation - silent accumulate, immediate send on text 2026-07-17 19:52:08 +08:00
10 changed files with 348 additions and 255 deletions

2
.gitignore vendored
View File

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

32
config.example.json Normal file
View File

@@ -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
}
}
]
}
]
}

View File

@@ -1,32 +0,0 @@
email_accounts:
- imap_server: "imap.gmail.com"
imap_port: 993
username: "your_email@gmail.com"
password: "your_app_password"
use_ssl: true
smtp:
server: "smtp.gmail.com"
port: 465
use_ssl: true
- imap_server: "imap.126.com"
imap_port: 993
username: "your_email@126.com"
password: "your_auth_code"
use_ssl: true
smtp:
server: "smtp.126.com"
port: 465
use_ssl: true
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

54
main.py
View File

@@ -6,12 +6,12 @@ import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from src.config import load_config
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
# Thread-safe logging: all log records go through a single QueueListener thread
# Thread-safe logging
_log_queue = queue_module.Queue(-1)
_queue_handler = logging.handlers.QueueHandler(_log_queue)
_console_handler = logging.StreamHandler()
@@ -31,9 +31,9 @@ logger = logging.getLogger("main")
_running = True
_shutdown_event = threading.Event()
# Queue 1: raw emails → AI processor
# 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
# Queue 2: AI summaries → TG sender (user, info_dict)
_tg_queue: queue_module.Queue = queue_module.Queue()
@@ -44,11 +44,14 @@ def _signal_handler(signum, frame):
_shutdown_event.set()
def _email_poller(cfg):
def _email_poller(cfg: Config):
logger.info("邮件轮询线程已启动")
while _running:
try:
for acct_idx, mail in poll_accounts(cfg):
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)
@@ -56,21 +59,21 @@ def _email_poller(cfg):
mark_fetched(uid_str, mail.account_email, mail.folder,
mail.subject, mail.sender)
try:
tg_msg_id = send_thinking(cfg.telegram, cfg.telegram.chat_id,
tg_msg_id = send_thinking(cfg.bot_token, user.chat_id,
f"📥 *获取邮件中…*\n{_mail_label(mail)}")
logger.info(" 获取邮件中消息已发送: msg_id=%d", tg_msg_id)
logger.info(" [%s] 获取邮件中消息已发送: msg_id=%d", user.chat_id, tg_msg_id)
except Exception as e:
logger.error(" 获取邮件中消息发送失败: %s", e)
logger.error(" [%s] 获取邮件中消息发送失败: %s", user.chat_id, e)
tg_msg_id = 0
_email_queue.put((acct_idx, mail, tg_msg_id))
_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_seconds)
_shutdown_event.wait(cfg.polling_interval)
def _ai_processor(cfg):
def _ai_processor(cfg: Config):
logger.info("AI 处理线程已启动 (并行 workers=10)")
max_workers = 10
with ThreadPoolExecutor(max_workers=max_workers) as executor:
@@ -78,21 +81,21 @@ def _ai_processor(cfg):
while _running:
while len(pending_futures) < max_workers:
try:
acct_idx, mail, tg_msg_id = _email_queue.get_nowait()
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, acct_idx, mail, tg_msg_id)
pending_futures[future] = (acct_idx, mail)
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:
acct_idx, mail = pending_futures.pop(future)
user, acct_idx, mail = pending_futures.pop(future)
try:
info = future.result()
if info:
_tg_queue.put(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)
@@ -101,24 +104,24 @@ def _ai_processor(cfg):
time.sleep(0.1 if pending_futures else 0.5)
def _tg_poller(cfg):
def _tg_poller(cfg: Config):
logger.info("TG 轮询线程已启动(按钮回调专用)")
last_update_id = 0
while _running:
try:
last_update_id = poll_telegram(cfg.telegram, cfg, last_update_id)
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):
def _tg_sender(cfg: Config):
logger.info("TG 发送线程已启动(邮件摘要专用)")
while _running:
try:
info = _tg_queue.get_nowait()
tg_send_and_mark(cfg, info)
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:
@@ -139,10 +142,13 @@ 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)
init_db()
logger.info(f"AI邮件摘要机器人已启动轮询间隔: {cfg.polling.interval_seconds}s")
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),

View File

@@ -8,7 +8,7 @@ from src.retry import retry
logger = logging.getLogger(__name__)
SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以 JSON 格式返回结构化摘要。所有输出subject、summary 等)必须使用简体中文,但 reply_suggestions智能回复请使用与邮件相同的语言以便用户直接回复
SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以 JSON 格式返回结构化摘要。所有输出subject、summary 等)必须使用简体中文。
返回格式必须严格遵循以下 JSON schema
{
@@ -21,12 +21,7 @@ SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以
"verification_code": "",
"app_name": "",
"links": [{"text": "按钮文字", "url": "https://..."}],
"can_reply": true/false,
"reply_suggestions": [
{"text": "回复内容", "is_simple": true/false},
{"text": "回复内容", "is_simple": true/false},
{"text": "回复内容", "is_simple": true/false}
]
"can_reply": true/false
}
其中:
@@ -38,11 +33,8 @@ SYSTEM_PROMPT = """你是一个邮件摘要助手。请分析邮件内容并以
- 退订链接提取规则推广邮件必须提取退订链接。退订链接通常在邮件最底部包含以下关键词之一unsubscribe、退订、取消订阅、opt out、manage preferences、email preferences。即使链接被包裹在 HTML 标签中,也要提取 href 属性里的完整 URL。如果邮件底部同时有多个退订相关链接只提取最主要的那个
- plain_text: 将邮件全文转换为可读的纯文本版本。去除所有HTML标签、CSS样式、脚本代码保留完整的文字内容和原始段落结构。不删减、不修改、不总结任何内容完整保留原文所有信息
- summary 要简洁,把需要知道的信息浓缩成一句话,不需要分条列出
- 当 category 为 "promotion"summary 直接写成一句话:「[发送方名称] 在推广 [推广的产品/服务/活动]」,例如「腾讯云在推广双十一云服务器折扣」。此时 can_reply 为 falsereply_suggestions 为空数组
- can_reply 为 false 表示无需回复或不适合建议回复(此时忽略 reply_suggestions
- is_simple 为 true 表示纯确认性回复(如"好的""收到"),用户选择后可直发
- is_simple 为 false 表示涉及实质内容,需要用户确认再发送
- 每条回复控制在 50 字以内
- 当 category 为 "promotion"summary 直接写成一句话:「[发送方名称] 在推广 [推广的产品/服务/活动]」,例如「腾讯云在推广双十一云服务器折扣」。此时 can_reply 为 false
- can_reply 为 false 表示无需回复或不适合建议回复
只返回 JSON不要包含任何其他文字。"""
@@ -86,7 +78,12 @@ def expand_promo_detail(ai_cfg: AIConfig, sender: str, subject: str, body: str)
return result.get("details", [])
MORE_REPLIES_PROMPT = """你是一个邮件回复助手。根据以下邮件内容生成3个完全不同的建议回复。
MORE_REPLIES_PROMPT_SHORT = """你是一个邮件回复助手。根据以下邮件内容生成3个完全不同的简洁回复。
规则:
- 简洁模式:直接说要点,不需要称呼、落款、邮件格式。像微信/短信一样简短。
- 必须使用与邮件相同的语言回复(英文邮件用英文,中文邮件用中文)。
- 不要添加任何占位符如 [Your Name]。
以下是一些已经生成过的回复,请生成全新的回复,不要与已有回复重复。
@@ -101,22 +98,44 @@ MORE_REPLIES_PROMPT = """你是一个邮件回复助手。根据以下邮件内
每条回复控制在 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 = "") -> list[dict]:
logger.info("AI 换批请求: sender=%s subject=%s user_hint=%s prev=%d",
sender, subject, user_hint or "(无)", len(previous_suggestions))
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": MORE_REPLIES_PROMPT},
{"role": "system", "content": prompt},
{"role": "user", "content": content},
],
"response_format": {"type": "json_object"},

View File

@@ -1,25 +1,30 @@
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import List, Optional
import yaml
import json
@dataclass
class ImapConfig:
server: str
port: int = 993
ssl: bool = True
starttls: bool = False
@dataclass
class SmtpConfig:
server: str
port: int
use_ssl: bool = True
use_starttls: bool = False
port: int = 465
ssl: bool = True
starttls: bool = False
@dataclass
class EmailAccount:
imap_server: str
imap_port: int
username: str
password: str
use_ssl: bool = True
use_starttls: bool = False
smtp: Optional[SmtpConfig] = None
imap: ImapConfig
smtp: SmtpConfig
@dataclass
@@ -30,40 +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 = []
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"])
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

View File

@@ -91,12 +91,13 @@ def _select_mailbox(conn, mailbox: str = "INBOX"):
def _login_and_prepare(account: EmailAccount, timeout: int = 30, select_inbox: bool = True):
logger.info("连接 %s:%d (超时%ds)", account.imap_server, account.imap_port, timeout)
if account.use_ssl:
conn = imaplib.IMAP4_SSL(account.imap_server, account.imap_port, timeout=timeout)
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(account.imap_server, account.imap_port, timeout=timeout)
if account.use_starttls:
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)

View File

@@ -23,12 +23,12 @@ def send_reply(account: EmailAccount, to_addr: str, subject: str, body: str,
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:
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.use_starttls:
if s.starttls:
conn.starttls()
conn.login(account.username, account.password)

View File

@@ -2,7 +2,7 @@ import logging
import re
import threading
from typing import Generator, Optional
from src.config import Config
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_summary, send_thinking, edit_message, format_summary
@@ -21,31 +21,32 @@ def _mail_label(mail: Email) -> str:
return f"{_escape_md(sender_short)} · {_escape_md(subject_short)}"
def poll_accounts(cfg: Config, skip_uids: set[bytes] | None = None) -> Generator[tuple[int, Email], None, None]:
for idx, acct in enumerate(cfg.email_accounts):
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:
logger.info(f"检查邮箱: {acct.username}")
logger.info("检查邮箱: %s (user=%s)", acct.username, user.chat_id)
emails = fetch_unseen_emails(acct, skip_uids=skip_uids)
if emails:
logger.info(f" 发现 {len(emails)} 封新邮件")
logger.info(" 发现 %d 封新邮件", len(emails))
for mail in emails:
yield idx, mail
except Exception as e:
logger.error(f"轮询 {acct.username} 失败: {e}", exc_info=True)
logger.error("轮询 %s 失败: %s", acct.username, e, exc_info=True)
def ai_process(cfg: Config, acct_idx: int, mail: Email, tg_msg_id: int = 0) -> Optional[dict]:
logger.info(f" 正在摘要: {mail.subject}")
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.telegram, cfg.telegram.chat_id, tg_msg_id, f"⏳ *等待处理…*\n{label}")
edit_message(cfg.bot_token, user.chat_id, tg_msg_id, f"⏳ *等待处理…*\n{label}")
else:
tg_msg_id = send_thinking(cfg.telegram, cfg.telegram.chat_id, f"⏳ *等待处理…*\n{label}")
tg_msg_id = send_thinking(cfg.bot_token, user.chat_id, f"⏳ *等待处理…*\n{label}")
# 阶段2AI 处理,更新为思考中
edit_message(cfg.telegram, cfg.telegram.chat_id, tg_msg_id, f"🤖 *AI思考中…*\n{label}")
summary = summarize_email(cfg.ai, mail.recipient, mail.subject, mail.sender, mail.body, mail.account_email)
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
@@ -66,19 +67,18 @@ def ai_process(cfg: Config, acct_idx: int, mail: Email, tg_msg_id: int = 0) -> O
}
def tg_send_and_mark(cfg: Config, info: dict):
acct = cfg.email_accounts[info["acct_idx"]]
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.telegram, cfg.telegram.chat_id, thinking_msg_id,
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"],
@@ -94,19 +94,18 @@ def tg_send_and_mark(cfg: Config, info: dict):
"can_reply": can_reply,
})
else:
# 兜底:没有思考中消息就正常发送
send_summary(cfg.telegram, cfg.telegram.chat_id,
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", ""))
# 标记已读放到后台线程,不阻塞 TG 发送
# 标记已读放到后台线程
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(f" TG 发送成功,标记已读已提交后台")
logger.info(" [%s] TG 发送成功,标记已读已提交后台", user.chat_id)

View File

@@ -4,7 +4,7 @@ import threading
from concurrent.futures import ThreadPoolExecutor
from email.utils import parseaddr
import requests
from src.config import TelegramConfig, Config
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
@@ -18,14 +18,14 @@ _ai_pool = ThreadPoolExecutor(max_workers=3, thread_name_prefix="ai_regen")
# ── Public API ──────────────────────────────────────────
@retry()
def send_summary(tg_cfg: TelegramConfig, chat_id: str, summary_text: str,
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(tg_cfg, "sendMessage", {
result = _tg_req(bot_token, "sendMessage", {
"chat_id": chat_id,
"text": summary_text,
"parse_mode": "MarkdownV2",
@@ -49,24 +49,24 @@ def send_summary(tg_cfg: TelegramConfig, chat_id: str, summary_text: str,
@retry()
def send_text(tg_cfg: TelegramConfig, chat_id: str, text: str,
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(tg_cfg, "sendMessage", payload)
result = _tg_req(bot_token, "sendMessage", payload)
return result["result"]["message_id"]
def send_thinking(tg_cfg: TelegramConfig, chat_id: str, text: str = "💭 思考中…") -> int:
result = _tg_req(tg_cfg, "sendMessage", {
def send_thinking(bot_token: str, chat_id: str, text: str = "💭 思考中…") -> int:
result = _tg_req(bot_token, "sendMessage", {
"chat_id": chat_id,
"text": text,
})
return result["result"]["message_id"]
def edit_message(tg_cfg: TelegramConfig, chat_id: str, msg_id: int,
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,
@@ -74,22 +74,22 @@ def edit_message(tg_cfg: TelegramConfig, chat_id: str, msg_id: int,
}
if reply_markup:
payload["reply_markup"] = reply_markup
_tg_req(tg_cfg, "editMessageText", payload)
_tg_req(bot_token, "editMessageText", payload)
def delete_message(tg_cfg: TelegramConfig, chat_id: str, msg_id: int):
def delete_message(bot_token: str, chat_id: str, msg_id: int):
try:
_tg_req(tg_cfg, "deleteMessage", {
_tg_req(bot_token, "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:
def poll_telegram(bot_token: str, cfg: Config, last_update_id: int) -> int:
try:
resp = requests.get(
f"https://api.telegram.org/bot{tg_cfg.bot_token}/getUpdates",
f"https://api.telegram.org/bot{bot_token}/getUpdates",
params={"offset": last_update_id, "timeout": 1},
timeout=10,
)
@@ -100,7 +100,7 @@ def poll_telegram(tg_cfg: TelegramConfig, cfg: Config, last_update_id: int) -> i
for update in data.get("result", []):
uid = update["update_id"]
try:
_handle_update(tg_cfg, cfg, update)
_handle_update(bot_token, cfg, update)
except Exception as e:
logger.warning("处理 update %d 失败: %s", uid, e)
last_update_id = uid + 1
@@ -169,8 +169,10 @@ def _escape(text: str) -> str:
CALLBACK_VIEW_ORIG = "v"
CALLBACK_VIEW_SUMM = "s"
CALLBACK_REPLY = "r"
CALLBACK_AI_REPLY = "a"
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"
@@ -190,7 +192,10 @@ def _summary_keyboard(can_reply: bool = True, is_promotion: bool = False, links:
],
]
if can_reply:
kb.append([{"text": "AI回复", "callback_data": CALLBACK_AI_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]
@@ -208,7 +213,10 @@ def _orig_keyboard(can_reply: bool = True) -> dict:
],
]
if can_reply:
kb.append([{"text": "AI回复", "callback_data": CALLBACK_AI_REPLY}])
kb.append([
{"text": "AI完整回复", "callback_data": CALLBACK_AI_REPLY_FULL},
{"text": "AI简洁回复", "callback_data": CALLBACK_AI_REPLY_SHORT},
])
return {"inline_keyboard": kb}
@@ -223,8 +231,12 @@ def _cancel_keyboard() -> dict:
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}])
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}"},
@@ -235,8 +247,8 @@ def _ai_reply_keyboard(msg_id: int, suggestions: list) -> dict:
# ── 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}"
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:
@@ -250,33 +262,41 @@ def _tg_req(tg_cfg: TelegramConfig, method: str, payload: dict) -> dict:
# ── Update handling ─────────────────────────────────────
def _do_regen(cfg: Config, chat_id: int, msg_id: int, ctx: dict, conv: dict):
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(
cfg.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
conv.get("all_suggestions", []),
user.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
conv.get("all_suggestions", []), style=style,
)
except Exception as e:
send_text(cfg.telegram, str(chat_id), f"生成失败: {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 条回复", len(new_suggestions))
_show_ai_suggestions(cfg.telegram, chat_id, msg_id, new_suggestions)
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(cfg: Config, chat_id: int, conv: dict, ctx: dict, hint: str):
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(
cfg.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
conv.get("all_suggestions", []), user_hint=hint,
user.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
conv.get("all_suggestions", []), user_hint=hint, style=style,
)
except Exception as e:
send_text(cfg.telegram, str(chat_id), f"生成失败: {e}")
send_text(bot_token, str(chat_id), f"生成失败: {e}")
delete_conversation(chat_id)
return
if new_suggestions:
@@ -286,24 +306,24 @@ def _do_ai_hint(cfg: Config, chat_id: int, conv: dict, ctx: dict, hint: str):
conv.pop("prompt_msg_id", None)
save_conversation(chat_id, conv)
logger.info("[ai_hint] 根据提示生成 %d 条新回复", len(new_suggestions))
_show_ai_suggestions(cfg.telegram, chat_id, conv["summary_msg_id"], new_suggestions)
_show_ai_suggestions(bot_token, chat_id, conv["summary_msg_id"], new_suggestions)
def _handle_update(tg_cfg: TelegramConfig, cfg: Config, update: dict):
def _handle_update(bot_token: str, cfg: Config, update: dict):
if "callback_query" in update:
_handle_callback(tg_cfg, cfg, update["callback_query"])
_handle_callback(bot_token, cfg, update["callback_query"])
elif "message" in update:
_handle_message(tg_cfg, cfg, update["message"])
_handle_message(bot_token, cfg, update["message"])
def _handle_callback(tg_cfg: TelegramConfig, cfg: Config, cb: dict):
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(tg_cfg, "answerCallbackQuery", {"callback_query_id": cb_id})
_tg_req(bot_token, "answerCallbackQuery", {"callback_query_id": cb_id})
except Exception:
pass
@@ -317,7 +337,7 @@ def _handle_callback(tg_cfg: TelegramConfig, cfg: Config, cb: dict):
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(tg_cfg, "editMessageText", {
_tg_req(bot_token, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": text, "parse_mode": "MarkdownV2",
"reply_markup": _orig_keyboard(can_reply),
@@ -326,16 +346,18 @@ def _handle_callback(tg_cfg: TelegramConfig, cfg: Config, cb: dict):
elif action == CALLBACK_VIEW_SUMM and ctx:
can_reply = ctx.get("can_reply", True)
_tg_req(tg_cfg, "editMessageText", {
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),
"reply_markup": _summary_keyboard(can_reply, is_promotion, links),
})
logger.info(" 切换回摘要视图")
elif action == CALLBACK_REPLY and ctx:
prompt_msg_id = send_text(
tg_cfg, str(chat_id), "请输入你的回复内容:",
bot_token, str(chat_id), "请输入你的回复内容:",
reply_markup=_cancel_keyboard(),
)
save_conversation(chat_id, {
@@ -344,21 +366,43 @@ def _handle_callback(tg_cfg: TelegramConfig, cfg: Config, cb: dict):
})
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
save_conversation(chat_id, {
"state": "ai_reply_selection",
"summary_msg_id": msg_id,
"ai_suggestions": suggestions,
"all_suggestions": list(suggestions),
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),
})
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])
@@ -369,33 +413,46 @@ def _handle_callback(tg_cfg: TelegramConfig, cfg: Config, cb: dict):
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']}")
delete_conversation(chat_id)
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),
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(tg_cfg, "editMessageText", {
_tg_req(bot_token, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": "🤖 *AI思考中…*", "parse_mode": "MarkdownV2",
})
# 丢到 AI 线程池处理,不阻塞按钮轮询
_ai_pool.submit(_do_regen, cfg, chat_id, msg_id, ctx, conv)
_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(
tg_cfg, str(chat_id), "请输入你对回复的提示/要求:",
bot_token, str(chat_id), "请输入你对回复的提示/要求:",
reply_markup=_cancel_keyboard(),
)
conv = load_conversation(chat_id)
@@ -407,20 +464,25 @@ def _handle_callback(tg_cfg: TelegramConfig, cfg: Config, cb: dict):
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)
_tg_req(tg_cfg, "editMessageText", {
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),
"reply_markup": _summary_keyboard(can_reply, is_promotion, links),
})
logger.info(" 取消, 恢复摘要视图")
else:
delete_message(tg_cfg, str(chat_id), msg_id)
delete_message(bot_token, str(chat_id), msg_id)
logger.info(" 取消, 删除提示消息")
elif action == CALLBACK_PROMO_DETAIL and ctx:
_tg_req(tg_cfg, "editMessageText", {
_tg_req(bot_token, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": "🤖 *AI思考中…*", "parse_mode": "MarkdownV2",
})
@@ -429,7 +491,7 @@ def _handle_callback(tg_cfg: TelegramConfig, cfg: Config, cb: dict):
cfg.ai, ctx["sender"], ctx["subject"], ctx["original_body"],
)
except Exception as e:
send_text(tg_cfg, str(chat_id), f"获取详情失败: {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 "暂无更多详情"
@@ -440,7 +502,7 @@ def _handle_callback(tg_cfg: TelegramConfig, cfg: Config, cb: dict):
f"\n{_escape(summary)}\n"
f"\n*📋 详情:*\n{detail_lines}"
)
_tg_req(tg_cfg, "editMessageText", {
_tg_req(bot_token, "editMessageText", {
"chat_id": chat_id, "message_id": msg_id,
"text": expanded, "parse_mode": "MarkdownV2",
"reply_markup": {"inline_keyboard": [
@@ -451,9 +513,10 @@ def _handle_callback(tg_cfg: TelegramConfig, cfg: Config, cb: dict):
logger.info(" 显示推广详情")
def _handle_message(tg_cfg: TelegramConfig, cfg: Config, msg: dict):
def _handle_message(bot_token: str, cfg: Config, msg: dict):
chat_id = msg["chat"]["id"]
text = msg.get("text", "").strip()
text = msg.get("text") or msg.get("caption") or ""
text = text.strip()
conv = load_conversation(chat_id)
if not conv:
@@ -471,38 +534,24 @@ def _handle_message(tg_cfg: TelegramConfig, cfg: Config, msg: dict):
ctx = load_email_context(conv["summary_msg_id"])
if not ctx:
delete_conversation(chat_id)
delete_message(tg_cfg, chat_id_str, user_msg_id)
delete_message(bot_token, chat_id_str, user_msg_id)
if prompt_msg_id:
delete_message(tg_cfg, chat_id_str, prompt_msg_id)
delete_message(bot_token, chat_id_str, prompt_msg_id)
logger.info(" 上下文已丢失,清理")
return
# 有附件但没文字:累积附件,等用户输入文字
if media and not text:
pending = conv.get("pending_attachments", [])
for att in media:
pending.append(att)
conv["pending_attachments"] = pending
save_conversation(chat_id, conv)
delete_message(tg_cfg, chat_id_str, user_msg_id)
send_text(tg_cfg, chat_id_str,
f"📎 已添加 {len(pending)} 个附件,请输入回复内容发送",
reply_markup=_cancel_keyboard())
return
# 有文字(可能也有附件):发送邮件
delete_message(tg_cfg, chat_id_str, user_msg_id)
delete_message(bot_token, chat_id_str, user_msg_id)
if prompt_msg_id:
delete_message(tg_cfg, chat_id_str, 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(tg_cfg, cfg, chat_id, conv["summary_msg_id"], ctx, text,
_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(tg_cfg, chat_id_str, f"✅ 回复已发送: {text}{attach_info}")
send_text(bot_token, chat_id_str, f"✅ 回复已发送{attach_info}")
delete_conversation(chat_id)
logger.info(" 回复发送完成")
@@ -510,50 +559,50 @@ def _handle_message(tg_cfg: TelegramConfig, cfg: Config, msg: dict):
ctx = load_email_context(conv["summary_msg_id"])
if not ctx:
delete_conversation(chat_id)
delete_message(tg_cfg, chat_id_str, user_msg_id)
delete_message(bot_token, chat_id_str, user_msg_id)
if prompt_msg_id:
delete_message(tg_cfg, chat_id_str, prompt_msg_id)
delete_message(bot_token, chat_id_str, prompt_msg_id)
return
delete_message(tg_cfg, chat_id_str, user_msg_id)
delete_message(bot_token, chat_id_str, user_msg_id)
if prompt_msg_id:
delete_message(tg_cfg, chat_id_str, prompt_msg_id)
delete_message(bot_token, chat_id_str, prompt_msg_id)
# 更新消息为 AI 思考中
_tg_req(tg_cfg, "editMessageText", {
_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, cfg, chat_id, conv, ctx, text)
_ai_pool.submit(_do_ai_hint, bot_token, cfg, chat_id, conv, ctx, text)
def _show_ai_suggestions(tg_cfg: TelegramConfig, chat_id: int, msg_id: int,
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(tg_cfg, "editMessageText", {
_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(tg_cfg: TelegramConfig, file_id: str, filename: str) -> str:
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{tg_cfg.bot_token}/getFile",
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{tg_cfg.bot_token}/{file_path}",
f"https://api.telegram.org/file/bot{bot_token}/{file_path}",
timeout=60,
)
dl.raise_for_status()
@@ -607,12 +656,16 @@ def _extract_media(msg: dict) -> list[dict]:
return attachments
def _do_send_reply(tg_cfg: TelegramConfig, cfg: Config,
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):
acct = cfg.email_accounts[ctx["account_idx"]]
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(tg_cfg, str(chat_id), "❌ 该邮箱未配置 SMTP无法发送回复。")
send_text(bot_token, str(chat_id), "❌ 该邮箱未配置 SMTP无法发送回复。")
return
to_addr = parseaddr(ctx.get("original_reply_to", ""))[1]
if not to_addr:
@@ -628,7 +681,7 @@ def _do_send_reply(tg_cfg: TelegramConfig, cfg: Config,
if attachments:
for att in attachments:
try:
path = _download_tg_file(tg_cfg, att["file_id"], att["filename"])
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)