nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-06-30 00:36:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix Telegram edit fallback

Browse Source
This commit is contained in:
jxxghp
2026-06-29 22:48:13 +08:00
parent 4d30dee74c
commit 73d5c95f4e
3 changed files with 140 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
app/modules/telegram/__init__.py
View File

@@ -21,6 +21,10 @@ from app.utils.structures import DictUtils
class TelegramModule(_ModuleBase, _MessageBase[Telegram]):
"""
Telegram 通知模块，负责模块生命周期、消息解析和通知发送。
"""
def init_module(self) -> None:
"""
初始化模块
@@ -32,6 +36,9 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]):
@staticmethod
def get_name() -> str:
"""
获取模块名称
"""
return "Telegram"
@staticmethod
@@ -75,6 +82,9 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]):
return True, ""
def init_setting(self) -> Tuple[str, Union[str, bool]]:
"""
获取模块初始化配置项。
"""
pass
@staticmethod

120
app/modules/telegram/telegram.py
View File

@@ -46,10 +46,18 @@ TELEGRAM_PARSE_MODE_ALIASES = {
class RetryException(Exception):
"""
Telegram 消息发送重试异常。
"""
pass
class Telegram:
"""
Telegram 消息客户端，负责发送、编辑、接收和转发 Telegram 消息。
"""
_ds_url = (
f"http://127.0.0.1:{settings.PORT}/api/v1/message?token={settings.API_TOKEN}"
)
@@ -1129,6 +1137,69 @@ class Telegram:
"""
return "there is no text in the message to edit" in str(err).lower()
@staticmethod
def __is_message_not_modified_error(err: Exception) -> bool:
"""
判断 Telegram 是否因为消息内容未变化而拒绝编辑。
"""
return "message is not modified" in str(err).lower()
@staticmethod
def __is_http_url_content_error(err: Exception) -> bool:
"""
判断 Telegram 是否因为无法获取远端图片 URL 而拒绝编辑。
"""
return "failed to get http url content" in str(err).lower()
def __edit_message_text_or_caption(
self,
chat_id: str,
message_id: int,
text: str,
reply_markup: Optional[InlineKeyboardMarkup] = None,
disable_web_page_preview: Optional[bool] = None,
parse_mode: Optional[str] = None,
) -> bool:
"""
编辑 Telegram 文本消息，原消息无文本时回退为 caption 编辑。
"""
prepared_text = self._prepare_text(text, parse_mode)
edit_text_kwargs: Dict[str, Any] = {
"chat_id": chat_id,
"message_id": message_id,
"text": prepared_text,
"parse_mode": parse_mode,
"reply_markup": reply_markup,
}
if disable_web_page_preview is not None:
edit_text_kwargs["disable_web_page_preview"] = (
disable_web_page_preview
)
try:
self._bot.edit_message_text(**edit_text_kwargs)
except Exception as err:
if self.__is_message_not_modified_error(err):
logger.debug(f"Telegram消息内容未变化，跳过编辑：{str(err)}")
return True
if not self.__is_no_text_edit_error(err):
raise
try:
self._bot.edit_message_caption(
chat_id=chat_id,
message_id=message_id,
caption=prepared_text,
parse_mode=parse_mode,
reply_markup=reply_markup,
)
except Exception as caption_err:
if self.__is_message_not_modified_error(caption_err):
logger.debug(
f"Telegram消息内容未变化，跳过编辑：{str(caption_err)}"
)
return True
raise
return True
def __edit_message(
self,
chat_id: str,
@@ -1175,31 +1246,34 @@ class Telegram:
)
else:
# 如果没有图片，使用edit_message_text
edit_text_kwargs: Dict[str, Any] = {
"chat_id": chat_id,
"message_id": message_id,
"text": self._prepare_text(text, parse_mode),
"parse_mode": parse_mode,
"reply_markup": reply_markup,
}
if disable_web_page_preview is not None:
edit_text_kwargs["disable_web_page_preview"] = (
disable_web_page_preview
)
try:
self._bot.edit_message_text(**edit_text_kwargs)
except Exception as err:
if not self.__is_no_text_edit_error(err):
raise
self._bot.edit_message_caption(
chat_id=chat_id,
message_id=message_id,
caption=self._prepare_text(text, parse_mode),
parse_mode=parse_mode,
reply_markup=reply_markup,
)
return self.__edit_message_text_or_caption(
chat_id=chat_id,
message_id=message_id,
text=text,
reply_markup=reply_markup,
disable_web_page_preview=disable_web_page_preview,
parse_mode=parse_mode,
)
return True
except Exception as e:
if self.__is_message_not_modified_error(e):
logger.debug(f"Telegram消息内容未变化，跳过编辑：{str(e)}")
return True
if image and self.__is_http_url_content_error(e):
logger.warning(
f"Telegram图片编辑失败，降级为文本编辑：{str(e)}"
)
try:
return self.__edit_message_text_or_caption(
chat_id=chat_id,
message_id=message_id,
text=text,
reply_markup=reply_markup,
disable_web_page_preview=disable_web_page_preview,
parse_mode=parse_mode,
)
except Exception as fallback_err:
e = fallback_err
logger.error(f"编辑消息失败：{str(e)}")
return False

36
tests/test_telegram.py
View File

@@ -354,8 +354,8 @@ def test_edit_msg_with_html_parse_mode_keeps_html(telegram):
assert edit_kwargs["text"] == "<b>标题</b>\n<blockquote>请选择</blockquote>"
def test_edit_msg_keeps_other_edit_errors_failed(telegram):
"""非图片 caption 场景的编辑错误不应被错误标记为成功。"""
def test_edit_msg_treats_message_not_modified_as_success(telegram):
"""重复编辑相同内容时应视为成功，避免记录错误日志。"""
telegram.bot.edit_message_text.side_effect = Exception(
"Bad Request: message is not modified"
)
@@ -367,6 +367,36 @@ def test_edit_msg_keeps_other_edit_errors_failed(telegram):
text="测试内容",
)
assert result is False
assert result is True
telegram.bot.edit_message_text.assert_called_once()
telegram.bot.edit_message_caption.assert_not_called()
def test_send_msg_edit_with_image_falls_back_to_text_when_image_url_unavailable(telegram):
"""编辑图片消息失败时应去掉图片并降级为文本编辑。"""
telegram.bot.edit_message_media.side_effect = Exception(
"Bad Request: failed to get HTTP URL content"
)
result = telegram.send_msg(
title="测试标题",
text="测试内容",
image="https://example.com/poster.jpg",
buttons=[[{"text": "确认", "callback_data": "confirm"}]],
original_chat_id="1051253579",
original_message_id=110502,
)
assert result == {
"success": True,
"message_id": 110502,
"chat_id": "1051253579",
}
telegram.bot.edit_message_media.assert_called_once()
telegram.bot.edit_message_text.assert_called_once()
edit_kwargs = telegram.bot.edit_message_text.call_args.kwargs
assert edit_kwargs["chat_id"] == "1051253579"
assert edit_kwargs["message_id"] == 110502
assert "测试标题" in edit_kwargs["text"]
assert "测试内容" in edit_kwargs["text"]
assert edit_kwargs["reply_markup"] is not None