# -*- coding: utf-8 -*- """ Music-Plugin fuer Trixy. Ermoeglicht Musikwiedergabe ueber YouTube und lokale Dateien. Registriert Intent-Handler fuer Sprachsteuerung. """ import random from pathlib import Path from typing import Any from trixy_core.plugins import TrixyPlugin from trixy_core.nlp.intent_registry import IntentRegistry, IntentDefinition, IntentSlot from trixy_core.nlp.decorators import intent, pattern, example from trixy_core.nlp.handler import IntentReceivedData, IntentResult from trixy_core.music.track import Track, TrackState from trixy_core.music.playlist import Playlist, ShuffleMode from trixy_core.music.sources.local import LocalFileSource from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn from plugins.music.sources.youtube import YouTubeSource class MusicPlugin(TrixyPlugin): """ Music-Plugin. Bietet: - YouTube-Suche und -Download (optional, via yt-dlp) - Lokale Musikdateien - Intent-Handler fuer Sprachsteuerung """ NAME = "music" VERSION = "1.0.0" DESCRIPTION = "Musikwiedergabe ueber YouTube und lokale Dateien" AUTHOR = "Trixy" def __init__(self, application, plugin_path, config=None): super().__init__(application, plugin_path, config) self._youtube: YouTubeSource | None = None self._local: LocalFileSource | None = None self._registry = IntentRegistry.get_instance() @property def _music_service(self): """Zugriff auf den MusicPlayerService.""" return self.application.services.get_service("music_player") async def on_load(self) -> None: """Initialisiert Quellen und registriert Intents.""" pinfo(f"MusicPlugin: Lade Plugin v{self.VERSION}") yt_available = False local_available = False # YouTube-Source initialisieren (optional) yt_config = self.get_config_value("youtube", {}) if yt_config.get("enabled", True): self._youtube = YouTubeSource( cache_dir=yt_config.get("cache_dir", "cache/youtube"), max_cache_size_mb=yt_config.get("max_cache_size_mb", 500), preferred_format=yt_config.get( "preferred_format", "bestaudio[ext=m4a]/bestaudio/best" ), max_results=yt_config.get("max_results", 5), search_mode=yt_config.get("search_mode", "music_fallback"), ) yt_available = await self._youtube.connect() if not yt_available: pwarn("MusicPlugin: YouTube nicht verfuegbar (yt-dlp nicht installiert?)") self._youtube = None # Lokale Source initialisieren local_config = self.get_config_value("local", {}) if local_config.get("enabled", True): music_dirs = self._get_music_dirs(local_config) if music_dirs: self._local = LocalFileSource( name="Lokale Musik", paths=music_dirs, recursive=True, ) local_available = await self._local.connect() if local_available and local_config.get("scan_on_load", True): count = await self._local.scan() pdebug(f"MusicPlugin: {count} lokale Tracks gefunden") # Intents werden automatisch vom PluginManager registriert # (via @intent/@pattern/@example Dekoratoren auf den Handler-Methoden) yt_status = "ja" if yt_available else "nein" local_status = "ja" if local_available else "nein" pinfo(f"MusicPlugin: Geladen (YouTube={yt_status}, Lokal={local_status})") async def on_unload(self) -> None: """Entlaedt das Plugin und deregistriert Intents.""" # Intents deregistrieren self._registry.unregister_plugin(self.NAME) # Quellen trennen if self._youtube: await self._youtube.disconnect() self._youtube = None if self._local: await self._local.disconnect() self._local = None pdebug("MusicPlugin: Entladen") async def on_enable(self) -> None: """Aktiviert das Plugin.""" pinfo("MusicPlugin: Aktiviert") async def on_disable(self) -> None: """Deaktiviert das Plugin.""" pinfo("MusicPlugin: Deaktiviert") # ========================================================================== # Intent-Handler # ========================================================================== @intent("play_music", description="Musik abspielen (mit optionalem Suchbegriff)") @pattern("(spiele|spiel|abspielen) {query}") @pattern("(spiele|spiel) [etwas] musik") @pattern("musik abspielen") @example("Spiele Bohemian Rhapsody", "Spiel Musik") @example("Spiele etwas von Mozart", "Musik abspielen") async def handle_play_music(self, data: IntentReceivedData, query: str = "", source: str = "auto") -> IntentResult: """Verarbeitet den play_music Intent.""" service = self._music_service if not service: return IntentResult.failure( "MusicPlayerService nicht verfuegbar", response_text="Der Musik-Dienst ist gerade nicht verfuegbar.", ) query = data.get_slot("query", "") source = data.get_slot("source", "auto") satellite_id = data.satellite_id # Falls kein query-Slot, aus artist/title/album/genre zusammenbauen if not query: parts = [] for slot_name in ("artist", "title", "song", "album", "genre"): val = data.get_slot(slot_name, "") if val: parts.append(val) if parts: query = " ".join(parts) # Kein Query + Musik pausiert → Resume if not query and service.is_paused: await self.application.events.emit("music_resumed", { "satellite_id": satellite_id, }) return IntentResult.success_with_response("Wird fortgesetzt.") # Kein Query → Shuffle lokale Bibliothek if not query: return await self._play_local_shuffle(satellite_id) # Query vorhanden → Suchen und abspielen return await self._search_and_play(query, source, satellite_id) @intent("next_track", description="Zum naechsten Track springen") @pattern("(naechster|naechstes) (song|lied|track)") @pattern("(skip|weiter)") @example("Naechster Song", "Skip", "Weiter", "Naechstes Lied") async def handle_next_track(self, data: IntentReceivedData) -> IntentResult: """Springt zum naechsten Track.""" service = self._music_service if not service or not service.player: return IntentResult.failure( "Kein aktiver Player", response_text="Es laeuft gerade keine Musik.", ) await self.application.events.emit("music_next", { "satellite_id": data.satellite_id, }) return IntentResult.success_with_response("Naechster Song.") @intent("previous_track", description="Zum vorherigen Track springen") @pattern("(vorheriger|vorheriges) (song|lied|track)") @pattern("zurueck") @example("Zurueck", "Vorheriger Song", "Vorheriges Lied") async def handle_previous_track(self, data: IntentReceivedData) -> IntentResult: """Springt zum vorherigen Track.""" service = self._music_service if not service or not service.player: return IntentResult.failure( "Kein aktiver Player", response_text="Es laeuft gerade keine Musik.", ) await self.application.events.emit("music_previous", { "satellite_id": data.satellite_id, }) return IntentResult.success_with_response("Vorheriger Song.") @intent("what_is_playing", description="Aktuellen Track abfragen") @pattern("was laeuft [gerade]") @pattern("wie heisst [der] (song|lied|track)") @pattern("welches lied ist [das]") @pattern("was hoere ich [gerade]") @example("Was laeuft?", "Wie heisst der Song?") @example("Welches Lied ist das?", "Was hoere ich gerade?") async def handle_what_is_playing(self, data: IntentReceivedData) -> IntentResult: """Gibt Informationen zum aktuellen Track zurueck.""" service = self._music_service if not service: return IntentResult.failure( "MusicPlayerService nicht verfuegbar", response_text="Der Musik-Dienst ist gerade nicht verfuegbar.", ) track = service.current_track if not track: return IntentResult.success_with_response( "Es laeuft gerade keine Musik." ) # Track-Info zusammenbauen title = track.metadata.display_title artist = track.metadata.display_artist duration = track.metadata.duration_formatted if artist and artist != "Unbekannter Kuenstler": response = f"Es laeuft '{title}' von {artist}." else: response = f"Es laeuft '{title}'." if duration and duration != "0:00": response += f" Dauer: {duration}." return IntentResult.success_with_response(response) # ========================================================================== # Hilfsmethoden # ========================================================================== async def _play_local_shuffle(self, satellite_id: str) -> IntentResult: """Spielt lokale Musik im Shuffle-Modus.""" if not self._local or not self._local.tracks: # Auch Assets-Musik pruefen service = self._music_service if service: success = await service._start_assets_playlist(satellite_id) if success: return IntentResult.success_with_response( "Musik wird abgespielt." ) return IntentResult.failure( "Keine lokale Musik gefunden", response_text="Ich habe keine Musik in der lokalen Bibliothek gefunden.", ) # Playlist aus lokalen Tracks erstellen tracks = list(self._local.tracks) random.shuffle(tracks) playlist = Playlist(name="Lokale Musik (Shuffle)") playlist.shuffle_mode = ShuffleMode.ON for track in tracks: playlist.add(track) service = self._music_service if not service: return IntentResult.failure( "MusicPlayerService nicht verfuegbar", response_text="Der Musik-Dienst ist gerade nicht verfuegbar.", ) satellite_ids = self._get_target_satellites(satellite_id) success = await service.play_playlist(playlist, satellite_ids) if success: return IntentResult.success_with_response("Musik wird abgespielt.") else: return IntentResult.failure( "Wiedergabe fehlgeschlagen", response_text="Die Wiedergabe konnte nicht gestartet werden.", ) async def _search_and_play( self, query: str, source: str, satellite_id: str ) -> IntentResult: """Sucht nach Musik und startet die Wiedergabe.""" tracks: list[Track] = [] # Quelle bestimmen search_youtube = source in ("auto", "youtube") and self._youtube search_local = source in ("auto", "lokal", "local") and self._local # Lokal suchen if search_local and self._local: local_results = await self._local.search(query, limit=5) tracks.extend(local_results) # YouTube suchen (wenn lokal nichts gefunden oder explizit YouTube) if search_youtube and self._youtube and (not tracks or source == "youtube"): yt_results = await self._youtube.search(query, limit=3) tracks.extend(yt_results) if not tracks: return IntentResult.failure( f"Keine Ergebnisse fuer: {query}", response_text=f"Ich konnte keine Musik fuer '{query}' finden.", ) # Ersten Track verwenden track = tracks[0] # YouTube-Tracks muessen vorbereitet werden (Download) if track.source_type == "youtube" and self._youtube: pinfo(f"MusicPlugin: Lade YouTube-Track: {track.title}") prepared = await self._youtube.prepare_track(track) if not prepared: return IntentResult.failure( "YouTube-Download fehlgeschlagen", response_text=f"Der Download von '{track.title}' ist fehlgeschlagen.", ) service = self._music_service if not service: return IntentResult.failure( "MusicPlayerService nicht verfuegbar", response_text="Der Musik-Dienst ist gerade nicht verfuegbar.", ) satellite_ids = self._get_target_satellites(satellite_id) success = await service.play_track(track, satellite_ids) if success: artist = track.metadata.display_artist title = track.metadata.display_title if artist and artist != "Unbekannter Kuenstler": return IntentResult.success_with_response( f"Spiele '{title}' von {artist}." ) return IntentResult.success_with_response(f"Spiele '{title}'.") else: return IntentResult.failure( "Wiedergabe fehlgeschlagen", response_text="Die Wiedergabe konnte nicht gestartet werden.", ) def _get_music_dirs(self, local_config: dict) -> list[Path]: """Ermittelt die Musik-Verzeichnisse.""" dirs: list[Path] = [] # Konfigurierte Verzeichnisse for dir_str in local_config.get("music_dirs", []): path = Path(dir_str) if path.exists() and path.is_dir(): dirs.append(path) # Assets-Musik als Fallback if hasattr(self.application, "server_config"): assets_dir = Path(self.application.server_config.assets_directory) profile = self.application.server_config.profile for search_dir in [ assets_dir / profile / "music", assets_dir / "default" / "music", ]: if search_dir.exists() and search_dir not in dirs: dirs.append(search_dir) return dirs def _get_target_satellites(self, satellite_id: str) -> list[str]: """Ermittelt Ziel-Satellites fuer die Wiedergabe.""" # Anfragenden Satellite bevorzugen if satellite_id: satellites = getattr(self.application, "satellites", None) if satellites: sat = satellites.get(satellite_id) if sat and sat.is_connected and sat.sockets.music_out: return [satellite_id] # Fallback: Alle verbundenen Satellites mit Musik-Socket satellites = getattr(self.application, "satellites", None) if satellites: return [ sat.id for sat in satellites.all() if sat.is_connected and sat.sockets.music_out ] return []