# -*- 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()