Находки прохода /code-review high по backend/app: - achievement_service: slug из URL шёл в путь без проверки, из-за чего DELETE /api/admin/achievements/%2E%2E удалял rmtree'ом родительскую папку каталога ачивок (в проде это /data — БД, uploads, ачивки целиком). - match_service/attachment_service: версия партии = updated_at, но onupdate срабатывает лишь при реальном UPDATE строки matches. Правка одних участников и работа с вложениями его не вызывали, и оптимистичная блокировка молча пропускала конкурентную запись — бампаем updated_at явно. - admin_service: удаление группы с партиями упиралось в RESTRICT и уходило наружу голым 500; теперь понятная ошибка. Админское удаление партии не чистило файлы вложений с тома — они оставались навсегда. - user_service: при повторной загрузке аватара с тем же расширением avatar_path не менялся, updated_at не двигался, и кэш-бастер оставлял старую картинку до часа. Плюс версия считалась из наивного времени как из локального и разъезжалась с лидербордом, где то же поле считает SQL. - membership_service: единственный владелец мог разжаловать сам себя и группа оставалась без владельца навсегда. - notification_service: mark_read не слал SSE-сигнал, и бейдж непрочитанных на других устройствах висел до перезагрузки. - routers/admin: created_at после правки пользователя отдавался без смещения, и дата «создан» прыгала на часовой пояс до следующего обновления списка. #8 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0186Fk74jkkszahEHSjBzTjD
204 lines
7.6 KiB
Python
204 lines
7.6 KiB
Python
"""Ачивки: определения хранятся файлами на диске (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)
|
||
|
||
|
||
# Формат slug — то, что выдаёт _slugify: только латиница, цифры и дефис.
|
||
_SLUG_RE = re.compile(r"[a-z0-9][a-z0-9-]{0,63}")
|
||
|
||
|
||
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:
|
||
"""Папка ачивки. Slug приходит из URL, поэтому формат проверяем здесь: без этого
|
||
`..` или `a/b` увели бы файловые операции (вплоть до rmtree в delete) за пределы
|
||
каталога ачивок."""
|
||
if not _SLUG_RE.fullmatch(slug or ""):
|
||
raise NotFoundError("Ачивка не найдена.")
|
||
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"
|
||
)
|