diff --git a/astrbot/core/platform/sources/telegram/tg_adapter.py b/astrbot/core/platform/sources/telegram/tg_adapter.py index b18cda9c0e6..b1b60cbda35 100644 --- a/astrbot/core/platform/sources/telegram/tg_adapter.py +++ b/astrbot/core/platform/sources/telegram/tg_adapter.py @@ -573,6 +573,17 @@ def _apply_caption() -> None: message.message_str = "" message.message = [] + if ( + message.type == MessageType.GROUP_MESSAGE + and update.message.text + and update.message.text.startswith("/") + ): + command_token = update.message.text.split(" ", 1)[0] + if "@" in command_token: + _, bot_name = command_token.split("@", 1) + if bot_name.casefold() != context.bot.username.casefold(): + return None + if update.message.reply_to_message and not ( update.message.is_topic_message and update.message.message_thread_id @@ -668,12 +679,12 @@ def _apply_caption() -> None: plain_text2 = f"/@{context.bot.username} " + plain_text plain_text = plain_text2 - # 群聊场景命令特殊处理 + # Normalize commands explicitly addressed to a Telegram bot. if plain_text.startswith("/"): command_parts = plain_text.split(" ", 1) if "@" in command_parts[0]: - command, bot_name = command_parts[0].split("@") - if bot_name == self.client.username: + command, bot_name = command_parts[0].split("@", 1) + if bot_name.casefold() == context.bot.username.casefold(): plain_text = command + ( f" {command_parts[1]}" if len(command_parts) > 1 else "" ) diff --git a/tests/test_telegram_adapter.py b/tests/test_telegram_adapter.py index 0c92ce301c9..47ca43b6bfa 100644 --- a/tests/test_telegram_adapter.py +++ b/tests/test_telegram_adapter.py @@ -85,6 +85,114 @@ def _build_context() -> MagicMock: return context +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("text", "expected"), + [ + ("/help@test_bot", "/help"), + ("/help@TEST_BOT details", "/help details"), + ("/help", "/help"), + ], +) +async def test_telegram_command_for_current_bot_is_normalized(text, expected): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + update = create_mock_update(message_text=text, chat_type="group") + + result = await adapter.convert_message(update, _build_context()) + + assert result is not None + assert result.message_str == expected + assert any( + isinstance(component, Comp.Plain) and component.text == expected + for component in result.message + ) + + +@pytest.mark.asyncio +async def test_telegram_command_for_another_bot_is_ignored(): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + update = create_mock_update( + message_text="/help@another_bot details", + chat_type="group", + ) + + result = await adapter.convert_message(update, _build_context()) + + assert result is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "text", + [ + "/help @some_user", + "hello @some_user", + ], +) +async def test_telegram_non_target_mentions_are_preserved(text): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + update = create_mock_update(message_text=text, chat_type="group") + + result = await adapter.convert_message(update, _build_context()) + + assert result is not None + assert result.message_str == text + + +@pytest.mark.asyncio +async def test_telegram_private_command_for_another_bot_is_preserved(): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + text = "/help@another_bot" + update = create_mock_update(message_text=text, chat_type="private") + + result = await adapter.convert_message(update, _build_context()) + + assert result is not None + assert result.message_str == text + + +@pytest.mark.asyncio +async def test_telegram_command_for_another_bot_skips_reply_side_effects(): + TelegramPlatformAdapter = _load_telegram_adapter() + adapter = TelegramPlatformAdapter( + make_platform_config("telegram"), + {}, + asyncio.Queue(), + ) + adapter.start = AsyncMock() + reply_to_message = create_mock_update(message_text="/start").message + update = create_mock_update( + message_text="/help@another_bot", + chat_type="group", + reply_to_message=reply_to_message, + ) + + result = await adapter.convert_message(update, _build_context()) + + assert result is None + adapter.start.assert_not_awaited() + + @pytest.mark.asyncio async def test_telegram_topic_with_missing_name_falls_back_to_group_name(): TelegramPlatformAdapter = _load_telegram_adapter()