"""MusicBrainz API Client mit Caching.""" import requests from typing import Dict, List, Optional from config import settings from cache import get, set MUSICBRAINZ_BASE_URL = "https://musicbrainz.org/ws/2" class MusicBrainzClient: def __init__(self): self.session = requests.Session() self.session.headers.update({ "User-Agent": "Rippy/1.0 (https://rippy.local)", "Content-Type": "application/json" }) def _request(self, endpoint: str, params: Dict = None) -> Optional[Dict]: """Mache API-Request mit Caching.""" cache_key = f"musicbrainz:{endpoint}:{params or {}}" cached = get(cache_key) if cached: return cached try: response = self.session.get( f"{MUSICBRAINZ_BASE_URL}/{endpoint}", params=params, timeout=10 ) response.raise_for_status() # XML-Response parsen if endpoint.endswith(".json"): data = response.json() else: # Fallback: XML als Text data = {"raw": response.text} set(cache_key, data) return data except Exception as e: print(f"MusicBrainz API Error: {e}") return None def search_artist(self, name: str) -> List[Dict]: """Suche Künstler.""" params = {"query": f"artist:{name}", "fmt": "json", "limit": 10} result = self._request("artist", params) if result and "artists" in result: return result["artists"] return [] def search_release(self, title: str, artist: str = None) -> List[Dict]: """Suche Releases.""" query = f"release:{title}" if artist: query += f" AND artist:{artist}" params = {"query": query, "fmt": "json", "limit": 10} result = self._request("release", params) if result and "releases" in result: return result["releases"] return [] def get_release_info(self, release_id: str) -> Optional[Dict]: """Hole Release-Details.""" params = {"fmt": "json", "inc": "artists+media+recordings"} return self._request(f"release/{release_id}", params) def get_artist_info(self, artist_id: str) -> Optional[Dict]: """Hole Künstler-Details.""" params = {"fmt": "json", "inc": "aliases+ratings"} return self._request(f"artist/{artist_id}", params)