From 765fb9c05fee36a7bfadee17cbd5e0242941045c Mon Sep 17 00:00:00 2001
From: noone <noone@noone.com>
Date: Wed, 24 Sep 2025 14:14:11 +0800
Subject: [PATCH] =?UTF-8?q?fix:=E6=9B=B4=E6=96=B0Telegram=E8=A7=A3?=
 =?UTF-8?q?=E6=9E=90=E6=A8=A1=E5=BC=8F=E4=B8=BAMarkdownV2=EF=BC=9BTelegram?=
 =?UTF-8?q?=E5=8F=91=E9=80=81=E7=9A=84=E5=86=85=E5=AE=B9=E6=8C=89=20Telegr?=
 =?UTF-8?q?am=20V2=20=E8=A7=84=E5=88=99=E8=BD=AC=E4=B9=89=E7=89=B9?=
 =?UTF-8?q?=E6=AE=8A=E5=AD=97=E7=AC=A6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/modules/telegram/telegram.py | 29 ++++++++++++++++++-----------
 1 file changed, 18 insertions(+), 11 deletions(-)

diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py
index 7c8982d3..d497ecde 100644
--- a/app/modules/telegram/telegram.py
+++ b/app/modules/telegram/telegram.py
@@ -31,7 +31,8 @@ class Telegram:
     _callback_handlers: Dict[str, Callable] = {}  # 存储回调处理器
     _user_chat_mapping: Dict[str, str] = {}  # userid -> chat_id mapping for reply targeting
     _bot_username: Optional[str] = None  # Bot username for mention detection
-
+    _escape_chars = r'_*[]()~`>#+-=|{}.!' # Telegram MarkdownV2
+    _markdown_escape_pattern = re.compile(f'([{re.escape(_escape_chars)}])') #Telegram MarkdownV2 规则转义特殊字符正则pattern
     def __init__(self, TELEGRAM_TOKEN: Optional[str] = None, TELEGRAM_CHAT_ID: Optional[str] = None, **kwargs):
         """
         初始化参数
@@ -52,7 +53,7 @@ class Telegram:
             else:
                 apihelper.proxy = settings.PROXY
             # bot
-            _bot = telebot.TeleBot(self._telegram_token, parse_mode="Markdown")
+            _bot = telebot.TeleBot(self._telegram_token, parse_mode="MarkdownV2")
             # 记录句柄
             self._bot = _bot
             # 获取并存储bot用户名用于@检测
@@ -238,7 +239,7 @@ class Telegram:
         try:
             if text:
                 # 对text进行Markdown特殊字符转义
-                text = re.sub(r"([_`])", r"\\\1", text)
+                text = self.escape_markdown(text)
                 caption = f"*{title}*\n{text}"
             else:
                 caption = f"*{title}*"
@@ -499,7 +500,7 @@ class Telegram:
 
             if image:
                 # 如果有图片，使用edit_message_media
-                media = InputMediaPhoto(media=image, caption=text, parse_mode="Markdown")
+                media = InputMediaPhoto(media=image, caption=text, parse_mode="MarkdownV2")
                 self._bot.edit_message_media(
                     chat_id=chat_id,
                     message_id=message_id,
@@ -512,7 +513,7 @@ class Telegram:
                     chat_id=chat_id,
                     message_id=message_id,
                     text=text,
-                    parse_mode="Markdown",
+                    parse_mode="MarkdownV2",
                     reply_markup=reply_markup
                 )
             return True
@@ -541,8 +542,8 @@ class Telegram:
                 # 发送图片到Telegram
                 ret = self._bot.send_photo(chat_id=userid or self._telegram_chat_id,
                                            photo=photo,
-                                           caption=caption,
-                                           parse_mode="Markdown",
+                                           caption=self.escape_markdown(caption),
+                                           parse_mode="MarkdownV2",
                                            reply_markup=reply_markup)
                 if ret is None:
                     raise RetryException("发送图片消息失败")
@@ -552,13 +553,13 @@ class Telegram:
         if len(caption) > 4095:
             for i in range(0, len(caption), 4095):
                 ret = self._bot.send_message(chat_id=userid or self._telegram_chat_id,
-                                             text=caption[i:i + 4095],
-                                             parse_mode="Markdown",
+                                             text=self.escape_markdown(caption[i:i + 4095]),
+                                             parse_mode="MarkdownV2",
                                              reply_markup=reply_markup if i == 0 else None)
         else:
             ret = self._bot.send_message(chat_id=userid or self._telegram_chat_id,
-                                         text=caption,
-                                         parse_mode="Markdown",
+                                         text=self.escape_markdown(caption),
+                                         parse_mode="MarkdownV2",
                                          reply_markup=reply_markup)
         if ret is None:
             raise RetryException("发送文本消息失败")
@@ -597,3 +598,9 @@ class Telegram:
             self._bot.stop_polling()
             self._polling_thread.join()
             logger.info("Telegram消息接收服务已停止")
+
+    def escape_markdown(self, text: str) -> str:
+        # 按 Telegram MarkdownV2 规则转义特殊字符
+        if not isinstance(text, str):
+            return str(text) if text is not None else ""
+        return self._markdown_escape_pattern.sub(r'\\\1', text)
\ No newline at end of file