"""MusicBrainz API client for verifying songs exist.""" import httpx MB_API_URL = "https://musicbrainz.org/ws/2" HEADERS = { "User-Agent": "Vynl/1.0 (chris.ryan@deepcutsai.com)", "Accept": "application/json", } # MusicBrainz rate limit: 1 request per second import time _last_request_time = 0.0 def _rate_limit(): global _last_request_time now = time.time() elapsed = now - _last_request_time if elapsed < 1.1: time.sleep(1.1 - elapsed) _last_request_time = time.time() def verify_track(artist: str, title: str) -> dict | None: """Verify a track exists on MusicBrainz and return canonical data. Returns dict with: artist, title, album, mb_id or None if not found. """ _rate_limit() try: resp = httpx.get( f"{MB_API_URL}/recording", params={ "query": f'artist:"{artist}" recording:"{title}"', "fmt": "json", "limit": 5, }, headers=HEADERS, timeout=10, ) if resp.status_code != 200: return None data = resp.json() recordings = data.get("recordings", []) for rec in recordings: score = rec.get("score", 0) if score < 70: continue rec_title = rec.get("title", "") rec_artists = rec.get("artist-credit", []) rec_artist = rec_artists[0]["name"] if rec_artists else "" # Get album from first release album = None releases = rec.get("releases", []) if releases: album = releases[0].get("title") return { "artist": rec_artist, "title": rec_title, "album": album, "mb_id": rec.get("id"), "score": score, } return None except Exception: return None def search_artist(artist: str) -> dict | None: """Verify an artist exists on MusicBrainz.""" _rate_limit() try: resp = httpx.get( f"{MB_API_URL}/artist", params={ "query": f'artist:"{artist}"', "fmt": "json", "limit": 3, }, headers=HEADERS, timeout=10, ) if resp.status_code != 200: return None data = resp.json() artists = data.get("artists", []) for a in artists: if a.get("score", 0) >= 80: return { "name": a.get("name"), "mb_id": a.get("id"), "score": a.get("score"), } return None except Exception: return None