Trixy7/plugins/spotify_example/main.py

# -*- coding: utf-8 -*-
"""
Spotify-Plugin (Beispiel).

Demonstriert wie ein Plugin neue Musikquellen über das Extension-System
hinzufügen kann. Dies ist ein Beispiel - für echte Spotify-Integration
wäre die Spotipy-Bibliothek und OAuth2-Authentifizierung nötig.
"""

import asyncio
import logging
from pathlib import Path
from typing import Any, AsyncIterator

from trixy_core.plugins import (
    TrixyPlugin,
    MusicSourceExtension,
    get_global_registry,
    MUSIC_SOURCE_POINT,
)
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

class SpotifySource(MusicSource):
    """
    Spotify als Musikquelle.

    Hinweis: Dies ist eine Beispiel-Implementierung.
    Für echte Nutzung wäre spotipy und OAuth2 nötig.
    """

    def __init__(
        self,
        client_id: str = "",
        client_secret: str = "",
        redirect_uri: str = "http://localhost:8888/callback",
    ) -> None:
        super().__init__(name="Spotify", source_type=SourceType.SPOTIFY)
        self.client_id = client_id
        self.client_secret = client_secret
        self.redirect_uri = redirect_uri
        self.logger = logging.getLogger(__name__)

        self._authenticated = False
        self._access_token = ""

    async def connect(self) -> bool:
        """
        Verbindet zu Spotify (OAuth2).

        In einer echten Implementierung würde hier OAuth2 stattfinden.
        """
        if not self.client_id or not self.client_secret:
            pwarn("Spotify: Keine Credentials konfiguriert")
            self._state = SourceState.UNAVAILABLE
            return False

        # Hier würde OAuth2-Flow stattfinden
        pinfo("Spotify: Verbindung wird hergestellt...")

        # Simuliert erfolgreiche Verbindung
        self._authenticated = True
        self._state = SourceState.AVAILABLE

        pinfo("Spotify: Verbunden (Beispiel-Modus)")
        return True

    async def disconnect(self) -> None:
        """Trennt die Verbindung zu Spotify."""
        self._authenticated = False
        self._access_token = ""
        self._state = SourceState.UNAVAILABLE
        pinfo("Spotify: Verbindung getrennt")

    async def scan(self) -> int:
        """
        Scannt ist bei Spotify nicht sinnvoll.
        Stattdessen wird bei Bedarf gesucht.
        """
        return 0

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

        In echter Implementierung: Spotify Web API aufrufen.
        """
        if not self._authenticated:
            return []

        self.logger.debug(f"Spotify-Suche: {query}")

        # Beispiel-Ergebnisse (würde von API kommen)
        tracks = []

        # Simulierte Ergebnisse für Demo
        for i in range(min(5, limit)):
            track = Track(
                id=f"spotify_track_{i}",
                uri=f"spotify:track:example{i}",
                source_type="spotify",
                source_id=f"example{i}",
                metadata=TrackMetadata(
                    title=f"Beispiel-Track {i + 1} für '{query}'",
                    artist="Demo Artist",
                    album="Demo Album",
                    duration_ms=180000 + i * 30000,
                ),
                state=TrackState.AVAILABLE,
            )
            tracks.append(track)

        return tracks

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

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

        Hinweis: Spotify erlaubt kein direktes Streaming über die API.
        Optionen:
        - Spotify Connect verwenden
        - Librespot für lokales Streaming
        """
        pwarn(
            "Spotify: Direktes Streaming nicht verfügbar. "
            "Verwende Spotify Connect oder Librespot."
        )
        # Leerer Generator - keine Audio-Daten
        return
        yield  # macht dies zu einem Generator

    async def get_metadata(self, track: Track) -> TrackMetadata | None:
        """Lädt Metadaten von Spotify."""
        # Würde API aufrufen
        return track.metadata


class SpotifySourceExtension(MusicSourceExtension):
    """
    Extension die SpotifySource bereitstellt.
    """

    def __init__(self, plugin_name: str) -> None:
        super().__init__(
            extension_id="spotify_source",
            name="Spotify",
            plugin_name=plugin_name,
            source_type="spotify",
            description="Spotify-Integration für Musiksuche und Wiedergabe",
        )
        self._info.priority = 100  # Hohe Priorität

    async def create_instance(self, config: dict[str, Any]) -> SpotifySource:
        """Erstellt SpotifySource mit Konfiguration."""
        return SpotifySource(
            client_id=config.get("client_id", ""),
            client_secret=config.get("client_secret", ""),
            redirect_uri=config.get("redirect_uri", "http://localhost:8888/callback"),
        )

    def get_default_config(self) -> dict[str, Any]:
        """Standard-Konfiguration."""
        return {
            "client_id": "",
            "client_secret": "",
            "redirect_uri": "http://localhost:8888/callback",
        }

    def get_config_schema(self) -> dict[str, Any]:
        """JSON-Schema für Konfiguration."""
        return {
            "type": "object",
            "properties": {
                "client_id": {
                    "type": "string",
                    "description": "Spotify Client ID",
                },
                "client_secret": {
                    "type": "string",
                    "description": "Spotify Client Secret",
                },
                "redirect_uri": {
                    "type": "string",
                    "description": "OAuth Redirect URI",
                },
            },
            "required": ["client_id", "client_secret"],
        }


class SpotifyPlugin(TrixyPlugin):
    """
    Spotify-Plugin.

    Registriert SpotifySource als Extension am music.source Extension Point.
    """

    NAME = "spotify"
    VERSION = "1.0.0"
    DESCRIPTION = "Spotify-Integration für Trixy"
    AUTHOR = "Trixy Team"

    def __init__(self, application, plugin_path: Path, config: dict | None = None) -> None:
        super().__init__(application, plugin_path, config)
        self._extension: SpotifySourceExtension | None = None

    async def on_load(self) -> None:
        """Plugin wird geladen - Extension registrieren."""
        pinfo("Spotify-Plugin: Lade...")

        # Extension erstellen
        self._extension = SpotifySourceExtension(self.NAME)

        # Am Extension Point registrieren
        registry = get_global_registry()
        point = registry.get_point(MUSIC_SOURCE_POINT)

        if point is not None:
            # Konfiguration aus Plugin-Config
            config = {
                "client_id": self.get_config_value("spotify.client_id", ""),
                "client_secret": self.get_config_value("spotify.client_secret", ""),
                "redirect_uri": self.get_config_value(
                    "spotify.redirect_uri",
                    "http://localhost:8888/callback"
                ),
            }

            success = await point.register(self._extension, config)
            if success:
                pinfo("Spotify-Plugin: Extension registriert")
            else:
                perror("Spotify-Plugin: Extension-Registrierung fehlgeschlagen")
        else:
            pwarn("Spotify-Plugin: music.source Extension Point nicht gefunden")

    async def on_unload(self) -> None:
        """Plugin wird entladen - Extension entfernen."""
        pinfo("Spotify-Plugin: Entlade...")

        if self._extension is not None:
            registry = get_global_registry()
            point = registry.get_point(MUSIC_SOURCE_POINT)

            if point is not None:
                await point.unregister(self._extension.id)

            self._extension = None

    async def on_enable(self) -> None:
        """Plugin wird aktiviert."""
        pinfo("Spotify-Plugin: Aktiviert")

        # Spotify-Source verbinden
        if self._extension and self._extension.instance:
            await self._extension.instance.connect()

    async def on_disable(self) -> None:
        """Plugin wird deaktiviert."""
        pinfo("Spotify-Plugin: Deaktiviert")

        # Spotify-Source trennen
        if self._extension and self._extension.instance:
            await self._extension.instance.disconnect()

    async def on_config_change(self, old_config: dict) -> None:
        """Konfiguration hat sich geändert."""
        pinfo("Spotify-Plugin: Konfiguration aktualisiert")

        # Bei Credential-Änderung neu verbinden
        if self._extension and self._extension.instance:
            new_client_id = self.get_config_value("spotify.client_id", "")
            old_client_id = old_config.get("spotify", {}).get("client_id", "")

            if new_client_id != old_client_id:
                pinfo("Spotify: Credentials geändert, verbinde neu...")
                await self._extension.instance.disconnect()
                self._extension.instance.client_id = new_client_id
                self._extension.instance.client_secret = self.get_config_value(
                    "spotify.client_secret", ""
                )
                await self._extension.instance.connect()