diff --git a/core/listenbrainz_client.py b/core/listenbrainz_client.py index 2ec6ed0e..0f21acfe 100644 --- a/core/listenbrainz_client.py +++ b/core/listenbrainz_client.py @@ -158,6 +158,81 @@ class ListenBrainzClient: return True + def create_playlist(self, title: str, tracks: List[Dict], public: bool = False) -> Dict: + """Create a playlist on ListenBrainz and add its tracks (#903). + + ``tracks`` are JSPF track dicts — each MUST carry an ``identifier`` of the form + ``https://musicbrainz.org/recording/`` (LB rejects text-only tracks; the + caller drops un-resolved tracks via ``core.exports.jspf_export``). + + Robust for large playlists: create an empty playlist (title only) to get the + playlist MBID, then add tracks in batches of ``MAX_TRACKS_PER_ADD`` (LB's + ``MAX_RECORDINGS_PER_ADD`` cap) via the ``item/add`` endpoint — so a 1k-track + playlist doesn't blow a single-request limit. + + Returns ``{success, playlist_mbid, playlist_url, added, requested, error}``. + Never raises — failures are reported in the dict so the export job can surface them. + """ + MAX_TRACKS_PER_ADD = 100 + PLAYLIST_EXT = "https://musicbrainz.org/doc/jspf#playlist" + result = {"success": False, "playlist_mbid": None, "playlist_url": None, + "added": 0, "requested": len(tracks or []), "error": None} + + if not self.is_authenticated(): + result["error"] = "ListenBrainz not authenticated (no token/username)" + return result + + headers = {"Authorization": f"Token {self.token}", "Content-Type": "application/json"} + + # 1) Create the (empty) playlist — title + visibility only. + create_body = {"playlist": { + "title": (title or "SoulSync Export").strip() or "SoulSync Export", + "extension": {PLAYLIST_EXT: {"public": bool(public)}}, + }} + try: + resp = self._make_request_with_retry( + "POST", f"{self.base_url}/playlist/create", json=create_body, headers=headers + ) + except Exception as e: + result["error"] = f"create request failed: {e}" + return result + if not resp or resp.status_code not in (200, 201): + result["error"] = f"create returned {resp.status_code if resp else 'no response'}" + return result + try: + playlist_mbid = (resp.json() or {}).get("playlist_mbid") + except Exception: + playlist_mbid = None + if not playlist_mbid: + result["error"] = "create succeeded but no playlist_mbid in response" + return result + + result["playlist_mbid"] = playlist_mbid + result["playlist_url"] = f"https://listenbrainz.org/playlist/{playlist_mbid}" + + # 2) Add tracks in capped batches. + added = 0 + for i in range(0, len(tracks or []), MAX_TRACKS_PER_ADD): + batch = tracks[i:i + MAX_TRACKS_PER_ADD] + add_body = {"playlist": {"track": batch}} + try: + r = self._make_request_with_retry( + "POST", f"{self.base_url}/playlist/{playlist_mbid}/item/add", + json=add_body, headers=headers, + ) + if r and r.status_code in (200, 201): + added += len(batch) + else: + logger.warning(f"ListenBrainz item/add batch failed: " + f"{r.status_code if r else 'no response'}") + except Exception as e: + logger.error(f"ListenBrainz item/add error: {e}") + + result["added"] = added + # Playlist was created even if some adds failed — report partial success honestly. + result["success"] = True + return result + def get_playlists_created_for_user(self, count: int = 25, offset: int = 0) -> List[Dict]: """ Fetch playlists created FOR the user (recommendations, personalized playlists) diff --git a/tests/exports/test_listenbrainz_create_playlist.py b/tests/exports/test_listenbrainz_create_playlist.py new file mode 100644 index 00000000..a87cb440 --- /dev/null +++ b/tests/exports/test_listenbrainz_create_playlist.py @@ -0,0 +1,86 @@ +"""ListenBrainz create_playlist client method (#903). + +Pins the create -> batched item/add flow against a mocked network: large playlists are +added in <=100 batches (LB's MAX_RECORDINGS_PER_ADD), the new playlist MBID/URL are +returned, and failures are reported (never raised) so the export job can surface them. +""" + +from __future__ import annotations + +from core.listenbrainz_client import ListenBrainzClient + + +class _Resp: + def __init__(self, status, body=None): + self.status_code = status + self._body = body or {} + + def json(self): + return self._body + + +def _client(): + # token='' (not None) skips the network token-validation in __init__; then fake auth. + c = ListenBrainzClient(token="") + c.token = "tok" + c.username = "user" + return c + + +def _tracks(n): + return [{"identifier": f"https://musicbrainz.org/recording/{i:08d}-0000-0000-0000-000000000000"} + for i in range(n)] + + +def test_create_then_batched_add(monkeypatch): + c = _client() + calls = [] + + def fake_req(method, url, **kw): + calls.append(url) + if url.endswith("/playlist/create"): + return _Resp(200, {"status": "ok", "playlist_mbid": "PL-MBID"}) + return _Resp(200, {"status": "ok"}) + + monkeypatch.setattr(c, "_make_request_with_retry", fake_req) + res = c.create_playlist("My Playlist", _tracks(250)) + + assert res["success"] is True + assert res["playlist_mbid"] == "PL-MBID" + assert res["playlist_url"] == "https://listenbrainz.org/playlist/PL-MBID" + assert res["added"] == 250 + # 1 create + 3 add batches (100 + 100 + 50) + assert sum(1 for u in calls if u.endswith("/playlist/create")) == 1 + assert sum(1 for u in calls if "/item/add" in u) == 3 + + +def test_not_authenticated_is_reported_not_raised(): + c = ListenBrainzClient(token="") # no username -> not authenticated + res = c.create_playlist("x", _tracks(1)) + assert res["success"] is False + assert "authenticated" in (res["error"] or "") + + +def test_create_failure_reported(monkeypatch): + c = _client() + monkeypatch.setattr(c, "_make_request_with_retry", lambda *a, **k: _Resp(400, {})) + res = c.create_playlist("x", _tracks(5)) + assert res["success"] is False + assert res["playlist_mbid"] is None + assert "400" in (res["error"] or "") + + +def test_create_ok_but_partial_add_failure(monkeypatch): + c = _client() + + def fake_req(method, url, **kw): + if url.endswith("/playlist/create"): + return _Resp(200, {"playlist_mbid": "PL"}) + return _Resp(500, {}) # every add fails + + monkeypatch.setattr(c, "_make_request_with_retry", fake_req) + res = c.create_playlist("x", _tracks(10)) + # Playlist WAS created -> success True, but added=0 (honest partial report) + assert res["success"] is True + assert res["playlist_mbid"] == "PL" + assert res["added"] == 0