"""Effektives Medien-Wurzelverzeichnis (MEDIA_ROOT + Superadmin-relativer Pfad). Siehe MEDIA_ASSETS_AND_ARCHIVE_SPEC.md §7.1.""" from __future__ import annotations import os import re import shutil from pathlib import Path from typing import Any, Optional def _default_media_root() -> Path: return Path(os.getenv("MEDIA_ROOT", str(Path(__file__).resolve().parent / "media"))) def normalize_local_relative_root(raw: str) -> str: s = (raw or "").strip().replace("\\", "/") s = s.strip("/") if not s: return "" if ".." in s.split("/"): raise ValueError("Pfad darf nicht '..' enthalten") if s.startswith("/"): raise ValueError("Nur relativer Pfad erlaubt") return s def get_effective_media_root(cur: Any) -> Path: """ MEDIA_ROOT aus ENV mit optionalem local_relative_root aus platform_media_storage (id=1). """ base = _default_media_root().resolve() rel = "" try: cur.execute( "SELECT local_relative_root FROM platform_media_storage WHERE id = 1", ) row = cur.fetchone() if row is not None: v = row["local_relative_root"] if isinstance(row, dict) else row[0] rel = normalize_local_relative_root(str(v or "")) except Exception: rel = "" if not rel: return base return (base / rel).resolve() def path_under_media_root(media_root: Path, storage_key: str) -> Optional[Path]: """Gibt absoluten Pfad zurück oder None bei Path-Traversal.""" key = (storage_key or "").strip().replace("\\", "/").lstrip("/") if not key or ".." in key.split("/"): return None p = (media_root / key).resolve() try: p.relative_to(media_root.resolve()) except ValueError: return None return p def library_storage_key( visibility: str, club_id: Optional[int], sha256_hex: str, ext: str, ) -> str: """ Relativer Speicherpfad unter MEDIA_ROOT für lokale media_assets. Layout (nach Verein gegliedert, globale Medien getrennt): - private → library/club/c{club_id}/private/{sha256}{ext} - club (Vereinssichtbarkeit) → library/club/c{club_id}/shared/{sha256}{ext} - official → library/official/{sha256}{ext} club_id ist bei private/club zwingend (Vereinsordner); bei official nicht genutzt. """ vis = (visibility or "private").strip().lower() if vis not in ("private", "club", "official"): raise ValueError(f"Ungültige Sichtbarkeit für Speicherpfad: {visibility!r}") sha = (sha256_hex or "").strip().lower() if len(sha) != 64 or any(c not in "0123456789abcdef" for c in sha): raise ValueError("sha256_hex muss 64 Hex-Zeichen sein") e = (ext or "").strip() if not e: e = ".bin" if not e.startswith("."): e = "." + e e = e[:16] if ".." in e or "/" in e or "\\" in e or "\x00" in e: raise ValueError("Ungültige Dateiendung") blob = f"{sha}{e}" if vis == "official": return f"library/official/{blob}" if club_id is None: raise ValueError("Verein (club_id) ist für diese Sichtbarkeit auf der Platte erforderlich") cid = int(club_id) if cid < 1: raise ValueError("club_id muss eine positive Ganzzahl sein") if vis == "private": return f"library/club/c{cid}/private/{blob}" return f"library/club/c{cid}/shared/{blob}" def relocate_local_media_file(media_root: Path, old_storage_key: str, new_storage_key: str) -> None: """ Physisches Verschieben bei geändertem library_*-Pfad (z. B. nach PATCH visibility/club_id). - Kein Op, wenn Quelle fehlt aber Ziel bereits existiert (idempotent). - Erwartet storage_backend=local; Aufrufer prüft das. """ if (old_storage_key or "").replace("\\", "/").lstrip("/") == (new_storage_key or "").replace( "\\", "/" ).lstrip("/"): return old_p = path_under_media_root(media_root, old_storage_key) new_p = path_under_media_root(media_root, new_storage_key) if old_p is None or new_p is None: raise ValueError("Ungültiger Speicherpfad (Path-Traversal)") if new_p.is_file(): if not old_p.is_file(): return if old_p.resolve() == new_p.resolve(): return raise FileExistsError(f"Zieldatei existiert bereits: {new_storage_key}") if not old_p.is_file(): raise FileNotFoundError(f"Medien-Quelldatei nicht gefunden: {old_storage_key}") new_p.parent.mkdir(parents=True, exist_ok=True) shutil.move(str(old_p), str(new_p))