Trixy7/trixy_core/music/sources/local.py

# -*- coding: utf-8 -*-
"""
Lokale Dateien als Musikquelle.

Scannt lokale Verzeichnisse nach Musikdateien.
"""

import asyncio
import logging
import os
from pathlib import Path
from typing import AsyncIterator

from trixy_core.music.sources.base import MusicSource, SourceType, SourceState
from trixy_core.music.track import Track, TrackMetadata, TrackState


# Unterstützte Audio-Formate
SUPPORTED_EXTENSIONS = {
    ".mp3", ".wav", ".ogg", ".flac",
    ".m4a", ".aac", ".wma", ".opus",
    ".aiff", ".ape", ".mka",
}


class LocalFileSource(MusicSource):
    """
    Lokale Dateien als Musikquelle.

    Scannt ein oder mehrere Verzeichnisse nach Musikdateien und
    extrahiert Metadaten aus ID3-Tags.
    """

    def __init__(
        self,
        name: str = "Lokale Musik",
        paths: list[Path] | None = None,
        recursive: bool = True,
    ) -> None:
        """
        Initialisiert die lokale Musikquelle.

        Args:
            name: Name der Quelle
            paths: Zu scannende Verzeichnisse
            recursive: Rekursiv scannen
        """
        super().__init__(name=name, source_type=SourceType.LOCAL)
        self.paths = paths or []
        self.recursive = recursive
        self.logger = logging.getLogger(__name__)

        # Statistiken
        self._total_size = 0
        self._scan_errors: list[tuple[Path, str]] = []

    @property
    def total_size_bytes(self) -> int:
        """Gesamtgröße aller Dateien."""
        return self._total_size

    @property
    def scan_errors(self) -> list[tuple[Path, str]]:
        """Fehler beim letzten Scan."""
        return self._scan_errors.copy()

    # ==========================================================================
    # Pfad-Management
    # ==========================================================================

    def add_path(self, path: Path) -> bool:
        """
        Fügt ein Verzeichnis hinzu.

        Args:
            path: Verzeichnispfad

        Returns:
            True wenn hinzugefügt
        """
        if path.exists() and path.is_dir():
            if path not in self.paths:
                self.paths.append(path)
                return True
        return False

    def remove_path(self, path: Path) -> bool:
        """
        Entfernt ein Verzeichnis.

        Args:
            path: Verzeichnispfad

        Returns:
            True wenn entfernt
        """
        if path in self.paths:
            self.paths.remove(path)
            return True
        return False

    # ==========================================================================
    # MusicSource Implementation
    # ==========================================================================

    async def connect(self) -> bool:
        """Verbindet zur Quelle (prüft Verzeichnisse)."""
        available = False

        for path in self.paths:
            if path.exists() and path.is_dir():
                available = True
                break

        self._state = SourceState.AVAILABLE if available else SourceState.UNAVAILABLE
        return available

    async def disconnect(self) -> None:
        """Trennt die Verbindung (Cleanup)."""
        self._state = SourceState.UNAVAILABLE

    async def scan(self) -> int:
        """
        Scannt alle Verzeichnisse nach Musikdateien.

        Returns:
            Anzahl der gefundenen Tracks
        """
        async with self._scan_lock:
            if self._scanning:
                return self.track_count

            self._scanning = True
            self._state = SourceState.SCANNING
            self._scan_errors.clear()
            self._total_size = 0

            try:
                self._clear_tracks()

                for base_path in self.paths:
                    if not base_path.exists():
                        continue

                    await self._scan_directory(base_path)

                self._state = SourceState.AVAILABLE
                self.logger.info(
                    f"Scan abgeschlossen: {self.track_count} Tracks, "
                    f"{self._total_size / (1024*1024):.1f} MB"
                )

            except Exception as e:
                self.logger.error(f"Scan-Fehler: {e}")
                self._state = SourceState.ERROR

            finally:
                self._scanning = False

            return self.track_count

    async def _scan_directory(self, directory: Path) -> None:
        """
        Scannt ein einzelnes Verzeichnis.

        Args:
            directory: Zu scannendes Verzeichnis
        """
        try:
            # Asynchron über Verzeichnis iterieren
            entries = await asyncio.to_thread(list, directory.iterdir())

            for entry in entries:
                try:
                    if entry.is_file():
                        if entry.suffix.lower() in SUPPORTED_EXTENSIONS:
                            track = await self._create_track_from_file(entry)
                            if track:
                                self._add_track(track)
                                self._total_size += entry.stat().st_size

                    elif entry.is_dir() and self.recursive:
                        # Test-Verzeichnisse ueberspringen
                        dir_name = entry.name.lower()
                        if dir_name.startswith("test") or dir_name.endswith("_test") or dir_name.endswith("_tests"):
                            continue
                        await self._scan_directory(entry)

                except PermissionError:
                    self._scan_errors.append((entry, "Keine Berechtigung"))
                except Exception as e:
                    self._scan_errors.append((entry, str(e)))

        except PermissionError:
            self._scan_errors.append((directory, "Keine Berechtigung"))
        except Exception as e:
            self._scan_errors.append((directory, str(e)))

    async def _create_track_from_file(self, file_path: Path) -> Track | None:
        """
        Erstellt Track aus Datei.

        Args:
            file_path: Pfad zur Audiodatei

        Returns:
            Track oder None bei Fehler
        """
        try:
            # Basis-Metadaten aus Dateinamen
            metadata = TrackMetadata(
                title=file_path.stem,
                format=file_path.suffix.lstrip(".").upper(),
            )

            # ID3-Tags laden (wenn Mutagen verfügbar)
            metadata = await self._load_metadata(file_path, metadata)

            track = Track(
                uri=f"file://{file_path}",
                source_type="local",
                source_id=str(file_path),
                path=file_path,
                metadata=metadata,
                state=TrackState.AVAILABLE,
            )

            return track

        except Exception as e:
            self.logger.debug(f"Fehler bei {file_path}: {e}")
            return None

    async def _load_metadata(
        self,
        file_path: Path,
        default: TrackMetadata,
    ) -> TrackMetadata:
        """
        Lädt Metadaten aus ID3-Tags.

        Args:
            file_path: Pfad zur Datei
            default: Standard-Metadaten

        Returns:
            Geladene Metadaten
        """
        try:
            # Mutagen für ID3-Tags verwenden (falls installiert)
            import mutagen
            from mutagen.easyid3 import EasyID3
            from mutagen.mp3 import MP3
            from mutagen.flac import FLAC
            from mutagen.oggvorbis import OggVorbis

            audio = mutagen.File(file_path)

            if audio is None:
                return default

            # Dauer
            if hasattr(audio.info, "length"):
                default.duration_ms = int(audio.info.length * 1000)

            # Technische Daten
            if hasattr(audio.info, "sample_rate"):
                default.sample_rate = audio.info.sample_rate
            if hasattr(audio.info, "bitrate"):
                default.bit_rate = getattr(audio.info, "bitrate", 0)
            if hasattr(audio.info, "channels"):
                default.channels = audio.info.channels

            # Tags
            tags = {}
            if hasattr(audio, "tags") and audio.tags:
                # EasyID3-Style Tags
                if isinstance(audio.tags, dict):
                    tags = dict(audio.tags)
                else:
                    # Versuche als dict zu behandeln
                    for key in audio.tags.keys():
                        try:
                            tags[key] = audio.tags[key]
                        except Exception:
                            pass

            # Standard-Tags extrahieren
            default.title = self._get_tag(tags, ["title", "TIT2"]) or default.title
            default.artist = self._get_tag(tags, ["artist", "TPE1"]) or default.artist
            default.album = self._get_tag(tags, ["album", "TALB"]) or default.album
            default.album_artist = self._get_tag(tags, ["albumartist", "TPE2"]) or ""
            default.genre = self._get_tag(tags, ["genre", "TCON"]) or ""
            default.composer = self._get_tag(tags, ["composer", "TCOM"]) or ""

            # Track-Nummer
            track_num = self._get_tag(tags, ["tracknumber", "TRCK"])
            if track_num:
                if "/" in track_num:
                    num, total = track_num.split("/")
                    default.track_number = int(num)
                    default.track_total = int(total)
                else:
                    default.track_number = int(track_num)

            # Jahr
            year = self._get_tag(tags, ["date", "year", "TDRC", "TYER"])
            if year:
                try:
                    default.year = int(str(year)[:4])
                except ValueError:
                    pass

            # Artwork
            artwork = self._extract_artwork(audio)
            if artwork:
                default.artwork_data = artwork

        except ImportError:
            # Mutagen nicht installiert
            pass
        except Exception as e:
            self.logger.debug(f"Metadaten-Fehler bei {file_path}: {e}")

        return default

    def _get_tag(self, tags: dict, keys: list[str]) -> str | None:
        """Holt Tag-Wert für verschiedene Schlüssel."""
        for key in keys:
            if key in tags:
                value = tags[key]
                if isinstance(value, list):
                    return str(value[0]) if value else None
                return str(value)
        return None

    def _extract_artwork(self, audio) -> bytes | None:
        """Extrahiert Artwork aus Audio-Datei."""
        try:
            # ID3
            if hasattr(audio, "tags"):
                tags = audio.tags
                if tags:
                    # APIC Frame (MP3)
                    for key in tags.keys():
                        if key.startswith("APIC"):
                            return tags[key].data

                    # FLAC/OGG Pictures
                    if hasattr(audio, "pictures"):
                        if audio.pictures:
                            return audio.pictures[0].data
        except Exception:
            pass
        return None

    async def get_track(self, track_id: str) -> Track | None:
        """Holt Track nach ID."""
        return self._tracks.get(track_id)

    async def get_audio_stream(self, track: Track) -> AsyncIterator[bytes]:
        """
        Liefert Audio-Daten als Stream.

        Args:
            track: Track zum Streamen

        Yields:
            Audio-Daten in Chunks (64KB)
        """
        if not track.path or not track.path.exists():
            return

        chunk_size = 65536  # 64 KB

        def read_file():
            with open(track.path, "rb") as f:
                while True:
                    chunk = f.read(chunk_size)
                    if not chunk:
                        break
                    yield chunk

        for chunk in await asyncio.to_thread(lambda: list(read_file())):
            yield chunk

    async def get_metadata(self, track: Track) -> TrackMetadata | None:
        """Lädt Metadaten für einen Track."""
        if track.path:
            return await self._load_metadata(track.path, track.metadata)
        return track.metadata

    def get_info(self):
        """Liefert erweiterte Informationen."""
        info = super().get_info()
        info.path = self.paths[0] if self.paths else None
        info.total_size_bytes = self._total_size
        info.description = f"{len(self.paths)} Verzeichnisse"
        return info