fix(enrichment): rank the canonical studio take over live/comp versions

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) <noreply@anthropic.com>

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 every candidate against the song and return them sorted best-first.
-    score (MusicBrainz's own search score is only a tiebreak). Each returned
+    The combined `score` caps at 1.0, so a perfect-text-match query (every "AC/DC
-    dict is a copy carrying `score` (rounded — it's displayed and stored)."""
+    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
+    """Pick the release used for canon album/year: prefer an OFFICIAL studio
-    an Album release-group, then the earliest date. Returns {} if none."""
+    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,
     }
 
 

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]:
+def _mb_search_recordings(artist, title, limit: int = 12) -> list[dict]:
-    """Text search (tier 2–4): denoised Lucene query over /recording."""
+    """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
+    text. `duration` (seconds) lets a caller that HAS the audio but no library
-    throttle's sleep never blocks the event loop."""
+    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)}
 
 

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():