fix: ignore Telegram commands addressed to other bots

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 ""
                         )

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()