"""Раздел «Последовательности»: сообщения по порядку, картинки и кнопки.""" from __future__ import annotations import logging import re import uuid from typing import Any from fastapi import APIRouter, HTTPException, Request, Response from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse from starlette.datastructures import UploadFile from admin.media import media_path from admin.security import check_csrf from common import __version__ from common.repositories import sequences_repo from common.texts.render import MAX_MESSAGE_LENGTH, TAG_REGISTRY, validate logger = logging.getLogger(__name__) router = APIRouter(prefix="/sequences", tags=["sequences"]) #: Формат ключа последовательности. KEY_RE = re.compile(r"^[a-z][a-z0-9_.-]{1,31}$") #: Лимит подписи к фото. MAX_CAPTION = 1024 #: Сколько кнопок можно повесить на одно сообщение. MAX_BUTTONS = 8 #: Допустимые типы картинок. ALLOWED_IMAGES = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"} #: Максимальный размер картинки. MAX_IMAGE_BYTES = 5 * 1024 * 1024 def _editor_id(request: Request) -> int: return int(request.state.admin["telegram_id"]) def _prepare_items(items: list[dict[str, Any]]) -> None: """Добавить к каждому сообщению строку кнопок для textarea.""" for item in items: item["buttons_text"] = "\n".join(f"{button['text']} | {button['url']}" for button in item["buttons"]) def parse_buttons(raw: str) -> tuple[list[tuple[str, str]], list[str]]: """Разобрать кнопки: по одной в строке, формат ``Текст | https://…``.""" buttons: list[tuple[str, str]] = [] errors: list[str] = [] for line in raw.splitlines(): line = line.strip() if not line: continue if "|" not in line: errors.append(f"Кнопка без разделителя «|»: {line}") continue text, url = (part.strip() for part in line.split("|", 1)) if not text: errors.append("У кнопки пустой текст") continue if not url.startswith(("https://", "http://")): errors.append(f"Ссылка кнопки должна начинаться с http(s): {url}") continue if len(text) > 64: errors.append("Текст кнопки длиннее 64 символов") continue buttons.append((text, url)) if len(buttons) > MAX_BUTTONS: errors.append(f"Не больше {MAX_BUTTONS} кнопок на сообщение") return buttons, errors def _image_error(upload: UploadFile) -> str | None: content_type = (upload.content_type or "").lower() if content_type not in ALLOWED_IMAGES: return "Картинка должна быть JPEG, PNG или WebP" return None async def _save_image(settings: Any, upload: UploadFile, key: str, item_id: int) -> str: error = _image_error(upload) if error: raise ValueError(error) data = await upload.read() if len(data) > MAX_IMAGE_BYTES: raise ValueError("Картинка больше 5 МБ") name = f"{key}_{item_id}_{uuid.uuid4().hex}{ALLOWED_IMAGES[(upload.content_type or '').lower()]}" path = media_path(settings, name) path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) return name def _delete_image(settings: Any, name: str | None) -> None: if not name: return try: media_path(settings, name).unlink(missing_ok=True) except OSError: logger.warning("Не удалось удалить картинку %s", name, exc_info=True) async def _list_response( request: Request, *, errors: list[str] | None = None, form: dict[str, str] | None = None, saved: str | None = None, status_code: int = 200, ) -> HTMLResponse: return request.app.state.templates.TemplateResponse( request=request, name="sequences/list.html", context={ "version": __version__, "sequences": await sequences_repo.list_sequences(), "errors": errors or [], "form": form or {"key": "", "title": "", "description": ""}, "saved": saved, }, status_code=status_code, ) async def _editor_response( request: Request, key: str, *, errors: list[str] | None = None, saved: str | None = None, overrides: dict[int, dict[str, Any]] | None = None, status_code: int = 200, ) -> HTMLResponse: sequence = await sequences_repo.get_sequence(key) if sequence is None: raise HTTPException(status_code=404, detail="Последовательность не найдена") items = await sequences_repo.get_items(key) _prepare_items(items) for item in items: override = (overrides or {}).get(int(item["id"])) if override: item.update(override) return request.app.state.templates.TemplateResponse( request=request, name="sequences/edit.html", context={ "version": __version__, "sequence": sequence, "items": items, "history": await sequences_repo.history(key, limit=5), "errors": errors or [], "saved": saved, "max_caption": MAX_CAPTION, "max_message": MAX_MESSAGE_LENGTH, "max_buttons": MAX_BUTTONS, "tags": TAG_REGISTRY, }, status_code=status_code, ) @router.get("", response_class=HTMLResponse) async def list_sequences(request: Request, saved: str | None = None) -> HTMLResponse: return await _list_response(request, saved=saved) @router.post("") async def create_sequence(request: Request) -> Response: form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) key = str(form.get("key", "")).strip() title = str(form.get("title", "")).strip() description = str(form.get("description", "")).strip() raw_form = {"key": key, "title": title, "description": description} errors: list[str] = [] if not KEY_RE.match(key): errors.append("Ключ: латиница в нижнем регистре, цифры, точка, подчёркивание или дефис (2–32)") if not title: errors.append("Название не может быть пустым") if errors: return await _list_response(request, errors=errors, form=raw_form, status_code=400) if not await sequences_repo.create_sequence( key, title, description, editor_id=_editor_id(request), ): return await _list_response( request, errors=["Последовательность с таким ключом уже существует"], form=raw_form, status_code=400, ) return RedirectResponse(url=f"/sequences/{key}", status_code=303) @router.get("/media/{filename}") async def serve_media(request: Request, filename: str) -> FileResponse: """Отдать картинку из media-volume (доступ только после входа).""" path = media_path(request.app.state.settings, filename) if not path.is_file(): raise HTTPException(status_code=404, detail="Файл не найден") return FileResponse(path) @router.get("/{key}", response_class=HTMLResponse) async def edit_sequence(request: Request, key: str, saved: str | None = None) -> HTMLResponse: return await _editor_response(request, key, saved=saved) @router.post("/{key}") async def update_sequence(request: Request, key: str) -> Response: form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) title = str(form.get("title", "")).strip() description = str(form.get("description", "")).strip() if not title: return await _editor_response(request, key, errors=["Название не может быть пустым"], status_code=400) if not await sequences_repo.update_sequence( key, title, description, editor_id=_editor_id(request), ): raise HTTPException(status_code=404, detail="Последовательность не найдена") return RedirectResponse(url=f"/sequences/{key}?saved=meta", status_code=303) @router.post("/{key}/items") async def add_item(request: Request, key: str) -> Response: form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) if await sequences_repo.add_item(key, editor_id=_editor_id(request)) is None: raise HTTPException(status_code=404, detail="Последовательность не найдена") return RedirectResponse(url=f"/sequences/{key}?saved=item", status_code=303) @router.post("/{key}/items/{item_id}") async def update_item(request: Request, key: str, item_id: int) -> Response: form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) items = await sequences_repo.get_items(key) item = next((row for row in items if int(row["id"]) == item_id), None) if item is None: raise HTTPException(status_code=404, detail="Сообщение не найдено") text = str(form.get("text", "")) enabled = form.get("enabled") is not None 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 new_image: str | None = None update_image = False saved_file: str | None = None upload = form.get("image") if isinstance(upload, UploadFile) and upload.filename: try: saved_file = await _save_image(settings, upload, key, item_id) new_image = saved_file update_image = True except ValueError as exc: errors.append(str(exc)) elif form.get("remove_image") is not None and item["image_name"]: update_image = True new_image = None effective_image = new_image if update_image else item["image_name"] limit = MAX_CAPTION if effective_image else MAX_MESSAGE_LENGTH if len(text) > limit: errors.append(f"С подписью к фото — не больше {MAX_CAPTION} символов, без фото — {MAX_MESSAGE_LENGTH}") if not text.strip() and not effective_image: errors.append("Нужен текст или картинка") if errors: if saved_file: _delete_image(settings, saved_file) return await _editor_response( request, key, errors=errors, overrides={ item_id: { "text": text, "enabled": enabled, "buttons_text": raw_buttons, } }, status_code=400, ) updated = await sequences_repo.update_item( item_id, text, enabled, buttons, image_name=new_image, update_image=update_image, editor_id=_editor_id(request), ) if updated is None: if saved_file: _delete_image(settings, saved_file) raise HTTPException(status_code=404, detail="Сообщение не найдено") _, old_image = updated if update_image and old_image and old_image != new_image: _delete_image(settings, old_image) return RedirectResponse(url=f"/sequences/{key}?saved=item", status_code=303) @router.post("/{key}/items/{item_id}/move") async def move_item(request: Request, key: str, item_id: int) -> Response: form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) direction = str(form.get("direction", "")) if direction not in {"up", "down"}: raise HTTPException(status_code=400, detail="Неверное направление") if await sequences_repo.move_item(item_id, direction, editor_id=_editor_id(request)) is None: raise HTTPException(status_code=404, detail="Сообщение не найдено") return RedirectResponse(url=f"/sequences/{key}?saved=order", status_code=303) @router.post("/{key}/items/{item_id}/delete") async def delete_item(request: Request, key: str, item_id: int) -> Response: form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) items = await sequences_repo.get_items(key) item = next((row for row in items if int(row["id"]) == item_id), None) if item is None: raise HTTPException(status_code=404, detail="Сообщение не найдено") if await sequences_repo.delete_item(item_id, editor_id=_editor_id(request)) is None: raise HTTPException(status_code=404, detail="Сообщение не найдено") _delete_image(request.app.state.settings, item["image_name"]) return RedirectResponse(url=f"/sequences/{key}?saved=item", status_code=303)