import json import logging from email.utils import parseaddr import requests 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__) _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 = "", 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" is_verification = summary_data.get("category") == "notification" and summary_data.get("verification_code") plain_text = is_promotion or is_verification result = _tg_req(tg_cfg, "sendMessage", { "chat_id": chat_id, "text": summary_text, "parse_mode": "MarkdownV2" if not plain_text else None, "reply_markup": _summary_keyboard(can_reply, is_promotion), }) 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", ""), "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(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, 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"๐Ÿ“ข {_escape(summary)}", f"่ดฆๆˆท: {_escape(account_email)}", ] return "\n".join(lines) def _format_verification(data: dict, account_email: str = "") -> str: code = data.get("verification_code", "") lines = [ f"๐Ÿ”‘ {_escape(code)}", f"่ดฆๆˆท: {_escape(account_email)}", ] return "\n".join(lines) def _format_normal(data: dict, account_email: str = "") -> str: priority = data.get("priority", "medium") icon = _priority_icon.get(priority, "โšช") lines = [ f"*๐Ÿ“ง {_escape(data.get('subject', '้‚ฎไปถๆ‘˜่ฆ'))}*", "โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”", f"*ๆ”ถไปถไบบ:* {_escape(data.get('recipient', 'ๆœช็Ÿฅ'))}", f"*ๅ‘ไปถไบบ:* {_escape(data.get('sender', 'ๆœช็Ÿฅ'))}", f"*่ดฆๆˆท:* {_escape(account_email)}", f"*ไผ˜ๅ…ˆ็บง:* {icon} {priority.upper()}", "", _escape(data.get("summary", "")), "", ] if data.get("action_required"): lines.append("*๐Ÿ“Œ ้œ€่ฆๅค„็†:* ๆ˜ฏ") for item in data.get("action_items", []): lines.append(f" \\- {_escape(item)}") lines.append("") for p in data.get("key_points", []): lines.append(f" \\- {_escape(p)}") 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 = "a" CALLBACK_SELECT_REPLY = "sr" CALLBACK_REGEN = "rg" CALLBACK_HINT = "h" CALLBACK_CANCEL = "c" def _summary_keyboard(can_reply: bool = True, is_promotion: bool = False) -> dict: if is_promotion: return {"inline_keyboard": [[{"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}]) 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.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"] send_reply(acct, to_addr, subject, reply_text)