"""Минимальный клиент Telegram Bot API для админки. Нужен для действий модерации (исключение из чата) и уведомлений в чат логов. Специально без aiogram: админке не нужен весь фреймворк бота. """ from __future__ import annotations import logging from typing import Any import httpx from common.config import AdminSettings logger = logging.getLogger(__name__) _API = "https://api.telegram.org/bot{token}/{method}" class TelegramError(RuntimeError): """Telegram отклонил запрос или недоступен.""" async def _call(settings: AdminSettings, method: str, payload: dict[str, Any]) -> Any: if not settings.bot_token: raise TelegramError("BOT_TOKEN не задан") url = _API.format(token=settings.bot_token, method=method) try: async with httpx.AsyncClient(timeout=10.0) as client: response = await client.post(url, json=payload) except httpx.HTTPError as exc: raise TelegramError(f"Telegram недоступен: {exc}") from exc try: data = response.json() except ValueError as exc: raise TelegramError("Telegram вернул некорректный ответ") from exc if not data.get("ok"): raise TelegramError(str(data.get("description") or "Telegram отклонил запрос")) return data.get("result") async def kick_member(settings: AdminSettings, user_id: int) -> None: """Исключить пользователя из основного чата (бан + разбан, чтобы мог вернуться).""" if not settings.chat_id: raise TelegramError("CHAT_ID не задан") await _call( settings, "banChatMember", {"chat_id": settings.chat_id, "user_id": user_id}, ) await _call( settings, "unbanChatMember", {"chat_id": settings.chat_id, "user_id": user_id, "only_if_banned": True}, ) async def chat_title(settings: AdminSettings) -> str: """Название основного чата (best-effort, для текста уведомления).""" try: chat = await _call(settings, "getChat", {"chat_id": settings.chat_id}) except TelegramError: return "чат" return str(chat.get("title") or chat.get("username") or "чат") async def send_alert(settings: AdminSettings, text: str) -> None: """Отправить служебное сообщение в чат логов (ошибки только в лог).""" if not settings.alert_chat_id: return try: await _call( settings, "sendMessage", {"chat_id": settings.alert_chat_id, "text": text}, ) except TelegramError: logger.warning("Не удалось отправить уведомление в чат логов", exc_info=True)