From fc52d1f9d2063040c66aa0759269f6203b4a67a1 Mon Sep 17 00:00:00 2001 From: ChrisBeWithYou Date: Sat, 4 Jul 2026 05:22:24 -0500 Subject: [PATCH] fix(enrichment): rank the canonical studio take over live/comp versions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A flat MusicBrainz /recording text search ties every take of a song at the same score, so "AC/DC — Highway to Hell" returns a wall of live bootlegs and compilations with the 1979 studio version buried (or below the fetch limit). - build_recording_query: drop live-ONLY recordings (`-secondarytype:Live`). Compilations are deliberately kept — they REUSE the studio recording, so filtering them cuts the very recording we want (verified against MB). - _best_release / parse_recording_doc: pick the canonical studio album (primary Album, no Live/Compilation/Remix/... secondary type) for the displayed album/year, and expose a `studio` flag. - rank_candidates: since the combined score caps at 1.0 (perfect text match ties), break ties on the studio flag and — when the caller knows the audio length — on duration proximity, so the studio take wins over live/extended cuts. The studio distinction is intentionally NOT scored (a live take is still the right SONG), only re-ordered. - /api/enrichment/search: accept an optional `duration` param so a caller that has the audio but no library row (the editor's create modal) can pass the master-track length for the duration tiebreak. Verified end-to-end against live MusicBrainz: AC/DC "Highway to Hell" now returns the 1979 studio recording at #1 with the correct album + year. Co-Authored-By: Claude Opus 4.8 (1M context) --- lib/mb_match.py | 67 +++++++++++++++++++++++++++++++++++------- server.py | 20 +++++++++---- tests/test_mb_match.py | 4 ++- 3 files changed, 75 insertions(+), 16 deletions(-) diff --git a/lib/mb_match.py b/lib/mb_match.py index 2efb233..2be80e4 100644 --- a/lib/mb_match.py +++ b/lib/mb_match.py @@ -39,6 +39,14 @@ DURATION_BONUS_LOOSE = 0.025 # …within 15s _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,29 @@ 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")) + + 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, # canonical studio take + -_dur_diff(c), # closest to the audio length + c.get("mb_score") or 0), + reverse=True) return ranked @@ -209,7 +235,16 @@ 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). + if q: + q += " AND -secondarytype:Live" + return q def _artist_credit(doc: dict) -> tuple[str, str, str]: @@ -226,19 +261,29 @@ 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) + return (clean, status_ok, date) return sorted(releases, key=sort_key)[0] @@ -261,6 +306,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 +327,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 94f282f..762c903 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 b2912c9..9af0abb 100644 --- a/tests/test_mb_match.py +++ b/tests/test_mb_match.py @@ -149,7 +149,9 @@ def test_rank_candidates_orders_by_our_score(): 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_escapes_and_handles_missing_artist():