lekss361
ca1015bd4e
All checks were successful
Deploy Trade-In / test (push) Successful in 4m59s
Deploy Trade-In / build-backend (push) Successful in 1m10s
Deploy Trade-In / deploy (push) Successful in 1m17s
Deploy Trade-In / changes (push) Successful in 14s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
265 lines
12 KiB
Python
265 lines
12 KiB
Python
"""Тонкая httpx-обёртка над Telegram Bot API (#tgsupport).
|
||
|
||
Зачем свой клиент, а не aiogram: единственные нужные методы — `getUpdates`
|
||
(long-polling), `copyMessage` (зеркалирование ЛЮБОГО типа контента без ре-аплоада)
|
||
и `sendMessage` (заголовки/приветствия/уведомления). aiogram — избыточная
|
||
зависимость (webhook-framework, dispatcher, FSM) ради трёх HTTP-вызовов; в стеке
|
||
уже есть httpx (см. `app.services.dadata`, `app.services.geocoder` — тот же паттерн
|
||
retry/timeout).
|
||
|
||
Docs: https://core.telegram.org/bots/api
|
||
|
||
Ретраи:
|
||
- HTTP 429 (Too Many Requests) — уважаем `parameters.retry_after` из тела ответа
|
||
(Telegram сам говорит сколько ждать), fallback на `_DEFAULT_RETRY_AFTER_S`.
|
||
- HTTP 5xx / сетевые ошибки (timeout/connect) — экспоненциальный backoff,
|
||
`capped` на `_MAX_BACKOFF_S`.
|
||
- Любая другая 4xx (400/401/403/404) — НЕ ретраится, сразу `TelegramApiError`
|
||
(запрос некорректен или прав нет — повтор не поможет).
|
||
|
||
БЕЗОПАСНОСТЬ: наши `logger.*`-вызовы здесь содержат только имя метода API,
|
||
HTTP-статус и `description` из ответа Telegram — токен туда не пишем.
|
||
Это НЕ гарантирует, что токен не утечёт по другим стокам: он живёт в
|
||
`self._base`/`url` (локальные переменные stack-фрейма `_request`), а GlitchTip
|
||
(sentry_sdk) по умолчанию прикладывает locals к traceback и Httpx-интеграция
|
||
кладёт полный URL в span data. Эти стоки закрываются НЕ здесь, а в
|
||
`app.tgbot_main` (`include_local_variables=False`, `before_send`-редактор,
|
||
`traces_sample_rate=0.0`) и подавлением INFO-логов самого `httpx`-логгера
|
||
(который печатает полный request URL, включая токен, на уровне INFO).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_DEFAULT_TIMEOUT_S = 15.0
|
||
_DEFAULT_RETRY_AFTER_S = 5.0
|
||
_MAX_BACKOFF_S = 30.0
|
||
_DEFAULT_MAX_RETRIES = 5
|
||
|
||
|
||
class TelegramApiError(Exception):
|
||
"""Telegram Bot API ответил `ok: false` (после исчерпания ретраев, если применимо)."""
|
||
|
||
def __init__(self, method: str, error_code: int, description: str) -> None:
|
||
self.method = method
|
||
self.error_code = error_code
|
||
self.description = description
|
||
super().__init__(f"Telegram API {method} failed: {error_code} {description}")
|
||
|
||
|
||
def _extract_retry_after(
|
||
response: httpx.Response, default: float = _DEFAULT_RETRY_AFTER_S
|
||
) -> float:
|
||
"""Достаёт `parameters.retry_after` из тела 429-ответа. Fallback — `default`."""
|
||
try:
|
||
data = response.json()
|
||
except ValueError:
|
||
return default
|
||
if not isinstance(data, dict):
|
||
return default
|
||
params = data.get("parameters")
|
||
if isinstance(params, dict):
|
||
retry_after = params.get("retry_after")
|
||
if isinstance(retry_after, int | float):
|
||
return float(retry_after)
|
||
return default
|
||
|
||
|
||
def _error_from_body(response: httpx.Response) -> tuple[int, str]:
|
||
"""Парсит (error_code, description) из тела ответа Telegram; fallback на HTTP-статус."""
|
||
try:
|
||
data = response.json()
|
||
except ValueError:
|
||
return response.status_code, (response.text or "")[:200]
|
||
if not isinstance(data, dict):
|
||
return response.status_code, str(data)[:200]
|
||
error_code = data.get("error_code", response.status_code)
|
||
description = data.get("description", "")
|
||
code = int(error_code) if isinstance(error_code, int | float) else response.status_code
|
||
return code, str(description)
|
||
|
||
|
||
class TelegramClient:
|
||
"""Bot API клиент на httpx.AsyncClient. Каждый вызов — отдельное короткоживущее соединение."""
|
||
|
||
def __init__(
|
||
self,
|
||
token: str,
|
||
base_url: str = "https://api.telegram.org",
|
||
timeout: float = _DEFAULT_TIMEOUT_S,
|
||
) -> None:
|
||
self._base = f"{base_url}/bot{token}"
|
||
self._timeout = timeout
|
||
|
||
async def _request(
|
||
self,
|
||
method: str,
|
||
payload: dict[str, Any],
|
||
*,
|
||
timeout: float | None = None,
|
||
max_retries: int = _DEFAULT_MAX_RETRIES,
|
||
) -> Any:
|
||
"""POST `method` с JSON-телом `payload`. Ретраит 429/5xx/network, иначе raise сразу."""
|
||
url = f"{self._base}/{method}"
|
||
effective_timeout = timeout if timeout is not None else self._timeout
|
||
attempt = 0
|
||
|
||
while True:
|
||
attempt += 1
|
||
try:
|
||
async with httpx.AsyncClient(timeout=effective_timeout) as client:
|
||
response = await client.post(url, json=payload)
|
||
except (httpx.TimeoutException, httpx.NetworkError) as exc:
|
||
if attempt > max_retries:
|
||
logger.error(
|
||
"tg client: %s — network error после %d попыток: %s", method, attempt, exc
|
||
)
|
||
raise
|
||
backoff = min(2.0**attempt, _MAX_BACKOFF_S)
|
||
logger.warning(
|
||
"tg client: %s — network error (попытка %d/%d): %s — retry через %.0fs",
|
||
method,
|
||
attempt,
|
||
max_retries,
|
||
exc,
|
||
backoff,
|
||
)
|
||
await asyncio.sleep(backoff)
|
||
continue
|
||
|
||
if response.status_code == 429:
|
||
retry_after = _extract_retry_after(response)
|
||
if attempt > max_retries:
|
||
error_code, description = _error_from_body(response)
|
||
logger.error("tg client: %s — 429 после %d попыток, сдаёмся", method, attempt)
|
||
raise TelegramApiError(method, error_code, description)
|
||
logger.warning(
|
||
"tg client: %s — HTTP 429 (попытка %d/%d), retry_after=%.0fs",
|
||
method,
|
||
attempt,
|
||
max_retries,
|
||
retry_after,
|
||
)
|
||
await asyncio.sleep(retry_after)
|
||
continue
|
||
|
||
if response.status_code >= 500:
|
||
if attempt > max_retries:
|
||
error_code, description = _error_from_body(response)
|
||
logger.error(
|
||
"tg client: %s — HTTP %d после %d попыток, сдаёмся",
|
||
method,
|
||
response.status_code,
|
||
attempt,
|
||
)
|
||
raise TelegramApiError(method, error_code, description)
|
||
backoff = min(2.0**attempt, _MAX_BACKOFF_S)
|
||
logger.warning(
|
||
"tg client: %s — HTTP %d (попытка %d/%d) — retry через %.0fs",
|
||
method,
|
||
response.status_code,
|
||
attempt,
|
||
max_retries,
|
||
backoff,
|
||
)
|
||
await asyncio.sleep(backoff)
|
||
continue
|
||
|
||
if response.status_code >= 400:
|
||
# 4xx кроме 429 — запрос некорректен/прав нет, повтор не поможет.
|
||
error_code, description = _error_from_body(response)
|
||
raise TelegramApiError(method, error_code, description)
|
||
|
||
try:
|
||
data = response.json()
|
||
except ValueError as exc:
|
||
raise TelegramApiError(
|
||
method, response.status_code, f"invalid json: {exc}"
|
||
) from exc
|
||
|
||
if not isinstance(data, dict) or not data.get("ok"):
|
||
error_code, description = _error_from_body(response)
|
||
raise TelegramApiError(method, error_code, description)
|
||
|
||
return data.get("result")
|
||
|
||
async def get_updates(
|
||
self,
|
||
offset: int,
|
||
timeout: int = 30,
|
||
allowed_updates: list[str] | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
"""Long-polling getUpdates. `timeout` — сколько Telegram держит запрос открытым (сек).
|
||
|
||
HTTP-таймаут запроса берётся с запасом (`timeout + 10s`), чтобы не обрывать
|
||
соединение раньше, чем ответит сам Telegram long-poll.
|
||
"""
|
||
payload: dict[str, Any] = {"offset": offset, "timeout": timeout}
|
||
if allowed_updates is not None:
|
||
payload["allowed_updates"] = allowed_updates
|
||
result = await self._request(
|
||
"getUpdates", payload, timeout=float(timeout) + 10.0, max_retries=3
|
||
)
|
||
return result if isinstance(result, list) else []
|
||
|
||
async def copy_message(
|
||
self,
|
||
*,
|
||
chat_id: int,
|
||
from_chat_id: int,
|
||
message_id: int,
|
||
message_thread_id: int | None = None,
|
||
reply_to_message_id: int | None = None,
|
||
) -> dict[str, Any]:
|
||
"""copyMessage — зеркалит ЛЮБОЙ тип контента без ре-аплоада файла."""
|
||
payload: dict[str, Any] = {
|
||
"chat_id": chat_id,
|
||
"from_chat_id": from_chat_id,
|
||
"message_id": message_id,
|
||
}
|
||
if message_thread_id:
|
||
payload["message_thread_id"] = message_thread_id
|
||
if reply_to_message_id:
|
||
payload["reply_to_message_id"] = reply_to_message_id
|
||
result = await self._request("copyMessage", payload)
|
||
return result if isinstance(result, dict) else {}
|
||
|
||
async def send_message(
|
||
self,
|
||
*,
|
||
chat_id: int,
|
||
text: str,
|
||
message_thread_id: int | None = None,
|
||
reply_to_message_id: int | None = None,
|
||
timeout: float | None = None,
|
||
max_retries: int | None = None,
|
||
) -> dict[str, Any]:
|
||
"""sendMessage — текстовое сообщение (заголовки, приветствия, уведомления об ошибке).
|
||
|
||
`timeout`/`max_retries` — по умолчанию наследуют воркерную политику
|
||
(`_DEFAULT_TIMEOUT_S`/`_DEFAULT_MAX_RETRIES`: на 429 спим Telegram-овский
|
||
`retry_after` — для группы это штатные 30-60с, на 5xx backoff до 30с).
|
||
Это ПРИЕМЛЕМО для `tgbot_main.py` (изолированный long-polling воркер), но
|
||
ФАТАЛЬНО для интерактивного HTTP-запроса (#tgsupport-web review H1) —
|
||
синхронный request/response путь не может легально висеть минуты. Вызывающая
|
||
сторона на interactive-пути ОБЯЗАНА передать узкий бюджет явно (см.
|
||
`app.api.v1.support.send_support_message`)."""
|
||
payload: dict[str, Any] = {"chat_id": chat_id, "text": text}
|
||
if message_thread_id:
|
||
payload["message_thread_id"] = message_thread_id
|
||
if reply_to_message_id:
|
||
payload["reply_to_message_id"] = reply_to_message_id
|
||
kwargs: dict[str, Any] = {}
|
||
if timeout is not None:
|
||
kwargs["timeout"] = timeout
|
||
if max_retries is not None:
|
||
kwargs["max_retries"] = max_retries
|
||
result = await self._request("sendMessage", payload, **kwargs)
|
||
return result if isinstance(result, dict) else {}
|