ba2429612a
- Doppelte Arcane-Einträge behoben (rippy statt Rippy) - Import-Fixes: relative → absolute imports in main.py, cache/__init__.py, prescan/__init__.py, clients/__init__.py - Neue Dateien: cache.py, prescan.py, Settings.tsx - API-Port 8000 in docker-compose.yml gemappt - UI mit Sidebar, Dark Mode, Einstellungen-Tabs Fixes: #2978 (doppelte Einträge), #2888 (Import-Fehler)
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""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)
|