146 lines
5.6 KiB
Python
146 lines
5.6 KiB
Python
"""Раздел «Админы»: назначение и управление доступом к веб-панели."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from fastapi import APIRouter, HTTPException, Request, Response
|
||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||
|
||
from admin.security import check_csrf
|
||
from common import __version__
|
||
from common.repositories import admins_repo
|
||
|
||
router = APIRouter(prefix="/admins", tags=["admins"])
|
||
|
||
#: Допустимые роли (обе сейчас дают полный доступ, moderator — задел).
|
||
ROLES = ("admin", "moderator")
|
||
|
||
|
||
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="admins/list.html",
|
||
context={
|
||
"version": __version__,
|
||
"admins": await admins_repo.list_admins(),
|
||
"roles": ROLES,
|
||
"errors": errors or [],
|
||
"form": form or {"telegram_id": "", "username": "", "role": "admin"},
|
||
"saved": saved,
|
||
"current_id": int(request.state.admin["telegram_id"]),
|
||
},
|
||
status_code=status_code,
|
||
)
|
||
|
||
|
||
@router.get("", response_class=HTMLResponse)
|
||
async def list_admins(request: Request, saved: str | None = None) -> HTMLResponse:
|
||
"""Список администраторов и форма добавления."""
|
||
return await _list_response(request, saved=saved)
|
||
|
||
|
||
@router.post("")
|
||
async def add_admin(request: Request) -> Response:
|
||
"""Добавить администратора по Telegram-ID."""
|
||
form = await request.form()
|
||
check_csrf(request, str(form.get("csrf_token", "")))
|
||
|
||
raw_id = str(form.get("telegram_id", "")).strip()
|
||
username = str(form.get("username", "")).strip().lstrip("@")
|
||
role = str(form.get("role", "admin"))
|
||
raw_form = {"telegram_id": raw_id, "username": username, "role": role}
|
||
|
||
errors: list[str] = []
|
||
try:
|
||
telegram_id = int(raw_id)
|
||
except (TypeError, ValueError):
|
||
telegram_id = 0
|
||
errors.append("Telegram-ID должен быть числом")
|
||
if telegram_id <= 0 and "Telegram-ID должен быть числом" not in errors:
|
||
errors.append("Telegram-ID должен быть положительным")
|
||
if role not in ROLES:
|
||
errors.append("Неизвестная роль")
|
||
if len(username) > 64:
|
||
errors.append("Username длиннее 64 символов")
|
||
|
||
if errors:
|
||
return await _list_response(request, errors=errors, form=raw_form, status_code=400)
|
||
|
||
await admins_repo.upsert_admin(telegram_id, username or None, role)
|
||
return RedirectResponse(url="/admins?saved=added", status_code=303)
|
||
|
||
|
||
@router.post("/{telegram_id}/role")
|
||
async def change_role(request: Request, telegram_id: int) -> Response:
|
||
form = await request.form()
|
||
check_csrf(request, str(form.get("csrf_token", "")))
|
||
role = str(form.get("role", ""))
|
||
|
||
if role not in ROLES:
|
||
return await _list_response(request, errors=["Неизвестная роль"], status_code=400)
|
||
if not await admins_repo.set_admin_role(telegram_id, role):
|
||
raise HTTPException(status_code=404, detail="Администратор не найден")
|
||
return RedirectResponse(url="/admins?saved=role", status_code=303)
|
||
|
||
|
||
@router.post("/{telegram_id}/toggle")
|
||
async def toggle_admin(request: Request, telegram_id: int) -> Response:
|
||
form = await request.form()
|
||
check_csrf(request, str(form.get("csrf_token", "")))
|
||
|
||
if telegram_id == int(request.state.admin["telegram_id"]):
|
||
return await _list_response(
|
||
request,
|
||
errors=["Нельзя выключить самого себя"],
|
||
status_code=400,
|
||
)
|
||
|
||
admins = await admins_repo.list_admins()
|
||
target = next((item for item in admins if int(item["telegram_id"]) == telegram_id), None)
|
||
if target is None:
|
||
raise HTTPException(status_code=404, detail="Администратор не найден")
|
||
|
||
enabled_count = sum(1 for item in admins if item["enabled"])
|
||
if target["enabled"] and enabled_count <= 1:
|
||
return await _list_response(
|
||
request,
|
||
errors=["Нельзя выключить последнего администратора"],
|
||
status_code=400,
|
||
)
|
||
|
||
await admins_repo.set_admin_enabled(telegram_id, not target["enabled"])
|
||
return RedirectResponse(url="/admins?saved=toggle", status_code=303)
|
||
|
||
|
||
@router.post("/{telegram_id}/delete")
|
||
async def delete_admin(request: Request, telegram_id: int) -> Response:
|
||
form = await request.form()
|
||
check_csrf(request, str(form.get("csrf_token", "")))
|
||
|
||
if telegram_id == int(request.state.admin["telegram_id"]):
|
||
return await _list_response(
|
||
request,
|
||
errors=["Нельзя удалить самого себя"],
|
||
status_code=400,
|
||
)
|
||
|
||
admins = await admins_repo.list_admins()
|
||
target = next((item for item in admins if int(item["telegram_id"]) == telegram_id), None)
|
||
if target is None:
|
||
raise HTTPException(status_code=404, detail="Администратор не найден")
|
||
if target["enabled"] and sum(1 for item in admins if item["enabled"]) <= 1:
|
||
return await _list_response(
|
||
request,
|
||
errors=["Нельзя удалить последнего администратора"],
|
||
status_code=400,
|
||
)
|
||
|
||
await admins_repo.delete_admin(telegram_id)
|
||
return RedirectResponse(url="/admins?saved=deleted", status_code=303)
|