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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
|
"""
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: the node owner's hub account email, fetched once at
login via ``GET /v1/users/me`` and passed to this client at construction.
If the owner has no email on file, lookups are inert (callers get an
empty result, never an exception).
"""
import asyncio
import difflib
import logging
import re
import time
import httpx
log = logging.getLogger(__name__)
_BASE_URL = "https://musicbrainz.org/ws/2/"
_COVER_ART_BASE = "https://coverartarchive.org/release/"
_TIMEOUT = 10.0
_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()
# Lucene/Solr query-syntax characters (the parser MusicBrainz's `ws/2` search
# runs on). A tag or folder-derived artist/album string is untrusted free
# text as far as this parser is concerned — a stray "(", ":" or bare '"'
# either breaks the surrounding quoted phrase or gets read as field/grouping
# syntax rather than a literal character. Backslash-escaping each one keeps
# it literal without changing what the analyzer tokenizes.
_LUCENE_SPECIAL_RE = re.compile(r'([+\-&|!(){}\[\]^"~*?:\\/])')
def _escape_lucene(s: str) -> str:
return _LUCENE_SPECIAL_RE.sub(r"\\\1", s)
def _similarity(query: str, val: str | None) -> float:
return (difflib.SequenceMatcher(None, _normalize(query), _normalize(str(val))).ratio()
if val else 0.0)
def _best_match_release(artist: str, album: str, results: list[dict]) -> 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.
Confidence is the average of the album/title and artist similarity,
not the title alone: `search_release`'s loose fallback query has no
field scoping at all, so a title-only ratio would happily call a
same-titled album by an unrelated artist a good match. Averaging both
still lets a strong single-field match (e.g. the artist matches
exactly but the local album string carries an edition suffix) clear
the caller's threshold, while an unrelated same-name result does not.
"""
if not results:
return None, 0.0
top = results[0]
title_ratio = _similarity(album, top.get("title"))
credit = top.get("artist-credit") or []
artist_name = credit[0].get("name") if credit else None
artist_ratio = _similarity(artist, artist_name)
return top, (title_ratio + artist_ratio) / 2
class MusicBrainzClient:
"""One instance per node, holding the resolved contact and an httpx client."""
def __init__(self, owner_email: str = "",
transport: httpx.AsyncBaseTransport | None = None):
self._owner_email = owner_email
# `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:
return self._owner_email 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.
Tries a field-scoped exact-phrase query first — cheap, and the
common case since tag data usually already matches MusicBrainz's
own spelling. Verified live against the real service: real tag/
folder text routinely carries something an exact phrase does not
tolerate at all — a year suffix, an edition tag, a stray
punctuation mark, or an artist credited under an older/aliased
name on that specific release — and MusicBrainz's phrase parser
then returns **zero** hits, not a low-scored one, so the caller's
confidence check never gets a chance to run at all.
`artist:"Groundation" AND release:"Hebron Gate"` finds it, score
100; appending "(2003)" — exactly how many rip folders name a
release — drops it to zero results outright. A second, unscoped,
unquoted query lets MusicBrainz's own relevance ranking find the
same release regardless of the extra text — confirmed with the
same pair, buried among 6000+ candidates by the plain query, still
ranked first. Only tried when the strict query comes back empty,
to keep the common case at one request.
"""
strict_query = f'artist:"{_escape_lucene(artist)}" AND release:"{_escape_lucene(album)}"'
data = await self._get(_BASE_URL + "release", {"query": strict_query})
results = (data or {}).get("releases", [])
if results:
return _best_match_release(artist, album, results)
loose_query = f"{_escape_lucene(artist)} {_escape_lucene(album)}"
data = await self._get(_BASE_URL + "release", {"query": loose_query})
results = (data or {}).get("releases", [])
return _best_match_release(artist, album, results)
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
|