Trixy7/plugins/music/sources/spotify.py

# -*- coding: utf-8 -*-
"""
Spotify-Quelle via Spotify Web API + spotdl.

Sucht Tracks ueber die offizielle Spotify Web API (kostenlos,
Client Credentials Flow) und laedt Audio via spotdl herunter.

spotdl nutzt intern YouTube als Audio-Quelle — das ist der
sauberste Ansatz, da Spotify selbst kein Audio-Streaming
fuer Drittanbieter erlaubt (DRM).

Abhaengigkeiten (alle optional):
    - spotipy: Spotify Web API Client
    - spotdl: Audio-Download (nutzt YouTube + FFmpeg)
"""

from __future__ import annotations

import asyncio
import os
import re
import shutil
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
from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn

# Optionale Abhaengigkeiten
_SPOTIPY_AVAILABLE = False
try:
    import spotipy
    from spotipy.oauth2 import SpotifyClientCredentials
    _SPOTIPY_AVAILABLE = True
except ImportError:
    pass


class SpotifySource(MusicSource):
    """Spotify-Quelle via Spotify Web API + spotdl."""

    def __init__(
        self,
        name: str = "Spotify",
        client_id: str = "",
        client_secret: str = "",
        cache_dir: str = "cache/spotify",
        max_cache_size_mb: int = 500,
        max_results: int = 5,
        market: str = "DE",
    ) -> None:
        """
        Initialisiert die Spotify-Quelle.

        Args:
            name: Name der Quelle
            client_id: Spotify API Client ID
            client_secret: Spotify API Client Secret
            cache_dir: Cache-Verzeichnis fuer Downloads
            max_cache_size_mb: Maximale Cache-Groesse in MB
            max_results: Maximale Suchergebnisse
            market: Spotify-Markt fuer regionale Verfuegbarkeit
        """
        super().__init__(name=name, source_type=SourceType.SPOTIFY)
        self._client_id = client_id
        self._client_secret = client_secret
        self._cache_dir = Path(cache_dir)
        self._max_cache_size_mb = max_cache_size_mb
        self._max_results = max_results
        self._market = market
        self._sp: spotipy.Spotify | None = None
        self._spotdl_available = False

    @property
    def has_api(self) -> bool:
        """Prueft ob die Spotify API verfuegbar ist."""
        return self._sp is not None

    @property
    def has_download(self) -> bool:
        """Prueft ob spotdl fuer Downloads verfuegbar ist."""
        return self._spotdl_available

    async def connect(self) -> bool:
        """
        Verbindet sich mit der Spotify Web API und prueft spotdl.

        Benoetigt:
        - spotipy (Python-Paket)
        - Gueltige client_id + client_secret
        - spotdl (optional, fuer Audio-Download)
        """
        if not _SPOTIPY_AVAILABLE:
            pwarn("SpotifySource: spotipy nicht installiert")
            self._state = SourceState.UNAVAILABLE
            return False

        if not self._client_id or not self._client_secret:
            pwarn("SpotifySource: client_id oder client_secret fehlt")
            self._state = SourceState.UNAVAILABLE
            return False

        # Spotify API Client erstellen
        try:
            auth_manager = SpotifyClientCredentials(
                client_id=self._client_id,
                client_secret=self._client_secret,
            )
            self._sp = spotipy.Spotify(auth_manager=auth_manager)

            # Verbindung testen
            await asyncio.to_thread(self._sp.search, q="test", limit=1)
            pdebug("SpotifySource: API-Verbindung erfolgreich")
        except Exception as e:
            perror(f"SpotifySource: API-Verbindung fehlgeschlagen: {e}")
            self._sp = None
            self._state = SourceState.ERROR
            return False

        # spotdl pruefen (fuer Audio-Download)
        self._spotdl_available = shutil.which("spotdl") is not None
        if not self._spotdl_available:
            pwarn(
                "SpotifySource: spotdl nicht installiert — "
                "Suche funktioniert, Download nicht"
            )

        # Cache-Verzeichnis erstellen
        try:
            self._cache_dir.mkdir(parents=True, exist_ok=True)
        except OSError as e:
            perror(f"SpotifySource: Cache-Verzeichnis Fehler: {e}")

        self._state = SourceState.AVAILABLE
        dl_status = "ja" if self._spotdl_available else "nein"
        pinfo(f"SpotifySource: Verbunden (Download={dl_status})")
        return True

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

    async def scan(self) -> int:
        """Spotify unterstuetzt kein Scanning — gibt 0 zurueck."""
        return 0

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

    async def get_audio_stream(self, track: Track) -> AsyncIterator[bytes]:
        """Liefert Audio-Daten als Stream (aus heruntergeladener Datei)."""
        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:
        """Liefert Metadaten eines Tracks."""
        return track.metadata

    async def search(self, query: str, limit: int = 5) -> list[Track]:
        """
        Sucht auf Spotify nach Tracks.

        Nutzt die offizielle Spotify Web API — schnell, zuverlaessig
        und mit hochwertigen Metadaten (Artist, Album, Artwork, Dauer).

        Args:
            query: Suchstring
            limit: Maximale Ergebnisse

        Returns:
            Liste gefundener Tracks
        """
        if not self._sp or self._state != SourceState.AVAILABLE:
            return []

        effective_limit = min(limit, self._max_results)

        try:
            results = await asyncio.to_thread(
                self._search_sync, query, effective_limit
            )
            return results
        except Exception as e:
            perror(f"SpotifySource: Suchfehler: {e}")
            return []

    def _search_sync(self, query: str, limit: int) -> list[Track]:
        """Synchrone Spotify-Suche (wird in Thread ausgefuehrt)."""
        tracks: list[Track] = []

        result = self._sp.search(q=query, limit=limit, type="track", market=self._market)
        if not result or "tracks" not in result:
            return tracks

        for item in result["tracks"].get("items", []):
            if not item:
                continue

            track_id = item.get("id", "")
            track_name = item.get("name", "Unbekannt")
            duration_ms = item.get("duration_ms", 0)

            # Artists
            artists = item.get("artists", [])
            artist_name = ", ".join(a.get("name", "") for a in artists) if artists else "Unbekannt"

            # Album
            album = item.get("album", {})
            album_name = album.get("name", "")
            album_images = album.get("images", [])
            artwork_url = album_images[0]["url"] if album_images else ""

            # Release-Jahr
            release_date = album.get("release_date", "")
            year = 0
            if release_date:
                try:
                    year = int(release_date[:4])
                except (ValueError, IndexError):
                    pass

            # Track-Nummer
            track_number = item.get("track_number", 0)
            disc_number = item.get("disc_number", 1)

            # Spotify URI fuer spotdl
            spotify_uri = item.get("uri", f"spotify:track:{track_id}")
            external_url = item.get("external_urls", {}).get("spotify", "")

            metadata = TrackMetadata(
                title=track_name,
                artist=artist_name,
                album=album_name,
                artwork_url=artwork_url,
                duration_ms=duration_ms,
                year=year,
                track_number=track_number,
                disc_number=disc_number,
            )

            track = Track(
                uri=f"spotify:{track_id}",
                source_type="spotify",
                source_id=track_id,
                metadata=metadata,
                state=TrackState.AVAILABLE,
                extra={
                    "spotify_uri": spotify_uri,
                    "spotify_url": external_url,
                    "preview_url": item.get("preview_url", ""),
                    "popularity": item.get("popularity", 0),
                    "explicit": item.get("explicit", False),
                },
            )
            tracks.append(track)

        return tracks

    async def prepare_track(self, track: Track) -> bool:
        """
        Laedt Audio herunter via spotdl und setzt track.path.

        spotdl sucht den Track auf YouTube und laedt die Audio-Version
        herunter. Das ist der empfohlene Weg — legal, zuverlaessig,
        gute Audioqualitaet.

        Args:
            track: Track zum Vorbereiten

        Returns:
            True wenn erfolgreich
        """
        if not self._spotdl_available:
            perror("SpotifySource: spotdl nicht installiert — Download nicht moeglich")
            return False

        spotify_url = track.extra.get("spotify_url", "")
        if not spotify_url:
            spotify_uri = track.extra.get("spotify_uri", "")
            if spotify_uri:
                # spotify:track:ABC123 → https://open.spotify.com/track/ABC123
                parts = spotify_uri.split(":")
                if len(parts) == 3:
                    spotify_url = f"https://open.spotify.com/track/{parts[2]}"

        if not spotify_url:
            perror("SpotifySource: Keine Spotify-URL fuer Download")
            return False

        # Cache pruefen
        cached_path = self._get_cached_path(track.source_id)
        if cached_path:
            track.path = cached_path
            track.state = TrackState.AVAILABLE
            pdebug(f"SpotifySource: Cache-Treffer: {cached_path.name}")
            return True

        # Download via spotdl
        track.state = TrackState.LOADING
        try:
            output_path = await asyncio.to_thread(
                self._download_sync, spotify_url, track.source_id
            )
            if output_path and output_path.exists():
                track.path = output_path
                track.state = TrackState.AVAILABLE
                pinfo(f"SpotifySource: Heruntergeladen: {output_path.name}")
                await self.cleanup_cache()
                return True
            else:
                track.state = TrackState.ERROR
                track.error_message = "Download fehlgeschlagen"
                return False

        except Exception as e:
            perror(f"SpotifySource: Download-Fehler: {e}")
            track.state = TrackState.ERROR
            track.error_message = str(e)
            return False

    def _download_sync(self, spotify_url: str, track_id: str) -> Path | None:
        """Synchroner Download via spotdl (wird in Thread ausgefuehrt)."""
        import subprocess

        output_path = self._cache_dir / f"{track_id}.mp3"

        # spotdl: Download mit Ausgabe-Format und Verzeichnis
        cmd = [
            "spotdl", "download", spotify_url,
            "--output", str(output_path),
            "--format", "mp3",
            "--bitrate", "192k",
        ]

        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                timeout=120,
                cwd=str(self._cache_dir),
            )

            if result.returncode != 0:
                stderr = result.stderr.decode(errors="replace")
                pdebug(f"SpotifySource: spotdl Fehler: {stderr[:200]}")

            # spotdl benennt die Datei manchmal nach Artist - Title
            if output_path.exists():
                return output_path

            # Fallback: Suche nach neuen Dateien im Cache-Verzeichnis
            for f in sorted(
                self._cache_dir.glob("*.mp3"),
                key=lambda p: p.stat().st_mtime,
                reverse=True,
            ):
                # Neueste Datei die nicht schon als Cache bekannt ist
                target = self._cache_dir / f"{track_id}.mp3"
                if f != target:
                    f.rename(target)
                    return target
                return f

        except subprocess.TimeoutExpired:
            perror("SpotifySource: spotdl Timeout (120s)")
        except FileNotFoundError:
            perror("SpotifySource: spotdl nicht gefunden")
        except Exception as e:
            perror(f"SpotifySource: Download-Fehler: {e}")

        return None

    def _get_cached_path(self, track_id: str) -> Path | None:
        """Prueft ob eine Datei im Cache liegt."""
        if not self._cache_dir.exists():
            return None
        cached = self._cache_dir / f"{track_id}.mp3"
        return cached if cached.exists() else None

    async def cleanup_cache(self) -> int:
        """
        Entfernt aelteste Dateien wenn Cache zu gross.

        Returns:
            Anzahl entfernter Dateien
        """
        if not self._cache_dir.exists():
            return 0
        return await asyncio.to_thread(self._cleanup_cache_sync)

    def _cleanup_cache_sync(self) -> int:
        """Synchrone Cache-Bereinigung (LRU)."""
        files: list[tuple[Path, float, int]] = []
        total_size = 0

        for f in self._cache_dir.iterdir():
            if f.is_file():
                stat = f.stat()
                files.append((f, stat.st_mtime, stat.st_size))
                total_size += stat.st_size

        max_size = self._max_cache_size_mb * 1024 * 1024
        if total_size <= max_size:
            return 0

        # Aelteste zuerst entfernen
        files.sort(key=lambda x: x[1])

        removed = 0
        for file_path, _, size in files:
            if total_size <= max_size:
                break
            try:
                file_path.unlink()
                total_size -= size
                removed += 1
            except OSError:
                pass

        if removed > 0:
            pdebug(f"SpotifySource: {removed} Cache-Dateien entfernt")
        return removed