feat(telegram): route channel notifications to configured topic

app/modules/telegram/telegram.py

@@ -89,6 +89,7 @@ class Telegram:
             self,
             TELEGRAM_TOKEN: Optional[str] = None,
             TELEGRAM_CHAT_ID: Optional[str] = None,
+            TELEGRAM_TOPIC_ID: Optional[str] = None,
             **kwargs,
     ):
         """
@@ -97,6 +98,13 @@ class Telegram:
         # 即使配置不完整也保留基础属性，便于测试和未启用实例安全调用发送方法。
         self._telegram_token = TELEGRAM_TOKEN
         self._telegram_chat_id = TELEGRAM_CHAT_ID
+        self._telegram_topic_id = None
+        if TELEGRAM_TOPIC_ID and str(TELEGRAM_TOPIC_ID).strip():
+            topic_id = str(TELEGRAM_TOPIC_ID).strip()
+            if topic_id.isdecimal() and int(topic_id) > 0:
+                self._telegram_topic_id = int(topic_id)
+            else:
+                logger.warning("Telegram话题ID无效，通知将发送到默认话题")
         self._polling_thread = None
         # 一个 Telegram 配置对应一个 SDK client，运行状态不能被其他配置共享。
         self._user_chat_mapping: Dict[str, str] = {}
@@ -845,6 +853,7 @@ class Telegram:
                     voice=fp,
                     caption=self._prepare_text(caption, parse_mode),
                     parse_mode=parse_mode if caption else None,
+                    **self._topic_kwargs(chat_id),
                 )
             self._stop_typing_if_needed(chat_id, stop_typing)
             if sent and hasattr(sent, "message_id"):
@@ -909,6 +918,7 @@ class Telegram:
                         photo=fp,
                         caption=self._prepare_text(caption, parse_mode),
                         parse_mode=parse_mode if caption else None,
+                        **self._topic_kwargs(chat_id),
                     )
                 else:
                     sent = self._bot.send_document(
@@ -916,6 +926,7 @@ class Telegram:
                         document=(send_name, fp),
                         caption=self._prepare_text(caption, parse_mode),
                         parse_mode=parse_mode if caption else None,
+                        **self._topic_kwargs(chat_id),
                     )
             self._stop_typing_if_needed(chat_id, stop_typing)
             if sent and hasattr(sent, "message_id"):
@@ -961,6 +972,13 @@ class Telegram:
         # 3. 最后使用默认聊天ID
         return self._telegram_chat_id
 
+    def _topic_kwargs(self, chat_id: Optional[Union[str, int]]) -> dict[str, int]:
+        """仅向本渠道配置的群组投递话题参数，避免影响私聊和其它会话。"""
+        topic_id = getattr(self, "_telegram_topic_id", None)
+        if topic_id and chat_id is not None and str(chat_id) == str(self._telegram_chat_id):
+            return {"message_thread_id": topic_id}
+        return {}
+
     def send_medias_msg(
             self,
             medias: List[MediaInfo],
@@ -1516,6 +1534,7 @@ class Telegram:
                     rich_message=chunk,
                     reply_markup=reply_markup if index == 0 else None,
                     reply_parameters=reply_parameters if index == 0 else None,
+                    **self._topic_kwargs(chat_id),
                 )
             return sent
         except Exception as err:
@@ -1545,6 +1564,7 @@ class Telegram:
             "parse_mode": parse_mode,
             "reply_markup": reply_markup,
         }
+        kwargs.update(self._topic_kwargs(kwargs["chat_id"]))
         if reply_to_message_id:
             kwargs["reply_to_message_id"] = reply_to_message_id
         # 处理图片

tests/test_telegram.py

@@ -54,6 +54,49 @@ def test_send_msg_success(telegram):
     assert result and result.get("success")
 
 
+def test_topic_id_is_scoped_to_channel_and_default_chat(tmp_path):
+    """同一 bot 和群组的不同渠道应各发各的话题，私聊与空配置不带话题参数。"""
+    bots = [MagicMock(), MagicMock(), MagicMock()]
+    for bot in bots:
+        bot.get_me.return_value = SimpleNamespace(username="test_bot")
+        bot.infinity_polling = lambda *args, **kwargs: None
+        bot.stop_polling = lambda *args, **kwargs: None
+    with patch("app.modules.telegram.telegram.TeleBot", side_effect=bots), patch(
+        "app.modules.telegram.telegram.ImageHelper"
+    ):
+        clients = [
+            Telegram(TELEGRAM_TOKEN="same-token", TELEGRAM_CHAT_ID="-1001", TELEGRAM_TOPIC_ID=topic)
+            for topic in ("101", "202", "")
+        ]
+        try:
+            for client, bot, topic in zip(clients, bots, (101, 202, None)):
+                assert client.send_msg(title="", text="通知")["success"]
+                assert bot.send_message.call_args.kwargs.get("message_thread_id") == topic
+                assert client.send_msg(title="", text="私聊", userid="123")["success"]
+                assert "message_thread_id" not in bot.send_message.call_args.kwargs
+
+            assert clients[0].send_msg(title="", rich_message="# 通知")["success"]
+            assert bots[0].send_rich_message.call_args.kwargs["message_thread_id"] == 101
+
+            attachment = tmp_path / "notice.txt"
+            attachment.write_text("通知")
+            assert clients[0].send_file(file_path=str(attachment))["success"]
+            assert bots[0].send_document.call_args.kwargs["message_thread_id"] == 101
+
+            photo = tmp_path / "notice.jpg"
+            photo.write_bytes(b"photo")
+            assert clients[0].send_file(file_path=str(photo))["success"]
+            assert bots[0].send_photo.call_args.kwargs["message_thread_id"] == 101
+
+            voice = tmp_path / "notice.ogg"
+            voice.write_bytes(b"voice")
+            assert clients[0].send_voice(voice_path=str(voice))["success"]
+            assert bots[0].send_voice.call_args.kwargs["message_thread_id"] == 101
+        finally:
+            for client in clients:
+                client.stop()
+
+
 def test_edit_msg_with_rich_message(telegram):
     """Telegram 流式编辑应继续使用 Rich Message 协议。"""
     telegram._bot.edit_message_text.return_value = SimpleNamespace(message_id=101)