"""Раздел «Расписания»: CRUD рассылок, которые исполняет бот.""" from __future__ import annotations from collections.abc import Mapping from typing import Any from fastapi import APIRouter, HTTPException, Request, Response from fastapi.responses import HTMLResponse, RedirectResponse from admin.forms import format_weekdays, parse_schedule_fields from admin.security import check_csrf from common import __version__ from common.repositories import schedules_repo, templates_repo router = APIRouter(prefix="/schedules", tags=["schedules"]) def _editor_id(request: Request) -> int: """Telegram-ID текущего администратора (ставит middleware).""" return int(request.state.admin["telegram_id"]) def _raw_from_form(form: Any) -> dict[str, Any]: return { "template_key": str(form.get("template_key", "")), "hour": str(form.get("hour", "")), "minute": str(form.get("minute", "")), "weekdays": [str(day) for day in form.getlist("weekdays")], "timezone": str(form.get("timezone", "")), "description": str(form.get("description", "")), "enabled": form.get("enabled") is not None, } def _view(values: Mapping[str, Any], *, is_new: bool) -> dict[str, Any]: """Привести значения (из БД или формы) к виду, удобному шаблону.""" days: list[int] = [] for day in values.get("weekdays") or (): try: days.append(int(day)) except (TypeError, ValueError): continue return { "template_key": values.get("template_key") or "", "hour": values.get("hour", 16), "minute": values.get("minute", 0), "weekdays": sorted(set(days)), "timezone": values.get("timezone") or "Europe/Moscow", "description": values.get("description") or "", "enabled": bool(values.get("enabled", True)), "id": values.get("id"), "is_new": is_new, "action": "/schedules" if is_new else f"/schedules/{values.get('id')}", } def _form_response( request: Request, schedule: dict[str, Any], message_templates: list[dict[str, Any]], *, errors: list[str] | None = None, history: list[dict[str, Any]] | None = None, status_code: int = 200, ) -> Response: return request.app.state.templates.TemplateResponse( request=request, name="schedules/form.html", context={ "version": __version__, "schedule": schedule, "message_templates": message_templates, "errors": errors or [], "history": history or [], "weekday_names": {1: "Пн", 2: "Вт", 3: "Ср", 4: "Чт", 5: "Пт", 6: "Сб", 7: "Вс"}, }, status_code=status_code, ) @router.get("", response_class=HTMLResponse) async def list_schedules(request: Request) -> HTMLResponse: """Список расписаний: время, дни, шаблон, таймзона и статус.""" return request.app.state.templates.TemplateResponse( request=request, name="schedules/list.html", context={ "version": __version__, "schedules": await schedules_repo.list_schedules(), "format_weekdays": format_weekdays, }, ) @router.get("/new", response_class=HTMLResponse) async def new_schedule(request: Request) -> HTMLResponse: """Форма создания расписания.""" message_templates = await templates_repo.list_templates() defaults = { "template_key": message_templates[0]["key"] if message_templates else "", "hour": 16, "minute": 0, "weekdays": [], "timezone": "Europe/Moscow", "description": "", "enabled": True, } return _form_response(request, _view(defaults, is_new=True), message_templates) @router.post("") async def create_schedule(request: Request) -> Response: """Создать расписание (с проверкой полей и существования шаблона).""" form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) raw = _raw_from_form(form) data, errors = parse_schedule_fields(**raw) message_templates = await templates_repo.list_templates() if data is not None and await templates_repo.get_template(data["template_key"]) is None: errors.append("Шаблон не найден") if errors: return _form_response( request, _view(raw, is_new=True), message_templates, errors=errors, status_code=400, ) await schedules_repo.create_schedule(data, editor_id=_editor_id(request)) return RedirectResponse(url="/schedules", status_code=303) @router.get("/{schedule_id}", response_class=HTMLResponse) async def edit_schedule(request: Request, schedule_id: int) -> HTMLResponse: """Форма редактирования и история изменений.""" schedule = await schedules_repo.get_schedule(schedule_id) if schedule is None: raise HTTPException(status_code=404, detail="Расписание не найдено") return _form_response( request, _view(schedule, is_new=False), await templates_repo.list_templates(), history=await schedules_repo.schedule_history(schedule_id), ) @router.post("/{schedule_id}") async def update_schedule(request: Request, schedule_id: int) -> Response: """Сохранить изменения расписания.""" if await schedules_repo.get_schedule(schedule_id) is None: raise HTTPException(status_code=404, detail="Расписание не найдено") form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) raw = _raw_from_form(form) data, errors = parse_schedule_fields(**raw) message_templates = await templates_repo.list_templates() if data is not None and await templates_repo.get_template(data["template_key"]) is None: errors.append("Шаблон не найден") if errors: view = _view({**raw, "id": schedule_id}, is_new=False) return _form_response( request, view, message_templates, errors=errors, history=await schedules_repo.schedule_history(schedule_id), status_code=400, ) await schedules_repo.update_schedule(schedule_id, data, editor_id=_editor_id(request)) return RedirectResponse(url="/schedules", status_code=303) @router.post("/{schedule_id}/toggle") async def toggle_schedule(request: Request, schedule_id: int) -> Response: """Включить или выключить расписание.""" form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) schedule = await schedules_repo.get_schedule(schedule_id) if schedule is None: raise HTTPException(status_code=404, detail="Расписание не найдено") await schedules_repo.set_enabled(schedule_id, not schedule["enabled"], editor_id=_editor_id(request)) return RedirectResponse(url="/schedules", status_code=303) @router.post("/{schedule_id}/delete") async def delete_schedule(request: Request, schedule_id: int) -> Response: """Удалить расписание (история при этом сохраняется).""" form = await request.form() check_csrf(request, str(form.get("csrf_token", ""))) if not await schedules_repo.delete_schedule(schedule_id, editor_id=_editor_id(request)): raise HTTPException(status_code=404, detail="Расписание не найдено") return RedirectResponse(url="/schedules", status_code=303)