690 lines
26 KiB
Python
690 lines
26 KiB
Python
import json
|
||
import logging
|
||
import threading
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from email.utils import parseaddr
|
||
import requests
|
||
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_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,
|
||
})
|
||
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",
|
||
}
|
||
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
|
||
|
||
|
||
# ── 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_promotion(data: dict, account_email: str = "") -> str:
|
||
summary = data.get("summary", "")
|
||
lines = [
|
||
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', '未知'))}",
|
||
"",
|
||
_escape(data.get("summary", "")),
|
||
]
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _escape(text: str) -> str:
|
||
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)
|