159 lines
5.2 KiB
Python
159 lines
5.2 KiB
Python
"""Минимальный клиент Telegram Bot API для админки.
|
||
|
||
Нужен для действий модерации (исключение из чата) и уведомлений в чат логов.
|
||
Специально без aiogram: админке не нужен весь фреймворк бота.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
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 отклонил запрос или недоступен."""
|
||
|
||
|
||
def _check(response: httpx.Response) -> Any:
|
||
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 _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
|
||
return _check(response)
|
||
|
||
|
||
async def _call_multipart(
|
||
settings: AdminSettings,
|
||
method: str,
|
||
data: dict[str, str],
|
||
files: dict[str, tuple[str, bytes, str]],
|
||
) -> 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=30.0) as client:
|
||
response = await client.post(url, data=data, files=files)
|
||
except httpx.HTTPError as exc:
|
||
raise TelegramError(f"Telegram недоступен: {exc}") from exc
|
||
return _check(response)
|
||
|
||
|
||
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)
|
||
|
||
|
||
async def send_to_chat(
|
||
settings: AdminSettings,
|
||
*,
|
||
text: str,
|
||
buttons: list[tuple[str, str]],
|
||
photo: tuple[bytes, str, str] | None = None,
|
||
) -> None:
|
||
"""Отправить сообщение в основной чат от имени бота.
|
||
|
||
``photo`` — (данные, mime, имя файла). При фото длинный текст уходит
|
||
отдельным сообщением (лимит подписи 1024).
|
||
"""
|
||
if not settings.chat_id:
|
||
raise TelegramError("CHAT_ID не задан")
|
||
|
||
rows = [[{"text": title, "url": url}] for title, url in buttons]
|
||
markup = {"inline_keyboard": rows} if rows else None
|
||
|
||
if photo is None:
|
||
await _call(
|
||
settings,
|
||
"sendMessage",
|
||
{
|
||
"chat_id": settings.chat_id,
|
||
"text": text,
|
||
"parse_mode": "HTML",
|
||
"reply_markup": markup,
|
||
},
|
||
)
|
||
return
|
||
|
||
data, mime, filename = photo
|
||
fields: dict[str, str] = {"chat_id": str(settings.chat_id), "parse_mode": "HTML"}
|
||
caption = text if text and len(text) <= 1024 else ""
|
||
if caption:
|
||
fields["caption"] = caption
|
||
if markup:
|
||
fields["reply_markup"] = json.dumps(markup)
|
||
|
||
await _call_multipart(settings, "sendPhoto", fields, {"photo": (filename, data, mime)})
|
||
|
||
if text and not caption:
|
||
await _call(
|
||
settings,
|
||
"sendMessage",
|
||
{
|
||
"chat_id": settings.chat_id,
|
||
"text": text,
|
||
"parse_mode": "HTML",
|
||
"reply_markup": markup,
|
||
},
|
||
)
|