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

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,
     }
 
 

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)}
 
 

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"] = [