GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-09-06 16:07:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

rollback telegram

Browse Source
This commit is contained in:
jxxghp
2025-11-18 21:56:55 +08:00
parent d64492bda5
commit c83589cac6
1 changed files with 26 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/telegram/telegram.py
+26 -40
View File
@@ -7,7 +7,6 @@ from typing import Optional, List, Dict, Callable
from urllib.parse import urljoin from urllib.parse import urljoin
import telebot import telebot
import telegramify_markdown
from telebot import apihelper from telebot import apihelper
from telebot.types import InputFile, InlineKeyboardMarkup, InlineKeyboardButton from telebot.types import InputFile, InlineKeyboardMarkup, InlineKeyboardButton
from telebot.types import InputMediaPhoto from telebot.types import InputMediaPhoto
@@ -53,7 +52,7 @@ class Telegram:
else: else:
apihelper.proxy = settings.PROXY apihelper.proxy = settings.PROXY
# bot # bot
_bot = telebot.TeleBot(self._telegram_token, parse_mode="MarkdownV2") _bot = telebot.TeleBot(self._telegram_token, parse_mode="Markdown")
# 记录句柄 # 记录句柄
self._bot = _bot self._bot = _bot
# 获取并存储bot用户名用于@检测 # 获取并存储bot用户名用于@检测
@@ -237,14 +236,12 @@ class Telegram:
return False return False
try: try:
if title and text: if text:
caption = f"**{title}**\n{text}" # 对text进行Markdown特殊字符转义
elif title: text = re.sub(r"([_`])", r"\\\1", text)
caption = f"**{title}**" caption = f"*{title}*\n{text}"
elif text:
caption = text
else: else:
caption = "" caption = f"*{title}*"
if link: if link:
caption = f"{caption}\n[查看详情]({link})" caption = f"{caption}\n[查看详情]({link})"
@@ -266,7 +263,7 @@ class Telegram:
return self.__send_request(userid=chat_id, image=image, caption=caption, reply_markup=reply_markup) return self.__send_request(userid=chat_id, image=image, caption=caption, reply_markup=reply_markup)
except Exception as msg_e: except Exception as msg_e:
logger.error(f"使用 send_msg 发送消息失败：{msg_e}") logger.error(f"发送消息失败：{msg_e}")
return False return False
def _determine_target_chat_id(self, userid: Optional[str] = None, def _determine_target_chat_id(self, userid: Optional[str] = None,
@@ -311,7 +308,7 @@ class Telegram:
return None return None
try: try:
index, image, caption = 1, "", f"**{title}**" if title else "" index, image, caption = 1, "", "*%s*" % title
for media in medias: for media in medias:
if not image: if not image:
image = media.get_message_image() image = media.get_message_image()
@@ -350,7 +347,7 @@ class Telegram:
return self.__send_request(userid=chat_id, image=image, caption=caption, reply_markup=reply_markup) return self.__send_request(userid=chat_id, image=image, caption=caption, reply_markup=reply_markup)
except Exception as msg_e: except Exception as msg_e:
logger.error(f"使用 send_medias_msg 发送消息失败：{msg_e}") logger.error(f"发送消息失败：{msg_e}")
return False return False
def send_torrents_msg(self, torrents: List[Context], def send_torrents_msg(self, torrents: List[Context],
@@ -372,7 +369,7 @@ class Telegram:
return None return None
try: try:
index, caption = 1, f"**{title}**" if title else "" index, caption = 1, "*%s*" % title
image = torrents[0].media_info.get_message_image() image = torrents[0].media_info.get_message_image()
for context in torrents: for context in torrents:
torrent = context.torrent_info torrent = context.torrent_info
@@ -410,7 +407,7 @@ class Telegram:
return self.__send_request(userid=chat_id, image=image, caption=caption, reply_markup=reply_markup) return self.__send_request(userid=chat_id, image=image, caption=caption, reply_markup=reply_markup)
except Exception as msg_e: except Exception as msg_e:
logger.error(f"使用 send_torrents_msg 发送消息失败：{msg_e}") logger.error(f"发送消息失败：{msg_e}")
return False return False
@staticmethod @staticmethod
@@ -502,7 +499,7 @@ class Telegram:
if image: if image:
# 如果有图片，使用edit_message_media # 如果有图片，使用edit_message_media
media = InputMediaPhoto(media=image, caption=text, parse_mode="MarkdownV2") media = InputMediaPhoto(media=image, caption=text, parse_mode="Markdown")
self._bot.edit_message_media( self._bot.edit_message_media(
chat_id=chat_id, chat_id=chat_id,
message_id=message_id, message_id=message_id,
@@ -515,7 +512,7 @@ class Telegram:
chat_id=chat_id, chat_id=chat_id,
message_id=message_id, message_id=message_id,
text=text, text=text,
parse_mode="MarkdownV2", parse_mode="Markdown",
reply_markup=reply_markup reply_markup=reply_markup
) )
return True return True
@@ -545,37 +542,26 @@ class Telegram:
ret = self._bot.send_photo(chat_id=userid or self._telegram_chat_id, ret = self._bot.send_photo(chat_id=userid or self._telegram_chat_id,
photo=photo, photo=photo,
caption=caption, caption=caption,
parse_mode="MarkdownV2", parse_mode="Markdown",
reply_markup=reply_markup) reply_markup=reply_markup)
if ret is None: if ret is None:
raise RetryException("发送图片消息失败") raise RetryException("发送图片消息失败")
return True return True
# 使用 telegramify-markdown 的 telegramify 函数智能分割长消息 # 按4096分段循环发送消息
# telegramify 会自动处理 Markdown 格式转换和智能分割，避免在格式标记中间分割
ret = None ret = None
try: if len(caption) > 4095:
# 使用 telegramify 处理原始 Markdown 文本 for i in range(0, len(caption), 4095):
chunks = list(telegramify_markdown.markdownify(
caption,
max_line_length=4095
))
if not chunks:
# 如果没有分割，使用 markdownify 转换后直接发送
converted = telegramify_markdown.markdownify(caption)
chunks = [converted]
# 发送所有消息块（telegramify 已经转换过格式，直接发送）
for i, chunk in enumerate(chunks):
ret = self._bot.send_message(chat_id=userid or self._telegram_chat_id, ret = self._bot.send_message(chat_id=userid or self._telegram_chat_id,
text=chunk, text=caption[i:i + 4095],
parse_mode="MarkdownV2", parse_mode="Markdown",
reply_markup=reply_markup if i == 0 else None) reply_markup=reply_markup if i == 0 else None)
if ret is None: else:
raise RetryException(f"发送文本消息失败（第 {i + 1}/{len(chunks)} 条）") ret = self._bot.send_message(chat_id=userid or self._telegram_chat_id,
except Exception as e: text=caption,
logger.warning(f"Telegram 发送消息失败：{e}") parse_mode="Markdown",
reply_markup=reply_markup)
if ret is None:
raise RetryException("发送文本消息失败")
return True if ret else False return True if ret else False
def register_commands(self, commands: Dict[str, dict]): def register_commands(self, commands: Dict[str, dict]):