Hitonabi
0ea5f10b31
Ampel / ampel (push) Failing after 12m31s
- Metadaten-Preview WIEDERHERGESTELLT (Stub ueberschattete echte prescan-Implementierung), tote Altmodule geloescht - Celery update_state statt Phantom-Task/erfundener API - abcde-Kommando korrigiert (CD-Ripping war nie funktionsfaehig) - JWT: fester Schluessel Pflicht, echtes Logout, Cleanup nur Abgelaufene - main.py: crashende Endpoints (Path/secrets/api_keys), year-Bug, Admin-Login aus .env - Ruff gruen (29 Funde), Tests: auth/cache_keys/ripping_helpers, Placebo-test_health raus - SAVEPOINT: offene MakeMKV-Entscheidung SICHTBAR gemacht (Regel B)
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""MusicBrainz API Client mit Caching."""
|
|
|
|
import requests
|
|
from typing import Dict, List, Optional
|
|
|
|
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)
|