perf(telegram): replace pyTelegramBotAPI with a lightweight Bot API client

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/modules/telegram/botapi.py

@@ -0,0 +1,456 @@
+"""
+Telegram Bot API 精简客户端。
+
+只实现 MoviePilot 用到的 Bot API 方法和 getUpdates 长轮询，请求统一经 RequestUtils 发出。
+方法名与参数沿用 pyTelegramBotAPI 的命名，入参中的键盘、回复参数等结构直接使用 Bot API
+的 JSON 字典；返回值是 Bot API 响应里的 ``result`` 原始结构（Message 为 dict）。
+"""
+
+import json
+import threading
+from typing import Any, Callable, Dict, List, Optional, Union, cast
+from urllib.parse import urljoin
+
+from app.adapters.network.http import RequestUtils
+from app.runtime.log import logger
+
+DEFAULT_API_URL = "https://api.telegram.org"
+
+# 普通请求的连接/读取超时，与原 SDK 默认值保持一致。
+_CONNECT_TIMEOUT_SECONDS = 15
+_READ_TIMEOUT_SECONDS = 30
+# 轮询只接收宿主处理的两类更新，其余更新类型不进入 getUpdates 结果。
+_ALLOWED_UPDATES = ["message", "callback_query"]
+# 群聊/私聊中宿主会转发的消息内容类型，与原 SDK message_handler 注册的类型一致。
+FORWARDED_CONTENT_TYPES = (
+    "text", "photo", "video", "document", "animation",
+    "audio", "voice", "sticker", "video_note",
+)
+
+ChatId = Union[str, int]
+# Bot API 请求参数与响应对象（Message、User 等）的 JSON 结构。
+JsonDict = Dict[str, Any]
+
+
+class TelegramApiError(Exception):
+    """
+    Telegram Bot API 调用失败。
+
+    ``str(err)`` 包含 Telegram 返回的 description，调用方据此识别
+    "message is not modified" 等可忽略错误。
+    """
+
+    def __init__(self, method: str, description: str, error_code: Optional[int] = None):
+        self.method = method
+        self.description = description
+        self.error_code = error_code
+        super().__init__(
+            f"Telegram API {method} 调用失败，错误码：{error_code}，描述：{description}"
+        )
+
+
+class TelegramNetworkError(Exception):
+    """
+    Telegram Bot API 请求未得到响应（连接失败、超时等）。
+
+    requests 的异常文本包含完整请求地址，而地址路径里有 Bot token；这里只保留脱敏后的
+    描述，并切断异常链，避免 token 经日志或 traceback 泄露。
+    """
+
+    def __init__(self, method: str, description: str):
+        self.method = method
+        self.description = description
+        super().__init__(f"Telegram API {method} 请求失败：{description}")
+
+
+class TelegramBotApi:
+    """
+    单个 Bot token 的 Bot API 客户端，持有一个复用连接的 HTTP 会话。
+
+    会话在发送线程、typing 线程和轮询线程间共享；requests 连接池本身线程安全，
+    只有 POST 且不会自动重放，避免网络抖动时重复发送消息。
+    """
+
+    def __init__(
+            self,
+            token: str,
+            api_url: Optional[str] = None,
+            proxy: Optional[JsonDict] = None,
+            parse_mode: Optional[str] = None,
+    ):
+        """
+        :param token: Bot token
+        :param api_url: 自建 Bot API 服务地址，为空时使用官方地址
+        :param proxy: requests 代理配置，仅在使用官方地址时生效
+        :param parse_mode: parse_mode 参数为 None 时使用的默认格式
+        """
+        self._token = token
+        base_url = api_url or DEFAULT_API_URL
+        # 与历史实现一致：自定义地址只取协议和主机，路径固定为 /bot<token>/<method>。
+        self._api_base = urljoin(base_url, f"/bot{token}/")
+        self._file_base = urljoin(base_url, f"/file/bot{token}/")
+        # 自定义地址通常是内网自建服务，不走全局代理。
+        self._proxy = None if api_url else proxy
+        self._default_parse_mode = parse_mode
+        self._http = RequestUtils(
+            headers={"Accept": "application/json"},
+            proxies=self._proxy,
+            use_session=True,
+            # RequestUtils 默认沿用旧宿主的不校验证书行为；Bot token 随请求发送，必须校验。
+            verify=True,
+        )
+        self._polling_stop = threading.Event()
+        # 分发临界区：持锁期间检查停止标记并执行 handler，stop 取得该锁即证明不会再分发。
+        self._dispatch_lock = threading.Lock()
+
+    @property
+    def proxy(self) -> Optional[JsonDict]:
+        """本客户端实际使用的代理配置。"""
+        return self._proxy
+
+    def file_url(self, file_path: str) -> str:
+        """拼接 getFile 返回路径对应的下载地址（包含 token，不可写入日志）。"""
+        return f"{self._file_base}{file_path}"
+
+    def close(self) -> None:
+        """关闭 HTTP 会话。"""
+        self._http.close()
+
+    # ---------------------------------------------------------------- 请求层
+
+    @staticmethod
+    def _encode_params(params: Dict[str, Any]) -> Dict[str, Any]:
+        """丢弃 None 值，并按 Bot API 约定把嵌套结构编码为 JSON 字符串。"""
+        encoded: Dict[str, Any] = {}
+        for key, value in params.items():
+            if value is None:
+                continue
+            if isinstance(value, (dict, list)):
+                encoded[key] = json.dumps(value, ensure_ascii=False)
+            elif isinstance(value, bool):
+                encoded[key] = "true" if value else "false"
+            else:
+                encoded[key] = value
+        return encoded
+
+    def call(
+            self,
+            method: str,
+            params: Optional[Dict[str, Any]] = None,
+            files: Optional[Dict[str, Any]] = None,
+            read_timeout: float = _READ_TIMEOUT_SECONDS,
+    ) -> Any:
+        """
+        调用 Bot API 方法并返回 ``result``。
+
+        :raises TelegramApiError: Telegram 返回 ok=false 或非 JSON 响应
+        :raises TelegramNetworkError: 网络失败，描述已去除 token
+        """
+        try:
+            response = self._http.post_res(
+                f"{self._api_base}{method}",
+                data=self._encode_params(params or {}),
+                files=files,
+                timeout=(_CONNECT_TIMEOUT_SECONDS, read_timeout),
+                raise_exception=True,
+            )
+        except Exception as err:  # RequestUtils 在 raise_exception=True 时抛出 requests 异常
+            description = f"{type(err).__name__}: {err}".replace(self._token, "<token>")
+            raise TelegramNetworkError(method, description) from None
+        if response is None:
+            raise TelegramNetworkError(method, "未收到响应")
+        try:
+            payload = response.json()
+        except ValueError:
+            raise TelegramApiError(
+                method, f"非 JSON 响应（HTTP {response.status_code}）", response.status_code
+            ) from None
+        if not payload.get("ok"):
+            raise TelegramApiError(
+                method, str(payload.get("description")), payload.get("error_code")
+            )
+        return payload.get("result")
+
+    def _call_dict(self, method: str, *args: Any, **kwargs: Any) -> JsonDict:
+        """调用返回对象（Message、User、File 等）的方法。"""
+        return cast(JsonDict, self.call(method, *args, **kwargs))
+
+    def _call_bool(self, method: str, *args: Any, **kwargs: Any) -> bool:
+        """调用返回 True 的方法。"""
+        return bool(self.call(method, *args, **kwargs))
+
+    def _call_list(self, method: str, *args: Any, **kwargs: Any) -> List[JsonDict]:
+        """调用返回对象数组的方法。"""
+        return cast(List[JsonDict], self.call(method, *args, **kwargs) or [])
+
+    def _parse_mode(self, parse_mode: Optional[str]) -> Optional[str]:
+        """None 使用默认格式，空字符串表示纯文本，不传 parse_mode。"""
+        resolved = self._default_parse_mode if parse_mode is None else parse_mode
+        return resolved or None
+
+    @staticmethod
+    def _reply_parameters(reply_to_message_id: Optional[int]) -> Optional[JsonDict]:
+        """旧式 reply_to_message_id 转为 Bot API 7.0 起的 reply_parameters。"""
+        if not reply_to_message_id:
+            return None
+        return {"message_id": int(reply_to_message_id)}
+
+    @staticmethod
+    def _link_preview_options(disable_web_page_preview: Optional[bool]) -> Optional[JsonDict]:
+        """旧式 disable_web_page_preview 转为 link_preview_options。"""
+        if disable_web_page_preview is None:
+            return None
+        return {"is_disabled": disable_web_page_preview}
+
+    @staticmethod
+    def _input_file(field: str, value: Any, params: Dict[str, Any], files: Dict[str, Any]) -> None:
+        """
+        把 photo/document/voice 参数放到正确位置。
+
+        字符串是 file_id 或 URL，走普通参数；bytes、文件对象或 ``(文件名, 内容)`` 走 multipart。
+        """
+        if isinstance(value, str):
+            params[field] = value
+        elif isinstance(value, tuple):
+            name, content = value
+            files[field] = (name or field, content)
+        else:
+            files[field] = value
+
+    # ------------------------------------------------------------- Bot API 方法
+
+    def get_me(self) -> JsonDict:
+        """获取 Bot 自身信息。"""
+        return self._call_dict("getMe")
+
+    def get_file(self, file_id: str) -> JsonDict:
+        """获取文件信息，返回值含 file_path。"""
+        return self._call_dict("getFile", {"file_id": file_id})
+
+    def send_message(
+            self,
+            chat_id: ChatId,
+            text: Optional[str],
+            parse_mode: Optional[str] = None,
+            reply_markup: Optional[JsonDict] = None,
+            disable_web_page_preview: Optional[bool] = None,
+            reply_to_message_id: Optional[int] = None,
+            message_thread_id: Optional[int] = None,
+    ) -> JsonDict:
+        """发送文本消息。"""
+        return self._call_dict("sendMessage", {
+            "chat_id": chat_id,
+            "text": text,
+            "parse_mode": self._parse_mode(parse_mode),
+            "reply_markup": reply_markup,
+            "link_preview_options": self._link_preview_options(disable_web_page_preview),
+            "reply_parameters": self._reply_parameters(reply_to_message_id),
+            "message_thread_id": message_thread_id,
+        })
+
+    def _send_media(
+            self,
+            method: str,
+            field: str,
+            chat_id: ChatId,
+            media: Any,
+            caption: Optional[str] = None,
+            parse_mode: Optional[str] = None,
+            reply_markup: Optional[JsonDict] = None,
+            reply_to_message_id: Optional[int] = None,
+            message_thread_id: Optional[int] = None,
+    ) -> JsonDict:
+        """发送单个图片、文件或语音，按类型选择 URL/file_id 或 multipart 上传。"""
+        params: Dict[str, Any] = {
+            "chat_id": chat_id,
+            "caption": caption,
+            "parse_mode": self._parse_mode(parse_mode),
+            "reply_markup": reply_markup,
+            "reply_parameters": self._reply_parameters(reply_to_message_id),
+            "message_thread_id": message_thread_id,
+        }
+        files: Dict[str, Any] = {}
+        self._input_file(field, media, params, files)
+        return self._call_dict(method, params, files=files or None)
+
+    def send_photo(self, chat_id: ChatId, photo: Any, **kwargs: Any) -> JsonDict:
+        """发送图片，参数见 ``_send_media``。"""
+        return self._send_media("sendPhoto", "photo", chat_id, photo, **kwargs)
+
+    def send_document(self, chat_id: ChatId, document: Any, **kwargs: Any) -> JsonDict:
+        """发送文件，参数见 ``_send_media``。"""
+        return self._send_media("sendDocument", "document", chat_id, document, **kwargs)
+
+    def send_voice(self, chat_id: ChatId, voice: Any, **kwargs: Any) -> JsonDict:
+        """发送语音，参数见 ``_send_media``。"""
+        return self._send_media("sendVoice", "voice", chat_id, voice, **kwargs)
+
+    def send_rich_message(
+            self,
+            chat_id: ChatId,
+            rich_message: JsonDict,
+            reply_markup: Optional[JsonDict] = None,
+            reply_parameters: Optional[JsonDict] = None,
+            message_thread_id: Optional[int] = None,
+    ) -> JsonDict:
+        """发送 Rich Message，rich_message 为 InputRichMessage 的 JSON 结构。"""
+        return self._call_dict("sendRichMessage", {
+            "chat_id": chat_id,
+            "rich_message": rich_message,
+            "reply_markup": reply_markup,
+            "reply_parameters": reply_parameters,
+            "message_thread_id": message_thread_id,
+        })
+
+    def send_chat_action(self, chat_id: ChatId, action: str) -> bool:
+        """发送 typing 等聊天状态。"""
+        return self._call_bool("sendChatAction", {"chat_id": chat_id, "action": action})
+
+    def edit_message_text(
+            self,
+            chat_id: ChatId,
+            message_id: int,
+            text: Optional[str] = None,
+            parse_mode: Optional[str] = None,
+            reply_markup: Optional[JsonDict] = None,
+            disable_web_page_preview: Optional[bool] = None,
+            rich_message: Optional[JsonDict] = None,
+    ) -> Any:
+        """编辑文本消息；传 rich_message 时替换为 Rich Message 且不带 parse_mode。"""
+        return self.call("editMessageText", {
+            "chat_id": chat_id,
+            "message_id": message_id,
+            "text": text,
+            "rich_message": rich_message,
+            "parse_mode": None if rich_message else self._parse_mode(parse_mode),
+            "reply_markup": reply_markup,
+            "link_preview_options": self._link_preview_options(disable_web_page_preview),
+        })
+
+    def edit_message_caption(
+            self,
+            chat_id: ChatId,
+            message_id: int,
+            caption: Optional[str] = None,
+            parse_mode: Optional[str] = None,
+            reply_markup: Optional[JsonDict] = None,
+    ) -> Any:
+        """编辑媒体消息的说明文字。"""
+        return self.call("editMessageCaption", {
+            "chat_id": chat_id,
+            "message_id": message_id,
+            "caption": caption,
+            "parse_mode": self._parse_mode(parse_mode),
+            "reply_markup": reply_markup,
+        })
+
+    def edit_message_media(
+            self,
+            chat_id: ChatId,
+            message_id: int,
+            media: JsonDict,
+            reply_markup: Optional[JsonDict] = None,
+    ) -> Any:
+        """替换消息媒体，media 为 InputMedia 的 JSON 结构（媒体须为 URL 或 file_id）。"""
+        return self.call("editMessageMedia", {
+            "chat_id": chat_id,
+            "message_id": message_id,
+            "media": media,
+            "reply_markup": reply_markup,
+        })
+
+    def delete_message(self, chat_id: ChatId, message_id: int) -> bool:
+        """删除消息。"""
+        return self._call_bool("deleteMessage", {"chat_id": chat_id, "message_id": message_id})
+
+    def answer_callback_query(
+            self,
+            callback_query_id: Union[str, int],
+            text: Optional[str] = None,
+            show_alert: Optional[bool] = None,
+    ) -> bool:
+        """回应按钮回调，结束客户端的加载状态。"""
+        return self._call_bool("answerCallbackQuery", {
+            "callback_query_id": callback_query_id,
+            "text": text,
+            "show_alert": show_alert,
+        })
+
+    def set_my_commands(self, commands: List[JsonDict]) -> bool:
+        """设置菜单命令，元素为 ``{"command": ..., "description": ...}``。"""
+        return self._call_bool("setMyCommands", {"commands": commands})
+
+    def delete_my_commands(self) -> bool:
+        """清除菜单命令。"""
+        return self._call_bool("deleteMyCommands")
+
+    # ------------------------------------------------------------------ 轮询
+
+    def get_updates(self, offset: Optional[int], timeout: int) -> List[JsonDict]:
+        """长轮询获取更新，读取超时比服务端等待时间多留出网络余量。"""
+        return self._call_list(
+            "getUpdates",
+            {"offset": offset, "timeout": timeout, "allowed_updates": _ALLOWED_UPDATES},
+            read_timeout=timeout + _CONNECT_TIMEOUT_SECONDS,
+        )
+
+    def polling(
+            self,
+            handler: Callable[[JsonDict], None],
+            long_polling_timeout: int,
+            retry_max_seconds: int = 60,
+    ) -> None:
+        """
+        在当前线程循环拉取更新并逐条交给 handler，直到 ``stop_polling`` 被调用。
+
+        更新按 update_id 顺序串行处理；handler 抛错只记录日志，仍确认该更新，
+        避免同一条坏消息被反复投递。网络或 API 失败按指数退避重试，失败期间只在
+        首次失败和恢复时各记录一条日志。
+
+        停止后仍在进行的 getUpdates 返回时，结果直接丢弃且不确认 offset，Telegram
+        会把这些更新重新投递给下一个轮询者；新轮询者的 getUpdates 也会让 Telegram
+        立即结束旧请求。
+        """
+        offset: Optional[int] = None
+        failures = 0
+        while not self._polling_stop.is_set():
+            try:
+                updates = self.get_updates(offset, long_polling_timeout)
+            except Exception as err:
+                if self._polling_stop.is_set():
+                    return
+                failures += 1
+                if failures == 1:
+                    logger.warning(f"Telegram消息拉取失败，将自动重试：{err}")
+                else:
+                    logger.debug(f"Telegram消息拉取第{failures}次失败：{err}")
+                self._polling_stop.wait(min(3 * 2 ** (failures - 1), retry_max_seconds))
+                continue
+            if failures:
+                logger.info("Telegram消息拉取已恢复")
+                failures = 0
+            for update in updates or []:
+                with self._dispatch_lock:
+                    # 停止后不再分发，未确认的更新会在下次启动时重新投递。
+                    if self._polling_stop.is_set():
+                        return
+                    offset = update["update_id"] + 1
+                    try:
+                        handler(update)
+                    except Exception as err:
+                        logger.error(f"处理Telegram更新失败：{err}")
+
+    def stop_polling(self, timeout: float) -> bool:
+        """
+        请求轮询循环退出，并等待正在执行的 handler 结束。
+
+        不等待进行中的长轮询请求：它返回后只会丢弃结果并退出，不再触发任何回调。
+
+        :param timeout: 等待 handler 的最长秒数
+        :return: 是否已确认之后不会再分发更新
+        """
+        self._polling_stop.set()
+        if not self._dispatch_lock.acquire(timeout=max(timeout, 0)):
+            return False
+        self._dispatch_lock.release()
+        return True

app/modules/telegram/compat.py

@@ -1,24 +0,0 @@
-def ensure_urllib3_header_param_compat() -> None:
-    """
-    pyTelegramBotAPI imports urllib3.fields.format_header_param at import time.
-    Some urllib3-future builds only expose newer formatter names.
-    RFC 2231 formatting is kept as the last fallback because it encodes
-    non-ASCII values differently from urllib3's old default.
-    """
-    try:
-        from urllib3 import fields
-    except ImportError:
-        return
-
-    if hasattr(fields, "format_header_param"):
-        return
-
-    for fallback_name in (
-        "format_header_param_html5",
-        "format_multipart_header_param",
-        "format_header_param_rfc2231",
-    ):
-        fallback = getattr(fields, fallback_name, None)
-        if fallback is not None:
-            fields.format_header_param = fallback
-            return

app/modules/telegram/telegram.py

@@ -1,57 +1,44 @@
 import asyncio
 import html as html_utils
-import json
 import re
 import threading
 import time
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Union
-from urllib.parse import urljoin
 
-from app.modules.telegram.compat import ensure_urllib3_header_param_compat
-
-# Must run before importing pyTelegramBotAPI.
-ensure_urllib3_header_param_compat()
-
-from telebot import TeleBot, apihelper  # noqa: E402
-from telebot.types import (  # noqa: E402
-    BotCommand,
-    InlineKeyboardMarkup,
-    InlineKeyboardButton,
-    InputMediaPhoto,
-    InputRichMessage as TelebotInputRichMessage,
-    ReplyParameters,
-)
+from telegramify_markdown import richify, split_rich, standardize, telegramify
 try:
-    from telebot.types import ForceReply  # noqa: E402
-except ImportError:
-    ForceReply = None
-from telegramify_markdown import richify, split_rich, standardize, telegramify  # noqa: E402
-try:
-    from telegramify_markdown import entities_to_markdownv2  # noqa: E402
+    from telegramify_markdown import entities_to_markdownv2
 except ImportError:
     entities_to_markdownv2 = None
 try:
-    from telegramify_markdown.content import ContentTypes, File, Photo, Text  # noqa: E402
+    from telegramify_markdown.content import ContentTypes, File, Photo, Text
 except ImportError:
-    from telegramify_markdown.type import ContentTypes, File, Photo, Text  # noqa: E402
+    from telegramify_markdown.type import ContentTypes, File, Photo, Text
 
-from app.runtime.settings import get_runtime_setting  # noqa: E402
+from app.runtime.settings import get_runtime_setting
 
-from app.domain.context import MediaInfo, Context  # noqa: E402
-from app.domain.metainfo import MetaInfo  # noqa: E402
-from app.application.image import ImageHelper  # noqa: E402
-from app.application.messaging.ingress import forward_message_to_host  # noqa: E402
-from app.runtime.thread import ThreadHelper  # noqa: E402
-from app.runtime.log import logger  # noqa: E402
-from app.runtime.execution import retry  # noqa: E402
-from app.adapters.network.http import RequestUtils  # noqa: E402
-from app.foundation import size as size_tools  # noqa: E402
+from app.domain.context import MediaInfo, Context
+from app.domain.metainfo import MetaInfo
+from app.application.image import ImageHelper
+from app.application.messaging.ingress import forward_message_to_host
+from app.runtime.thread import ThreadHelper
+from app.runtime.log import logger
+from app.runtime.execution import retry
+from app.adapters.network.http import RequestUtils
+from app.modules.telegram.botapi import (
+    FORWARDED_CONTENT_TYPES,
+    JsonDict,
+    TelegramBotApi,
+)
+from app.foundation import size as size_tools
 
 
 TELEGRAM_PARSE_MODE_MARKDOWN = "MarkdownV2"
 TELEGRAM_PARSE_MODE_HTML = "HTML"
 TELEGRAM_PARSE_MODE_PLAIN = ""
+# InputRichMessage 在 Bot API 中定义的字段，转换结果中的其他键不发送。
+_RICH_MESSAGE_FIELDS = ("html", "markdown", "is_rtl", "skip_entity_detection")
 TELEGRAM_PARSE_MODE_ALIASES = {
     "markdownv2": TELEGRAM_PARSE_MODE_MARKDOWN,
     "mdv2": TELEGRAM_PARSE_MODE_MARKDOWN,
@@ -74,7 +61,7 @@ class Telegram:
     Telegram 消息客户端，负责发送、编辑、接收和转发 Telegram 消息。
     """
 
-    _bot: TeleBot = None
+    _bot: Optional[TelegramBotApi] = None
     _callback_handlers: Dict[str, Callable] = {}  # 存储回调处理器
     _bot_username: Optional[str] = None  # Bot username for mention detection
     _typing_interval_seconds = 5
@@ -84,6 +71,8 @@ class Telegram:
     _typing_callback_max_duration_seconds = 60
     _typing_join_timeout_seconds = 1
     _shutdown_timeout_seconds = 10
+    # getUpdates 长轮询等待秒数；停止时不等待进行中的请求，因此不受关闭预算约束。
+    _long_polling_timeout_seconds = 30
 
     def __init__(
             self,
@@ -116,119 +105,126 @@ class Telegram:
         if not TELEGRAM_TOKEN or not TELEGRAM_CHAT_ID:
             logger.error("Telegram配置不完整！")
             return
-        # 初始化机器人
-        if self._telegram_token and self._telegram_chat_id:
-            # telegram bot api 地址，格式：https://api.telegram.org
-            api_url = kwargs.get("API_URL")
-            if api_url:
-                # 如果提供了自定义API地址，使用它
-                apihelper.API_URL = urljoin(api_url, "/bot{0}/{1}")
-                apihelper.FILE_URL = urljoin(api_url, "/file/bot{0}/{1}")
-                # 使用自定义地址时，不设置代理
-                apihelper.proxy = None
-            else:
-                # 使用默认Telegram API地址
-                apihelper.API_URL = "https://api.telegram.org/bot{0}/{1}"
-                apihelper.FILE_URL = "https://api.telegram.org/file/bot{0}/{1}"
-                # 设置代理
-                apihelper.proxy = get_runtime_setting('PROXY')
-            # bot
-            _bot = TeleBot(self._telegram_token, parse_mode=TELEGRAM_PARSE_MODE_MARKDOWN)
-            # 记录句柄
-            self._bot = _bot
-            # 获取并存储bot用户名用于@检测
-            try:
-                bot_info = _bot.get_me()
-                self._bot_username = bot_info.username
-                logger.info(f"Telegram bot用户名: @{self._bot_username}")
-            except Exception as e:
-                logger.error(f"获取bot信息失败: {e}")
-                self._bot_username = None
+        # telegram bot api 地址，格式：https://api.telegram.org；自定义地址不走代理。
+        self._bot = TelegramBotApi(
+            TELEGRAM_TOKEN,
+            api_url=kwargs.get("API_URL"),
+            proxy=get_runtime_setting('PROXY'),
+            parse_mode=TELEGRAM_PARSE_MODE_MARKDOWN,
+        )
+        self._config_name = kwargs.get("name")
+        # 获取并存储bot用户名用于@检测
+        try:
+            self._bot_username = self._api.get_me().get("username")
+            logger.info(f"Telegram bot用户名: @{self._bot_username}")
+        except Exception as e:
+            logger.error(f"获取bot信息失败: {e}")
+            self._bot_username = None
 
-            self._config_name = kwargs.get("name")
+        self._polling_thread = threading.Thread(
+            target=self._run_polling,
+            name="MoviePilot-TelegramPolling",
+            daemon=True,
+        )
+        self._polling_thread.start()
+        logger.info("Telegram消息接收服务启动")
 
-            @_bot.message_handler(commands=["start", "help"])
-            def send_welcome(message):
-                _bot.reply_to(
-                    message,
-                    "温馨提示：直接发送名称或`订阅`+名称，搜索或订阅电影、电视剧",
-                )
+    @property
+    def _api(self) -> TelegramBotApi:
+        """
+        已初始化的 Bot API 客户端。
 
-            @_bot.message_handler(content_types=[
-                "text", "photo", "video", "document", "animation",
-                "audio", "voice", "sticker", "video_note",
-            ], func=lambda message: True)
-            def echo_all(message):
-                # Update user-chat mapping when receiving messages
-                self._update_user_chat_mapping(message.from_user.id, message.chat.id)
+        未配置或已停止时抛错，由各发送方法的异常处理记录失败，与历史实现访问空句柄的结果一致。
+        """
+        if self._bot is None:
+            raise RuntimeError("Telegram客户端未初始化")
+        return self._bot
 
-                # Check if we should process this message
-                if self._should_process_message(message):
-                    payload = self._serialize_update_payload(message)
-                    if not payload:
-                        logger.warn("Telegram消息序列化失败，跳过转发")
-                        return
-                    if not self._forward_to_message_chain(payload):
-                        logger.warn("Telegram消息转发失败")
+    def _run_polling(self) -> None:
+        """轮询线程入口，逐条处理 getUpdates 返回的更新。"""
+        try:
+            self._api.polling(
+                self._handle_update,
+                long_polling_timeout=self._long_polling_timeout_seconds,
+            )
+        except Exception as err:
+            logger.error(f"Telegram消息接收服务异常：{str(err)}")
 
-            @_bot.callback_query_handler(func=lambda call: True)
-            def callback_query(call):
-                """
-                处理按钮点击回调
-                """
-                chat_id = None
-                try:
-                    # Update user-chat mapping for callbacks too
-                    chat_id = call.message.chat.id
-                    self._update_user_chat_mapping(
-                        call.from_user.id, chat_id
-                    )
+    def _handle_update(self, update: JsonDict) -> None:
+        """按更新类型分发：普通消息转发主程序，按钮回调先应答再转发。"""
+        if "callback_query" in update:
+            self._handle_callback_query(update["callback_query"])
+            return
+        message = update.get("message")
+        if not message or not any(key in message for key in FORWARDED_CONTENT_TYPES):
+            return
+        if self._reply_welcome_if_needed(message):
+            return
+        sender = message.get("from") or {}
+        self._update_user_chat_mapping(sender.get("id"), message["chat"]["id"])
+        if self._should_process_message(message):
+            if not self._forward_to_message_chain(message):
+                logger.warn("Telegram消息转发失败")
 
-                    # 解析回调数据
-                    callback_data = call.data
-                    user_id = str(call.from_user.id)
+    def _reply_welcome_if_needed(self, message: JsonDict) -> bool:
+        """
+        /start、/help 命令只回复使用提示，不转发主程序。
 
-                    logger.info(f"收到按钮回调：{callback_data}，用户：{user_id}")
+        与原 SDK 命令过滤一致：忽略 @bot 后缀和参数，例如 ``/start@mp_bot payload``。
+        """
+        text = message.get("text") or ""
+        if not text.startswith("/"):
+            return False
+        command = text.split(maxsplit=1)[0][1:].split("@", 1)[0]
+        if command not in ("start", "help"):
+            return False
+        self._api.send_message(
+            chat_id=message["chat"]["id"],
+            text="温馨提示：直接发送名称或`订阅`+名称，搜索或订阅电影、电视剧",
+            reply_to_message_id=message["message_id"],
+        )
+        return True
 
-                    # 发送回调数据给主程序处理
-                    callback_json = {
-                        "callback_query": {
-                            "id": call.id,
-                            "from": call.from_user.to_dict(),
-                            "message": {
-                                "message_id": call.message.message_id,
-                                "chat": {
-                                    "id": chat_id,
-                                },
-                            },
-                            "data": callback_data,
-                        }
-                    }
+    def _handle_callback_query(self, call: JsonDict) -> None:
+        """
+        处理按钮点击回调
+        """
+        try:
+            # Update user-chat mapping for callbacks too
+            chat_id = call["message"]["chat"]["id"]
+            self._update_user_chat_mapping(call["from"]["id"], chat_id)
 
-                    # 先确认回调，避免用户看到loading状态
-                    _bot.answer_callback_query(call.id)
+            # 解析回调数据
+            callback_data = call.get("data")
+            user_id = str(call["from"]["id"])
 
-                    # 发送给主程序处理
-                    if not self._forward_to_message_chain(callback_json):
-                        logger.warn("Telegram按钮回调转发失败")
+            logger.info(f"收到按钮回调：{callback_data}，用户：{user_id}")
 
-                except Exception as err:
-                    logger.error(f"处理按钮回调失败：{str(err)}")
-                    _bot.answer_callback_query(call.id, "处理失败，请重试")
+            # 发送回调数据给主程序处理
+            callback_json = {
+                "callback_query": {
+                    "id": call["id"],
+                    "from": call["from"],
+                    "message": {
+                        "message_id": call["message"]["message_id"],
+                        "chat": {
+                            "id": chat_id,
+                        },
+                    },
+                    "data": callback_data,
+                }
+            }
 
-            def run_polling():
-                """
-                定义线程函数来运行 infinity_polling
-                """
-                try:
-                    _bot.infinity_polling(long_polling_timeout=30, logger_level=None)
-                except Exception as err:
-                    logger.error(f"Telegram消息接收服务异常：{str(err)}")
+            # 先确认回调，避免用户看到loading状态
+            self._api.answer_callback_query(call["id"])
 
-            # 启动线程来运行 infinity_polling
-            self._polling_thread = threading.Thread(target=run_polling, daemon=True)
-            self._polling_thread.start()
-            logger.info("Telegram消息接收服务启动")
+            # 发送给主程序处理
+            if not self._forward_to_message_chain(callback_json):
+                logger.warn("Telegram按钮回调转发失败")
+
+        except Exception as err:
+            logger.error(f"处理按钮回调失败：{str(err)}")
+            self._api.answer_callback_query(call["id"], "处理失败，请重试")
 
     def _forward_to_message_chain(self, payload: dict) -> bool:
         """把 Telegram SDK 回调同步转交统一消息入口。"""
@@ -238,7 +234,7 @@ class Telegram:
     def bot(self):
         """
         获取Telegram Bot实例
-        :return: TeleBot实例或None
+        :return: Bot API 客户端或None
         """
         return self._bot
 
@@ -259,27 +255,27 @@ class Telegram:
         if not self._bot:
             return None
         try:
-            file_info = self._bot.get_file(file_id)
-            file_url = apihelper.FILE_URL.format(
-                self._telegram_token, file_info.file_path
-            )
+            file_path = self._api.get_file(file_id).get("file_path")
+            if not file_path:
+                logger.warn(f"Telegram文件信息缺少下载路径: file_id={file_id}")
+                return None
+            file_url = self._api.file_url(file_path)
             resp = RequestUtils(
-                proxies=apihelper.proxy, timeout=30
+                proxies=self._api.proxy, timeout=30
             ).get_res(file_url)
             if resp and resp.content:
                 logger.info(
                     "Telegram文件下载成功: file_id=%s, file_path=%s, content_bytes=%s",
                     file_id,
-                    file_info.file_path,
+                    file_path,
                     len(resp.content),
                 )
                 return resp.content
             logger.warn(
-                "Telegram文件下载失败: file_id=%s, file_path=%s, file_url=%s, proxy_enabled=%s",
+                "Telegram文件下载失败: file_id=%s, file_path=%s, proxy_enabled=%s",
                 file_id,
-                getattr(file_info, "file_path", None),
-                file_url,
-                bool(apihelper.proxy),
+                file_path,
+                bool(self._api.proxy),
             )
         except Exception as e:
             logger.error(f"下载Telegram文件失败: {e}")
@@ -392,24 +388,9 @@ class Telegram:
             remaining = remaining[split_at:].lstrip("\n")
         return chunks
 
-    @staticmethod
-    def _serialize_update_payload(message: Any) -> Optional[dict]:
-        """
-        将 Telegram Message 对象稳定序列化为 dict，避免 requests 的 json 参数再次包一层字符串。
-        """
-        try:
-            if hasattr(message, "to_dict"):
-                payload = message.to_dict()
-            else:
-                payload = getattr(message, "json", None) or message
-            if isinstance(payload, str):
-                payload = json.loads(payload)
-            return payload if isinstance(payload, dict) else None
-        except Exception as e:
-            logger.error(f"序列化Telegram消息失败: {e}")
-            return None
-
-    def _update_user_chat_mapping(self, userid: int, chat_id: int) -> None:
+    def _update_user_chat_mapping(
+            self, userid: Optional[Union[str, int]], chat_id: Optional[Union[str, int]]
+    ) -> None:
         """
         更新用户与聊天的映射关系
         :param userid: 用户ID
@@ -426,19 +407,20 @@ class Telegram:
         """
         return self._user_chat_mapping.get(str(userid)) if userid else None
 
-    def _should_process_message(self, message) -> bool:
+    def _should_process_message(self, message: JsonDict) -> bool:
         """
         判断是否应该处理这条消息
-        :param message: Telegram消息对象
+        :param message: Bot API Message 结构
         :return: 是否处理
         """
+        chat_type = message["chat"].get("type")
         # 私聊消息总是处理
-        if message.chat.type == "private":
-            logger.debug(f"处理私聊消息：用户 {message.from_user.id}")
+        if chat_type == "private":
+            logger.debug(f"处理私聊消息：用户 {(message.get('from') or {}).get('id')}")
             return True
 
         # 消息内容：文本消息用 text，媒体消息的说明文字用 caption
-        message_text = message.text or message.caption or ""
+        message_text = message.get("text") or message.get("caption") or ""
 
         # 群聊中的命令消息总是处理（以/开头）
         if message_text.startswith("/"):
@@ -446,7 +428,7 @@ class Telegram:
             return True
 
         # 群聊中检查是否@了机器人
-        if message.chat.type in ["group", "supergroup"]:
+        if chat_type in ["group", "supergroup"]:
             if not self._bot_username:
                 # 如果没有获取到bot用户名，为了安全起见处理所有消息
                 logger.debug("未获取到bot用户名，处理所有群聊消息")
@@ -458,12 +440,17 @@ class Telegram:
                 return True
 
             # 检查消息实体（文本消息用 entities，媒体消息用 caption_entities）
-            entities = (message.entities if message.text is not None else message.caption_entities) or []
+            entities = (
+                message.get("entities")
+                if message.get("text") is not None
+                else message.get("caption_entities")
+            ) or []
             for entity in entities:
-                if entity.type == "mention":
+                if entity.get("type") == "mention":
                     # Telegram offset/length 基于 UTF-16 代码单元，需用字节编码处理含 emoji 的消息
+                    offset, length = entity["offset"], entity["length"]
                     mention_text = message_text.encode('utf-16-le')[
-                        entity.offset * 2: (entity.offset + entity.length) * 2
+                        offset * 2: (offset + length) * 2
                     ].decode('utf-16-le')
                     if mention_text == f"@{self._bot_username}":
                         logger.debug(
@@ -478,7 +465,7 @@ class Telegram:
             return False
 
         # 其他类型的聊天默认处理
-        logger.debug(f"处理其他类型聊天消息：{message.chat.type}")
+        logger.debug(f"处理其他类型聊天消息：{chat_type}")
         return True
 
     def get_state(self) -> bool:
@@ -537,7 +524,7 @@ class Telegram:
                             break
                         try:
                             if self._bot:
-                                self._bot.send_chat_action(chat_id, "typing")
+                                self._api.send_chat_action(chat_id, "typing")
                         except Exception as e:
                             logger.debug(f"发送typing状态失败: {e}")
                         # Telegram 客户端约 5-6 秒后会隐藏 typing，需要周期性续发。
@@ -653,7 +640,7 @@ class Telegram:
             parse_mode: Optional[str] = None,
             rich_message: Optional[str] = None,
             private_delivery: bool = False,
-    ) -> Optional[dict]:
+    ) -> Optional[JsonDict]:
         """
         发送Telegram消息
         :param title: 消息标题
@@ -705,7 +692,7 @@ class Telegram:
             reply_markup = None
             if buttons:
                 reply_markup = self._create_inline_keyboard(buttons)
-            elif force_reply and ForceReply:
+            elif force_reply:
                 reply_markup = self._create_force_reply_markup()
 
             if rich_message:
@@ -737,15 +724,7 @@ class Telegram:
                     ),
                 )
                 self._stop_typing_if_needed(chat_id, stop_typing)
-                if sent and hasattr(sent, "message_id"):
-                    return {
-                        "success": True,
-                        "message_id": sent.message_id,
-                        "chat_id": sent.chat.id if hasattr(sent, "chat") else chat_id,
-                    }
-                if sent:
-                    return {"success": True}
-                return {"success": False}
+                return self._sent_result(sent, chat_id)
 
             # 判断是编辑消息还是发送新消息
             if original_message_id and original_chat_id:
@@ -760,15 +739,7 @@ class Telegram:
                         reply_to_message_id=original_message_id,
                     )
                     self._stop_typing_if_needed(chat_id, stop_typing)
-                    if sent and hasattr(sent, "message_id"):
-                        return {
-                            "success": True,
-                            "message_id": sent.message_id,
-                            "chat_id": sent.chat.id if hasattr(sent, "chat") else chat_id,
-                        }
-                    elif sent:
-                        return {"success": True}
-                    return {"success": False}
+                    return self._sent_result(sent, chat_id)
                 # 编辑消息
                 result = self.__edit_message(
                     original_chat_id,
@@ -796,15 +767,7 @@ class Telegram:
                     parse_mode=parse_mode,
                 )
                 self._stop_typing_if_needed(chat_id, stop_typing)
-                if sent and hasattr(sent, "message_id"):
-                    return {
-                        "success": True,
-                        "message_id": sent.message_id,
-                        "chat_id": sent.chat.id if hasattr(sent, "chat") else chat_id,
-                    }
-                elif sent:
-                    return {"success": True}
-                return {"success": False}
+                return self._sent_result(sent, chat_id)
 
         except Exception as msg_e:
             logger.error(f"发送消息失败：{msg_e}")
@@ -812,16 +775,24 @@ class Telegram:
             return {"success": False}
 
     @staticmethod
-    def _create_force_reply_markup():
-        if not ForceReply:
-            return None
-        try:
-            return ForceReply(selective=True, input_field_placeholder="请输入内容")
-        except TypeError:
-            try:
-                return ForceReply(selective=True)
-            except TypeError:
-                return ForceReply()
+    def _sent_result(sent: Any, chat_id: Union[str, int]) -> JsonDict:
+        """把 Bot API 返回的 Message 转为发送结果；非 Message 的真值只表示成功。"""
+        if isinstance(sent, dict) and "message_id" in sent:
+            return {
+                "success": True,
+                "message_id": sent["message_id"],
+                "chat_id": sent.get("chat", {}).get("id", chat_id),
+            }
+        return {"success": bool(sent)}
+
+    @staticmethod
+    def _create_force_reply_markup() -> JsonDict:
+        """生成只对被回复用户生效的 ForceReply 标记。"""
+        return {
+            "force_reply": True,
+            "selective": True,
+            "input_field_placeholder": "请输入内容",
+        }
 
     def send_voice(
             self,
@@ -831,7 +802,7 @@ class Telegram:
             original_chat_id: Optional[str] = None,
             stop_typing: bool = False,
             parse_mode: Optional[str] = None,
-    ) -> Optional[dict]:
+    ) -> Optional[JsonDict]:
         """
         发送Telegram语音消息。
         """
@@ -848,7 +819,7 @@ class Telegram:
 
         try:
             with voice_file.open("rb") as fp:
-                sent = self._bot.send_voice(
+                sent = self._api.send_voice(
                     chat_id=chat_id,
                     voice=fp,
                     caption=self._prepare_text(caption, parse_mode),
@@ -856,13 +827,7 @@ class Telegram:
                     **self._topic_kwargs(chat_id),
                 )
             self._stop_typing_if_needed(chat_id, stop_typing)
-            if sent and hasattr(sent, "message_id"):
-                return {
-                    "success": True,
-                    "message_id": sent.message_id,
-                    "chat_id": sent.chat.id if hasattr(sent, "chat") else chat_id,
-                }
-            return {"success": bool(sent)}
+            return self._sent_result(sent, chat_id)
         except Exception as err:
             logger.error(f"发送语音消息失败：{err}")
             self._stop_typing_if_needed(chat_id, stop_typing)
@@ -883,7 +848,7 @@ class Telegram:
             original_chat_id: Optional[str] = None,
             stop_typing: bool = False,
             parse_mode: Optional[str] = None,
-    ) -> Optional[dict]:
+    ) -> Optional[JsonDict]:
         """
         发送本地图片或文件给 Telegram 用户。
         """
@@ -913,7 +878,7 @@ class Telegram:
 
             with local_file.open("rb") as fp:
                 if is_image:
-                    sent = self._bot.send_photo(
+                    sent = self._api.send_photo(
                         chat_id=chat_id,
                         photo=fp,
                         caption=self._prepare_text(caption, parse_mode),
@@ -921,7 +886,7 @@ class Telegram:
                         **self._topic_kwargs(chat_id),
                     )
                 else:
-                    sent = self._bot.send_document(
+                    sent = self._api.send_document(
                         chat_id=chat_id,
                         document=(send_name, fp),
                         caption=self._prepare_text(caption, parse_mode),
@@ -929,13 +894,7 @@ class Telegram:
                         **self._topic_kwargs(chat_id),
                     )
             self._stop_typing_if_needed(chat_id, stop_typing)
-            if sent and hasattr(sent, "message_id"):
-                return {
-                    "success": True,
-                    "message_id": sent.message_id,
-                    "chat_id": sent.chat.id if hasattr(sent, "chat") else chat_id,
-                }
-            return {"success": bool(sent)}
+            return self._sent_result(sent, chat_id)
         except Exception as err:
             logger.error(f"发送本地附件失败: {err}")
             self._stop_typing_if_needed(chat_id, stop_typing)
@@ -1163,11 +1122,11 @@ class Telegram:
             self._stop_typing_if_needed(chat_id, stop_typing)
 
     @staticmethod
-    def _create_inline_keyboard(buttons: List[List[Dict]]) -> InlineKeyboardMarkup:
+    def _create_inline_keyboard(buttons: List[List[Dict]]) -> JsonDict:
         """
         创建内联键盘
         :param buttons: 按钮配置，格式：[[{"text": "按钮文本", "callback_data": "回调数据", "url": "链接"}]]
-        :return: InlineKeyboardMarkup对象
+        :return: Bot API InlineKeyboardMarkup 结构
         """
         keyboard = []
         for row in buttons:
@@ -1175,15 +1134,13 @@ class Telegram:
             for button in row:
                 if "url" in button:
                     # URL按钮
-                    btn = InlineKeyboardButton(text=button["text"], url=button["url"])
+                    btn = {"text": button["text"], "url": button["url"]}
                 else:
                     # 回调按钮
-                    btn = InlineKeyboardButton(
-                        text=button["text"], callback_data=button["callback_data"]
-                    )
+                    btn = {"text": button["text"], "callback_data": button["callback_data"]}
                 button_row.append(btn)
             keyboard.append(button_row)
-        return InlineKeyboardMarkup(keyboard)
+        return {"inline_keyboard": keyboard}
 
     def answer_callback_query(
             self,
@@ -1198,7 +1155,7 @@ class Telegram:
             return None
 
         try:
-            self._bot.answer_callback_query(
+            self._api.answer_callback_query(
                 callback_query_id, text=text, show_alert=show_alert
             )
             return True
@@ -1226,7 +1183,7 @@ class Telegram:
                 target_chat_id = self._telegram_chat_id
 
             # 删除消息
-            result = self._bot.delete_message(
+            result = self._api.delete_message(
                 chat_id=target_chat_id, message_id=int(message_id)
             )
             if result:
@@ -1330,7 +1287,7 @@ class Telegram:
             chat_id: str,
             message_id: int,
             text: str,
-            reply_markup: Optional[InlineKeyboardMarkup] = None,
+            reply_markup: Optional[JsonDict] = None,
             disable_web_page_preview: Optional[bool] = None,
             parse_mode: Optional[str] = None,
     ) -> bool:
@@ -1350,7 +1307,7 @@ class Telegram:
                 disable_web_page_preview
             )
         try:
-            self._bot.edit_message_text(**edit_text_kwargs)
+            self._api.edit_message_text(**edit_text_kwargs)
         except Exception as err:
             if self.__is_message_not_modified_error(err):
                 logger.debug(f"Telegram消息内容未变化，跳过编辑：{str(err)}")
@@ -1358,7 +1315,7 @@ class Telegram:
             if not self.__is_no_text_edit_error(err):
                 raise
             try:
-                self._bot.edit_message_caption(
+                self._api.edit_message_caption(
                     chat_id=chat_id,
                     message_id=message_id,
                     caption=prepared_text,
@@ -1407,12 +1364,14 @@ class Telegram:
 
             if image:
                 # 如果有图片，使用edit_message_media
-                media = InputMediaPhoto(
-                    media=image,
-                    caption=self._prepare_text(text, parse_mode),
-                    parse_mode=parse_mode,
-                )
-                self._bot.edit_message_media(
+                media = {
+                    "type": "photo",
+                    "media": image,
+                    "caption": self._prepare_text(text, parse_mode),
+                }
+                if parse_mode:
+                    media["parse_mode"] = parse_mode
+                self._api.edit_message_media(
                     chat_id=chat_id,
                     message_id=message_id,
                     media=media,
@@ -1454,16 +1413,20 @@ class Telegram:
     @staticmethod
     def _build_rich_message_chunks(
             rich_message: str,
-    ) -> List[TelebotInputRichMessage]:
+    ) -> List[JsonDict]:
         """
         将 GitHub 风格 Markdown 转换并拆分为 Telegram Rich Message。
 
         :param rich_message: 完整的 Rich Markdown 正文
-        :return: 满足 Telegram 字节数和块数量限制的消息片段
+        :return: 满足 Telegram 字节数和块数量限制的 InputRichMessage 结构
         """
         converted = richify(rich_message, mode="html")
         return [
-            TelebotInputRichMessage(**chunk.to_dict())
+            {
+                key: value
+                for key, value in chunk.to_dict().items()
+                if key in _RICH_MESSAGE_FIELDS and value is not None
+            }
             for chunk in split_rich(converted)
         ]
 
@@ -1472,7 +1435,7 @@ class Telegram:
             chat_id: Union[str, int],
             message_id: Union[str, int],
             rich_message: str,
-            reply_markup: Optional[InlineKeyboardMarkup] = None,
+            reply_markup: Optional[JsonDict] = None,
     ) -> bool:
         """
         编辑 Telegram Rich Message。
@@ -1488,7 +1451,7 @@ class Telegram:
             logger.warning("Telegram Rich Message 超出单条限制，无法编辑原消息")
             return False
         try:
-            self._bot.edit_message_text(
+            self._api.edit_message_text(
                 chat_id=chat_id,
                 message_id=int(message_id),
                 text=None,
@@ -1508,7 +1471,7 @@ class Telegram:
             self,
             chat_id: Union[str, int],
             rich_message: str,
-            reply_markup: Optional[InlineKeyboardMarkup] = None,
+            reply_markup: Optional[JsonDict] = None,
             reply_to_message_id: Optional[Union[str, int]] = None,
     ) -> Any:
         """
@@ -1522,14 +1485,14 @@ class Telegram:
         """
         chunks = self._build_rich_message_chunks(rich_message)
         reply_parameters = (
-            ReplyParameters(message_id=int(reply_to_message_id))
+            {"message_id": int(reply_to_message_id)}
             if reply_to_message_id is not None
             else None
         )
         sent = None
         try:
             for index, chunk in enumerate(chunks):
-                sent = self._bot.send_rich_message(
+                sent = self._api.send_rich_message(
                     chat_id=chat_id,
                     rich_message=chunk,
                     reply_markup=reply_markup if index == 0 else None,
@@ -1545,7 +1508,7 @@ class Telegram:
             userid: Optional[str] = None,
             image="",
             caption="",
-            reply_markup: Optional[InlineKeyboardMarkup] = None,
+            reply_markup: Optional[JsonDict] = None,
             disable_web_page_preview: Optional[bool] = None,
             parse_mode: Optional[str] = None,
             reply_to_message_id: Optional[int] = None,
@@ -1559,7 +1522,7 @@ class Telegram:
         :return: 发送成功返回消息对象，失败返回None
         """
         parse_mode = self._normalize_parse_mode(parse_mode)
-        kwargs = {
+        kwargs: Dict[str, Any] = {
             "chat_id": userid or self._telegram_chat_id,
             "parse_mode": parse_mode,
             "reply_markup": reply_markup,
@@ -1620,13 +1583,13 @@ class Telegram:
         parse_mode = kwargs.get("parse_mode")
         try:
             if image:
-                return self._bot.send_photo(
+                return self._api.send_photo(
                     photo=image,
                     caption=self._prepare_text(caption, parse_mode),
                     **kwargs,
                 )
             else:
-                return self._bot.send_message(
+                return self._api.send_message(
                     text=self._prepare_text(caption, parse_mode),
                     disable_web_page_preview=disable_web_page_preview,
                     **kwargs
@@ -1653,7 +1616,7 @@ class Telegram:
             for index, chunk in enumerate(chunks):
                 current_reply_markup = reply_markup if index == 0 else None
                 if image and index == 0:
-                    ret = self._bot.send_photo(
+                    ret = self._api.send_photo(
                         **kwargs,
                         photo=image,
                         caption=chunk,
@@ -1663,7 +1626,7 @@ class Telegram:
                 msg_kwargs = dict(**kwargs)
                 if disable_web_page_preview is not None:
                     msg_kwargs["disable_web_page_preview"] = disable_web_page_preview
-                ret = self._bot.send_message(
+                ret = self._api.send_message(
                     **msg_kwargs,
                     text=chunk,
                     reply_markup=current_reply_markup,
@@ -1702,14 +1665,14 @@ class Telegram:
                     msg_kwargs = dict(**kwargs)
                     if disable_web_page_preview is not None:
                         msg_kwargs["disable_web_page_preview"] = disable_web_page_preview
-                    ret = self._bot.send_message(
+                    ret = self._api.send_message(
                         **msg_kwargs,
                         text=self._telegramify_item_text(item),
                         reply_markup=current_reply_markup,
                     )
 
                 elif item.content_type == ContentTypes.PHOTO or (image and i == 0):
-                    ret = self._bot.send_photo(
+                    ret = self._api.send_photo(
                         **kwargs,
                         photo=(
                             getattr(item, "file_name", ""),
@@ -1720,7 +1683,7 @@ class Telegram:
                     )
 
                 elif item.content_type == ContentTypes.FILE:
-                    ret = self._bot.send_document(
+                    ret = self._api.send_document(
                         **kwargs,
                         document=(item.file_name, item.file_data),
                         caption=self._telegramify_item_caption(item),
@@ -1745,10 +1708,10 @@ class Telegram:
             return
         # 设置bot命令
         if commands:
-            self._bot.delete_my_commands()
-            self._bot.set_my_commands(
+            self._api.delete_my_commands()
+            self._api.set_my_commands(
                 commands=[
-                    BotCommand(cmd[1:], str(desc.get("description")))
+                    {"command": cmd[1:], "description": str(desc.get("description"))}
                     for cmd, desc in commands.items()
                 ]
             )
@@ -1760,27 +1723,11 @@ class Telegram:
         if not self._bot:
             return
         # 清理菜单命令
-        self._bot.delete_my_commands()
-
-    @staticmethod
-    def _stop_bot_with_deadline(bot: TeleBot, deadline: float) -> bool:
-        """停止 SDK polling，并在共享 deadline 内等待 worker 收敛。"""
-        bot.stop_polling()
-        if not bot.threaded or not bot.worker_pool:
-            return True
-
-        workers = tuple(bot.worker_pool.workers)
-        for worker in workers:
-            worker.stop()
-        for worker in workers:
-            if worker is threading.current_thread():
-                continue
-            worker.join(timeout=max(0.0, deadline - time.monotonic()))
-        return all(not worker.is_alive() for worker in workers)
+        self._api.delete_my_commands()
 
     def stop(self) -> bool:
         """
-        停止 Telegram 消息接收服务，并返回 SDK/polling/typing owner 是否收敛。
+        停止 Telegram 消息接收服务，并返回 polling/typing owner 是否收敛。
         """
         converged = True
         with self._typing_lifecycle_lock:
@@ -1792,25 +1739,18 @@ class Telegram:
 
         bot = self._bot
         polling_thread = self._polling_thread
-        deadline = time.monotonic() + self._shutdown_timeout_seconds
-        transport_converged = True
         if bot:
-            if not self._stop_bot_with_deadline(bot, deadline):
-                converged = False
-                transport_converged = False
-                logger.error("Telegram SDK worker 未在关闭预算内退出")
-        if (
-            polling_thread
-            and polling_thread.is_alive()
-            and polling_thread is not threading.current_thread()
-        ):
-            polling_thread.join(timeout=max(0.0, deadline - time.monotonic()))
-        if polling_thread and polling_thread.is_alive():
-            logger.error("Telegram polling 线程未在关闭预算内退出")
-            converged = False
-            transport_converged = False
-        if not transport_converged:
-            return False
+            # 在 handler 内触发的 stop 不等待自身：handler 返回后轮询循环即看到停止标记。
+            in_polling_thread = polling_thread is threading.current_thread()
+            dispatch_idle = bot.stop_polling(
+                0 if in_polling_thread else self._shutdown_timeout_seconds
+            )
+            if not dispatch_idle and not in_polling_thread:
+                # 保留 owner，调用方重试 stop 时继续等待同一个 handler。
+                logger.error("Telegram 消息处理未在关闭预算内结束")
+                return False
+            # 进行中的 getUpdates 不再等待；它返回后只会丢弃结果并退出线程。
+            bot.close()
         self._polling_thread = None
         self._bot = None
         logger.info("Telegram消息接收服务已停止")

app/startup/lifecycle/__init__.py

@@ -13,18 +13,6 @@ from app.startup.initializers.cache import configure_cache_dependencies
 
 # 缓存装饰器会在业务模块导入时创建后端，必须先完成适配器装配。
 configure_cache_dependencies()
-# urllib3-future 覆盖 urllib3 命名空间后删除了 format_header_param，导致 telebot 崩溃，需在加载模块前打补丁
-try:
-    import urllib3.fields as _urllib3_fields
-
-    if not hasattr(_urllib3_fields, "format_header_param") and hasattr(
-        _urllib3_fields, "format_header_param_rfc2231"
-    ):
-        _urllib3_fields.format_header_param = (
-            _urllib3_fields.format_header_param_rfc2231
-        )
-except Exception:
-    pass
 
 from app.adapters.external.server import MoviePilotServerHelper
 from app.adapters.network.http import (

docs/architecture-overview.md

@@ -743,7 +743,7 @@ flowchart LR
   `--write-host` 不会替代人工决策。
 - 同一 baseline 的 `direct_adapter_imports` 记录现存原始直连；Application 与 Chain 均已清零，
   policy 目标为空集合。新增、替换、删除后未清理 policy 都会失败。
-- `direct_egress` 记录全宿主 53 条 raw transport、network SDK 和协议操作 identity；普通 HTTP/
+- `direct_egress` 记录全宿主 50 条 raw transport、network SDK 和协议操作 identity；普通 HTTP/
   Session bridge 与 Application DNS I/O 债务已清零，其余 canonical transport、SDK、
   stream/vendor/diagnostic/control-plane 事实是精确 containment。每条初始边的指纹由测试独立冻结，
   bindings/uses 变化、分类互换、通配导入和初始边增长都会失败；债务删除时同步删除冻结项以禁止恢复，
@@ -763,17 +763,17 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 1056 |
-| 内部导入边 | 9,022 |
+| 内部导入边 | 9,027 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
-| Direct egress | 53（债务已清零，53 条精确 containment） |
+| Direct egress | 50（债务已清零，50 条精确 containment） |
 | Module Contract V2 spec | 215（其中 211 个进入 `run_module` 观察面） |
 | Event Contract | 53 |
 | Event producer / consumer | 86（85 静态、1 动态）/ 17（16 静态、1 动态） |
 | Model/Oper 自动事务与自建 Session | 0 |
 | 组合根外 `SystemConfigOper()` | 0 |
 
-整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责；宿主依赖边现为 9,022，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。当前 `app.startup.lifecycle` 为 576、`app.factory` 为 588、`app.main` 为 590。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
+整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责；宿主依赖边现为 9,027，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。当前 `app.startup.lifecycle` 为 576、`app.factory` 为 588、`app.main` 为 590。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。
 
 架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
 语义门禁，`Check host architecture snapshot` 再执行快照测试及一次

docs/refactor/optimization-checklist.md

@@ -96,7 +96,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 1056 / 9,022 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档 |
+| 宿主 Python 模块 / 内部依赖边 | 1056 / 9,027 | `dependency-baseline.json` 当前快照；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
@@ -104,8 +104,8 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
 | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 |
 | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 |
-| 全量 mypy 历史债务 | 9,243 / 508 文件 | 删除独立活动日志中间件后的现状基线；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
-| Ruff 历史诊断 | 516 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
+| 全量 mypy 历史债务 | 9,232 / 508 文件 | 删除独立活动日志中间件后的现状基线；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 |
+| Ruff 历史诊断 | 515 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
 | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件
@@ -543,7 +543,7 @@ S4/ARCH-110/111 的取消不撤销现有防回退门禁。后续按路线图的
   具体 cache adapter；Memory/Redis 均实现严格 `AtomicCacheBackend.store/consume`。
 - S0-L2.4b 已为 LLM streaming、第三方 SDK、移植库和本地控制面建立完整 egress identity 与
   zero-growth policy；11 条普通 HTTP/Session bridge 和 1 条 Application DNS I/O 债务已清零。
-  当前 53 条出口事实均为精确 containment，每条指纹由独立上界冻结，不能靠同时刷新
+  当前 50 条出口事实均为精确 containment，每条指纹由独立上界冻结，不能靠同时刷新
   baseline/policy 掩盖调用面增长，已删除债务也不得恢复。
 
 **目标与步骤**

pyproject.toml

@@ -65,6 +65,7 @@ dependencies = [
     "pyparsing~=3.3.2",
     "pyquery~=2.0.1",
     "pystray~=0.19.5",
+    # 宿主 Telegram 渠道已改用自研 Bot API 客户端，此 SDK 暂时保留给插件使用。
     "pytelegrambotapi~=4.36.0",
     "python-dateutil~=2.9.0.post0",
     "python-dotenv~=1.2.2",

tests/fixtures/architecture/complexity-baseline.json

@@ -114,8 +114,6 @@
       "app/modules/rtorrent/__init__.py:RtorrentModule.list_torrents": 20,
       "app/modules/slack/__init__.py:SlackModule.message_parser": 16,
       "app/modules/slack/slack.py:Slack.send_medias_msg": 18,
-      "app/modules/telegram/telegram.py:Telegram.__init__": 17,
-      "app/modules/telegram/telegram.py:Telegram.send_msg": 20,
       "app/modules/transmission/__init__.py:TransmissionModule.download": 19,
       "app/modules/transmission/__init__.py:TransmissionModule.list_torrents": 26,
       "app/modules/trimemedia/api.py:Api.request": 16,

tests/fixtures/architecture/dependency-baseline.json

@@ -43,11 +43,11 @@
       "app.application": 0,
       "app.chain": 0
     },
-    "count": 53,
+    "count": 50,
     "counts_by_kind": {
-      "network_sdk": 26,
+      "network_sdk": 25,
       "protocol_operation": 7,
-      "raw_transport": 20
+      "raw_transport": 18
     },
     "entries": [
       {
@@ -860,77 +860,6 @@
           "Slack.__init__|call:WebClient"
         ]
       },
-      {
-        "bindings": [
-          "from:urllib3.fields as fields"
-        ],
-        "fingerprint": "18862bbdb252b59573f57ea776b5d64bfb775e7739603fecd23cc8f2c38a7e0d",
-        "kind": "raw_transport",
-        "source": "app.modules.telegram.compat",
-        "target": "urllib3",
-        "uses": []
-      },
-      {
-        "bindings": [
-          "from:telebot.TeleBot as TeleBot",
-          "from:telebot.apihelper as apihelper",
-          "from:telebot.types.BotCommand as BotCommand",
-          "from:telebot.types.ForceReply as ForceReply",
-          "from:telebot.types.InlineKeyboardButton as InlineKeyboardButton",
-          "from:telebot.types.InlineKeyboardMarkup as InlineKeyboardMarkup",
-          "from:telebot.types.InputMediaPhoto as InputMediaPhoto",
-          "from:telebot.types.InputRichMessage as TelebotInputRichMessage",
-          "from:telebot.types.ReplyParameters as ReplyParameters"
-        ],
-        "fingerprint": "78f5ab18bfd67ba4fa0f3c0fc4a1a561a7b1d9e81e335fb80de86edb480b84b8",
-        "kind": "network_sdk",
-        "source": "app.modules.telegram.telegram",
-        "target": "telebot",
-        "uses": [
-          "Telegram.__edit_message_text_or_caption|call:edit_message_caption",
-          "Telegram.__edit_message_text_or_caption|call:edit_message_text",
-          "Telegram.__edit_message|call:edit_message_media",
-          "Telegram.__edit_message|call:types.InputMediaPhoto",
-          "Telegram.__edit_rich_message|call:edit_message_text",
-          "Telegram.__init__.callback_query|call:answer_callback_query",
-          "Telegram.__init__.callback_query|call:answer_callback_query",
-          "Telegram.__init__.run_polling|call:infinity_polling",
-          "Telegram.__init__.send_welcome|call:reply_to",
-          "Telegram.__init__|call:TeleBot",
-          "Telegram.__init__|call:callback_query_handler",
-          "Telegram.__init__|call:get_me",
-          "Telegram.__init__|call:message_handler",
-          "Telegram.__init__|call:message_handler",
-          "Telegram.__send_long_message|call:send_document",
-          "Telegram.__send_long_message|call:send_message",
-          "Telegram.__send_long_message|call:send_photo",
-          "Telegram.__send_long_plain_message|call:send_message",
-          "Telegram.__send_long_plain_message|call:send_photo",
-          "Telegram.__send_rich_message|call:send_rich_message",
-          "Telegram.__send_rich_message|call:types.ReplyParameters",
-          "Telegram.__send_short_message|call:send_message",
-          "Telegram.__send_short_message|call:send_photo",
-          "Telegram._build_rich_message_chunks|call:types.InputRichMessage",
-          "Telegram._create_force_reply_markup|call:types.ForceReply",
-          "Telegram._create_force_reply_markup|call:types.ForceReply",
-          "Telegram._create_force_reply_markup|call:types.ForceReply",
-          "Telegram._create_inline_keyboard|call:types.InlineKeyboardButton",
-          "Telegram._create_inline_keyboard|call:types.InlineKeyboardButton",
-          "Telegram._create_inline_keyboard|call:types.InlineKeyboardMarkup",
-          "Telegram._start_typing_task.typing_worker|call:send_chat_action",
-          "Telegram.answer_callback_query|call:answer_callback_query",
-          "Telegram.delete_commands|call:delete_my_commands",
-          "Telegram.delete_msg|call:delete_message",
-          "Telegram.download_file|call:apihelper.FILE_URL.format",
-          "Telegram.download_file|call:get_file",
-          "Telegram.register_commands|call:delete_my_commands",
-          "Telegram.register_commands|call:set_my_commands",
-          "Telegram.register_commands|call:types.BotCommand",
-          "Telegram.send_file|call:send_document",
-          "Telegram.send_file|call:send_photo",
-          "Telegram.send_voice|call:send_voice"
-        ]
-      },
       {
         "bindings": [
           "import:requests as requests",
@@ -1013,16 +942,6 @@
           "_NetworkTestTransportAdapter.websocket.connect|call:create_connection"
         ]
       },
-      {
-        "bindings": [
-          "import:urllib3.fields as _urllib3_fields"
-        ],
-        "fingerprint": "cb6f0a314aeb1e2d3e76c240aa20460ac0c36d9f5c18c1a6ea3170f64dd3366b",
-        "kind": "raw_transport",
-        "source": "app.startup.lifecycle",
-        "target": "urllib3",
-        "uses": []
-      },
       {
         "bindings": [
           "import:socket as socket"
@@ -1102,8 +1021,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 9022,
-  "edge_sha256": "bd56aa30ca580c8305cf18ef0d2433449239e012b99764ac43e28becbdac712c",
+  "edge_count": 9027,
+  "edge_sha256": "3f8eaae519865c26335851f18585d73b8a1d7536e20b3d4cc3e8c64bc8ec098b",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -7861,6 +7780,11 @@
     "app.modules.synologychat.synologychat -> app.foundation.url",
     "app.modules.synologychat.synologychat -> app.runtime",
     "app.modules.synologychat.synologychat -> app.runtime.log",
+    "app.modules.telegram.botapi -> app.adapters",
+    "app.modules.telegram.botapi -> app.adapters.network",
+    "app.modules.telegram.botapi -> app.adapters.network.http",
+    "app.modules.telegram.botapi -> app.runtime",
+    "app.modules.telegram.botapi -> app.runtime.log",
     "app.modules.telegram.module -> app.application",
     "app.modules.telegram.module -> app.application.messaging",
     "app.modules.telegram.module -> app.application.messaging.channel",
@@ -7893,7 +7817,7 @@
     "app.modules.telegram.telegram -> app.foundation.size",
     "app.modules.telegram.telegram -> app.modules",
     "app.modules.telegram.telegram -> app.modules.telegram",
-    "app.modules.telegram.telegram -> app.modules.telegram.compat",
+    "app.modules.telegram.telegram -> app.modules.telegram.botapi",
     "app.modules.telegram.telegram -> app.runtime",
     "app.modules.telegram.telegram -> app.runtime.execution",
     "app.modules.telegram.telegram -> app.runtime.log",
@@ -10885,7 +10809,7 @@
     "app.modules.synologychat",
     "app.modules.synologychat.synologychat",
     "app.modules.telegram",
-    "app.modules.telegram.compat",
+    "app.modules.telegram.botapi",
     "app.modules.telegram.module",
     "app.modules.telegram.telegram",
     "app.modules.theaudiodb",

tests/fixtures/architecture/dependency-policy.json

@@ -147,7 +147,6 @@
           {"source": "app.modules.qbittorrent.qbittorrent", "target": "qbittorrentapi", "kind": "network_sdk", "fingerprint": "b2f5a27f0c54cf95ed42fe99848fd6c9c0641caca8bc4ff8006a411cb427066d"},
           {"source": "app.modules.slack.slack", "target": "slack_bolt", "kind": "network_sdk", "fingerprint": "be5dcb032ece8d8627abeb243f98143aaf60f26751ab6e5e098813d4048419e6"},
           {"source": "app.modules.slack.slack", "target": "slack_sdk", "kind": "network_sdk", "fingerprint": "7559f31e4172ad3bbbaf161e1164ea48b997c56d05c77b49b82626298c39aa14"},
-          {"source": "app.modules.telegram.telegram", "target": "telebot", "kind": "network_sdk", "fingerprint": "78f5ab18bfd67ba4fa0f3c0fc4a1a561a7b1d9e81e335fb80de86edb480b84b8"},
           {"source": "app.modules.transmission.transmission", "target": "transmission_rpc", "kind": "network_sdk", "fingerprint": "1652e661cb17dbadb039fdc4ab73d6d06e47eb118292ba3693313e45834959cf"},
           {"source": "app.modules.webpush", "target": "pywebpush", "kind": "network_sdk", "fingerprint": "389c73b06150e3d5bcaf31f35a25178873d2ed38ef9a28354cb9bab691eeab76"}
         ]
@@ -205,9 +204,7 @@
           {"source": "app.modules.filemanager.storages.smb", "target": "smbprotocol", "kind": "network_sdk", "fingerprint": "f02686afd99c59820dffa7b4c2627a0be1ad62980c98a9cd33b7564697842b22"},
           {"source": "app.modules.jellyfin.jellyfin", "target": "requests", "kind": "raw_transport", "fingerprint": "5c46d09ca9a4bcc0bae21ca5d554ef09baa3f901c562b27c5f5ee1439c7746b5"},
           {"source": "app.modules.rtorrent.rtorrent", "target": "http.client", "kind": "raw_transport", "fingerprint": "6adc93b3bc479dfe81197554c7977930a60abc51714f14ef29298eb40730aaf7"},
-          {"source": "app.modules.telegram.compat", "target": "urllib3", "kind": "raw_transport", "fingerprint": "18862bbdb252b59573f57ea776b5d64bfb775e7739603fecd23cc8f2c38a7e0d"},
-          {"source": "app.modules.zspace.zspace", "target": "requests", "kind": "raw_transport", "fingerprint": "9df3fd27b9696d45a72e7c8f67b5a9ad79a7371d1fe690bbaa17485bd1960d51"},
-          {"source": "app.startup.lifecycle", "target": "urllib3", "kind": "raw_transport", "fingerprint": "cb6f0a314aeb1e2d3e76c240aa20460ac0c36d9f5c18c1a6ea3170f64dd3366b"}
+          {"source": "app.modules.zspace.zspace", "target": "requests", "kind": "raw_transport", "fingerprint": "9df3fd27b9696d45a72e7c8f67b5a9ad79a7371d1fe690bbaa17485bd1960d51"}
         ]
       },
       {

tests/fixtures/architecture/mypy-baseline.json

@@ -2047,10 +2047,10 @@
     "assignment": 2,
     "misc": 4,
     "no-any-return": 8,
-    "no-untyped-call": 2,
-    "no-untyped-def": 19,
+    "no-untyped-call": 1,
+    "no-untyped-def": 13,
     "return-value": 1,
-    "type-arg": 14,
+    "type-arg": 10,
     "union-attr": 1,
     "var-annotated": 1
   },

tests/fixtures/architecture/ruff-baseline.json

@@ -350,7 +350,7 @@
     "I001": 1
   },
   "app/modules/telegram/telegram.py": {
-    "I001": 3
+    "I001": 2
   },
   "app/modules/themoviedb/cache.py": {
     "I001": 1

tests/test_architecture_egress.py

@@ -47,7 +47,6 @@ FROZEN_EGRESS_EDGES_BY_REASON = {
         ("app.modules.qbittorrent.qbittorrent", "qbittorrentapi"),
         ("app.modules.slack.slack", "slack_bolt"),
         ("app.modules.slack.slack", "slack_sdk"),
-        ("app.modules.telegram.telegram", "telebot"),
         ("app.modules.transmission.transmission", "transmission_rpc"),
         ("app.modules.webpush", "pywebpush"),
     },
@@ -75,9 +74,7 @@ FROZEN_EGRESS_EDGES_BY_REASON = {
         ("app.modules.filemanager.storages.smb", "smbprotocol"),
         ("app.modules.jellyfin.jellyfin", "requests"),
         ("app.modules.rtorrent.rtorrent", "http.client"),
-        ("app.modules.telegram.compat", "urllib3"),
         ("app.modules.zspace.zspace", "requests"),
-        ("app.startup.lifecycle", "urllib3"),
     },
     "test_network_guard": {
         ("app.testing.network", "socket.getaddrinfo"),
@@ -132,15 +129,12 @@ FROZEN_EGRESS_FINGERPRINT_BY_EDGE = {
     ("app.modules.rtorrent.rtorrent", "xmlrpc.client.ServerProxy"): "2ac64b5670c930bc28bb2135cc4b73891c1cae01a74a9586f74e302878c45c85",
     ("app.modules.slack.slack", "slack_bolt"): "be5dcb032ece8d8627abeb243f98143aaf60f26751ab6e5e098813d4048419e6",
     ("app.modules.slack.slack", "slack_sdk"): "7559f31e4172ad3bbbaf161e1164ea48b997c56d05c77b49b82626298c39aa14",
-    ("app.modules.telegram.compat", "urllib3"): "18862bbdb252b59573f57ea776b5d64bfb775e7739603fecd23cc8f2c38a7e0d",
-    ("app.modules.telegram.telegram", "telebot"): "78f5ab18bfd67ba4fa0f3c0fc4a1a561a7b1d9e81e335fb80de86edb480b84b8",
     ("app.modules.themoviedb.tmdbv3api.tmdb", "requests"): "d605eb176a203b3f4d205c5d183469b3002426682eb9cb4e7408bb6013652484",
     ("app.modules.transmission.transmission", "transmission_rpc"): "1652e661cb17dbadb039fdc4ab73d6d06e47eb118292ba3693313e45834959cf",
     ("app.modules.webpush", "pywebpush"): "389c73b06150e3d5bcaf31f35a25178873d2ed38ef9a28354cb9bab691eeab76",
     ("app.modules.wechat.wechatbot", "websocket"): "1bae78270eadce0571e2caaa111a5c0a9065ba2da97ebb26b8a5b76d3ed5eef6",
     ("app.startup.composition.network", "websocket"): "09e1909ac750fe40b145dde8fee52eedf84aaf2123f6a1bc38109252c89d6000",
     ("app.modules.zspace.zspace", "requests"): "9df3fd27b9696d45a72e7c8f67b5a9ad79a7371d1fe690bbaa17485bd1960d51",
-    ("app.startup.lifecycle", "urllib3"): "cb6f0a314aeb1e2d3e76c240aa20460ac0c36d9f5c18c1a6ea3170f64dd3366b",
     ("app.testing.network", "socket.getaddrinfo"): "ebf33718b54c81e1f575401da4a0bef5aa0e1ddc284e201f557897feeca71c6d",
 }
 

tests/test_module_lifecycle.py

@@ -5,7 +5,6 @@ import time
 from unittest.mock import Mock, patch
 
 import pytest
-from telebot import TeleBot
 
 from app.modules import _MessageBase
 from app.modules.discord import DiscordModule
@@ -19,6 +18,7 @@ from app.modules.qqbot.module import QQBotModule
 from app.modules.qqbot.qqbot import QQBot
 from app.modules.slack import SlackModule
 from app.modules.slack.slack import Slack
+from app.modules.telegram.botapi import TelegramBotApi
 from app.modules.telegram.module import TelegramModule
 from app.modules.telegram.telegram import Telegram
 from app.modules.themoviedb import TheMovieDbModule
@@ -182,104 +182,116 @@ def test_module_stop_isolates_each_service_instance(
     getattr(healthy_client, stop_method).assert_called_once_with()
 
 
-def test_telegram_stop_closes_sdk_and_waits_for_polling_thread():
-    """客户端停止完成后不得保留 SDK worker 或 polling 线程句柄。"""
+def _polling_telegram_client(bot, handler=None):
+    """构造只含停止路径所需状态的 Telegram 客户端，并启动真实轮询线程。"""
     client = Telegram.__new__(Telegram)
-    bot = Mock()
-    bot.threaded = False
-    bot.worker_pool = None
     client._bot = bot
-    polling_thread = Mock()
-    polling_thread.is_alive.side_effect = [True, False]
-    client._polling_thread = polling_thread
     client._typing_tasks = {}
     client._typing_stop_flags = {}
     client._typing_lock = threading.RLock()
     client._typing_lifecycle_lock = threading.RLock()
     client._typing_accepting = True
-
-    assert client.stop() is True
-    assert client.stop() is True
-
-    bot.stop_polling.assert_called_once_with()
-    polling_thread.join.assert_called_once_with(
-        timeout=pytest.approx(client._shutdown_timeout_seconds, abs=0.1)
-    )
-    assert client._bot is None
-    assert client._polling_thread is None
-
-
-def test_telegram_stop_keeps_polling_owner_when_thread_misses_deadline():
-    """polling 超过关闭预算时必须返回未收敛并保留原 owner。"""
-    client = Telegram.__new__(Telegram)
-    bot = Mock()
-    bot.threaded = False
-    bot.worker_pool = None
-    polling_thread = Mock()
-    polling_thread.is_alive.return_value = True
-    client._bot = bot
-    client._polling_thread = polling_thread
-    client._shutdown_timeout_seconds = 0.01
-    client._typing_tasks = {}
-    client._typing_stop_flags = {}
-    client._typing_lock = threading.RLock()
-    client._typing_lifecycle_lock = threading.RLock()
-    client._typing_accepting = True
-
-    assert client.stop() is False
-
-    polling_thread.join.assert_called_once()
-    remaining_timeout = polling_thread.join.call_args.kwargs["timeout"]
-    assert 0 <= remaining_timeout <= client._shutdown_timeout_seconds
-    assert client._bot is bot
-    assert client._polling_thread is polling_thread
+    client._long_polling_timeout_seconds = 1
+    if handler is not None:
+        client._handle_update = handler
+    client._polling_thread = threading.Thread(target=client._run_polling, daemon=True)
+    client._polling_thread.start()
+    return client
 
 
 @pytest.mark.asyncio
-async def test_telegram_stop_bounds_real_sdk_worker_and_retries_after_release():
-    """真实 SDK worker 阻塞时应保留 owner，释放后重试可以完整收敛。"""
-    bot = TeleBot("123:test", threaded=True, num_threads=1)
+async def test_telegram_stop_does_not_wait_for_in_flight_long_poll():
+    """长轮询请求进行中时 stop 应立即收敛，请求返回的更新不得再分发。"""
+    bot = TelegramBotApi("123:test")
     entered = threading.Event()
     release = threading.Event()
+    handled = []
 
-    def blocking_callback() -> None:
+    def blocking_get_updates(offset, timeout):
         entered.set()
-        release.wait(timeout=1.0)
+        release.wait(timeout=2.0)
+        return [{"update_id": 1, "message": {"text": "late"}}]
 
-    bot.worker_pool.put(blocking_callback)
-    assert await asyncio.to_thread(entered.wait, 0.2)
-
-    client = Telegram.__new__(Telegram)
-    client._bot = bot
-    client._polling_thread = None
-    client._shutdown_timeout_seconds = 0.02
-    client._typing_tasks = {}
-    client._typing_stop_flags = {}
-    client._typing_lock = threading.RLock()
-    client._typing_lifecycle_lock = threading.RLock()
-    client._typing_accepting = True
-
-    heartbeat = asyncio.create_task(asyncio.sleep(0.005))
-    started_at = time.monotonic()
-    try:
-        assert await run_in_threadpool_to_completion(client.stop) is False
-        assert time.monotonic() - started_at < 0.2
-        assert heartbeat.done()
-        assert client._bot is bot
-        assert any(worker.is_alive() for worker in bot.worker_pool.workers)
-
-        release.set()
-        for worker in bot.worker_pool.workers:
-            await asyncio.to_thread(worker.join, 0.2)
+    with patch.object(bot, "get_updates", side_effect=blocking_get_updates), \
+            patch.object(bot, "close", wraps=bot.close) as close:
+        client = _polling_telegram_client(bot, handler=handled.append)
+        polling_thread = client._polling_thread
+        assert await asyncio.to_thread(entered.wait, 0.5)
 
+        started_at = time.monotonic()
         assert await run_in_threadpool_to_completion(client.stop) is True
+        assert time.monotonic() - started_at < 0.5
         assert client._bot is None
         assert client._polling_thread is None
-    finally:
+        close.assert_called_once_with()
+
         release.set()
-        for worker in bot.worker_pool.workers:
-            worker.stop()
-            await asyncio.to_thread(worker.join, 0.2)
+        await asyncio.to_thread(polling_thread.join, 1.0)
+    assert not polling_thread.is_alive()
+    assert handled == []
+
+
+@pytest.mark.asyncio
+async def test_telegram_stop_keeps_owner_until_running_handler_finishes():
+    """handler 超过关闭预算时返回未收敛并保留 owner，handler 结束后重试可收敛。"""
+    bot = TelegramBotApi("123:test")
+    entered = threading.Event()
+    release = threading.Event()
+    batches = [[{"update_id": 1}]]
+
+    def get_updates(offset, timeout):
+        if batches:
+            return batches.pop(0)
+        release.wait(timeout=2.0)
+        return []
+
+    def blocking_handler(update):
+        entered.set()
+        release.wait(timeout=2.0)
+
+    with patch.object(bot, "get_updates", side_effect=get_updates):
+        client = _polling_telegram_client(bot, handler=blocking_handler)
+        client._shutdown_timeout_seconds = 0.02
+        polling_thread = client._polling_thread
+        assert await asyncio.to_thread(entered.wait, 0.5)
+
+        heartbeat = asyncio.create_task(asyncio.sleep(0.005))
+        try:
+            assert await run_in_threadpool_to_completion(client.stop) is False
+            assert heartbeat.done()
+            assert client._bot is bot
+            assert client._polling_thread is polling_thread
+
+            release.set()
+            await asyncio.to_thread(polling_thread.join, 1.0)
+            assert await run_in_threadpool_to_completion(client.stop) is True
+            assert client._bot is None
+        finally:
+            release.set()
+            bot.stop_polling(0)
+            await asyncio.to_thread(polling_thread.join, 1.0)
+
+
+def test_telegram_stop_from_handler_does_not_wait_for_itself():
+    """handler 内触发的 stop（例如配置重载）不能等待自身持有的分发锁。"""
+    bot = TelegramBotApi("123:test")
+    results = []
+    holder = {}
+    ready = threading.Event()
+    done = threading.Event()
+
+    def handler(update):
+        ready.wait(1.0)
+        results.append(holder["client"].stop())
+        done.set()
+
+    with patch.object(bot, "get_updates", side_effect=[[{"update_id": 1}], []]):
+        holder["client"] = _polling_telegram_client(bot, handler=handler)
+        holder["client"]._shutdown_timeout_seconds = 5
+        ready.set()
+        assert done.wait(1.0)
+    assert results == [True]
+    assert holder["client"]._bot is None
 
 
 @pytest.mark.parametrize(

tests/test_telegram.py

@@ -20,21 +20,21 @@ from app.schemas.types import MediaType
 
 @pytest.fixture
 def telegram():
-    """构造 TeleBot 与 ImageHelper 均已打桩的 Telegram 实例。
+    """构造 Bot API 客户端与 ImageHelper 均已打桩的 Telegram 实例。
 
     空 token 会让 Telegram.__init__ 提前返回、致 send_* 抛错，故用假 bot 让初始化完整、
     消息发送走内存桩；ImageHelper 打桩避免 send_medias/send_msg 按 poster_path 真实下载海报
     （否则对 raw.githubusercontent.com 等外链发起真实 HTTP，外部 IO 不可接受且拖慢用例）。
     with 上下文在 fixture 结束时自动停桩，即使实例化失败也不泄漏 patch。
     """
-    with patch("app.modules.telegram.telegram.TeleBot") as mock_telebot_cls, \
+    with patch("app.modules.telegram.telegram.TelegramBotApi") as mock_telebot_cls, \
             patch("app.modules.telegram.telegram.ImageHelper") as mock_image_cls:
         bot_instance = MagicMock()
         # get_me 用于初始化 bot 用户名，需返回带 username 的对象
-        bot_instance.get_me.return_value = MagicMock(username="test_bot")
+        bot_instance.get_me.return_value = {"username": "test_bot"}
         # polling/stop 使用普通函数，避免后台线程执行 MagicMock 时在退出阶段产生锁竞争。
-        bot_instance.infinity_polling = lambda *args, **kwargs: None
-        bot_instance.stop_polling = lambda *args, **kwargs: None
+        bot_instance.polling = lambda *args, **kwargs: None
+        bot_instance.stop_polling = lambda *args, **kwargs: True
         mock_telebot_cls.return_value = bot_instance
         mock_image_cls.return_value.fetch_image.return_value = b"fake-image-bytes"
         telegram = Telegram(TELEGRAM_TOKEN="fake_token", TELEGRAM_CHAT_ID="fake_chat_id")
@@ -58,10 +58,10 @@ 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(
+        bot.get_me.return_value = {"username": "test_bot"}
+        bot.polling = lambda *args, **kwargs: None
+        bot.stop_polling = lambda *args, **kwargs: True
+    with patch("app.modules.telegram.telegram.TelegramBotApi", side_effect=bots), patch(
         "app.modules.telegram.telegram.ImageHelper"
     ):
         clients = [
@@ -99,7 +99,7 @@ def test_topic_id_is_scoped_to_channel_and_default_chat(tmp_path):
 
 def test_edit_msg_with_rich_message(telegram):
     """Telegram 流式编辑应继续使用 Rich Message 协议。"""
-    telegram._bot.edit_message_text.return_value = SimpleNamespace(message_id=101)
+    telegram._bot.edit_message_text.return_value = {"message_id": 101}
 
     result = telegram.edit_msg(
         chat_id="10001",
@@ -111,7 +111,7 @@ def test_edit_msg_with_rich_message(telegram):
     assert result is True
     kwargs = telegram._bot.edit_message_text.call_args.kwargs
     assert kwargs["text"] is None
-    assert kwargs["rich_message"].html == (
+    assert kwargs["rich_message"].get("html") == (
         '<h1>流式结果</h1><ul><li><b>完成</b></li></ul>'
     )
 
@@ -400,10 +400,10 @@ def test_send_msg_with_html_parse_mode_keeps_html(telegram):
 
 def test_send_msg_uses_rich_message_api(telegram):
     """Rich Markdown 应转换后通过 Telegram sendRichMessage 发送。"""
-    telegram.bot.send_rich_message.return_value = SimpleNamespace(
-        message_id=101,
-        chat=SimpleNamespace(id=10001),
-    )
+    telegram.bot.send_rich_message.return_value = {
+        "message_id": 101,
+        "chat": {"id": 10001},
+    }
 
     result = telegram.send_msg(
         title="",
@@ -420,9 +420,9 @@ def test_send_msg_uses_rich_message_api(telegram):
     telegram.bot.send_message.assert_not_called()
     send_kwargs = telegram.bot.send_rich_message.call_args.kwargs
     assert send_kwargs["chat_id"] == "fake_chat_id"
-    assert send_kwargs["rich_message"].markdown is None
-    assert "<h1>处理完成</h1>" in send_kwargs["rich_message"].html
-    assert "<table>" in send_kwargs["rich_message"].html
+    assert send_kwargs["rich_message"].get("markdown") is None
+    assert "<h1>处理完成</h1>" in send_kwargs["rich_message"].get("html")
+    assert "<table>" in send_kwargs["rich_message"].get("html")
     assert send_kwargs["reply_markup"] is not None
 
 
@@ -439,7 +439,7 @@ def test_send_msg_edits_rich_message(telegram):
     edit_kwargs = telegram.bot.edit_message_text.call_args.kwargs
     assert edit_kwargs["text"] is None
     assert edit_kwargs["message_id"] == 101
-    assert "<h1>更新结果</h1>" in edit_kwargs["rich_message"].html
+    assert "<h1>更新结果</h1>" in edit_kwargs["rich_message"].get("html")
 
 
 def test_telegram_module_passes_parse_mode_to_client():
@@ -786,12 +786,8 @@ def test_send_msg_with_force_reply_uses_force_reply_when_no_buttons(telegram):
     assert result and result.get("success")
     send_kwargs = telegram.bot.send_message.call_args.kwargs
     reply_markup = send_kwargs["reply_markup"]
-    assert reply_markup.__class__.__name__ == "ForceReply"
-    if hasattr(reply_markup, "to_dict"):
-        assert reply_markup.to_dict()["force_reply"] is True
-        assert reply_markup.to_dict().get("selective") is True
-    else:
-        assert getattr(reply_markup, "selective", None) is True
+    assert reply_markup["force_reply"] is True
+    assert reply_markup["selective"] is True
 
 
 def test_send_msg_with_force_reply_keeps_inline_keyboard_when_buttons_exist(telegram):
@@ -806,7 +802,9 @@ def test_send_msg_with_force_reply_keeps_inline_keyboard_when_buttons_exist(tele
     assert result and result.get("success")
     send_kwargs = telegram.bot.send_message.call_args.kwargs
     reply_markup = send_kwargs["reply_markup"]
-    assert reply_markup.__class__.__name__ == "InlineKeyboardMarkup"
+    assert reply_markup == {
+        "inline_keyboard": [[{"text": "默认", "callback_data": "default"}]]
+    }
 
 
 def test_send_msg_with_force_reply_and_original_message_sends_new_prompt(telegram):
@@ -824,7 +822,7 @@ def test_send_msg_with_force_reply_and_original_message_sends_new_prompt(telegra
     send_kwargs = telegram.bot.send_message.call_args.kwargs
     assert send_kwargs["chat_id"] == "group-1"
     assert send_kwargs["reply_to_message_id"] == 123
-    assert send_kwargs["reply_markup"].__class__.__name__ == "ForceReply"
+    assert send_kwargs["reply_markup"]["force_reply"] is True
 
 
 def test_send_msg_new_direct_context_message_prefers_original_chat(telegram):
@@ -934,3 +932,107 @@ def test_send_msg_edit_with_image_falls_back_to_text_when_image_url_unavailable(
     assert "测试标题" in edit_kwargs["text"]
     assert "测试内容" in edit_kwargs["text"]
     assert edit_kwargs["reply_markup"] is not None
+
+
+def _group_message(text, **extra):
+    """构造群聊文本消息的 Bot API 结构。"""
+    message = {
+        "message_id": 5,
+        "from": {"id": 42, "is_bot": False, "first_name": "u"},
+        "chat": {"id": -1001, "type": "supergroup"},
+        "text": text,
+    }
+    message.update(extra)
+    return message
+
+
+def test_handle_update_forwards_raw_message_dict(telegram):
+    """普通消息以 Bot API 原始 dict 转发主程序，并记录用户最近会话。"""
+    message = {
+        "message_id": 1,
+        "from": {"id": 42, "is_bot": False, "first_name": "u"},
+        "chat": {"id": 42, "type": "private"},
+        "photo": [{"file_id": "image-1"}],
+        "caption": "看看",
+    }
+    with patch.object(telegram, "_forward_to_message_chain", return_value=True) as forward:
+        telegram._handle_update({"update_id": 1, "message": message})
+
+    forward.assert_called_once_with(message)
+    assert telegram._get_user_chat_id("42") == "42"
+
+
+@pytest.mark.parametrize("text", ["/start", "/help@test_bot", "/start@other_bot payload"])
+def test_handle_update_replies_welcome_without_forwarding(telegram, text):
+    """/start、/help 只回复使用提示，与原 SDK 一样忽略 @bot 后缀和参数。"""
+    message = _group_message(text)
+    with patch.object(telegram, "_forward_to_message_chain") as forward:
+        telegram._handle_update({"update_id": 1, "message": message})
+
+    forward.assert_not_called()
+    kwargs = telegram.bot.send_message.call_args.kwargs
+    assert kwargs["chat_id"] == -1001
+    assert kwargs["reply_to_message_id"] == 5
+
+
+def test_handle_update_ignores_unsupported_content(telegram):
+    """入群通知等非转发内容类型不进入主程序。"""
+    message = _group_message(None)
+    del message["text"]
+    message["new_chat_members"] = [{"id": 7}]
+    with patch.object(telegram, "_forward_to_message_chain") as forward:
+        telegram._handle_update({"update_id": 1, "message": message})
+    forward.assert_not_called()
+
+
+def test_group_message_requires_mention_entity(telegram):
+    """群聊消息只在 @bot 时处理，实体偏移按 UTF-16 计算。"""
+    assert telegram._should_process_message(_group_message("随便聊聊")) is False
+    text = "😀 @test_bot 搜索"
+    mentioned = _group_message(
+        text, entities=[{"type": "mention", "offset": 3, "length": 9}]
+    )
+    assert telegram._should_process_message(mentioned) is True
+    caption_message = _group_message(None, caption="/search 片名")
+    assert telegram._should_process_message(caption_message) is True
+
+
+def test_handle_callback_query_answers_then_forwards(telegram):
+    """按钮回调先应答结束加载状态，再以精简结构转发主程序。"""
+    call = {
+        "id": "cb-1",
+        "from": {"id": 42, "is_bot": False, "first_name": "u"},
+        "message": {"message_id": 9, "chat": {"id": -1001, "type": "supergroup"}},
+        "data": "page:2",
+    }
+    with patch.object(telegram, "_forward_to_message_chain", return_value=True) as forward:
+        telegram._handle_update({"update_id": 2, "callback_query": call})
+
+    telegram.bot.answer_callback_query.assert_called_once_with("cb-1")
+    payload = forward.call_args.args[0]["callback_query"]
+    assert payload["from"]["id"] == 42
+    assert payload["message"] == {"message_id": 9, "chat": {"id": -1001}}
+    assert payload["data"] == "page:2"
+    assert telegram._get_user_chat_id("42") == "-1001"
+
+
+def test_send_file_returns_message_identity(telegram, tmp_path):
+    """发送结果从 Bot API Message dict 读取 message_id 和 chat id。"""
+    file_path = tmp_path / "report.txt"
+    file_path.write_text("ok", encoding="utf-8")
+    telegram.bot.send_document.return_value = {"message_id": 88, "chat": {"id": 10001}}
+
+    result = telegram.send_file(str(file_path), userid="10001")
+
+    assert result == {"success": True, "message_id": 88, "chat_id": 10001}
+    name, _ = telegram.bot.send_document.call_args.kwargs["document"]
+    assert name == "report.txt"
+
+
+def test_register_commands_uses_bot_command_dicts(telegram):
+    """菜单命令去掉前导斜杠，按 BotCommand 结构提交。"""
+    telegram.register_commands({"/search": {"description": "搜索"}})
+    telegram.bot.delete_my_commands.assert_called_once_with()
+    telegram.bot.set_my_commands.assert_called_once_with(
+        commands=[{"command": "search", "description": "搜索"}]
+    )

tests/test_telegram_botapi.py

@@ -0,0 +1,207 @@
+"""
+Telegram Bot API 精简客户端测试：请求编码、错误语义、文件上传和长轮询循环。
+"""
+import json
+import threading
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+import requests
+
+from app.modules.telegram.botapi import TelegramApiError, TelegramBotApi, TelegramNetworkError
+
+
+def _response(payload=None, status_code=200, text=None):
+    """构造 requests.Response 替身；payload 为 None 时 json() 抛 ValueError。"""
+    def _json():
+        if payload is None:
+            raise ValueError(text)
+        return payload
+
+    return SimpleNamespace(status_code=status_code, json=_json)
+
+
+@pytest.fixture
+def post_res():
+    """拦截客户端的全部出站 POST，默认返回成功结果。"""
+    with patch("app.modules.telegram.botapi.RequestUtils.post_res") as mock_post:
+        mock_post.return_value = _response({"ok": True, "result": {"message_id": 7}})
+        yield mock_post
+
+
+def test_default_api_uses_official_host_and_proxy(post_res):
+    """官方地址走全局代理，自定义地址只取主机并跳过代理。"""
+    proxy = {"https": "http://127.0.0.1:7890"}
+    client = TelegramBotApi("1:abc", proxy=proxy)
+    assert client.proxy == proxy
+    assert client.file_url("a/b.jpg") == "https://api.telegram.org/file/bot1:abc/a/b.jpg"
+    client.get_me()
+    assert post_res.call_args.args[0] == "https://api.telegram.org/bot1:abc/getMe"
+
+    custom = TelegramBotApi("1:abc", api_url="http://tg.lan:8081/ignored/path", proxy=proxy)
+    assert custom.proxy is None
+    custom.get_me()
+    assert post_res.call_args.args[0] == "http://tg.lan:8081/bot1:abc/getMe"
+    assert custom.file_url("x") == "http://tg.lan:8081/file/bot1:abc/x"
+
+
+def test_request_verifies_tls():
+    """Bot token 随每次请求发送，客户端不能沿用 RequestUtils 的默认不校验证书。"""
+    with patch("app.modules.telegram.botapi.RequestUtils") as mock_utils:
+        TelegramBotApi("1:abc")
+    assert mock_utils.call_args.kwargs["verify"] is True
+    assert mock_utils.call_args.kwargs["use_session"] is True
+
+
+def test_send_message_encodes_bot_api_parameters(post_res):
+    """嵌套结构编码为 JSON，旧式回复和预览参数转换为当前 Bot API 字段，None 不发送。"""
+    client = TelegramBotApi("1:abc", parse_mode="MarkdownV2")
+    result = client.send_message(
+        chat_id=10001,
+        text="你好",
+        reply_markup={"inline_keyboard": [[{"text": "按钮", "callback_data": "x"}]]},
+        disable_web_page_preview=True,
+        reply_to_message_id="42",
+        message_thread_id=None,
+    )
+
+    assert result == {"message_id": 7}
+    data = post_res.call_args.kwargs["data"]
+    assert data["chat_id"] == 10001
+    assert data["parse_mode"] == "MarkdownV2"
+    assert json.loads(data["reply_markup"])["inline_keyboard"][0][0]["text"] == "按钮"
+    assert json.loads(data["link_preview_options"]) == {"is_disabled": True}
+    assert json.loads(data["reply_parameters"]) == {"message_id": 42}
+    assert "message_thread_id" not in data
+    assert post_res.call_args.kwargs["files"] is None
+    assert post_res.call_args.kwargs["raise_exception"] is True
+
+
+def test_plain_parse_mode_is_omitted_and_rich_edit_has_no_parse_mode(post_res):
+    """空 parse_mode 表示纯文本；Rich Message 编辑不带默认 parse_mode。"""
+    client = TelegramBotApi("1:abc", parse_mode="MarkdownV2")
+    client.send_message(chat_id=1, text="plain", parse_mode="")
+    assert "parse_mode" not in post_res.call_args.kwargs["data"]
+
+    client.edit_message_text(chat_id=1, message_id=2, rich_message={"html": "<b>x</b>"})
+    data = post_res.call_args.kwargs["data"]
+    assert "parse_mode" not in data and "text" not in data
+    assert json.loads(data["rich_message"]) == {"html": "<b>x</b>"}
+
+    client.send_message(chat_id=1, text="flag", disable_web_page_preview=False)
+    assert json.loads(post_res.call_args.kwargs["data"]["link_preview_options"]) == {
+        "is_disabled": False
+    }
+
+
+def test_media_upload_routes_by_input_type(post_res):
+    """字符串按 file_id/URL 发送，bytes 和 (文件名, 内容) 走 multipart。"""
+    client = TelegramBotApi("1:abc")
+    client.send_photo(chat_id=1, photo="https://img.example/a.jpg", caption="c")
+    assert post_res.call_args.kwargs["data"]["photo"] == "https://img.example/a.jpg"
+    assert post_res.call_args.kwargs["files"] is None
+
+    client.send_photo(chat_id=1, photo=b"raw")
+    assert post_res.call_args.kwargs["files"] == {"photo": b"raw"}
+
+    client.send_document(chat_id=1, document=("报告.txt", b"doc"))
+    assert post_res.call_args.args[0].endswith("/sendDocument")
+    assert post_res.call_args.kwargs["files"] == {"document": ("报告.txt", b"doc")}
+
+    client.send_photo(chat_id=1, photo=("", b"img"))
+    assert post_res.call_args.kwargs["files"] == {"photo": ("photo", b"img")}
+
+
+def test_api_error_keeps_telegram_description(post_res):
+    """ok=false 抛出含 description 的异常，调用方据此识别可忽略的编辑错误。"""
+    post_res.return_value = _response({
+        "ok": False,
+        "error_code": 400,
+        "description": "Bad Request: message is not modified",
+    })
+    client = TelegramBotApi("1:abc")
+    with pytest.raises(TelegramApiError) as exc:
+        client.edit_message_text(chat_id=1, message_id=2, text="same")
+    assert exc.value.error_code == 400
+    assert "message is not modified" in str(exc.value).lower()
+
+    post_res.return_value = _response(None, status_code=502, text="<html>")
+    with pytest.raises(TelegramApiError) as exc:
+        client.get_me()
+    assert exc.value.error_code == 502
+
+
+def test_polling_dispatches_in_order_and_survives_handler_errors():
+    """更新按顺序分发，offset 确认到最后一条；单条处理失败不影响后续更新。"""
+    client = TelegramBotApi("1:abc")
+    batches = [
+        [{"update_id": 10, "n": 1}, {"update_id": 11, "n": 2}],
+        [{"update_id": 12, "n": 3}],
+    ]
+    offsets = []
+    handled = []
+
+    def fake_get_updates(offset, timeout):
+        offsets.append(offset)
+        if batches:
+            return batches.pop(0)
+        client.stop_polling(0)
+        return []
+
+    def handler(update):
+        handled.append(update["n"])
+        if update["n"] == 1:
+            raise RuntimeError("boom")
+
+    with patch.object(client, "get_updates", side_effect=fake_get_updates):
+        client.polling(handler, long_polling_timeout=1)
+
+    assert handled == [1, 2, 3]
+    assert offsets == [None, 12, 13]
+
+
+def test_polling_backs_off_on_failure_and_stops_promptly():
+    """拉取失败进入退避等待，stop_polling 能立即打断等待并退出循环。"""
+    client = TelegramBotApi("1:abc")
+    calls = threading.Event()
+
+    def failing_get_updates(offset, timeout):
+        calls.set()
+        raise ConnectionError("network down")
+
+    with patch.object(client, "get_updates", side_effect=failing_get_updates):
+        thread = threading.Thread(
+            target=client.polling, args=(lambda update: None,), kwargs={"long_polling_timeout": 1}
+        )
+        thread.start()
+        assert calls.wait(1)
+        client.stop_polling(0)
+        thread.join(1)
+    assert not thread.is_alive()
+
+
+def test_get_updates_requests_only_handled_update_types(post_res):
+    """getUpdates 只订阅消息和按钮回调，读取超时大于长轮询等待。"""
+    post_res.return_value = _response({"ok": True, "result": []})
+    client = TelegramBotApi("1:abc")
+    assert client.get_updates(offset=5, timeout=8) == []
+    data = post_res.call_args.kwargs["data"]
+    assert data["offset"] == 5 and data["timeout"] == 8
+    assert json.loads(data["allowed_updates"]) == ["message", "callback_query"]
+    connect_timeout, read_timeout = post_res.call_args.kwargs["timeout"]
+    assert read_timeout > 8 and connect_timeout > 0
+
+
+def test_network_error_hides_bot_token(post_res):
+    """requests 异常文本含完整地址，抛出前必须去掉 token 且不保留异常链。"""
+    token = "123456:SECRET-token"
+    post_res.side_effect = requests.ConnectTimeout(
+        f"HTTPSConnectionPool(host='api.telegram.org', port=443): url: /bot{token}/getMe"
+    )
+    client = TelegramBotApi(token)
+    with pytest.raises(TelegramNetworkError) as exc:
+        client.get_me()
+    assert token not in str(exc.value)
+    assert "<token>" in str(exc.value)
+    assert exc.value.__cause__ is None and exc.value.__suppress_context__

tests/test_telegram_urllib3_compat.py

@@ -1,112 +0,0 @@
-import importlib
-import sys
-from types import SimpleNamespace
-
-import pytest
-
-
-def test_ensure_urllib3_header_param_compat_adds_best_available_alias(monkeypatch):
-    """urllib3 future 缺少旧别名时，应补齐最接近的新入口。"""
-    from app.modules.telegram.compat import ensure_urllib3_header_param_compat
-
-    def multipart_formatter(name, value):
-        return f"multipart:{name}={value}"
-
-    def rfc2231_formatter(name, value):
-        return f"rfc2231:{name}={value}"
-
-    fake_fields = SimpleNamespace(
-        format_multipart_header_param=multipart_formatter,
-        format_header_param_rfc2231=rfc2231_formatter,
-    )
-    fake_urllib3 = SimpleNamespace(fields=fake_fields)
-    monkeypatch.setitem(sys.modules, "urllib3", fake_urllib3)
-
-    ensure_urllib3_header_param_compat()
-
-    assert fake_fields.format_header_param is multipart_formatter
-
-
-def test_ensure_urllib3_header_param_compat_prefers_html5_formatter(monkeypatch):
-    """旧版本 urllib3 的 html5 formatter 存在时，应优先保持原别名语义。"""
-    from app.modules.telegram.compat import ensure_urllib3_header_param_compat
-
-    def html5_formatter(name, value):
-        return f"html5:{name}={value}"
-
-    def multipart_formatter(name, value):
-        return f"multipart:{name}={value}"
-
-    fake_fields = SimpleNamespace(
-        format_header_param_html5=html5_formatter,
-        format_multipart_header_param=multipart_formatter,
-    )
-    fake_urllib3 = SimpleNamespace(fields=fake_fields)
-    monkeypatch.setitem(sys.modules, "urllib3", fake_urllib3)
-
-    ensure_urllib3_header_param_compat()
-
-    assert fake_fields.format_header_param is html5_formatter
-
-
-def test_ensure_urllib3_header_param_compat_keeps_existing_formatter(monkeypatch):
-    """已有 format_header_param 时不应覆盖，避免改变 urllib3 正常行为。"""
-    from app.modules.telegram.compat import ensure_urllib3_header_param_compat
-
-    def existing_formatter(name, value):
-        return f"existing:{name}={value}"
-
-    def rfc2231_formatter(name, value):
-        return f"fallback:{name}={value}"
-
-    fake_fields = SimpleNamespace(
-        format_header_param=existing_formatter,
-        format_header_param_rfc2231=rfc2231_formatter,
-    )
-    fake_urllib3 = SimpleNamespace(fields=fake_fields)
-    monkeypatch.setitem(sys.modules, "urllib3", fake_urllib3)
-
-    ensure_urllib3_header_param_compat()
-
-    assert fake_fields.format_header_param is existing_formatter
-
-
-def test_ensure_urllib3_header_param_compat_noops_without_fallback(monkeypatch):
-    """没有任何可用 fallback 时保持 no-op，让原始导入错误暴露。"""
-    from app.modules.telegram.compat import ensure_urllib3_header_param_compat
-
-    fake_fields = SimpleNamespace()
-    fake_urllib3 = SimpleNamespace(fields=fake_fields)
-    monkeypatch.setitem(sys.modules, "urllib3", fake_urllib3)
-
-    ensure_urllib3_header_param_compat()
-
-    assert not hasattr(fake_fields, "format_header_param")
-
-
-def test_telegram_module_imports_when_urllib3_header_alias_missing(monkeypatch):
-    """导入 Telegram 模块前旧别名缺失时，应先补齐再导入 telebot。"""
-    from urllib3 import fields
-
-    if not any(
-        hasattr(fields, name)
-        for name in (
-            "format_header_param_html5",
-            "format_multipart_header_param",
-            "format_header_param_rfc2231",
-        )
-    ):
-        pytest.skip("urllib3 has no compatible header formatter fallback")
-
-    monkeypatch.delattr(fields, "format_header_param", raising=False)
-    for module_name in list(sys.modules):
-        if (
-            module_name == "app.modules.telegram.telegram"
-            or module_name.startswith("telebot")
-        ):
-            monkeypatch.delitem(sys.modules, module_name, raising=False)
-
-    telegram_module = importlib.import_module("app.modules.telegram.telegram")
-
-    assert telegram_module.Telegram
-    assert hasattr(fields, "format_header_param")