1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
"""
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
|