diff --git a/lib/mb_match.py b/lib/mb_match.py index 2efb2339..92309615 100644 --- a/lib/mb_match.py +++ b/lib/mb_match.py @@ -39,6 +39,14 @@ _DURATION_TIGHT = 5 _DURATION_LOOSE = 15 +# Release-group secondary types that mark a NON-canonical release (a live album, +# a greatest-hits comp, a remix/DJ set, …). Used both to pick the canonical +# studio album for display and to reward studio recordings in ranking. +_SECONDARY_SKIP = { + "live", "compilation", "remix", "dj-mix", "mixtape/street", + "demo", "interview", "audiobook", "spokenword", +} + # ── Denoise ─────────────────────────────────────────────────────────────────── # A parenthetical/bracketed group is dropped when it contains any of these # noise terms as a whole word (chart-variant markers, tuning/pitch notes, @@ -154,6 +162,10 @@ def score_candidate(song: dict, cand: dict) -> float: score += DURATION_BONUS elif diff <= _DURATION_LOOSE: score += DURATION_BONUS_LOOSE + # NB: the studio-vs-live distinction is deliberately NOT scored here — a live + # take is still the RIGHT SONG (same title/artist), so it must not change the + # auto/review confidence. Canonical-version preference lives in the RANK sort + # (rank_candidates) instead, where it only reorders same-song candidates. return min(score, 1.0) @@ -179,15 +191,34 @@ def classify(song: dict, cand: dict, score: float, auto_min: float | None = None def rank_candidates(song: dict, candidates: list[dict]) -> list[dict]: - """Score every candidate against the song and return them sorted by our - score (MusicBrainz's own search score is only a tiebreak). Each returned - dict is a copy carrying `score` (rounded — it's displayed and stored).""" + """Score every candidate against the song and return them sorted best-first. + The combined `score` caps at 1.0, so a perfect-text-match query (every "AC/DC + Highway to Hell" recording) ties at the top — there the studio flag and, when + the caller knows the audio length, the duration match break the tie so the + canonical studio take wins over live/promo/extended cuts. Each returned dict + is a copy carrying `score` (rounded — it's displayed and stored).""" + sd = _duration_int(song.get("duration")) + # For a chart that IS a live take (build_recording_query keeps live + # recordings for these) the studio take is the WRONG recording, so drop the + # studio tiebreak — duration proximity + text/mb score then pick the right + # live version instead of auto-matching the studio one. + prefer_studio = not _LIVE_GROUP_RE.search(str(song.get("title") or "")) + + def _dur_diff(c): + cd = _duration_int(c.get("duration")) + return abs(sd - cd) if (sd and cd) else 10 ** 6 + ranked = [] for cand in candidates or []: c = dict(cand) c["score"] = round(score_candidate(song, cand), 4) ranked.append(c) - ranked.sort(key=lambda c: (c["score"], c.get("mb_score") or 0), reverse=True) + ranked.sort( + key=lambda c: (c["score"], + (1 if c.get("studio") else 0) if prefer_studio else 0, + -_dur_diff(c), # closest to the audio length + c.get("mb_score") or 0), + reverse=True) return ranked @@ -198,6 +229,11 @@ def _lucene_escape_phrase(s: str) -> str: return s.replace("\\", "\\\\").replace('"', '\\"') +# A parenthetical/bracketed "(Live …)" marker — the live signal denoise() strips +# from the title. Mirrors _NOISE_GROUP_RE but for the `live` term only. +_LIVE_GROUP_RE = re.compile(r"[(\[][^)\]]*\blive\b[^)\]]*[)\]]", re.IGNORECASE) + + def build_recording_query(artist, title) -> str: """Lucene query for /ws/2/recording. Built from the DENOISED fields — the noise we strip (author credits, "(Live)", "(v2)") would otherwise @@ -209,7 +245,22 @@ def build_recording_query(artist, title) -> str: parts.append('recording:"%s"' % _lucene_escape_phrase(t)) if a: parts.append('artist:"%s"' % _lucene_escape_phrase(a)) - return " AND ".join(parts) + q = " AND ".join(parts) + # Drop live-ONLY recordings (bootlegs, live albums) — the canonical studio + # take is never tagged Live, and this is the single biggest source of junk in + # a flat recording search. Compilations are deliberately NOT excluded: they + # REUSE the studio recording, so filtering them would drop the very recording + # we want (verified against MusicBrainz — `-secondarytype:Compilation` cut the + # AC/DC studio "Highway to Hell" recording entirely). + # + # EXCEPT when the source chart is itself a live take: denoise() strips the + # "(Live at …)" qualifier from the query, so filtering Live would leave the + # genuinely-live chart with NO correct recording. Only a parenthetical marker + # counts — a bare title word ("Live and Let Die") is a real word, not a live + # tag — mirroring what denoise removes. + if q and not _LIVE_GROUP_RE.search(str(title or "")): + q += " AND -secondarytype:Live" + return q def _artist_credit(doc: dict) -> tuple[str, str, str]: @@ -226,19 +277,33 @@ def _artist_credit(doc: dict) -> tuple[str, str, str]: return name, str(artist.get("id", "") or ""), str(artist.get("sort-name", "") or "") +def _is_clean_studio_album(rg: dict) -> bool: + """A release-group that is a primary-type Album with NO non-canonical + secondary type (Live / Compilation / Remix / …) — i.e. a studio album.""" + if str(rg.get("primary-type", "")).lower() != "album": + return False + secs = {str(s).lower() for s in (rg.get("secondary-types") or [])} + return not (secs & _SECONDARY_SKIP) + + def _best_release(doc: dict) -> dict: - """Pick the release used for canon album/year: prefer Official status and - an Album release-group, then the earliest date. Returns {} if none.""" + """Pick the release used for canon album/year: prefer an OFFICIAL studio + Album (primary Album with no Live/Compilation/… secondary type), then the + earliest date. Falls back to any release when none is clean. {} if none.""" releases = [r for r in (doc.get("releases") or []) if isinstance(r, dict)] if not releases: return {} def sort_key(r): - status_ok = 0 if str(r.get("status", "")).lower() == "official" else 1 rg = r.get("release-group") or {} - album_ok = 0 if str(rg.get("primary-type", "")).lower() == "album" else 1 + clean = 0 if _is_clean_studio_album(rg) else 1 + status_ok = 0 if str(r.get("status", "")).lower() == "official" else 1 date = str(r.get("date", "") or "9999") - return (status_ok, album_ok, date) + # Official FIRST, then prefer a clean studio album: this still surfaces + # the studio album over an (official) live/comp album for the display + # album/year, but never lets an UNofficial bootleg album outrank an + # official single/EP/comp — which `(clean, status_ok, …)` would. + return (status_ok, clean, date) return sorted(releases, key=sort_key)[0] @@ -261,6 +326,7 @@ def parse_recording_doc(doc: dict) -> dict | None: return None artist_name, artist_id, artist_sort = _artist_credit(doc) release = _best_release(doc) + studio = _is_clean_studio_album(release.get("release-group") or {}) length = doc.get("length") try: duration = int(round(float(length) / 1000.0)) if length else None @@ -281,6 +347,7 @@ def parse_recording_doc(doc: dict) -> dict | None: "isrc": isrcs[0] if isrcs else "", "genres": _genres(doc), "mb_score": int(doc.get("score") or 0), + "studio": studio, } diff --git a/server.py b/server.py index 94f282f4..762c903d 100644 --- a/server.py +++ b/server.py @@ -6235,8 +6235,11 @@ def _mb_http_get(path: str, params: dict) -> dict | None: raise EnrichTransportError("bad JSON from musicbrainz") from e -def _mb_search_recordings(artist, title, limit: int = 8) -> list[dict]: - """Text search (tier 2–4): denoised Lucene query over /recording.""" +def _mb_search_recordings(artist, title, limit: int = 12) -> list[dict]: + """Text search (tier 2–4): denoised Lucene query over /recording. The query + now drops live-only recordings and our ranker rewards the studio take, so a + slightly larger default result set gives the re-ranker room to surface the + canonical version (one request per song regardless of limit).""" query = mb_match.build_recording_query(artist, title) if not query: return [] @@ -7413,13 +7416,16 @@ def api_enrichment_pick(filename: str, data: dict = Body(...)): @app.get("/api/enrichment/search") def api_enrichment_search(artist: str = "", title: str = "", limit: int = 8, - filename: str = ""): + filename: str = "", duration: float = 0.0): """Manual-search proxy to MusicBrainz (throttled + identified like the background matcher — a user typing in the drawer must not sidestep the rate limit). `filename` optionally scores results against that song's stored identity (year/duration corroboration) instead of just the typed - text. Sync route on purpose: FastAPI runs it in the threadpool, so the - throttle's sleep never blocks the event loop.""" + text. `duration` (seconds) lets a caller that HAS the audio but no library + row — e.g. the editor's create modal, which holds the master track — pass + its length so the studio take ranks above live/extended cuts. Sync route on + purpose: FastAPI runs it in the threadpool, so the throttle's sleep never + blocks the event loop.""" if not (artist.strip() or title.strip()): raise HTTPException(status_code=400, detail="artist or title required") limit = max(1, min(int(limit), 25)) @@ -7433,6 +7439,10 @@ def api_enrichment_search(artist: str = "", title: str = "", limit: int = 8, ref = meta_db.enrichment_song_row(filename) if ref is None: ref = {"artist": artist, "title": title} + # A caller-supplied duration corroborates the take even without a library row. + if duration and duration > 0 and not ref.get("duration"): + ref = dict(ref) + ref["duration"] = duration return {"candidates": mb_match.rank_candidates(ref, cands)} diff --git a/tests/test_mb_match.py b/tests/test_mb_match.py index b2912c9f..7f0da984 100644 --- a/tests/test_mb_match.py +++ b/tests/test_mb_match.py @@ -145,11 +145,40 @@ def test_rank_candidates_orders_by_our_score(): assert all("score" in c for c in ranked) +def test_rank_candidates_studio_preference_is_dropped_for_live_charts(): + """Tied-score candidates: a studio chart prefers the studio take, but a + LIVE chart must NOT be forced to the studio recording.""" + studio = {"recording_id": "studio", "artist": "AC/DC", "title": "Highway to Hell", + "studio": True, "mb_score": 90} + live = {"recording_id": "live", "artist": "AC/DC", "title": "Highway to Hell", + "studio": False, "mb_score": 95} + # Studio chart -> studio take wins the tie (studio flag), despite lower mb_score. + studio_song = {"artist": "AC/DC", "title": "Highway to Hell"} + assert m.rank_candidates(studio_song, [live, studio])[0]["recording_id"] == "studio" + # Live chart -> studio preference dropped, so the higher-mb_score live take wins. + live_song = {"artist": "AC/DC", "title": "Highway to Hell (Live at Donington)"} + assert m.rank_candidates(live_song, [studio, live])[0]["recording_id"] == "live" + + # ── query building ──────────────────────────────────────────────────────────── def test_build_recording_query_denoises_and_quotes(): q = m.build_recording_query("ACDC", 'Thunderstruck (v2)') - assert q == 'recording:"thunderstruck" AND artist:"acdc"' + # Live-only recordings are excluded — the studio take is never tagged Live, + # and it's the biggest source of junk in a flat recording search. + assert q == 'recording:"thunderstruck" AND artist:"acdc" AND -secondarytype:Live' + + +def test_build_recording_query_keeps_live_for_live_charts(): + """A chart that IS a live take must NOT get the live filter, or its only + correct recording is excluded. A bare title word ("Live and Let Die") is a + real word, not a marker, so it still filters.""" + live = m.build_recording_query("AC/DC", "Highway to Hell (Live at Donington)") + assert "-secondarytype:Live" not in live + assert 'recording:"highway to hell"' in live + # A real word "live" in the title is not a live marker → still filtered. + bare = m.build_recording_query("Wings", "Live and Let Die") + assert "-secondarytype:Live" in bare def test_build_recording_query_escapes_and_handles_missing_artist(): @@ -200,6 +229,29 @@ def test_parse_recording_doc_normalizes(): assert c["mb_score"] == 98 +def test_best_release_prefers_official_single_over_unofficial_album(): + """An OFFICIAL single/EP must outrank an UNofficial bootleg album for the + canonical album/year: official comes before the studio-album preference, so + a single-only song is never seeded from a bootleg. (`(clean, status_ok, …)` + would wrongly pick the bootleg.)""" + doc = { + "id": "rec-x", "title": "One-Off", "score": 90, + "artist-credit": [ + {"name": "A", "joinphrase": "", + "artist": {"id": "a", "name": "A", "sort-name": "A"}}], + "releases": [ + {"id": "rel-boot", "title": "Boot LP", "status": "Bootleg", + "date": "1990-01-01", "release-group": {"primary-type": "Album"}}, + {"id": "rel-single", "title": "The Single", "status": "Official", + "date": "1988-01-01", "release-group": {"primary-type": "Single"}}, + ], + } + c = m.parse_recording_doc(doc) + assert c["release_id"] == "rel-single" + assert c["album"] == "The Single" + assert c["studio"] is False # a Single isn't a clean studio ALBUM + + def test_parse_recording_doc_joined_artist_credit(): doc = dict(MB_DOC) doc["artist-credit"] = [