diff --git a/app/modules/telegram/botapi.py b/app/modules/telegram/botapi.py index a1847f181..6c004d499 100644 --- a/app/modules/telegram/botapi.py +++ b/app/modules/telegram/botapi.py @@ -72,8 +72,9 @@ class TelegramBotApi: """ 单个 Bot token 的 Bot API 客户端,持有一个复用连接的 HTTP 会话。 - 会话在发送线程、typing 线程和轮询线程间共享;requests 连接池本身线程安全, - 只有 POST 且不会自动重放,避免网络抖动时重复发送消息。 + 发送线程和 typing 线程共享一个会话,轮询线程使用独立会话。requests 连接池本身 + 线程安全;轮询请求虽然也是 POST,但只读取更新且不会自动重放,避免网络抖动时 + 重复发送消息。轮询连接断开后可以单独重建,不影响正在发送的消息。 """ def __init__( @@ -97,13 +98,8 @@ class TelegramBotApi: # 自定义地址通常是内网自建服务,不走全局代理。 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._http = self._create_http() + self._polling_http: Optional[RequestUtils] = None self._polling_stop = threading.Event() # 分发临界区:持锁期间检查停止标记并执行 handler,stop 取得该锁即证明不会再分发。 self._dispatch_lock = threading.Lock() @@ -117,9 +113,36 @@ class TelegramBotApi: """拼接 getFile 返回路径对应的下载地址(包含 token,不可写入日志)。""" return f"{self._file_base}{file_path}" + def _create_http(self) -> RequestUtils: + """创建一个 Bot API 专用 HTTP 会话。""" + return RequestUtils( + headers={"Accept": "application/json"}, + proxies=self._proxy, + use_session=True, + # RequestUtils 默认沿用旧宿主的不校验证书行为;Bot token 随请求发送,必须校验。 + verify=True, + ) + + def _get_polling_http(self) -> RequestUtils: + """按需取得轮询专用会话,避免长轮询故障影响出站消息。""" + if self._polling_http is None: + self._polling_http = self._create_http() + return self._polling_http + + def _reset_polling_http(self) -> None: + """重建轮询连接池;只由轮询线程调用。""" + old_http = self._polling_http + new_http = self._create_http() + self._polling_http = new_http + if old_http is not None: + old_http.close() + def close(self) -> None: """关闭 HTTP 会话。""" self._http.close() + if self._polling_http is not None: + self._polling_http.close() + self._polling_http = None # ---------------------------------------------------------------- 请求层 @@ -144,6 +167,7 @@ class TelegramBotApi: params: Optional[Dict[str, Any]] = None, files: Optional[Dict[str, Any]] = None, read_timeout: float = _READ_TIMEOUT_SECONDS, + _http: Optional[RequestUtils] = None, ) -> Any: """ 调用 Bot API 方法并返回 ``result``。 @@ -151,8 +175,9 @@ class TelegramBotApi: :raises TelegramApiError: Telegram 返回 ok=false 或非 JSON 响应 :raises TelegramNetworkError: 网络失败,描述已去除 token """ + request_http = _http or self._http try: - response = self._http.post_res( + response = request_http.post_res( f"{self._api_base}{method}", data=self._encode_params(params or {}), files=files, @@ -397,6 +422,7 @@ class TelegramBotApi: "getUpdates", {"offset": offset, "timeout": timeout, "allowed_updates": _ALLOWED_UPDATES}, read_timeout=timeout + _CONNECT_TIMEOUT_SECONDS, + _http=self._get_polling_http(), ) def polling( @@ -426,6 +452,11 @@ class TelegramBotApi: except Exception as err: if self._polling_stop.is_set(): return + if isinstance(err, TelegramNetworkError): + try: + self._reset_polling_http() + except Exception as reset_err: + logger.warning(f"Telegram轮询连接重建失败,将继续重试:{reset_err}") failures += 1 now = time.monotonic() retry_seconds = min(3 * 2 ** (failures - 1), retry_max_seconds) diff --git a/tests/test_telegram_botapi.py b/tests/test_telegram_botapi.py index e6bc97eb2..a7ff95c27 100644 --- a/tests/test_telegram_botapi.py +++ b/tests/test_telegram_botapi.py @@ -185,6 +185,32 @@ def test_polling_backs_off_on_failure_and_stops_promptly(): assert not thread.is_alive() +def test_polling_rebuilds_only_the_long_polling_session_after_network_failure(): + """长轮询连接失败后应重建独立会话,避免后续更新继续复用坏连接池。""" + client = TelegramBotApi("1:abc") + client._get_polling_http() + replies = [TelegramNetworkError("getUpdates", "connection reset"), []] + polling_sessions = [] + + def fake_get_updates(offset, timeout): + """记录每次拉取使用的会话,并在恢复后结束测试轮询。""" + polling_sessions.append(client._polling_http) + reply = replies.pop(0) + if isinstance(reply, Exception): + raise reply + client.stop_polling(0) + return reply + + with patch.object(client, "get_updates", side_effect=fake_get_updates), \ + patch.object(client._polling_stop, "wait"): + client.polling(lambda update: None, long_polling_timeout=1) + + assert polling_sessions[0] is not None + assert polling_sessions[0] is not polling_sessions[1] + assert client._http is not polling_sessions[1] + client.close() + + def test_polling_reports_persistent_failures_and_resets_after_recovery(): """失败每分钟警告,恢复后再次中断立即警告,offset 和退避计数均保持正确。""" client = TelegramBotApi("1:abc")