Trixy7/trixy_core/music/sources/usb.py

# -*- coding: utf-8 -*-
"""
USB-Speicher als Musikquelle.

Erkennt und scannt USB-Speichermedien nach Musikdateien.
"""

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

from trixy_core.music.sources.base import MusicSource, SourceType, SourceState
from trixy_core.music.sources.local import LocalFileSource, SUPPORTED_EXTENSIONS
from trixy_core.music.track import Track, TrackMetadata


class USBSource(MusicSource):
    """
    USB-Speicher als Musikquelle.

    Erkennt automatisch eingesteckte USB-Speichermedien und scannt
    diese nach Musikdateien. Basiert intern auf LocalFileSource.
    """

    def __init__(
        self,
        name: str = "USB-Speicher",
        mount_base: Path | None = None,
        auto_scan: bool = True,
    ) -> None:
        """
        Initialisiert die USB-Quelle.

        Args:
            name: Name der Quelle
            mount_base: Basis-Pfad für Mount-Points
            auto_scan: Automatisch scannen bei Erkennung
        """
        super().__init__(name=name, source_type=SourceType.USB)
        self.auto_scan = auto_scan
        self.logger = logging.getLogger(__name__)

        # Mount-Pfad je nach Betriebssystem
        if mount_base:
            self.mount_base = mount_base
        else:
            system = platform.system()
            if system == "Linux":
                # Standard Linux mount points
                if Path("/media").exists():
                    self.mount_base = Path("/media")
                elif Path("/mnt").exists():
                    self.mount_base = Path("/mnt")
                else:
                    self.mount_base = Path("/run/media")
            elif system == "Darwin":  # macOS
                self.mount_base = Path("/Volumes")
            else:  # Windows
                self.mount_base = Path("D:/")

        # Interne LocalFileSource für Scanning
        self._local_source = LocalFileSource(
            name=f"{name} (Scanner)",
            recursive=True,
        )

        # Erkannte USB-Geräte
        self._devices: dict[str, Path] = {}

    @property
    def devices(self) -> dict[str, Path]:
        """Erkannte USB-Geräte (Name → Pfad)."""
        return self._devices.copy()

    @property
    def device_count(self) -> int:
        """Anzahl erkannter Geräte."""
        return len(self._devices)

    # ==========================================================================
    # USB-Erkennung
    # ==========================================================================

    async def detect_devices(self) -> list[Path]:
        """
        Erkennt angeschlossene USB-Speichermedien.

        Returns:
            Liste der gefundenen Mount-Punkte
        """
        devices = []

        try:
            if not self.mount_base.exists():
                return devices

            # Verzeichnisse im Mount-Pfad durchsuchen
            for entry in await asyncio.to_thread(list, self.mount_base.iterdir()):
                if entry.is_dir() and self._is_usb_device(entry):
                    devices.append(entry)

            # Spezielle Linux-Behandlung: /media/$USER/
            if platform.system() == "Linux":
                user = os.environ.get("USER", "")
                if user:
                    user_media = self.mount_base / user
                    if user_media.exists():
                        for entry in await asyncio.to_thread(list, user_media.iterdir()):
                            if entry.is_dir():
                                devices.append(entry)

        except PermissionError:
            self.logger.warning(f"Keine Berechtigung für {self.mount_base}")
        except Exception as e:
            self.logger.error(f"USB-Erkennung fehlgeschlagen: {e}")

        # Geräte-Dict aktualisieren
        self._devices.clear()
        for device in devices:
            self._devices[device.name] = device

        self.logger.info(f"{len(devices)} USB-Geräte erkannt")
        return devices

    def _is_usb_device(self, path: Path) -> bool:
        """
        Prüft ob Pfad ein USB-Gerät ist.

        Einfache Heuristik: Ist ein Verzeichnis und nicht leer.

        Args:
            path: Zu prüfender Pfad

        Returns:
            True wenn wahrscheinlich USB-Gerät
        """
        try:
            # Ist gemountet und hat Inhalt
            if path.is_mount() or (path.is_dir() and any(path.iterdir())):
                # Ausschlüsse
                excluded = {"System", "Recovery", "Boot", "EFI"}
                if path.name not in excluded:
                    return True
        except (PermissionError, OSError):
            pass
        return False

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

    async def connect(self) -> bool:
        """Verbindet zur Quelle (sucht USB-Geräte)."""
        devices = await self.detect_devices()

        if devices:
            self._state = SourceState.AVAILABLE
            return True

        self._state = SourceState.UNAVAILABLE
        return False

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

    async def scan(self) -> int:
        """
        Scannt alle USB-Geräte 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

            try:
                # Zuerst Geräte erkennen
                devices = await self.detect_devices()

                if not devices:
                    self._state = SourceState.UNAVAILABLE
                    return 0

                # Alle Geräte zur LocalFileSource hinzufügen
                self._local_source.paths.clear()
                for device_path in devices:
                    self._local_source.add_path(device_path)

                # Scannen
                await self._local_source.scan()

                # Tracks übernehmen und Source-Type anpassen
                self._clear_tracks()
                for track in self._local_source.tracks:
                    track.source_type = "usb"
                    self._add_track(track)

                self._state = SourceState.AVAILABLE
                self.logger.info(f"USB-Scan: {self.track_count} Tracks")

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

            finally:
                self._scanning = False

            return self.track_count

    async def scan_device(self, device_name: str) -> int:
        """
        Scannt ein einzelnes USB-Gerät.

        Args:
            device_name: Name des Geräts

        Returns:
            Anzahl der gefundenen Tracks
        """
        if device_name not in self._devices:
            return 0

        device_path = self._devices[device_name]

        # Temporäre LocalFileSource für einzelnes Gerät
        scanner = LocalFileSource(
            name=f"USB: {device_name}",
            paths=[device_path],
            recursive=True,
        )

        await scanner.scan()

        # Tracks übernehmen
        count = 0
        for track in scanner.tracks:
            track.source_type = "usb"
            if track.id not in self._tracks:
                self._add_track(track)
                count += 1

        return count

    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."""
        async for chunk in self._local_source.get_audio_stream(track):
            yield chunk

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

    # ==========================================================================
    # Hilfsmethoden
    # ==========================================================================

    async def eject(self, device_name: str) -> bool:
        """
        Wirft USB-Gerät aus (nur Linux/macOS).

        Args:
            device_name: Name des Geräts

        Returns:
            True wenn erfolgreich
        """
        if device_name not in self._devices:
            return False

        device_path = self._devices[device_name]
        system = platform.system()

        try:
            if system == "Linux":
                # udisksctl unmount
                process = await asyncio.create_subprocess_exec(
                    "udisksctl", "unmount", "-b", str(device_path),
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
                await process.wait()

            elif system == "Darwin":
                # diskutil eject
                process = await asyncio.create_subprocess_exec(
                    "diskutil", "eject", str(device_path),
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
                await process.wait()

            else:
                return False

            # Aus Liste entfernen
            del self._devices[device_name]

            # Tracks dieses Geräts entfernen
            to_remove = [
                track_id for track_id, track in self._tracks.items()
                if track.path and str(device_path) in str(track.path)
            ]
            for track_id in to_remove:
                self._remove_track(track_id)

            self.logger.info(f"USB-Gerät ausgeworfen: {device_name}")
            return True

        except Exception as e:
            self.logger.error(f"Auswerfen fehlgeschlagen: {e}")
            return False

    def get_info(self):
        """Liefert erweiterte Informationen."""
        info = super().get_info()
        info.path = self.mount_base
        info.description = f"{self.device_count} Geräte erkannt"
        info.metadata["devices"] = list(self._devices.keys())
        return info