Trixy7/trixy_core/assets/manager.py

# -*- coding: utf-8 -*-
"""
Asset-Manager für Trixy.

Verwaltet Assets mit Profil-basiertem Fallback-System:
1. Suche im aktuellen Profil-Verzeichnis
2. Fallback auf "default" Profil
3. Optional: Caching für Performance

Beispiel:
    manager = AssetManager("./assets", profile="custom_profile")

    # Sucht: ./assets/custom_profile/audio/success.wav
    # Fallback: ./assets/default/audio/success.wav
    path = manager.get("audio/success.wav")
"""

from dataclasses import dataclass, field
from enum import Enum, auto
from pathlib import Path
from typing import Iterator
import hashlib
import mimetypes
import threading
import time

from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn


class AssetType(Enum):
    """Asset-Kategorien für typisierte Zugriffe."""
    AUDIO = auto()          # Audio-Dateien (WAV, MP3, etc.)
    MUSIC = auto()          # Musik-Dateien
    IMAGE = auto()          # Bilder (PNG, JPG, etc.)
    VIDEO = auto()          # Video-Dateien
    CONFIG = auto()         # Konfigurationsdateien
    DATA = auto()           # Datendateien (JSON, XML, etc.)
    MODEL = auto()          # ML-Modelle
    UNKNOWN = auto()        # Unbekannter Typ


# Dateiendungen zu Asset-Typen
_EXTENSION_MAP: dict[str, AssetType] = {
    # Audio
    ".wav": AssetType.AUDIO,
    ".mp3": AssetType.AUDIO,
    ".ogg": AssetType.AUDIO,
    ".flac": AssetType.AUDIO,
    ".aac": AssetType.AUDIO,
    ".m4a": AssetType.AUDIO,
    # Musik (gleiche Formate, aber in music/ Verzeichnis)
    # Image
    ".png": AssetType.IMAGE,
    ".jpg": AssetType.IMAGE,
    ".jpeg": AssetType.IMAGE,
    ".gif": AssetType.IMAGE,
    ".bmp": AssetType.IMAGE,
    ".svg": AssetType.IMAGE,
    ".webp": AssetType.IMAGE,
    ".ico": AssetType.IMAGE,
    # Video
    ".mp4": AssetType.VIDEO,
    ".avi": AssetType.VIDEO,
    ".mkv": AssetType.VIDEO,
    ".mov": AssetType.VIDEO,
    ".webm": AssetType.VIDEO,
    # Config
    ".json": AssetType.CONFIG,
    ".yaml": AssetType.CONFIG,
    ".yml": AssetType.CONFIG,
    ".toml": AssetType.CONFIG,
    ".ini": AssetType.CONFIG,
    # Data
    ".xml": AssetType.DATA,
    ".csv": AssetType.DATA,
    ".txt": AssetType.DATA,
    # Model
    ".pth": AssetType.MODEL,
    ".pt": AssetType.MODEL,
    ".onnx": AssetType.MODEL,
    ".h5": AssetType.MODEL,
    ".keras": AssetType.MODEL,
}


class AssetNotFoundError(FileNotFoundError):
    """
    Wird ausgelöst, wenn ein Asset weder im Profil noch im Default gefunden wird.
    """

    def __init__(self, asset_path: str, profile: str, base_dir: Path) -> None:
        self.asset_path = asset_path
        self.profile = profile
        self.base_dir = base_dir
        self.searched_paths = [
            base_dir / profile / asset_path,
            base_dir / "default" / asset_path,
        ]
        super().__init__(
            f"Asset nicht gefunden: '{asset_path}' "
            f"(Profil: {profile}, gesucht in: {', '.join(str(p) for p in self.searched_paths)})"
        )


@dataclass
class AssetInfo:
    """
    Informationen über ein Asset.

    Attributes:
        path: Absoluter Pfad zum Asset
        relative_path: Relativer Pfad (z.B. "audio/success.wav")
        profile: Profil, aus dem das Asset stammt
        is_default: True wenn aus dem Default-Profil
        asset_type: Typ des Assets
        size: Dateigröße in Bytes
        modified: Letzte Änderung (Unix-Timestamp)
        mime_type: MIME-Typ der Datei
        checksum: MD5-Hash (optional, lazy berechnet)
    """
    path: Path
    relative_path: str
    profile: str
    is_default: bool
    asset_type: AssetType
    size: int
    modified: float
    mime_type: str | None = None
    _checksum: str | None = field(default=None, repr=False)

    @property
    def checksum(self) -> str:
        """Berechnet und cached den MD5-Hash des Assets."""
        if self._checksum is None:
            self._checksum = self._calculate_checksum()
        return self._checksum

    def _calculate_checksum(self) -> str:
        """Berechnet den MD5-Hash."""
        hash_md5 = hashlib.md5()
        try:
            with open(self.path, "rb") as f:
                for chunk in iter(lambda: f.read(4096), b""):
                    hash_md5.update(chunk)
            return hash_md5.hexdigest()
        except Exception:
            return ""

    @property
    def exists(self) -> bool:
        """Prüft, ob das Asset noch existiert."""
        return self.path.exists()

    @property
    def extension(self) -> str:
        """Gibt die Dateiendung zurück."""
        return self.path.suffix.lower()

    @property
    def name(self) -> str:
        """Gibt den Dateinamen zurück."""
        return self.path.name

    @property
    def stem(self) -> str:
        """Gibt den Dateinamen ohne Endung zurück."""
        return self.path.stem


@dataclass
class CacheEntry:
    """Eintrag im Asset-Cache."""
    info: AssetInfo
    timestamp: float

    def is_valid(self, ttl: float) -> bool:
        """Prüft, ob der Cache-Eintrag noch gültig ist."""
        return (time.time() - self.timestamp) < ttl


class AssetManager:
    """
    Verwaltet Assets mit Profil-basiertem Fallback.

    Der AssetManager sucht Assets zuerst im konfigurierten Profil-Verzeichnis
    und fällt auf das "default" Profil zurück, wenn das Asset dort nicht
    gefunden wird.

    Beispiel:
        manager = AssetManager("./assets", profile="my_profile")

        # Einfacher Zugriff
        path = manager.get("audio/success.wav")

        # Mit Info
        info = manager.get_info("audio/success.wav")
        print(f"Gefunden in: {info.profile}, Größe: {info.size}")

        # Typ-spezifisch
        audio_path = manager.get_audio("success.wav")
        music_path = manager.get_music("song.mp3")

        # Alle Assets eines Typs
        for info in manager.list_assets(AssetType.AUDIO):
            print(info.name)
    """

    DEFAULT_PROFILE = "default"

    def __init__(
        self,
        base_dir: str | Path = "./assets",
        profile: str = "default",
        cache_enabled: bool = True,
        cache_ttl: float = 60.0,
    ) -> None:
        """
        Initialisiert den AssetManager.

        Args:
            base_dir: Basis-Verzeichnis für Assets
            profile: Aktives Profil (Fallback: "default")
            cache_enabled: Cache für Asset-Lookups aktivieren
            cache_ttl: Cache Time-to-Live in Sekunden
        """
        self._base_dir = Path(base_dir).resolve()
        self._profile = profile
        self._cache_enabled = cache_enabled
        self._cache_ttl = cache_ttl
        self._cache: dict[str, CacheEntry] = {}
        self._lock = threading.RLock()

        # Verzeichnisse erstellen falls nicht vorhanden
        self._ensure_directories()

        pdebug(f"AssetManager initialisiert: base={self._base_dir}, profile={profile}")

    def _ensure_directories(self) -> None:
        """Stellt sicher, dass die Verzeichnisse existieren."""
        default_dir = self._base_dir / self.DEFAULT_PROFILE
        profile_dir = self._base_dir / self._profile

        default_dir.mkdir(parents=True, exist_ok=True)
        if self._profile != self.DEFAULT_PROFILE:
            profile_dir.mkdir(parents=True, exist_ok=True)

    @property
    def base_dir(self) -> Path:
        """Gibt das Basis-Verzeichnis zurück."""
        return self._base_dir

    @property
    def profile(self) -> str:
        """Gibt das aktive Profil zurück."""
        return self._profile

    @profile.setter
    def profile(self, value: str) -> None:
        """Setzt das aktive Profil und leert den Cache."""
        with self._lock:
            if value != self._profile:
                self._profile = value
                self._cache.clear()
                self._ensure_directories()
                pinfo(f"Profil gewechselt zu: {value}")

    @property
    def profile_dir(self) -> Path:
        """Gibt das Profil-Verzeichnis zurück."""
        return self._base_dir / self._profile

    @property
    def default_dir(self) -> Path:
        """Gibt das Default-Verzeichnis zurück."""
        return self._base_dir / self.DEFAULT_PROFILE

    def get(self, asset_path: str, raise_on_missing: bool = False) -> Path | None:
        """
        Gibt den Pfad zu einem Asset zurück.

        Sucht zuerst im aktiven Profil, dann im Default-Profil.

        Args:
            asset_path: Relativer Pfad zum Asset (z.B. "audio/success.wav")
            raise_on_missing: Bei True wird AssetNotFoundError geworfen

        Returns:
            Absoluter Pfad zum Asset oder None wenn nicht gefunden

        Raises:
            AssetNotFoundError: Wenn raise_on_missing=True und Asset nicht gefunden
        """
        info = self.get_info(asset_path)

        if info is not None:
            return info.path

        if raise_on_missing:
            raise AssetNotFoundError(asset_path, self._profile, self._base_dir)

        return None

    def get_info(self, asset_path: str) -> AssetInfo | None:
        """
        Gibt detaillierte Informationen über ein Asset zurück.

        Args:
            asset_path: Relativer Pfad zum Asset

        Returns:
            AssetInfo oder None wenn nicht gefunden
        """
        # Normalisiere Pfad
        asset_path = asset_path.replace("\\", "/").lstrip("/")

        # Cache prüfen
        with self._lock:
            if self._cache_enabled and asset_path in self._cache:
                entry = self._cache[asset_path]
                if entry.is_valid(self._cache_ttl):
                    return entry.info
                else:
                    del self._cache[asset_path]

        # Suche in Profil
        profile_path = self._base_dir / self._profile / asset_path
        if profile_path.exists() and profile_path.is_file():
            info = self._create_info(profile_path, asset_path, self._profile, False)
            self._cache_info(asset_path, info)
            return info

        # Fallback auf Default
        if self._profile != self.DEFAULT_PROFILE:
            default_path = self._base_dir / self.DEFAULT_PROFILE / asset_path
            if default_path.exists() and default_path.is_file():
                info = self._create_info(default_path, asset_path, self.DEFAULT_PROFILE, True)
                self._cache_info(asset_path, info)
                return info

        return None

    def _create_info(
        self,
        path: Path,
        relative_path: str,
        profile: str,
        is_default: bool
    ) -> AssetInfo:
        """Erstellt ein AssetInfo-Objekt."""
        stat = path.stat()
        asset_type = self._detect_type(path, relative_path)
        mime_type, _ = mimetypes.guess_type(str(path))

        return AssetInfo(
            path=path,
            relative_path=relative_path,
            profile=profile,
            is_default=is_default,
            asset_type=asset_type,
            size=stat.st_size,
            modified=stat.st_mtime,
            mime_type=mime_type,
        )

    def _detect_type(self, path: Path, relative_path: str) -> AssetType:
        """Erkennt den Asset-Typ basierend auf Pfad und Endung."""
        # Musik-Verzeichnis hat Vorrang
        if relative_path.startswith("music/"):
            return AssetType.MUSIC

        extension = path.suffix.lower()
        return _EXTENSION_MAP.get(extension, AssetType.UNKNOWN)

    def _cache_info(self, asset_path: str, info: AssetInfo) -> None:
        """Fügt Info zum Cache hinzu."""
        if self._cache_enabled:
            with self._lock:
                self._cache[asset_path] = CacheEntry(info, time.time())

    def exists(self, asset_path: str) -> bool:
        """
        Prüft, ob ein Asset existiert.

        Args:
            asset_path: Relativer Pfad zum Asset

        Returns:
            True wenn das Asset existiert
        """
        return self.get_info(asset_path) is not None

    def get_audio(self, filename: str) -> Path | None:
        """
        Gibt den Pfad zu einer Audio-Datei zurück.

        Args:
            filename: Dateiname (z.B. "success.wav")

        Returns:
            Pfad oder None
        """
        return self.get(f"audio/{filename}")

    def get_music(self, filename: str) -> Path | None:
        """
        Gibt den Pfad zu einer Musik-Datei zurück.

        Args:
            filename: Dateiname (z.B. "song.mp3")

        Returns:
            Pfad oder None
        """
        return self.get(f"music/{filename}")

    def get_image(self, filename: str) -> Path | None:
        """
        Gibt den Pfad zu einer Bild-Datei zurück.

        Args:
            filename: Dateiname (z.B. "logo.png")

        Returns:
            Pfad oder None
        """
        return self.get(f"images/{filename}")

    def list_assets(
        self,
        asset_type: AssetType | None = None,
        directory: str | None = None,
        include_default: bool = True,
    ) -> Iterator[AssetInfo]:
        """
        Listet alle Assets auf.

        Args:
            asset_type: Optional: Nur Assets dieses Typs
            directory: Optional: Nur in diesem Verzeichnis
            include_default: Default-Assets einschließen

        Yields:
            AssetInfo für jedes gefundene Asset
        """
        seen: set[str] = set()

        # Zuerst Profil scannen
        profile_dir = self.profile_dir
        if directory:
            profile_dir = profile_dir / directory

        if profile_dir.exists():
            for path in profile_dir.rglob("*"):
                if path.is_file():
                    relative = str(path.relative_to(self._base_dir / self._profile))
                    info = self._create_info(path, relative, self._profile, False)

                    if asset_type is None or info.asset_type == asset_type:
                        seen.add(relative)
                        yield info

        # Dann Default (wenn gewünscht)
        if include_default and self._profile != self.DEFAULT_PROFILE:
            default_dir = self.default_dir
            if directory:
                default_dir = default_dir / directory

            if default_dir.exists():
                for path in default_dir.rglob("*"):
                    if path.is_file():
                        relative = str(path.relative_to(self._base_dir / self.DEFAULT_PROFILE))

                        # Überspringe wenn bereits im Profil gefunden
                        if relative in seen:
                            continue

                        info = self._create_info(path, relative, self.DEFAULT_PROFILE, True)

                        if asset_type is None or info.asset_type == asset_type:
                            yield info

    def list_audio(self) -> Iterator[AssetInfo]:
        """Listet alle Audio-Assets."""
        return self.list_assets(AssetType.AUDIO, "audio")

    def list_music(self) -> Iterator[AssetInfo]:
        """Listet alle Musik-Assets."""
        return self.list_assets(AssetType.MUSIC, "music")

    def list_images(self) -> Iterator[AssetInfo]:
        """Listet alle Bild-Assets."""
        return self.list_assets(AssetType.IMAGE, "images")

    def list_profiles(self) -> list[str]:
        """
        Listet alle verfügbaren Profile auf.

        Returns:
            Liste der Profil-Namen
        """
        profiles = []

        if self._base_dir.exists():
            for path in self._base_dir.iterdir():
                if path.is_dir() and not path.name.startswith("."):
                    profiles.append(path.name)

        # "default" sollte immer zuerst sein
        if self.DEFAULT_PROFILE in profiles:
            profiles.remove(self.DEFAULT_PROFILE)
            profiles.insert(0, self.DEFAULT_PROFILE)

        return profiles

    def copy_to_profile(
        self,
        asset_path: str,
        target_profile: str | None = None
    ) -> Path | None:
        """
        Kopiert ein Default-Asset in ein Profil.

        Nützlich um ein Default-Asset anzupassen ohne das Original zu ändern.

        Args:
            asset_path: Relativer Pfad zum Asset
            target_profile: Ziel-Profil (Standard: aktuelles Profil)

        Returns:
            Pfad zur kopierten Datei oder None
        """
        import shutil

        target_profile = target_profile or self._profile

        # Hole Info (muss aus Default kommen)
        info = self.get_info(asset_path)
        if info is None:
            pwarn(f"Asset nicht gefunden: {asset_path}")
            return None

        if not info.is_default and info.profile == target_profile:
            pdebug(f"Asset bereits im Profil: {asset_path}")
            return info.path

        # Ziel-Pfad
        target_path = self._base_dir / target_profile / asset_path
        target_path.parent.mkdir(parents=True, exist_ok=True)

        try:
            shutil.copy2(info.path, target_path)
            pinfo(f"Asset kopiert: {asset_path} -> {target_profile}")

            # Cache invalidieren
            with self._lock:
                if asset_path in self._cache:
                    del self._cache[asset_path]

            return target_path

        except Exception as e:
            perror(f"Fehler beim Kopieren: {e}")
            return None

    def clear_cache(self) -> None:
        """Leert den Asset-Cache."""
        with self._lock:
            self._cache.clear()
            pdebug("Asset-Cache geleert")

    def get_stats(self) -> dict:
        """
        Gibt Statistiken über die Assets zurück.

        Returns:
            Dictionary mit Statistiken
        """
        stats = {
            "profile": self._profile,
            "base_dir": str(self._base_dir),
            "cache_enabled": self._cache_enabled,
            "cache_size": len(self._cache),
            "profiles": self.list_profiles(),
            "counts": {
                "audio": 0,
                "music": 0,
                "images": 0,
                "other": 0,
                "total": 0,
            },
            "sizes": {
                "profile": 0,
                "default": 0,
                "total": 0,
            }
        }

        for info in self.list_assets():
            stats["counts"]["total"] += 1
            stats["sizes"]["total"] += info.size

            if info.is_default:
                stats["sizes"]["default"] += info.size
            else:
                stats["sizes"]["profile"] += info.size

            if info.asset_type == AssetType.AUDIO:
                stats["counts"]["audio"] += 1
            elif info.asset_type == AssetType.MUSIC:
                stats["counts"]["music"] += 1
            elif info.asset_type == AssetType.IMAGE:
                stats["counts"]["images"] += 1
            else:
                stats["counts"]["other"] += 1

        return stats

    def __repr__(self) -> str:
        return f"AssetManager(base_dir='{self._base_dir}', profile='{self._profile}')"