#903: playlist export to ListenBrainz — pure cores (JSPF builder + MBID resolver)

Phase 1 of exporting mirrored playlists to ListenBrainz. Two pure, fully-tested seams,
zero runtime wiring yet (additive, no regression):

- core/exports/jspf_export.py: build_jspf(title, tracks) -> ({"playlist": {...}}, summary).
  LB's POST /1/playlist/create requires every track to carry a string identifier
  'https://musicbrainz.org/recording/<mbid>' (text-only tracks are rejected), so tracks
  without a valid recording-MBID UUID are dropped and counted in the coverage summary.
- core/exports/mbid_resolver.py: resolve_recording_mbid(artist, title, sources) — the
  cheapest-first waterfall (cache -> DB -> file tag -> MusicBrainz) as a pure function over
  injected (label, fn) sources. Short-circuits expensive lookups, treats a raising source
  as a miss (one flaky MB call can't fail the export), reports the resolving source label.

API spec confirmed against LB docs: POST /1/playlist/create, 'Authorization: Token <t>',
{"playlist": {"title", "track": [{"identifier": "<mb recording url>", title, creator, album}]}}.

13 tests.

core/exports/jspf_export.py

@@ -0,0 +1,89 @@
+"""Build a JSPF playlist (ListenBrainz-compatible) from resolved SoulSync tracks.
+
+ListenBrainz's ``POST /1/playlist/create`` requires JSPF where **every track carries a
+``identifier`` of ``https://musicbrainz.org/recording/<recording-mbid>``** — text-only
+entries (title/creator alone) are rejected. So a track can only be exported once we've
+resolved its MusicBrainz *recording* MBID (see ``mbid_resolver``); tracks without one are
+dropped here and surfaced to the user as "unmatched".
+
+Pure + I/O-free: callers pass already-resolved track dicts, this returns the JSPF dict
+(and a small coverage summary). The same JSPF is used for both the downloadable ``.jspf``
+file and the direct create-playlist POST, so there's one source of truth for the shape.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, List, Tuple
+
+MB_RECORDING_PREFIX = "https://musicbrainz.org/recording/"
+
+# A MusicBrainz MBID is a canonical UUID. Validate to avoid emitting garbage identifiers
+# that LB would reject (or, worse, that silently point nowhere).
+_UUID_RE = re.compile(
+    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.IGNORECASE
+)
+
+
+def is_valid_recording_mbid(mbid: Any) -> bool:
+    """True when ``mbid`` is a well-formed MusicBrainz UUID."""
+    return bool(mbid) and isinstance(mbid, str) and bool(_UUID_RE.match(mbid.strip()))
+
+
+def _track_entry(track: Dict[str, Any]) -> Dict[str, Any] | None:
+    """Build one JSPF track entry, or None if the track has no valid recording MBID."""
+    mbid = (track.get("recording_mbid") or "").strip() if isinstance(track.get("recording_mbid"), str) else ""
+    if not is_valid_recording_mbid(mbid):
+        return None
+    entry: Dict[str, Any] = {"identifier": f"{MB_RECORDING_PREFIX}{mbid}"}
+    # Optional, human-friendly fields — LB ignores them on create but they make the
+    # downloaded .jspf readable and round-trippable.
+    if track.get("title"):
+        entry["title"] = str(track["title"])
+    if track.get("artist"):
+        entry["creator"] = str(track["artist"])
+    if track.get("album"):
+        entry["album"] = str(track["album"])
+    return entry
+
+
+def build_jspf(
+    title: str,
+    tracks: List[Dict[str, Any]],
+    *,
+    creator: str = "",
+) -> Tuple[Dict[str, Any], Dict[str, int]]:
+    """Build a ListenBrainz-compatible JSPF dict from resolved tracks.
+
+    ``tracks`` is an ordered list of dicts with ``recording_mbid`` (required to be
+    included), plus optional ``title`` / ``artist`` / ``album``. Tracks without a valid
+    recording MBID are skipped (LB rejects them).
+
+    Returns ``(jspf, summary)`` where ``jspf`` is ``{"playlist": {...}}`` and ``summary``
+    is ``{"total", "included", "skipped"}`` for the coverage display.
+    """
+    jspf_tracks: List[Dict[str, Any]] = []
+    for t in tracks or []:
+        if not isinstance(t, dict):
+            continue
+        entry = _track_entry(t)
+        if entry is not None:
+            jspf_tracks.append(entry)
+
+    playlist: Dict[str, Any] = {
+        "title": (title or "SoulSync Export").strip() or "SoulSync Export",
+        "track": jspf_tracks,
+    }
+    if creator:
+        playlist["creator"] = str(creator)
+
+    total = sum(1 for t in (tracks or []) if isinstance(t, dict))
+    summary = {
+        "total": total,
+        "included": len(jspf_tracks),
+        "skipped": total - len(jspf_tracks),
+    }
+    return {"playlist": playlist}, summary
+
+
+__all__ = ["build_jspf", "is_valid_recording_mbid", "MB_RECORDING_PREFIX"]

core/exports/mbid_resolver.py

@@ -0,0 +1,88 @@
+"""Resolve a playlist track's MusicBrainz *recording* MBID, cheapest source first.
+
+A ListenBrainz playlist export needs each track's recording MBID (``jspf_export``). A
+SoulSync track can supply it from several places, in increasing cost:
+
+1. **resolution cache** — a prior (artist,title)->mbid result (persistent; reused across
+   playlists and runs, so the same song never costs twice).
+2. **library DB** — ``tracks.musicbrainz_recording_id`` (set by the MusicBrainz
+   enrichment worker).
+3. **file tags** — ``MUSICBRAINZ_RECORDING_ID`` written into the audio file on import
+   post-processing (catches tracks enriched at import but not via the worker).
+4. **MusicBrainz lookup** — a live ``match_recording(artist, title)`` (rate-limited
+   ~1 req/s; the slow tail — only hit when 1–3 miss).
+
+This module is the **pure waterfall**: the caller passes ordered ``(label, fn)`` sources,
+each ``fn(artist, title) -> mbid | None``, and ``resolve_recording_mbid`` returns the
+first valid hit plus its label (for the live status / stats). The actual I/O (DB query,
+mutagen read, MB request, cache read/write) lives in the export job that wires the real
+sources — so this stays trivially unit-testable and short-circuits correctly.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Callable, List, Optional, Tuple
+
+# Source labels (also used in the live-status breakdown).
+SRC_CACHE = "cache"
+SRC_DB = "db"
+SRC_FILE = "file"
+SRC_MUSICBRAINZ = "musicbrainz"
+SRC_NONE = None
+
+_UUID_RE = re.compile(
+    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.IGNORECASE
+)
+
+Source = Tuple[str, Callable[[str, str], Optional[str]]]
+
+
+def _valid(mbid: Any) -> Optional[str]:
+    """Return the trimmed MBID if it's a well-formed UUID, else None."""
+    if not isinstance(mbid, str):
+        return None
+    m = mbid.strip()
+    return m if _UUID_RE.match(m) else None
+
+
+def normalize_key(artist: Any, title: Any) -> str:
+    """Stable cache key for an (artist, title) pair — lower, punctuation-stripped,
+    whitespace-collapsed — so trivial variations share a cache entry."""
+    def _n(v: Any) -> str:
+        s = re.sub(r"[^\w\s]", "", str(v or "").lower())
+        return re.sub(r"\s+", " ", s).strip()
+    return f"{_n(artist)}␟{_n(title)}"
+
+
+def resolve_recording_mbid(
+    artist: str,
+    title: str,
+    sources: List[Source],
+) -> Tuple[Optional[str], Optional[str]]:
+    """Walk ``sources`` in order; return ``(mbid, label)`` of the first that yields a
+    valid recording MBID, or ``(None, None)`` when every source misses.
+
+    Each source is ``(label, fn)`` and ``fn(artist, title)`` returns an MBID or None. A
+    source that raises is treated as a miss (never aborts the waterfall) — so one flaky
+    lookup (e.g. a MusicBrainz timeout) can't fail the whole export. Short-circuits: a
+    later/expensive source isn't called once an earlier one hits.
+    """
+    for label, fn in sources or []:
+        try:
+            mbid = _valid(fn(artist, title))
+        except Exception:
+            mbid = None
+        if mbid:
+            return (mbid, label)
+    return (None, None)
+
+
+__all__ = [
+    "resolve_recording_mbid",
+    "normalize_key",
+    "SRC_CACHE",
+    "SRC_DB",
+    "SRC_FILE",
+    "SRC_MUSICBRAINZ",
+]

tests/exports/test_jspf_export.py

@@ -0,0 +1,72 @@
+"""JSPF builder for ListenBrainz playlist export (#903).
+
+LB's create-playlist requires every track to carry a recording-MBID identifier; text-only
+tracks are rejected. These pin the shape (top-level {"playlist": {...}}, string identifier
+in the exact MB recording URL form), that tracks without a valid MBID are dropped, and that
+the coverage summary counts included vs skipped.
+"""
+
+from __future__ import annotations
+
+from core.exports.jspf_export import MB_RECORDING_PREFIX, build_jspf, is_valid_recording_mbid
+
+MBID_A = "e8f9b188-f819-4e43-ab0f-4bd26ce9ff56"
+MBID_B = "8f3471b5-7e6a-4c1f-9c1a-2b2b2b2b2b2b"
+
+
+def test_valid_mbid_check():
+    assert is_valid_recording_mbid(MBID_A) is True
+    assert is_valid_recording_mbid("not-a-uuid") is False
+    assert is_valid_recording_mbid("") is False
+    assert is_valid_recording_mbid(None) is False
+
+
+def test_top_level_shape_and_identifier_format():
+    jspf, summary = build_jspf("My Playlist", [
+        {"recording_mbid": MBID_A, "title": "Gold", "artist": "Spandau Ballet", "album": "True"},
+    ])
+    assert set(jspf.keys()) == {"playlist"}
+    pl = jspf["playlist"]
+    assert pl["title"] == "My Playlist"
+    assert len(pl["track"]) == 1
+    t = pl["track"][0]
+    # identifier is a STRING in the exact MB recording form (per the LB/JSPF spec)
+    assert t["identifier"] == f"{MB_RECORDING_PREFIX}{MBID_A}"
+    assert isinstance(t["identifier"], str)
+    assert t["title"] == "Gold"
+    assert t["creator"] == "Spandau Ballet"  # artist -> creator
+    assert t["album"] == "True"
+    assert summary == {"total": 1, "included": 1, "skipped": 0}
+
+
+def test_tracks_without_valid_mbid_are_dropped():
+    jspf, summary = build_jspf("P", [
+        {"recording_mbid": MBID_A, "title": "Keep"},
+        {"recording_mbid": "", "title": "No MBID"},
+        {"recording_mbid": "garbage", "title": "Bad MBID"},
+        {"title": "Missing key entirely"},
+    ])
+    assert [t["title"] for t in jspf["playlist"]["track"]] == ["Keep"]
+    assert summary == {"total": 4, "included": 1, "skipped": 3}
+
+
+def test_order_is_preserved():
+    jspf, _ = build_jspf("P", [
+        {"recording_mbid": MBID_A, "title": "first"},
+        {"recording_mbid": MBID_B, "title": "second"},
+    ])
+    assert [t["title"] for t in jspf["playlist"]["track"]] == ["first", "second"]
+
+
+def test_optional_fields_omitted_when_absent():
+    jspf, _ = build_jspf("P", [{"recording_mbid": MBID_A}])
+    t = jspf["playlist"]["track"][0]
+    assert "title" not in t and "creator" not in t and "album" not in t
+
+
+def test_creator_and_title_defaults():
+    jspf, summary = build_jspf("", [], creator="SoulSync")
+    assert jspf["playlist"]["title"] == "SoulSync Export"  # blank -> default
+    assert jspf["playlist"]["creator"] == "SoulSync"
+    assert jspf["playlist"]["track"] == []
+    assert summary == {"total": 0, "included": 0, "skipped": 0}

tests/exports/test_mbid_resolver.py

@@ -0,0 +1,75 @@
+"""MBID resolution waterfall for playlist export (#903).
+
+Pins: cheapest-source-first short-circuit (don't hit MusicBrainz when the DB has it),
+invalid MBIDs are treated as misses, a raising source doesn't abort the export, and the
+resolving source label is reported (for the live status breakdown).
+"""
+
+from __future__ import annotations
+
+from core.exports.mbid_resolver import (
+    SRC_CACHE,
+    SRC_DB,
+    SRC_MUSICBRAINZ,
+    normalize_key,
+    resolve_recording_mbid,
+)
+
+MBID = "e8f9b188-f819-4e43-ab0f-4bd26ce9ff56"
+MBID2 = "8f3471b5-7e6a-4c1f-9c1a-2b2b2b2b2b2b"
+
+
+def _src(label, value):
+    return (label, lambda a, t: value)
+
+
+def test_returns_first_hit_with_label():
+    mbid, label = resolve_recording_mbid("A", "T", [_src(SRC_DB, MBID), _src(SRC_MUSICBRAINZ, MBID2)])
+    assert (mbid, label) == (MBID, SRC_DB)
+
+
+def test_short_circuits_expensive_sources():
+    called = {"mb": False}
+    def mb(a, t):
+        called["mb"] = True
+        return MBID2
+    mbid, label = resolve_recording_mbid("A", "T", [_src(SRC_CACHE, MBID), (SRC_MUSICBRAINZ, mb)])
+    assert (mbid, label) == (MBID, SRC_CACHE)
+    assert called["mb"] is False  # cache hit -> MusicBrainz never queried
+
+
+def test_falls_through_misses_to_later_source():
+    mbid, label = resolve_recording_mbid("A", "T", [
+        _src(SRC_CACHE, None),
+        _src(SRC_DB, ""),
+        _src(SRC_MUSICBRAINZ, MBID2),
+    ])
+    assert (mbid, label) == (MBID2, SRC_MUSICBRAINZ)
+
+
+def test_invalid_mbid_is_a_miss():
+    mbid, label = resolve_recording_mbid("A", "T", [
+        _src(SRC_DB, "not-a-uuid"),
+        _src(SRC_MUSICBRAINZ, MBID),
+    ])
+    assert (mbid, label) == (MBID, SRC_MUSICBRAINZ)
+
+
+def test_raising_source_does_not_abort():
+    def boom(a, t):
+        raise RuntimeError("MusicBrainz timeout")
+    mbid, label = resolve_recording_mbid("A", "T", [
+        (SRC_DB, boom),
+        _src(SRC_MUSICBRAINZ, MBID),
+    ])
+    assert (mbid, label) == (MBID, SRC_MUSICBRAINZ)
+
+
+def test_all_miss_returns_none():
+    assert resolve_recording_mbid("A", "T", [_src(SRC_DB, None), _src(SRC_MUSICBRAINZ, None)]) == (None, None)
+    assert resolve_recording_mbid("A", "T", []) == (None, None)
+
+
+def test_normalize_key_is_stable_across_variations():
+    assert normalize_key("The Beatles", "Hey Jude!") == normalize_key("the beatles", "hey  jude")
+    assert normalize_key("A", "X") != normalize_key("B", "X")