# -*- coding: utf-8 -*- """ Deezer-Quelle via Deezer Public API + yt-dlp. Die Deezer Public API ist kostenlos und braucht KEINEN API-Key fuer Suchoperationen: https://api.deezer.com/search?q=... Audio-Download: - 30-Sekunden-Preview direkt von Deezer (immer verfuegbar) - Voller Track via yt-dlp (sucht den Song auf YouTube) - Konfigurierbar: "preview" oder "full" (Default: "full") Vorteile: - Kein API-Key / Registrierung noetig - Hochwertige Metadaten (Artist, Album, Cover, Dauer, ISRC) - Gute Suchqualitaet (besser als reine YouTube-Suche) Abhaengigkeiten: - aiohttp: Fuer API-Anfragen (oft schon installiert) - yt-dlp: Fuer vollen Audio-Download (optional, Fallback auf Preview) """ from __future__ import annotations import asyncio import re 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 _AIOHTTP_AVAILABLE = False try: import aiohttp _AIOHTTP_AVAILABLE = True except ImportError: pass _YT_DLP_AVAILABLE = False try: import yt_dlp _YT_DLP_AVAILABLE = True except ImportError: pass DEEZER_API_BASE = "https://api.deezer.com" class DeezerSource(MusicSource): """Deezer-Quelle via Public API + yt-dlp.""" def __init__( self, name: str = "Deezer", cache_dir: str = "cache/deezer", max_cache_size_mb: int = 500, max_results: int = 5, download_mode: str = "full", ) -> None: """ Initialisiert die Deezer-Quelle. Args: name: Name der Quelle cache_dir: Cache-Verzeichnis max_cache_size_mb: Maximale Cache-Groesse max_results: Maximale Suchergebnisse download_mode: "full" (yt-dlp, voller Track) oder "preview" (30s Deezer-Preview) """ super().__init__(name=name, source_type=SourceType.URL) self._cache_dir = Path(cache_dir) self._max_cache_size_mb = max_cache_size_mb self._max_results = max_results self._download_mode = download_mode async def connect(self) -> bool: """Prueft aiohttp Verfuegbarkeit und API-Erreichbarkeit.""" if not _AIOHTTP_AVAILABLE: pwarn("DeezerSource: aiohttp nicht installiert") self._state = SourceState.UNAVAILABLE return False # API-Erreichbarkeit testen try: async with aiohttp.ClientSession() as session: async with session.get( f"{DEEZER_API_BASE}/search?q=test&limit=1", timeout=aiohttp.ClientTimeout(total=5), ) as resp: if resp.status != 200: perror(f"DeezerSource: API nicht erreichbar (HTTP {resp.status})") self._state = SourceState.ERROR return False except Exception as e: perror(f"DeezerSource: API-Test fehlgeschlagen: {e}") self._state = SourceState.ERROR return False try: self._cache_dir.mkdir(parents=True, exist_ok=True) except OSError as e: perror(f"DeezerSource: Cache-Verzeichnis Fehler: {e}") self._state = SourceState.AVAILABLE dl_mode = self._download_mode if dl_mode == "full" and not _YT_DLP_AVAILABLE: dl_mode = "preview (yt-dlp nicht installiert)" pinfo(f"DeezerSource: Verbunden (Download={dl_mode})") return True async def disconnect(self) -> None: self._state = SourceState.UNAVAILABLE async def scan(self) -> int: return 0 async def get_track(self, track_id: str) -> Track | None: return self._tracks.get(track_id) async def get_audio_stream(self, track: Track) -> AsyncIterator[bytes]: if not track.path or not track.path.exists(): return chunk_size = 65536 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: return track.metadata async def search(self, query: str, limit: int = 5) -> list[Track]: """ Sucht auf Deezer via Public API. Kein API-Key noetig — die Deezer Search API ist oeffentlich zugaenglich. """ if not _AIOHTTP_AVAILABLE or self._state != SourceState.AVAILABLE: return [] effective_limit = min(limit, self._max_results) try: return await self._search_async(query, effective_limit) except Exception as e: perror(f"DeezerSource: Suchfehler: {e}") return [] async def _search_async(self, query: str, limit: int) -> list[Track]: """Async Deezer-Suche.""" tracks: list[Track] = [] url = f"{DEEZER_API_BASE}/search" params = {"q": query, "limit": limit} async with aiohttp.ClientSession() as session: async with session.get( url, params=params, timeout=aiohttp.ClientTimeout(total=10), ) as resp: if resp.status != 200: return [] data = await resp.json() for item in data.get("data", []): if not item: continue track_id = str(item.get("id", "")) title = item.get("title", "Unbekannt") duration = item.get("duration", 0) preview_url = item.get("preview", "") # Artist artist_data = item.get("artist", {}) artist_name = artist_data.get("name", "Unbekannt") artist_picture = artist_data.get("picture_medium", "") # Album album_data = item.get("album", {}) album_name = album_data.get("title", "") album_cover = album_data.get("cover_big", album_data.get("cover_medium", "")) metadata = TrackMetadata( title=title, artist=artist_name, album=album_name, artwork_url=album_cover or artist_picture, duration_ms=duration * 1000, ) track = Track( uri=f"deezer:{track_id}", source_type="deezer", source_id=track_id, metadata=metadata, state=TrackState.AVAILABLE, extra={ "deezer_id": track_id, "preview_url": preview_url, "deezer_url": item.get("link", ""), "rank": item.get("rank", 0), "explicit": item.get("explicit_lyrics", False), }, ) tracks.append(track) return tracks async def prepare_track(self, track: Track) -> bool: """ Bereitet einen Track fuer die Wiedergabe vor. Strategie: 1. Cache pruefen 2. download_mode == "full" + yt-dlp: Suche auf YouTube, lade vollen Track 3. Fallback: 30-Sekunden-Preview von Deezer """ # Cache pruefen cached = self._get_cached_path(track.source_id) if cached: track.path = cached track.state = TrackState.AVAILABLE pdebug(f"DeezerSource: Cache-Treffer: {cached.name}") return True track.state = TrackState.LOADING # Voller Track via yt-dlp (YouTube-Suche nach Artist + Title) if self._download_mode == "full" and _YT_DLP_AVAILABLE: search_query = f"{track.metadata.artist} {track.metadata.title}" try: output_path = await asyncio.to_thread( self._download_full_sync, search_query, track.source_id ) if output_path and output_path.exists(): track.path = output_path track.state = TrackState.AVAILABLE pinfo(f"DeezerSource: Heruntergeladen: {output_path.name}") await self.cleanup_cache() return True except Exception as e: pdebug(f"DeezerSource: yt-dlp Download fehlgeschlagen: {e}") # Fallthrough zu Preview # Fallback: 30s Preview von Deezer preview_url = track.extra.get("preview_url", "") if preview_url and _AIOHTTP_AVAILABLE: try: output_path = await self._download_preview(preview_url, track.source_id) if output_path and output_path.exists(): track.path = output_path track.state = TrackState.AVAILABLE pdebug(f"DeezerSource: Preview heruntergeladen: {output_path.name}") return True except Exception as e: pdebug(f"DeezerSource: Preview-Download fehlgeschlagen: {e}") track.state = TrackState.ERROR track.error_message = "Download fehlgeschlagen" return False def _download_full_sync(self, search_query: str, track_id: str) -> Path | None: """Voller Track via yt-dlp YouTube-Suche.""" output_template = str(self._cache_dir / f"{track_id}.%(ext)s") ydl_opts = { "format": "bestaudio/best", "outtmpl": output_template, "quiet": True, "no_warnings": True, "default_search": "ytsearch1", "postprocessors": [{ "key": "FFmpegExtractAudio", "preferredcodec": "mp3", "preferredquality": "192", }], } with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.download([search_query]) expected = self._cache_dir / f"{track_id}.mp3" if expected.exists(): return expected for f in self._cache_dir.iterdir(): if f.stem == track_id and f.is_file(): return f return None async def _download_preview(self, url: str, track_id: str) -> Path | None: """Laedt 30-Sekunden-Preview von Deezer herunter.""" output_path = self._cache_dir / f"{track_id}_preview.mp3" async with aiohttp.ClientSession() as session: async with session.get( url, timeout=aiohttp.ClientTimeout(total=30) ) as resp: if resp.status != 200: return None data = await resp.read() output_path.write_bytes(data) return output_path def _get_cached_path(self, track_id: str) -> Path | None: if not self._cache_dir.exists(): return None # Voller Track bevorzugt full = self._cache_dir / f"{track_id}.mp3" if full.exists(): return full # Preview als Fallback preview = self._cache_dir / f"{track_id}_preview.mp3" if preview.exists(): return preview return None async def cleanup_cache(self) -> int: if not self._cache_dir.exists(): return 0 return await asyncio.to_thread(self._cleanup_cache_sync) def _cleanup_cache_sync(self) -> int: 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 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"DeezerSource: {removed} Cache-Dateien entfernt") return removed