84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""Минимальный клиент 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)
|