""" MusicBrainz (musicbrainz.org) + Cover Art Archive (coverartarchive.org) client for the Music group app. Called only by the node, never by a client (docs/musicbay.md §3): the node makes the one lookup per unique release, shared by every member, and self- paces against MusicBrainz's shared rate limit rather than letting several members' tile requests multiply it. Unlike tmdb.py, there is **no API key here** — MusicBrainz's read-only search/lookup endpoints and Cover Art Archive need no credential and no account, only: 1. a descriptive `User-Agent` (application name/version + a contact), which MusicBrainz's usage policy asks for — not a secret; 2. a self-imposed ~1 request/second pace, since that's a courtesy limit enforced by convention (and by MusicBrainz throttling abusive clients), not a token bucket handed out by the server. Contact resolution order (docs/musicbay.md §3.2), same shape as tmdb.py's token resolution: 1. an operator-supplied contact string (roster.py group_settings, group_id="") 2. the MESHBAY_MUSICBRAINZ_CONTACT_DEFAULT environment variable 3. none — MusicBrainz lookups are inert (callers get an empty result, never an exception). Deliberately **not** falling back to a generic User-Agent: sending an unidentified client to a service that polices its User-Agent policy risks the node's IP being blocked, which is a worse failure than "no music metadata yet". No literal contact value lives in this file, for the same reason tmdb.py carries no literal token — see docs/musicbay.md §3.2 on why a personal address must never land in source control. """ import asyncio import difflib import logging import os import re import time import httpx from meshbay_node.roster import Roster log = logging.getLogger(__name__) _BASE_URL = "https://musicbrainz.org/ws/2/" _COVER_ART_BASE = "https://coverartarchive.org/release/" _TIMEOUT = 10.0 _DEFAULT_CONTACT_ENV = "MESHBAY_MUSICBRAINZ_CONTACT_DEFAULT" _APP_NAME = "MeshBay-Node" # MusicBrainz's own stated courtesy limit for unauthenticated use. Enforced # here, not requested from the server — there is nothing to request. _MIN_INTERVAL_SECS = 1.0 def _normalize(s: str) -> str: s = s.lower() s = re.sub(r"[^a-z0-9àâäéèêëïîôöùûüçñ ]+", " ", s) return re.sub(r"\s+", " ", s).strip() def _best_match(query: str, results: list[dict], key: str) -> tuple[dict | None, float]: """ Same "trust the search's own ranking" shape as tmdb.py's `_best_match` (§3.3 of mediacenter.md found a locally-recomputed re-rank pick a coincidentally closer-looking wrong result once — no reason to expect MusicBrainz's own scored search to fare differently under the same treatment). MusicBrainz already returns results ordered by its own `score`; only the top one is considered, and the similarity ratio is a confidence signal for the caller's fallback decision, not a re-rank. """ if not results: return None, 0.0 top = results[0] val = top.get(key) ratio = (difflib.SequenceMatcher(None, _normalize(query), _normalize(str(val))).ratio() if val else 0.0) return top, ratio class MusicBrainzClient: """One instance per node, holding the resolved contact and an httpx client.""" def __init__(self, roster: Roster | None = None, transport: httpx.AsyncBaseTransport | None = None): self._roster = roster # `transport` is a test-only seam (httpx.MockTransport) — production # callers never pass it. self._client = httpx.AsyncClient(timeout=_TIMEOUT, transport=transport) self._rate_lock = asyncio.Lock() self._last_request_monotonic: float | None = None async def close(self) -> None: await self._client.aclose() async def _resolve_contact(self) -> str | None: if self._roster is not None: contact = await self._roster.musicbrainz_contact() else: contact = None return contact or os.environ.get(_DEFAULT_CONTACT_ENV) or None async def _pace(self) -> None: """Serializes every call through this client to >= _MIN_INTERVAL_SECS apart.""" async with self._rate_lock: now = time.monotonic() if self._last_request_monotonic is not None: wait = _MIN_INTERVAL_SECS - (now - self._last_request_monotonic) if wait > 0: await asyncio.sleep(wait) self._last_request_monotonic = time.monotonic() async def _get(self, url: str, params: dict) -> dict | None: contact = await self._resolve_contact() if not contact: return None await self._pace() try: resp = await self._client.get( url, params={**params, "fmt": "json"}, headers={"User-Agent": f"{_APP_NAME}/1.0 ( {contact} )"}, ) resp.raise_for_status() return resp.json() except httpx.HTTPError as e: log.warning("MusicBrainz request failed (%s): %s", url, e) return None async def search_release(self, artist: str, album: str) -> tuple[dict | None, float]: """Top MusicBrainz release match for an (artist, album) pair.""" query = f'artist:"{artist}" AND release:"{album}"' data = await self._get(_BASE_URL + "release", {"query": query}) results = (data or {}).get("releases", []) return _best_match(album, results, "title") async def release_details(self, mbid: str) -> dict | None: """Full release details, including recordings (tracklist).""" return await self._get(_BASE_URL + f"release/{mbid}", {"inc": "recordings+artist-credits"}) async def fetch_cover_art(self, mbid: str) -> bytes | None: """ The release's front cover, or None if Cover Art Archive has nothing for it (a real, common outcome — most releases have no scan). Paced the same as a metadata call: Cover Art Archive is served from the same courtesy-limit policy. """ contact = await self._resolve_contact() if not contact: return None await self._pace() try: resp = await self._client.get( f"{_COVER_ART_BASE}{mbid}/front-500", headers={"User-Agent": f"{_APP_NAME}/1.0 ( {contact} )"}, follow_redirects=True, ) if resp.status_code == 404: return None resp.raise_for_status() return resp.content except httpx.HTTPError as e: log.warning("Cover Art Archive fetch failed (%s): %s", mbid, e) return None