"""Раздел «Сообщения»: список шаблонов, редактор и предпросмотр (HTMX).""" from __future__ import annotations import html import re from fastapi import APIRouter, HTTPException, Request, Response from fastapi.responses import HTMLResponse, RedirectResponse from admin.preview import SAMPLE_CONTEXT from admin.security import check_csrf from common import __version__ from common.repositories import templates_repo from common.texts.render import MAX_MESSAGE_LENGTH, TAG_REGISTRY, TemplateError, render, validate router = APIRouter(prefix="/messages", tags=["messages"]) #: Формат ключа нового шаблона (латиница, цифры, точка, подчёркивание, дефис). KEY_RE = re.compile(r"^[a-z][a-z0-9_.-]{1,63}$") #: Максимальная длина описания шаблона. MAX_DESCRIPTION_LENGTH = 200 def _editor_context( request: Request, template: dict[str, object], *, text: str | None = None, errors: list[str] | None = None, saved: bool = False, history: list[dict[str, object]] | None = None, ) -> dict[str, object]: """Контекст страницы редактора (общий для GET и повторного показа после ошибки).""" return { "version": __version__, "template": template, "text": text if text is not None else template["text"], "errors": errors or [], "saved": saved, "history": history or [], "tags": TAG_REGISTRY, "max_length": MAX_MESSAGE_LENGTH, } @router.get("", response_class=HTMLResponse) async def list_messages(request: Request) -> HTMLResponse: """Список шаблонов: ключ, описание, длина и время последней правки.""" templates = request.app.state.templates return templates.TemplateResponse( request=request, name="messages/list.html", context={"version": __version__, "templates": await templates_repo.list_templates()}, ) @router.get("/new", response_class=HTMLResponse) async def new_template(request: Request) -> HTMLResponse: """Форма создания нового шаблона.""" return request.app.state.templates.TemplateResponse( request=request, name="messages/new.html", context={ "version": __version__, "key": "", "description": "", "text": "", "enabled": True, "errors": [], "tags": TAG_REGISTRY, "max_length": MAX_MESSAGE_LENGTH, "max_description_length": MAX_DESCRIPTION_LENGTH, }, ) @router.post("") async def create_template(request: Request) -> Response: """Создать шаблон (ключ проверяется на формат и занятость).""" form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) key = str(form.get("key", "")).strip() description = str(form.get("description", "")).strip() text = str(form.get("text", "")) enabled = form.get("enabled") is not None errors: list[str] = [] if not KEY_RE.match(key): errors.append("Ключ: латиница в нижнем регистре, цифры, точка, подчёркивание или дефис (2–64 символа)") if not text.strip(): errors.append("Текст не может быть пустым") unknown = validate(text) if unknown: errors.append("Неизвестные теги: " + ", ".join(unknown)) if len(text) > MAX_MESSAGE_LENGTH: errors.append(f"Длина {len(text)} превышает лимит Telegram ({MAX_MESSAGE_LENGTH})") if len(description) > MAX_DESCRIPTION_LENGTH: errors.append(f"Описание длиннее {MAX_DESCRIPTION_LENGTH} символов") if errors: return request.app.state.templates.TemplateResponse( request=request, name="messages/new.html", context={ "version": __version__, "key": key, "description": description, "text": text, "enabled": enabled, "errors": errors, "tags": TAG_REGISTRY, "max_length": MAX_MESSAGE_LENGTH, "max_description_length": MAX_DESCRIPTION_LENGTH, }, status_code=400, ) created = await templates_repo.create_template( key, text, description, enabled=enabled, editor_id=int(request.state.admin["telegram_id"]), ) if not created: return request.app.state.templates.TemplateResponse( request=request, name="messages/new.html", context={ "version": __version__, "key": key, "description": description, "text": text, "enabled": enabled, "errors": ["Шаблон с таким ключом уже существует"], "tags": TAG_REGISTRY, "max_length": MAX_MESSAGE_LENGTH, "max_description_length": MAX_DESCRIPTION_LENGTH, }, status_code=400, ) return RedirectResponse(url=f"/messages/{key}", status_code=303) @router.post("/preview", response_class=HTMLResponse) async def preview_message(request: Request) -> HTMLResponse: """Отрендерить текст с тестовыми значениями (HTMX-фрагмент). Результат кладётся в ``iframe sandbox``: разметка Telegram видна «как в чате», но скрипты из введённого текста не выполняются. """ form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) text = str(form.get("text", "")) unknown = validate(text) if unknown: return HTMLResponse( '

Неизвестные теги: ' + html.escape(", ".join(unknown)) + "

" ) try: rendered = render(text, SAMPLE_CONTEXT) except TemplateError as exc: # pragma: no cover - validate уже отсекает неизвестные теги return HTMLResponse(f'

{html.escape(str(exc))}

') if len(rendered) > MAX_MESSAGE_LENGTH: return HTMLResponse( f'

Текст длиннее лимита Telegram: ' f"{len(rendered)} из {MAX_MESSAGE_LENGTH} символов.

" ) return HTMLResponse( '' ) @router.post("/{key}") async def save_message(request: Request, key: str) -> Response: """Сохранить текст шаблона; при ошибках вернуть редактор с подсветкой.""" form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) text = str(form.get("text", "")) template = await templates_repo.get_template(key) if template is None: raise HTTPException(status_code=404, detail="Шаблон не найден") errors: list[str] = [] unknown = validate(text) if unknown: errors.append("Неизвестные теги: " + ", ".join(unknown)) if not text.strip(): errors.append("Текст не может быть пустым") if len(text) > MAX_MESSAGE_LENGTH: errors.append(f"Длина {len(text)} превышает лимит Telegram ({MAX_MESSAGE_LENGTH})") if errors: history = await templates_repo.template_history(key) return request.app.state.templates.TemplateResponse( request=request, name="messages/edit.html", context=_editor_context(request, template, text=text, errors=errors, history=history), status_code=400, ) editor_id = int(request.state.admin["telegram_id"]) await templates_repo.update_template(key, text, editor_id=editor_id) return RedirectResponse(url=f"/messages/{key}?saved=1", status_code=303) @router.get("/{key}", response_class=HTMLResponse) async def edit_message(request: Request, key: str, saved: bool = False) -> HTMLResponse: """Редактор шаблона: текст, подсказка по тегам, предпросмотр и история.""" template = await templates_repo.get_template(key) if template is None: raise HTTPException(status_code=404, detail="Шаблон не найден") history = await templates_repo.template_history(key) return request.app.state.templates.TemplateResponse( request=request, name="messages/edit.html", context=_editor_context(request, template, saved=saved, history=history), )