151 lines
4.6 KiB
Python
151 lines
4.6 KiB
Python
"""Ручная отправка сообщения в основной чат от имени бота."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, Request, Response
|
||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||
from starlette.datastructures import UploadFile
|
||
|
||
from admin import telegram
|
||
from admin.buttons import MAX_BUTTONS, parse_buttons
|
||
from admin.media import read_image
|
||
from admin.security import check_csrf
|
||
from common import __version__
|
||
from common.repositories import admin_messages_repo
|
||
from common.texts.render import MAX_MESSAGE_LENGTH, TAG_REGISTRY, validate
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/send", tags=["send"])
|
||
|
||
#: Лимит подписи к фото.
|
||
MAX_CAPTION = 1024
|
||
|
||
|
||
async def _recent() -> list[dict[str, Any]]:
|
||
try:
|
||
return await admin_messages_repo.recent(limit=5)
|
||
except Exception:
|
||
logger.warning("Не удалось прочитать журнал отправок", exc_info=True)
|
||
return []
|
||
|
||
|
||
async def _render(
|
||
request: Request,
|
||
*,
|
||
text: str = "",
|
||
buttons_text: str = "",
|
||
errors: list[str] | None = None,
|
||
saved: bool = False,
|
||
status_code: int = 200,
|
||
) -> HTMLResponse:
|
||
settings = request.app.state.settings
|
||
return request.app.state.templates.TemplateResponse(
|
||
request=request,
|
||
name="send/form.html",
|
||
context={
|
||
"version": __version__,
|
||
"text": text,
|
||
"buttons_text": buttons_text,
|
||
"errors": errors or [],
|
||
"saved": saved,
|
||
"chat_id": settings.chat_id,
|
||
"max_message": MAX_MESSAGE_LENGTH,
|
||
"max_caption": MAX_CAPTION,
|
||
"max_buttons": MAX_BUTTONS,
|
||
"tags": TAG_REGISTRY,
|
||
"recent": await _recent(),
|
||
},
|
||
status_code=status_code,
|
||
)
|
||
|
||
|
||
@router.get("", response_class=HTMLResponse)
|
||
async def send_form(request: Request, saved: bool = False) -> HTMLResponse:
|
||
return await _render(request, saved=saved)
|
||
|
||
|
||
@router.post("")
|
||
async def send_message(request: Request) -> Response:
|
||
form = await request.form()
|
||
check_csrf(request, str(form.get("csrf_token", "")))
|
||
|
||
text = str(form.get("text", ""))
|
||
raw_buttons = str(form.get("buttons", ""))
|
||
buttons, errors = parse_buttons(raw_buttons)
|
||
|
||
unknown = validate(text)
|
||
if unknown:
|
||
errors.append("Неизвестные теги: " + ", ".join(unknown))
|
||
|
||
settings = request.app.state.settings
|
||
photo: tuple[bytes, str, str] | None = None
|
||
upload = form.get("image")
|
||
if isinstance(upload, UploadFile) and upload.filename:
|
||
try:
|
||
data, mime, extension = await read_image(upload)
|
||
photo = (data, mime, f"upload{extension}")
|
||
except ValueError as exc:
|
||
errors.append(str(exc))
|
||
|
||
limit = MAX_CAPTION if photo else MAX_MESSAGE_LENGTH
|
||
if len(text) > limit:
|
||
errors.append(f"С фото — не больше {MAX_CAPTION} символов, без фото — {MAX_MESSAGE_LENGTH}")
|
||
if not text.strip() and photo is None:
|
||
errors.append("Нужен текст или картинка")
|
||
if not settings.chat_id:
|
||
errors.append("CHAT_ID не задан")
|
||
|
||
if errors:
|
||
return await _render(
|
||
request,
|
||
text=text,
|
||
buttons_text=raw_buttons,
|
||
errors=errors,
|
||
status_code=400,
|
||
)
|
||
|
||
success = True
|
||
error_text: str | None = None
|
||
try:
|
||
await telegram.send_to_chat(
|
||
settings,
|
||
text=text,
|
||
buttons=buttons,
|
||
photo=photo,
|
||
)
|
||
except telegram.TelegramError as exc:
|
||
success = False
|
||
error_text = str(exc)
|
||
except Exception:
|
||
success = False
|
||
error_text = "Не удалось отправить сообщение"
|
||
logger.exception("Ошибка отправки сообщения из админки")
|
||
|
||
try:
|
||
await admin_messages_repo.log_message(
|
||
editor_id=int(request.state.admin["telegram_id"]),
|
||
chat_id=settings.chat_id,
|
||
text=text,
|
||
buttons=buttons,
|
||
has_image=photo is not None,
|
||
success=success,
|
||
error=error_text,
|
||
)
|
||
except Exception:
|
||
logger.warning("Не удалось записать журнал отправки", exc_info=True)
|
||
|
||
if not success:
|
||
return await _render(
|
||
request,
|
||
text=text,
|
||
buttons_text=raw_buttons,
|
||
errors=[f"Telegram: {error_text}"],
|
||
status_code=400,
|
||
)
|
||
|
||
return RedirectResponse(url="/send?saved=true", status_code=303)
|