From fb9d88ea6ac646352f96c1b302949c72b22adb5f Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Mon, 22 Jun 2026 20:25:44 -0700 Subject: [PATCH] =?UTF-8?q?#903:=20playlist=20export=20to=20ListenBrainz?= =?UTF-8?q?=20=E2=80=94=20pure=20cores=20(JSPF=20builder=20+=20MBID=20reso?= =?UTF-8?q?lver)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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/' (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 ', {"playlist": {"title", "track": [{"identifier": "", title, creator, album}]}}. 13 tests. --- core/exports/jspf_export.py | 89 +++++++++++++++++++++++++++++ core/exports/mbid_resolver.py | 88 ++++++++++++++++++++++++++++ tests/exports/__init__.py | 0 tests/exports/test_jspf_export.py | 72 +++++++++++++++++++++++ tests/exports/test_mbid_resolver.py | 75 ++++++++++++++++++++++++ 5 files changed, 324 insertions(+) create mode 100644 core/exports/jspf_export.py create mode 100644 core/exports/mbid_resolver.py create mode 100644 tests/exports/__init__.py create mode 100644 tests/exports/test_jspf_export.py create mode 100644 tests/exports/test_mbid_resolver.py diff --git a/core/exports/jspf_export.py b/core/exports/jspf_export.py new file mode 100644 index 00000000..3d77cd16 --- /dev/null +++ b/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/``** — 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"] diff --git a/core/exports/mbid_resolver.py b/core/exports/mbid_resolver.py new file mode 100644 index 00000000..f5b11972 --- /dev/null +++ b/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", +] diff --git a/tests/exports/__init__.py b/tests/exports/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/exports/test_jspf_export.py b/tests/exports/test_jspf_export.py new file mode 100644 index 00000000..684265d5 --- /dev/null +++ b/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} diff --git a/tests/exports/test_mbid_resolver.py b/tests/exports/test_mbid_resolver.py new file mode 100644 index 00000000..9be12a1d --- /dev/null +++ b/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")