"""Вложения партии (фото). Файлы — на томе settings.upload_dir, в БД метаданные. Общий сервис для игрового и админского путей (роутеры лишь строят URL с нужным префиксом). """ from __future__ import annotations import shutil from pathlib import Path from sqlmodel import Session, select from app.core.config import settings from app.core.errors import ConflictError, NotFoundError from app.models import Match, MatchAttachment, User MAX_ATTACHMENTS = 10 _SUBDIR = "matches" def list_for_match(session: Session, match_id: int) -> list[MatchAttachment]: return list( session.exec( select(MatchAttachment) .where(MatchAttachment.match_id == match_id) .order_by(MatchAttachment.created_at, MatchAttachment.id) ).all() ) def count(session: Session, match_id: int) -> int: return len( session.exec( select(MatchAttachment.id).where(MatchAttachment.match_id == match_id) ).all() ) def file_path(att: MatchAttachment) -> Path: return Path(settings.upload_dir) / att.storage_path def add_photo( session: Session, match: Match, user: User, content: bytes, ext: str, mime: str ) -> MatchAttachment: if count(session, match.id) >= MAX_ATTACHMENTS: # type: ignore[arg-type] raise ConflictError(f"Можно прикрепить не более {MAX_ATTACHMENTS} файлов.") att = MatchAttachment( match_id=match.id, # type: ignore[arg-type] uploaded_by=user.id, kind="photo", storage_path="", mime_type=mime, size_bytes=len(content), ) session.add(att) session.flush() # получаем id для имени файла rel = f"{_SUBDIR}/{match.id}/{att.id}.{ext}" abs_path = Path(settings.upload_dir) / rel abs_path.parent.mkdir(parents=True, exist_ok=True) abs_path.write_bytes(content) att.storage_path = rel session.add(att) session.commit() session.refresh(att) return att def get_for_match(session: Session, match_id: int, att_id: int) -> MatchAttachment: att = session.get(MatchAttachment, att_id) if att is None or att.match_id != match_id: raise NotFoundError("Вложение не найдено.") return att def delete(session: Session, match: Match, att_id: int) -> None: att = get_for_match(session, match.id, att_id) # type: ignore[arg-type] path = file_path(att) if path.exists(): try: path.unlink() except OSError: pass session.delete(att) session.commit() def delete_match_files(match_id: int) -> None: """Удалить всю папку медиа партии (при удалении партии; строки БД уйдут каскадом).""" folder = Path(settings.upload_dir) / _SUBDIR / str(match_id) if folder.is_dir(): shutil.rmtree(folder, ignore_errors=True)