diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py index 5d1bf90f..ca872b82 100644 --- a/app/modules/telegram/telegram.py +++ b/app/modules/telegram/telegram.py @@ -44,7 +44,6 @@ class Telegram: _bot_username: Optional[str] = None # Bot username for mention detection _typing_tasks: Dict[str, threading.Thread] = {} # chat_id -> typing任务 _typing_stop_flags: Dict[str, bool] = {} # chat_id -> 停止标志 - _message_payload_cache: Dict[str, Dict[str, Any]] = {} # message payload cache def __init__( self, @@ -82,7 +81,6 @@ class Telegram: _bot = TeleBot(self._telegram_token, parse_mode="MarkdownV2") # 记录句柄 self._bot = _bot - self._message_payload_cache = {} # 获取并存储bot用户名用于@检测 try: bot_info = _bot.get_me() @@ -365,66 +363,6 @@ class Telegram: if task and task.is_alive(): task.join(timeout=1) - @staticmethod - def _message_cache_key(chat_id: Union[str, int], message_id: Union[str, int]) -> str: - """ - 构造消息缓存键。 - """ - return f"{chat_id}:{message_id}" - - @staticmethod - def _serialize_reply_markup( - reply_markup: Optional[InlineKeyboardMarkup], - ) -> Optional[str]: - """ - 将 reply_markup 稳定序列化，用于判重。 - """ - if not reply_markup: - return None - if hasattr(reply_markup, "to_dict"): - return json.dumps( - reply_markup.to_dict(), ensure_ascii=False, sort_keys=True - ) - return json.dumps(reply_markup, ensure_ascii=False, sort_keys=True) - - def _build_message_payload_signature( - self, - text: str, - reply_markup: Optional[InlineKeyboardMarkup] = None, - image: Optional[str] = None, - disable_web_page_preview: Optional[bool] = None, - ) -> Dict[str, Any]: - """ - 构造 Telegram 消息内容签名，基于实际发送到 API 的标准化内容。 - """ - return { - "text": standardize(text) if text else "", - "reply_markup": self._serialize_reply_markup(reply_markup), - "image": image or None, - "disable_web_page_preview": disable_web_page_preview, - } - - def _remember_message_payload( - self, - chat_id: Union[str, int], - message_id: Union[str, int], - text: str, - reply_markup: Optional[InlineKeyboardMarkup] = None, - image: Optional[str] = None, - disable_web_page_preview: Optional[bool] = None, - ) -> None: - """ - 记录消息最近一次成功发送/编辑的内容签名。 - """ - self._message_payload_cache[self._message_cache_key(chat_id, message_id)] = ( - self._build_message_payload_signature( - text=text, - reply_markup=reply_markup, - image=image, - disable_web_page_preview=disable_web_page_preview, - ) - ) - def send_msg( self, title: str, @@ -510,21 +448,10 @@ class Telegram: ) self._stop_typing_task(chat_id) if sent and hasattr(sent, "message_id"): - target_chat_id = ( - sent.chat.id if hasattr(sent, "chat") else chat_id - ) - self._remember_message_payload( - chat_id=target_chat_id, - message_id=sent.message_id, - text=caption, - reply_markup=reply_markup, - image=image, - disable_web_page_preview=disable_web_page_preview, - ) return { "success": True, "message_id": sent.message_id, - "chat_id": target_chat_id, + "chat_id": sent.chat.id if hasattr(sent, "chat") else chat_id, } elif sent: return {"success": True} @@ -892,9 +819,6 @@ class Telegram: logger.info( f"成功删除Telegram消息: chat_id={target_chat_id}, message_id={message_id}" ) - self._message_payload_cache.pop( - self._message_cache_key(target_chat_id, message_id), None - ) return True else: logger.error( @@ -973,25 +897,10 @@ class Telegram: if buttons: reply_markup = self._create_inline_keyboard(buttons) - payload_signature = self._build_message_payload_signature( - text=text, - reply_markup=reply_markup, - image=image, - disable_web_page_preview=disable_web_page_preview, - ) - cache_key = self._message_cache_key(chat_id, message_id) - if self._message_payload_cache.get(cache_key) == payload_signature: - logger.debug( - f"跳过重复编辑Telegram消息: chat_id={chat_id}, message_id={message_id}" - ) - return True - if image: # 如果有图片，使用edit_message_media media = InputMediaPhoto( - media=image, - caption=payload_signature["text"], - parse_mode="MarkdownV2", + media=image, caption=standardize(text), parse_mode="MarkdownV2" ) self._bot.edit_message_media( chat_id=chat_id, @@ -1004,7 +913,7 @@ class Telegram: edit_text_kwargs: Dict[str, Any] = { "chat_id": chat_id, "message_id": message_id, - "text": payload_signature["text"], + "text": standardize(text), "parse_mode": "MarkdownV2", "reply_markup": reply_markup, } @@ -1013,15 +922,8 @@ class Telegram: disable_web_page_preview ) self._bot.edit_message_text(**edit_text_kwargs) - self._message_payload_cache[cache_key] = payload_signature return True except Exception as e: - if "message is not modified" in str(e): - self._message_payload_cache[cache_key] = payload_signature - logger.debug( - f"Telegram消息内容未变化，跳过重复编辑: chat_id={chat_id}, message_id={message_id}" - ) - return True logger.error(f"编辑消息失败：{str(e)}") return False diff --git a/tests/test_telegram_message_editing.py b/tests/test_telegram_message_editing.py deleted file mode 100644 index b9d4a06d..00000000 --- a/tests/test_telegram_message_editing.py +++ /dev/null @@ -1,52 +0,0 @@ -import unittest -from types import SimpleNamespace -from unittest.mock import Mock - -from app.modules.telegram.telegram import Telegram - - -class TestTelegramMessageEditing(unittest.TestCase): - def _build_telegram(self) -> Telegram: - telegram = Telegram.__new__(Telegram) - telegram._bot = Mock() - telegram._telegram_token = "token-123" - telegram._telegram_chat_id = "456" - telegram._message_payload_cache = {} - telegram._typing_tasks = {} - telegram._typing_stop_flags = {} - telegram._user_chat_mapping = {} - return telegram - - def test_edit_msg_skips_duplicate_payload_after_initial_send(self): - telegram = self._build_telegram() - telegram._bot.send_message.return_value = SimpleNamespace( - message_id=11, chat=SimpleNamespace(id="456") - ) - - result = telegram.send_msg(title="", text="abc") - self.assertTrue(result["success"]) - - edited = telegram.edit_msg(chat_id="456", message_id=11, text="abc ") - - self.assertTrue(edited) - telegram._bot.edit_message_text.assert_not_called() - - def test_edit_msg_treats_message_not_modified_as_success(self): - telegram = self._build_telegram() - telegram._bot.edit_message_text.side_effect = Exception( - "A request to the Telegram API was unsuccessful. " - "Error code: 400. Description: Bad Request: message is not modified: " - "specified new message content and reply markup are exactly the same as " - "a current content and reply markup of the message" - ) - - first = telegram.edit_msg(chat_id="456", message_id=12, text="abc") - second = telegram.edit_msg(chat_id="456", message_id=12, text="abc") - - self.assertTrue(first) - self.assertTrue(second) - telegram._bot.edit_message_text.assert_called_once() - - -if __name__ == "__main__": - unittest.main()