Fix Telegram duplicate message edits

app/modules/telegram/telegram.py

@@ -44,6 +44,7 @@ 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,
@@ -81,6 +82,7 @@ class Telegram:
             _bot = TeleBot(self._telegram_token, parse_mode="MarkdownV2")
             # 记录句柄
             self._bot = _bot
+            self._message_payload_cache = {}
             # 获取并存储bot用户名用于@检测
             try:
                 bot_info = _bot.get_me()
@@ -363,6 +365,66 @@ 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,
@@ -448,10 +510,21 @@ 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": sent.chat.id if hasattr(sent, "chat") else chat_id,
+                        "chat_id": target_chat_id,
                     }
                 elif sent:
                     return {"success": True}
@@ -819,6 +892,9 @@ 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(
@@ -897,10 +973,25 @@ 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=standardize(text), parse_mode="MarkdownV2"
+                    media=image,
+                    caption=payload_signature["text"],
+                    parse_mode="MarkdownV2",
                 )
                 self._bot.edit_message_media(
                     chat_id=chat_id,
@@ -913,7 +1004,7 @@ class Telegram:
                 edit_text_kwargs: Dict[str, Any] = {
                     "chat_id": chat_id,
                     "message_id": message_id,
-                    "text": standardize(text),
+                    "text": payload_signature["text"],
                     "parse_mode": "MarkdownV2",
                     "reply_markup": reply_markup,
                 }
@@ -922,8 +1013,15 @@ 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
 

tests/test_telegram_message_editing.py

@@ -0,0 +1,52 @@
+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()