Редактор ачивок в админке + файловое хранилище ачивок (задел под титулы)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-17 17:41:20 +03:00
parent 96daddae05
commit c4065962e2
19 changed files with 1165 additions and 6 deletions
@@ -45,6 +45,10 @@ class Settings(BaseSettings):
    dev_upload_dir: str = "./data/dev/uploads"
    prod_upload_dir: str = "/data/uploads"

+    # Каталог ачивок (папка на ачивку: meta.json, condition.py, icon.*), раздельно.
+    dev_achievements_dir: str = "./data/dev/achievements"
+    prod_achievements_dir: str = "/data/achievements"
+
    # JWT / cookie
    secret_key: str = "change-me-dev-secret-not-for-production"
    jwt_algorithm: str = "HS256"
@@ -97,6 +101,10 @@ class Settings(BaseSettings):
    def upload_dir(self) -> str:
        return self.dev_upload_dir if self.is_development else self.prod_upload_dir

+    @property
+    def achievements_dir(self) -> str:
+        return self.dev_achievements_dir if self.is_development else self.prod_achievements_dir
+
    @property
    def cookie_secure(self) -> bool:
        """Secure-cookie нужен везде, где снаружи HTTPS (домен). Исключение —
@@ -15,7 +15,7 @@ from starlette.middleware.base import BaseHTTPMiddleware
 from app.core import security
 from app.core.config import settings
 from app.core.errors import AppError, app_error_handler
-from app.routers import admin, auth, groups, matches, reference, stats, users
+from app.routers import achievements, admin, auth, groups, matches, reference, stats, users

 # Каталог со сборкой фронта (в Docker — backend/static; локально может отсутствовать).
 _STATIC_DIR = Path(os.getenv("STATIC_DIR", str(Path(__file__).resolve().parent.parent / "static")))
@@ -107,7 +107,7 @@ def create_app() -> FastAPI:

    # API-роутеры под /api.
    api_routers = [auth.router, users.router, groups.router, matches.router,
-                   reference.router, stats.router, admin.router]
+                   reference.router, stats.router, achievements.router, admin.router]
    for r in api_routers:
        app.include_router(r, prefix="/api")

@@ -98,6 +98,9 @@ class User(SQLModel, table=True):
        )
    )
    is_active: bool = Field(sa_column=Column(Boolean, nullable=False, server_default="1"))
+    # Выбранный титул (slug ачивки), отображаемый под ником. Задел: пока всегда NULL
+    # (выдача ачивок игрокам — следующий этап).
+    title_achievement_slug: str | None = Field(sa_column=Column(String(64), nullable=True))
    created_at: datetime = Field(default_factory=_utcnow, nullable=False)
    updated_at: datetime = Field(
        default_factory=_utcnow,
@@ -106,6 +109,26 @@ class User(SQLModel, table=True):
    )


+class UserAchievement(SQLModel, table=True):
+    """Полученные игроком ачивки. Задел под будущую выдачу — сейчас не заполняется.
+
+    Ссылается на ачивку по slug (источник правды — папки на диске, не таблица БД)."""
+
+    __tablename__ = "user_achievements"
+    __table_args__ = (
+        UniqueConstraint("user_id", "achievement_slug", name="uq_user_achievement"),
+    )
+
+    id: int | None = Field(default=None, primary_key=True)
+    user_id: int = Field(
+        sa_column=Column(
+            Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
+        )
+    )
+    achievement_slug: str = Field(sa_column=Column(String(64), nullable=False))
+    earned_at: datetime = Field(default_factory=_utcnow, nullable=False)
+
+
 class AuthIdentity(SQLModel, table=True):
    __tablename__ = "auth_identity"
    __table_args__ = (
@@ -0,0 +1,27 @@
+"""Публичное чтение ачивок: список и иконки (для отображения; выдача — на будущее)."""
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends
+from fastapi.responses import FileResponse
+
+from app.auth.deps import get_current_user
+from app.models import User
+from app.schemas import api as s
+from app.services import achievement_service
+
+router = APIRouter(prefix="/achievements", tags=["achievements"])
+
+
+@router.get("", response_model=list[s.AchievementRead])
+def list_achievements(_user: User = Depends(get_current_user)) -> list[dict]:
+    return achievement_service.list_achievements()
+
+
+# Иконка — без auth: косметический ассет, нужен и игроку, и в админ-панели (cookie
+# админа scoped на /api/admin и сюда не отправляется). Отдаётся по известному slug.
+@router.get("/{slug}/icon")
+def get_icon(slug: str) -> FileResponse:
+    path, media_type = achievement_service.icon_file(slug)
+    return FileResponse(
+        path, media_type=media_type, headers={"Cache-Control": "public, max-age=3600"}
+    )
@@ -1,7 +1,7 @@
 """Админ-роутер: отдельный вход (логин+пароль) и управление сущностями."""
 from __future__ import annotations

-from fastapi import APIRouter, Depends, Query, Request, Response
+from fastapi import APIRouter, Depends, File, Query, Request, Response, UploadFile
 from sqlmodel import Session

 from app.auth.deps import get_current_admin
@@ -11,7 +11,15 @@ from app.db.session import get_session
 from app.models import User
 from app.routers.matches import build_match_read
 from app.schemas import api as s
-from app.services import admin_service, audit_service, faction_service, match_service
+from app.services import (
+    achievement_service,
+    admin_service,
+    audit_service,
+    faction_service,
+    match_service,
+)
+
+_ACHIEVEMENT_ICON_MAX_BYTES = 2 * 1024 * 1024  # 2 МБ
 from app.services.match_service import ParticipantInput

 router = APIRouter(prefix="/admin", tags=["admin"])
@@ -284,6 +292,82 @@ def delete_match(
    return s.OkResponse()


+# ─── Ачивки (определения; выдача игрокам — на будущее) ────────────────────────
+
+@router.get("/achievements", response_model=list[s.AchievementRead])
+def list_achievements(
+    _admin: User = Depends(get_current_admin),
+) -> list[dict]:
+    return achievement_service.list_achievements()
+
+
+@router.post("/achievements", response_model=s.AchievementRead)
+def create_achievement(
+    body: s.AchievementCreate,
+    request: Request,
+    session: Session = Depends(get_session),
+    admin: User = Depends(get_current_admin),
+) -> dict:
+    ach = achievement_service.create(body.name, body.description, body.condition)
+    audit_service.record(
+        session, actor_id=admin.id, action="create", entity_type="achievement",
+        payload={"slug": ach["slug"], "name": ach["name"]},
+        ip=request.client.host if request.client else None,
+    )
+    session.commit()
+    return ach
+
+
+@router.patch("/achievements/{slug}", response_model=s.AchievementRead)
+def update_achievement(
+    slug: str,
+    body: s.AchievementUpdate,
+    request: Request,
+    session: Session = Depends(get_session),
+    admin: User = Depends(get_current_admin),
+) -> dict:
+    ach = achievement_service.update(
+        slug, name=body.name, description=body.description, condition=body.condition
+    )
+    audit_service.record(
+        session, actor_id=admin.id, action="update", entity_type="achievement",
+        payload={"slug": slug}, ip=request.client.host if request.client else None,
+    )
+    session.commit()
+    return ach
+
+
+@router.put("/achievements/{slug}/icon", response_model=s.AchievementRead)
+def upload_achievement_icon(
+    slug: str,
+    file: UploadFile = File(...),
+    _admin: User = Depends(get_current_admin),
+) -> dict:
+    from app.core.errors import ValidationError
+
+    content = file.file.read(_ACHIEVEMENT_ICON_MAX_BYTES + 1)
+    if len(content) > _ACHIEVEMENT_ICON_MAX_BYTES:
+        raise ValidationError("Файл слишком большой (макс. 2 МБ).")
+    ext = achievement_service.validate_icon(content)
+    return achievement_service.set_icon(slug, content, ext)
+
+
+@router.delete("/achievements/{slug}", response_model=s.OkResponse)
+def delete_achievement(
+    slug: str,
+    request: Request,
+    session: Session = Depends(get_session),
+    admin: User = Depends(get_current_admin),
+) -> s.OkResponse:
+    achievement_service.delete(slug)
+    audit_service.record(
+        session, actor_id=admin.id, action="delete", entity_type="achievement",
+        payload={"slug": slug}, ip=request.client.host if request.client else None,
+    )
+    session.commit()
+    return s.OkResponse()
+
+
 # ─── Журнал аудита ───────────────────────────────────────────────────────────

@router.get("/audit-logs", response_model=s.AuditLogList)
@@ -410,6 +410,27 @@ class AdminMatchRead(BaseModel):
    created_at: str


+class AchievementRead(BaseModel):
+    slug: str
+    name: str
+    description: str = ""
+    condition: str = ""  # текст condition.py (задел; не исполняется)
+    has_condition: bool = False
+    icon_url: str | None = None
+
+
+class AchievementCreate(BaseModel):
+    name: str
+    description: str | None = None
+    condition: str | None = None
+
+
+class AchievementUpdate(BaseModel):
+    name: str | None = None
+    description: str | None = None
+    condition: str | None = None
+
+
 class AuditLogItem(BaseModel):
    id: int
    actor_id: int | None = None
@@ -0,0 +1,194 @@
+"""Ачивки: определения хранятся файлами на диске (settings.achievements_dir).
+
+Папка на ачивку: achievements/<slug>/
+  meta.json     — {"name", "description", "created_at"}
+  condition.py  — скрипт-условие выдачи (ЗАДЕЛ: только хранится как текст, НЕ исполняется)
+  icon.<ext>    — картинка-герб (png/jpg/webp), опционально
+
+Источник правды — файлы. В БД (user_achievements / users.title_achievement_slug) — только
+будущая выдача игрокам (сейчас не используется). Выдача и исполнение condition.py — следующий
+этап; здесь админ лишь создаёт/именует/описывает ачивки.
+"""
+from __future__ import annotations
+
+import json
+import re
+import shutil
+from datetime import datetime, timezone
+from pathlib import Path
+
+from app.core.config import settings
+from app.core.errors import ConflictError, NotFoundError, ValidationError
+from app.services.user_service import avatar_media_type, sniff_image_ext
+
+_CONDITION_FILE = "condition.py"
+_META_FILE = "meta.json"
+_CONDITION_PLACEHOLDER = (
+    "# Условие выдачи ачивки (ЗАДЕЛ на будущее — пока не исполняется).\n"
+    "# Здесь будет описана логика автоматической выдачи этого титула игроку.\n"
+)
+
+# Транслитерация для slug из русского названия.
+_TRANSLIT = {
+    "а": "a", "б": "b", "в": "v", "г": "g", "д": "d", "е": "e", "ё": "e", "ж": "zh",
+    "з": "z", "и": "i", "й": "i", "к": "k", "л": "l", "м": "m", "н": "n", "о": "o",
+    "п": "p", "р": "r", "с": "s", "т": "t", "у": "u", "ф": "f", "х": "h", "ц": "ts",
+    "ч": "ch", "ш": "sh", "щ": "sch", "ъ": "", "ы": "y", "ь": "", "э": "e", "ю": "yu",
+    "я": "ya",
+}
+
+
+def _root() -> Path:
+    return Path(settings.achievements_dir)
+
+
+def _slugify(name: str) -> str:
+    text = "".join(_TRANSLIT.get(ch, ch) for ch in (name or "").strip().lower())
+    slug = re.sub(r"[^a-z0-9]+", "-", text).strip("-")
+    return slug[:48]
+
+
+def _dir(slug: str) -> Path:
+    return _root() / slug
+
+
+def _icon_path(slug: str) -> Path | None:
+    folder = _dir(slug)
+    if not folder.is_dir():
+        return None
+    for f in sorted(folder.glob("icon.*")):
+        if f.is_file():
+            return f
+    return None
+
+
+def _read(slug: str) -> dict:
+    folder = _dir(slug)
+    meta_file = folder / _META_FILE
+    if not meta_file.is_file():
+        raise NotFoundError("Ачивка не найдена.")
+    meta = json.loads(meta_file.read_text(encoding="utf-8"))
+    condition = ""
+    cond_file = folder / _CONDITION_FILE
+    if cond_file.is_file():
+        condition = cond_file.read_text(encoding="utf-8")
+    icon = _icon_path(slug)
+    icon_url = None
+    if icon is not None:
+        version = int(icon.stat().st_mtime)
+        icon_url = f"/api/achievements/{slug}/icon?v={version}"
+    return {
+        "slug": slug,
+        "name": meta.get("name", slug),
+        "description": meta.get("description", ""),
+        "condition": condition,
+        "has_condition": bool(condition.strip()),
+        "icon_url": icon_url,
+    }
+
+
+def list_achievements() -> list[dict]:
+    root = _root()
+    if not root.is_dir():
+        return []
+    out = []
+    for folder in sorted(root.iterdir()):
+        if folder.is_dir() and (folder / _META_FILE).is_file():
+            out.append(_read(folder.name))
+    return out
+
+
+def get(slug: str) -> dict:
+    return _read(slug)
+
+
+def _validate_name(name: str) -> str:
+    name = (name or "").strip()
+    if not (1 <= len(name) <= 64):
+        raise ValidationError("Название ачивки: 1–64 символа.")
+    return name
+
+
+def create(name: str, description: str | None, condition: str | None) -> dict:
+    name = _validate_name(name)
+    slug = _slugify(name)
+    if not slug:
+        raise ValidationError("Не удалось сформировать идентификатор из названия.")
+    folder = _dir(slug)
+    if folder.exists():
+        raise ConflictError("Ачивка с таким названием уже существует.")
+    folder.mkdir(parents=True, exist_ok=True)
+    _write_meta(slug, name, (description or "").strip())
+    (folder / _CONDITION_FILE).write_text(
+        condition if condition else _CONDITION_PLACEHOLDER, encoding="utf-8"
+    )
+    return _read(slug)
+
+
+def update(
+    slug: str,
+    *,
+    name: str | None = None,
+    description: str | None = None,
+    condition: str | None = None,
+) -> dict:
+    folder = _dir(slug)
+    meta_file = folder / _META_FILE
+    if not meta_file.is_file():
+        raise NotFoundError("Ачивка не найдена.")
+    meta = json.loads(meta_file.read_text(encoding="utf-8"))
+    # slug не меняем (он стабильный идентификатор/путь), даже если правят название.
+    new_name = _validate_name(name) if name is not None else meta.get("name", slug)
+    new_desc = description.strip() if description is not None else meta.get("description", "")
+    _write_meta(slug, new_name, new_desc, created_at=meta.get("created_at"))
+    if condition is not None:
+        (folder / _CONDITION_FILE).write_text(condition, encoding="utf-8")
+    return _read(slug)
+
+
+def set_icon(slug: str, content: bytes, ext: str) -> dict:
+    folder = _dir(slug)
+    if not (folder / _META_FILE).is_file():
+        raise NotFoundError("Ачивка не найдена.")
+    old = _icon_path(slug)
+    if old is not None:
+        try:
+            old.unlink()
+        except OSError:
+            pass
+    (folder / f"icon.{ext}").write_bytes(content)
+    return _read(slug)
+
+
+def icon_file(slug: str) -> tuple[Path, str]:
+    """Путь к файлу иконки и его media_type; NotFound, если иконки нет."""
+    icon = _icon_path(slug)
+    if icon is None:
+        raise NotFoundError("Иконка не найдена.")
+    return icon, avatar_media_type(icon.suffix.lstrip("."))
+
+
+def delete(slug: str) -> None:
+    folder = _dir(slug)
+    if not folder.is_dir():
+        raise NotFoundError("Ачивка не найдена.")
+    shutil.rmtree(folder)
+
+
+def validate_icon(content: bytes) -> str:
+    """Проверяет, что это поддерживаемая картинка, и возвращает расширение."""
+    ext = sniff_image_ext(content)
+    if ext is None:
+        raise ValidationError("Поддерживаются только изображения PNG, JPEG или WebP.")
+    return ext
+
+
+def _write_meta(slug: str, name: str, description: str, created_at: str | None = None) -> None:
+    meta = {
+        "name": name,
+        "description": description,
+        "created_at": created_at or datetime.now(timezone.utc).isoformat(),
+    }
+    (_dir(slug) / _META_FILE).write_text(
+        json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8"
+    )