#903: persistent recording-MBID cache + export orchestrator

Phase 3. Additive backbone for the export job:
- mb_recording_cache table (IF NOT EXISTS) + core/exports/recording_mbid_cache.py: persistent
  (artist,title)->recording_mbid cache, mirrors album_mbid_cache (lazy DB, error-degrades to
  miss). The MusicBrainz tail is ~1 req/s, so a resolved MBID is remembered once and reused
  across every export/playlist.
- core/exports/playlist_export.py: resolve_playlist_tracks(tracks, resolve_fn) — walks tracks,
  dedups repeated songs within a run (resolve once), builds the ordered pseudo-playlist, tallies
  live stats (resolved/unmatched/deduped/by_source). Pure (I/O injected via resolve_fn + progress
  callback), so dedup + accounting are unit-tested with no DB/network. 5 tests.

No wiring into runtime yet; nothing existing touched except the additive table.

core/exports/playlist_export.py

@@ -0,0 +1,92 @@
+"""Orchestrate resolving a playlist's tracks to recording MBIDs for export (#903).
+
+This is the testable heart of the export job: walk the playlist's tracks, resolve each to a
+MusicBrainz recording MBID via an injected ``resolve_fn`` (which the job wires to the
+cache -> DB -> file -> MusicBrainz waterfall), dedup repeated songs within the run so they
+only cost one resolution, build the ordered "pseudo-playlist" of resolved tracks, and tally
+live stats (resolved / unmatched / per-source / deduped) for the on-card status display.
+
+Pure: all I/O (DB, file reads, MusicBrainz, cache) is behind ``resolve_fn`` and the optional
+``on_progress`` callback, so the dedup + accounting logic is unit-testable without any
+network or database. The returned ``resolved`` list feeds straight into ``jspf_export``.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+from core.exports.mbid_resolver import normalize_key
+
+# resolve_fn(artist, title) -> (recording_mbid|None, source_label|None)
+ResolveFn = Callable[[str, str], Tuple[Optional[str], Optional[str]]]
+ProgressFn = Callable[[int, int, Dict[str, Any]], None]
+
+
+def _field(track: Dict[str, Any], *names: str) -> str:
+    """First non-empty value among ``names`` (handles both playlist + LB-cache shapes)."""
+    for n in names:
+        v = track.get(n)
+        if v:
+            return str(v)
+    return ""
+
+
+def resolve_playlist_tracks(
+    tracks: List[Dict[str, Any]],
+    resolve_fn: ResolveFn,
+    *,
+    on_progress: Optional[ProgressFn] = None,
+) -> Dict[str, Any]:
+    """Resolve every track to a recording MBID and build the export pseudo-playlist.
+
+    ``tracks`` items may use ``artist``/``artist_name`` and ``title``/``track_name`` and
+    ``album``/``album_name`` (both the mirrored-playlist and LB-cache shapes are accepted).
+
+    Returns ``{"resolved": [...], "stats": {...}}`` where each resolved entry is
+    ``{artist, title, album, recording_mbid}`` (recording_mbid is None when unmatched),
+    in original order, and stats carries ``total, resolved, unmatched, deduped,
+    by_source`` for the live display.
+    """
+    total = len(tracks or [])
+    memo: Dict[str, Tuple[Optional[str], Optional[str]]] = {}
+    resolved: List[Dict[str, Any]] = []
+    stats: Dict[str, Any] = {
+        "total": total, "resolved": 0, "unmatched": 0, "deduped": 0, "by_source": {},
+    }
+
+    for i, t in enumerate(tracks or []):
+        if not isinstance(t, dict):
+            t = {}
+        artist = _field(t, "artist", "artist_name", "creator")
+        title = _field(t, "title", "track_name", "name")
+        album = _field(t, "album", "album_name", "release_name")
+        key = normalize_key(artist, title)
+
+        if key in memo:
+            mbid, source = memo[key]
+            stats["deduped"] += 1
+            fresh = False
+        else:
+            mbid, source = resolve_fn(artist, title)
+            memo[key] = (mbid, source)
+            fresh = True
+
+        resolved.append({"artist": artist, "title": title, "album": album, "recording_mbid": mbid})
+
+        if mbid:
+            stats["resolved"] += 1
+            if fresh and source:
+                stats["by_source"][source] = stats["by_source"].get(source, 0) + 1
+        else:
+            stats["unmatched"] += 1
+
+        if on_progress is not None:
+            try:
+                on_progress(i + 1, total, stats)
+            except Exception:  # noqa: S110 — a progress-display error must never fail the export
+                pass
+
+    return {"resolved": resolved, "stats": stats}
+
+
+__all__ = ["resolve_playlist_tracks"]

core/exports/recording_mbid_cache.py

@@ -0,0 +1,126 @@
+"""Persistent (artist,title) -> MusicBrainz recording-MBID cache for playlist export.
+
+The export waterfall (``core.exports.mbid_resolver``) ends in a live MusicBrainz lookup
+that's rate-limited to ~1 req/s — the slow tail of exporting a big playlist. Remembering a
+resolved recording MBID ONCE means the same song never costs a second lookup, across every
+future export and every playlist it appears in.
+
+Mirrors ``core.metadata.album_mbid_cache`` exactly: a tiny SQLite table, lazy DB accessor,
+every function wrapped so any DB error degrades to a cache miss / no-op. If this module
+breaks, exports still work — they just re-resolve via the live waterfall like a cold cache.
+Key is the normalized ``track_key`` from ``mbid_resolver.normalize_key(artist, title)``.
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Optional
+
+from utils.logging_config import get_logger
+
+logger = get_logger("exports.recording_mbid_cache")
+
+_db_factory_lock = threading.Lock()
+_db_factory = None
+
+
+def _get_database():
+    """Resolve the MusicDatabase singleton lazily; None on any failure (treated as miss)."""
+    global _db_factory
+    with _db_factory_lock:
+        if _db_factory is None:
+            try:
+                from database.music_database import get_database
+                _db_factory = get_database
+            except Exception as exc:
+                logger.warning(f"Recording-MBID cache: could not load database module: {exc}")
+                return None
+    try:
+        return _db_factory()
+    except Exception as exc:
+        logger.warning(f"Recording-MBID cache: database accessor failed: {exc}")
+        return None
+
+
+def lookup(track_key: str) -> Optional[str]:
+    """Read a cached recording MBID for ``track_key``; None on miss or any DB error."""
+    if not track_key:
+        return None
+    db = _get_database()
+    if db is None:
+        return None
+    conn = None
+    try:
+        conn = db._get_connection()
+        cursor = conn.cursor()
+        cursor.execute(
+            "SELECT recording_mbid FROM mb_recording_cache WHERE track_key = ? LIMIT 1",
+            (track_key,),
+        )
+        row = cursor.fetchone()
+        if row:
+            return (row[0] if not hasattr(row, "keys") else row["recording_mbid"]) or None
+    except Exception as exc:
+        logger.debug(f"Recording-MBID cache lookup failed: {exc}")
+    finally:
+        if conn is not None:
+            try:
+                conn.close()
+            except Exception:  # noqa: S110 — finally cleanup
+                pass
+    return None
+
+
+def record(track_key: str, recording_mbid: str) -> bool:
+    """Persist ``track_key`` -> ``recording_mbid`` (idempotent). False on any failure."""
+    if not track_key or not recording_mbid:
+        return False
+    db = _get_database()
+    if db is None:
+        return False
+    conn = None
+    try:
+        conn = db._get_connection()
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT OR REPLACE INTO mb_recording_cache "
+            "(track_key, recording_mbid, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)",
+            (track_key, recording_mbid),
+        )
+        conn.commit()
+        return True
+    except Exception as exc:
+        logger.debug(f"Recording-MBID cache record failed: {exc}")
+        return False
+    finally:
+        if conn is not None:
+            try:
+                conn.close()
+            except Exception:  # noqa: S110 — finally cleanup
+                pass
+
+
+def clear_all() -> bool:
+    """Wipe the cache (tests / forced re-resolve)."""
+    db = _get_database()
+    if db is None:
+        return False
+    conn = None
+    try:
+        conn = db._get_connection()
+        cursor = conn.cursor()
+        cursor.execute("DELETE FROM mb_recording_cache")
+        conn.commit()
+        return True
+    except Exception as exc:
+        logger.warning(f"Recording-MBID cache clear failed: {exc}")
+        return False
+    finally:
+        if conn is not None:
+            try:
+                conn.close()
+            except Exception:  # noqa: S110 — finally cleanup
+                pass
+
+
+__all__ = ["lookup", "record", "clear_all"]

database/music_database.py

@@ -2085,6 +2085,18 @@ class MusicDatabase:
                 "ON mb_album_release_cache (release_mbid)"
             )
 
+            # Persistent (artist,title) -> recording MBID cache for playlist export (#903).
+            # The MusicBrainz tail of the export waterfall is rate-limited (~1 req/s), so a
+            # resolved recording MBID is remembered ONCE and reused for that song across every
+            # future export and playlist. Additive: empty until the export feature writes it.
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS mb_recording_cache (
+                    track_key TEXT PRIMARY KEY,
+                    recording_mbid TEXT NOT NULL,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+
             # Discovery artist blacklist — artists users never want to see in discovery
             cursor.execute("""
                 CREATE TABLE IF NOT EXISTS discovery_artist_blacklist (

tests/exports/test_playlist_export.py

@@ -0,0 +1,74 @@
+"""Playlist export orchestrator (#903): dedup + stats + progress accounting.
+
+Pins: repeated songs resolve once (deduped), per-source breakdown counts only fresh
+resolutions, unmatched tracks carry recording_mbid=None but stay in order, alternate
+field shapes (artist_name/track_name) are accepted, and a throwing progress callback
+never fails the export.
+"""
+
+from __future__ import annotations
+
+from core.exports.playlist_export import resolve_playlist_tracks
+
+MBID = "e8f9b188-f819-4e43-ab0f-4bd26ce9ff56"
+MBID2 = "8f3471b5-7e6a-4c1f-9c1a-2b2b2b2b2b2b"
+
+
+def test_resolves_and_keeps_order_and_unmatched():
+    table = {("A", "T1"): (MBID, "db"), ("A", "T2"): (None, None)}
+    rf = lambda a, t: table.get((a, t), (None, None))
+    out = resolve_playlist_tracks(
+        [{"artist": "A", "title": "T1"}, {"artist": "A", "title": "T2"}], rf
+    )
+    res = out["resolved"]
+    assert [r["title"] for r in res] == ["T1", "T2"]
+    assert res[0]["recording_mbid"] == MBID
+    assert res[1]["recording_mbid"] is None
+    assert out["stats"]["resolved"] == 1
+    assert out["stats"]["unmatched"] == 1
+    assert out["stats"]["by_source"] == {"db": 1}
+
+
+def test_dedup_resolves_repeated_song_once():
+    calls = {"n": 0}
+    def rf(a, t):
+        calls["n"] += 1
+        return (MBID, "musicbrainz")
+    tracks = [{"artist": "A", "title": "Song"}, {"artist": "a", "title": "song"},  # same (normalized)
+              {"artist": "A", "title": "Song"}]
+    out = resolve_playlist_tracks(tracks, rf)
+    assert calls["n"] == 1                       # resolve_fn called once for 3 identical
+    assert out["stats"]["deduped"] == 2
+    assert out["stats"]["resolved"] == 3         # all three tracks still get the mbid
+    assert out["stats"]["by_source"] == {"musicbrainz": 1}  # counted once (fresh only)
+    assert all(r["recording_mbid"] == MBID for r in out["resolved"])
+
+
+def test_accepts_alternate_field_names():
+    rf = lambda a, t: (MBID, "db") if (a, t) == ("Artist", "Track") else (None, None)
+    out = resolve_playlist_tracks(
+        [{"artist_name": "Artist", "track_name": "Track", "album_name": "Alb"}], rf
+    )
+    r = out["resolved"][0]
+    assert r["artist"] == "Artist" and r["title"] == "Track" and r["album"] == "Alb"
+    assert r["recording_mbid"] == MBID
+
+
+def test_progress_called_per_track_and_safe_when_throwing():
+    seen = []
+    def prog(done, total, stats):
+        seen.append((done, total))
+        raise RuntimeError("display blew up")
+    out = resolve_playlist_tracks(
+        [{"artist": "A", "title": "x"}, {"artist": "B", "title": "y"}],
+        lambda a, t: (MBID, "db"),
+        on_progress=prog,
+    )
+    assert seen == [(1, 2), (2, 2)]              # called each track despite raising
+    assert out["stats"]["resolved"] == 2          # export still completed
+
+
+def test_empty_playlist():
+    out = resolve_playlist_tracks([], lambda a, t: (None, None))
+    assert out["resolved"] == []
+    assert out["stats"]["total"] == 0