danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

#903: re-export updates the same ListenBrainz playlist in place (no duplicates)

Browse source 
Re-running an export created a new LB playlist every time (LB keys on MBID, not name, and
create always mints a new one). Now remember which LB playlist a mirror was pushed to and
update it in place:

- listenbrainz_client: refactor batched-add into _add_tracks_in_batches; add
  get_playlist_track_count, delete_playlist, update_playlist (verify exists -> clear items via
  item/delete -> re-add -> edit title; reports gone=True if deleted on LB), and
  create_or_update_playlist (update when we have a prior MBID, else create; falls back to
  create if the remembered one was deleted). Stable URL/MBID across re-syncs.
- playlist_export_targets table + get/set_playlist_export_target: remember (mirror, target) -> LB MBID.
- export job consults/stores the target so push updates in place.

+6 mocked tests (clear+re-add same mbid, gone-fallback, create-or-update branches, delete). API
endpoints (item/delete, playlist/edit, playlist/delete, GET count) confirmed against LB docs;
live round-trip pending explicit auth.
This commit is contained in:
BoulderBadgeDad 2026-06-22 22:36:29 -07:00
parent 5f1ec9ed7e
commit ab8f82af2e
4 changed files with 249 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

162
core/listenbrainz_client.py
View file

@ -158,40 +158,78 @@ class ListenBrainzClient:
return True
_MAX_TRACKS_PER_ADD = 100 # ListenBrainz MAX_RECORDINGS_PER_ADD
_PLAYLIST_EXT = "https://musicbrainz.org/doc/jspf#playlist"
def _lb_headers(self) -> Dict:
return {"Authorization": f"Token {self.token}", "Content-Type": "application/json"}
def _add_tracks_in_batches(self, playlist_mbid: str, tracks: List[Dict]) -> int:
"""Add JSPF tracks to a playlist in <=100-track batches; return how many were added."""
added = 0
headers = self._lb_headers()
for i in range(0, len(tracks or []), self._MAX_TRACKS_PER_ADD):
batch = tracks[i:i + self._MAX_TRACKS_PER_ADD]
try:
r = self._make_request_with_retry(
"POST", f"{self.base_url}/playlist/{playlist_mbid}/item/add",
json={"playlist": {"track": batch}}, 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}")
return added
def get_playlist_track_count(self, playlist_mbid: str):
"""Current track count of an LB playlist, or None if it can't be fetched (gone/404)."""
try:
r = self._make_request_with_retry(
"GET", f"{self.base_url}/playlist/{playlist_mbid}",
params={"fetch_metadata": "false"},
headers={"Authorization": f"Token {self.token}"},
)
if r and r.status_code == 200:
return len(((r.json() or {}).get("playlist") or {}).get("track", []))
except Exception as e:
logger.debug(f"ListenBrainz get playlist count failed: {e}")
return None
def delete_playlist(self, playlist_mbid: str) -> bool:
"""Delete an LB playlist. True on success."""
try:
r = self._make_request_with_retry(
"POST", f"{self.base_url}/playlist/{playlist_mbid}/delete", headers=self._lb_headers()
)
return bool(r and r.status_code in (200, 201))
except Exception as e:
logger.error(f"ListenBrainz delete playlist error: {e}")
return False
def create_playlist(self, title: str, tracks: List[Dict], public: bool = False) -> Dict:
"""Create a playlist on ListenBrainz and add its tracks (#903).
"""Create a NEW 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/<mbid>`` (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.
``https://musicbrainz.org/recording/<mbid>`` (LB rejects text-only tracks). Creates
an empty playlist for the MBID, then adds tracks in <=100 batches. Returns
``{success, playlist_mbid, playlist_url, added, requested, error, updated}``. Never raises.
"""
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}
"added": 0, "requested": len(tracks or []), "error": None, "updated": False}
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)}},
"extension": {self._PLAYLIST_EXT: {"public": bool(public)}},
}}
try:
resp = self._make_request_with_retry(
"POST", f"{self.base_url}/playlist/create", json=create_body, headers=headers
"POST", f"{self.base_url}/playlist/create", json=create_body, headers=self._lb_headers()
)
except Exception as e:
result["error"] = f"create request failed: {e}"
@ -209,30 +247,72 @@ class ListenBrainzClient:
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["added"] = self._add_tracks_in_batches(playlist_mbid, tracks)
result["success"] = True
return result
def update_playlist(self, playlist_mbid: str, title: str, tracks: List[Dict], public: bool = False) -> Dict:
"""Replace an existing LB playlist's contents IN PLACE (stable URL/MBID) (#903).
Verifies the playlist still exists, clears its current items, re-adds the new tracks,
and updates the title. If the playlist is gone (deleted on LB), returns success=False
with ``gone=True`` so the caller can fall back to creating a fresh one.
Returns the same shape as ``create_playlist`` plus ``updated=True``.
"""
result = {"success": False, "playlist_mbid": playlist_mbid,
"playlist_url": f"https://listenbrainz.org/playlist/{playlist_mbid}",
"added": 0, "requested": len(tracks or []), "error": None,
"updated": True, "gone": False}
if not self.is_authenticated():
result["error"] = "ListenBrainz not authenticated (no token/username)"
return result
count = self.get_playlist_track_count(playlist_mbid)
if count is None:
result["error"] = "playlist not found on ListenBrainz"
result["gone"] = True
return result
headers = self._lb_headers()
# Clear existing items (one range delete from the top).
if count > 0:
try:
self._make_request_with_retry(
"POST", f"{self.base_url}/playlist/{playlist_mbid}/item/delete",
json={"index": 0, "count": count}, headers=headers,
)
except Exception as e:
logger.warning(f"ListenBrainz item/delete (clear) failed: {e}")
result["added"] = self._add_tracks_in_batches(playlist_mbid, tracks)
# Refresh the title (best-effort — content already replaced).
try:
self._make_request_with_retry(
"POST", f"{self.base_url}/playlist/edit/{playlist_mbid}",
json={"playlist": {"title": (title or "SoulSync Export").strip() or "SoulSync Export"}},
headers=headers,
)
except Exception as e:
logger.debug(f"ListenBrainz playlist title edit failed: {e}")
result["success"] = True
return result
def create_or_update_playlist(self, title: str, tracks: List[Dict],
existing_mbid: str = None, public: bool = False) -> Dict:
"""Update the existing LB playlist in place when we've pushed this one before, else
create a fresh one — so re-exporting the same SoulSync playlist never duplicates it.
Falls back to create if the remembered playlist was deleted on LB."""
if existing_mbid:
res = self.update_playlist(existing_mbid, title, tracks, public)
if res.get("success"):
return res
# Remembered playlist gone/failed -> create a new one instead of erroring out.
logger.info(f"ListenBrainz playlist {existing_mbid} unavailable for update "
f"({res.get('error')}); creating a new one.")
return self.create_playlist(title, tracks, public)
def get_playlists_created_for_user(self, count: int = 25, offset: int = 0) -> List[Dict]:
"""
Fetch playlists created FOR the user (recommendations, personalized playlists)

49
database/music_database.py
View file

@ -2097,6 +2097,19 @@ class MusicDatabase:
)
""")
# Remember which external playlist a mirrored playlist was exported to, so a
# re-export UPDATES it in place instead of creating a duplicate (#903). Keyed by
# (mirrored playlist, target service) -> the target's playlist id (LB recording MBID).
cursor.execute("""
CREATE TABLE IF NOT EXISTS playlist_export_targets (
mirrored_playlist_id INTEGER NOT NULL,
target TEXT NOT NULL,
target_playlist_mbid TEXT NOT NULL,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (mirrored_playlist_id, target)
)
""")
# Discovery artist blacklist — artists users never want to see in discovery
cursor.execute("""
CREATE TABLE IF NOT EXISTS discovery_artist_blacklist (
@ -14167,6 +14180,42 @@ class MusicDatabase:
logger.error(f"Error updating custom_name for playlist {playlist_id}: {e}")
return False
def get_playlist_export_target(self, mirrored_playlist_id: int, target: str) -> Optional[str]:
"""The external playlist id this mirror was last exported to (or None). #903."""
try:
with self._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"SELECT target_playlist_mbid FROM playlist_export_targets "
"WHERE mirrored_playlist_id = ? AND target = ? LIMIT 1",
(int(mirrored_playlist_id), target),
)
row = cur.fetchone()
if row:
return (row[0] if not hasattr(row, "keys") else row["target_playlist_mbid"]) or None
except Exception as e:
logger.debug(f"get_playlist_export_target failed: {e}")
return None
def set_playlist_export_target(self, mirrored_playlist_id: int, target: str, target_mbid: str) -> bool:
"""Remember the external playlist id for this mirror (idempotent). #903."""
if not target_mbid:
return False
try:
with self._get_connection() as conn:
cur = conn.cursor()
cur.execute(
"INSERT OR REPLACE INTO playlist_export_targets "
"(mirrored_playlist_id, target, target_playlist_mbid, updated_at) "
"VALUES (?, ?, ?, CURRENT_TIMESTAMP)",
(int(mirrored_playlist_id), target, target_mbid),
)
conn.commit()
return True
except Exception as e:
logger.debug(f"set_playlist_export_target failed: {e}")
return False
def get_mirrored_playlist_tracks(self, playlist_id: int) -> List[Dict]:
"""Return all tracks for a mirrored playlist ordered by position."""
try:

74
tests/exports/test_listenbrainz_create_playlist.py
View file

@ -84,3 +84,77 @@ def test_create_ok_but_partial_add_failure(monkeypatch):
assert res["success"] is True
assert res["playlist_mbid"] == "PL"
assert res["added"] == 0
# ── update-in-place + create-or-update (#903 no-duplicate re-export) ──────────
def _route(existing_count=None):
"""Build a fake _make_request_with_retry; records calls. existing_count=None -> GET 404."""
calls = []
def fake(method, url, **kw):
calls.append((method, url, kw.get("json")))
if method == "GET" and "/playlist/" in url:
if existing_count is None:
return _Resp(404, {})
return _Resp(200, {"playlist": {"track": [{} for _ in range(existing_count)]}})
if url.endswith("/playlist/create"):
return _Resp(200, {"playlist_mbid": "NEW-MBID"})
return _Resp(200, {}) # item/add, item/delete, edit, delete
return fake, calls
def test_update_playlist_clears_then_re_adds_same_mbid(monkeypatch):
c = _client()
fake, calls = _route(existing_count=3)
monkeypatch.setattr(c, "_make_request_with_retry", fake)
res = c.update_playlist("EXIST-MBID", "New Title", _tracks(5))
assert res["success"] is True and res["updated"] is True
assert res["playlist_mbid"] == "EXIST-MBID" # stable URL/MBID
assert res["added"] == 5
urls = [u for _, u, _ in calls]
assert any("/item/delete" in u for u in urls) # cleared existing
assert any("/item/add" in u for u in urls) # re-added
assert any("/playlist/edit/EXIST-MBID" in u for u in urls) # title refreshed
def test_update_playlist_reports_gone(monkeypatch):
c = _client()
fake, _ = _route(existing_count=None) # GET 404 -> gone
monkeypatch.setattr(c, "_make_request_with_retry", fake)
res = c.update_playlist("DEAD-MBID", "T", _tracks(2))
assert res["success"] is False
assert res["gone"] is True
def test_create_or_update_updates_when_existing(monkeypatch):
c = _client()
fake, _ = _route(existing_count=2)
monkeypatch.setattr(c, "_make_request_with_retry", fake)
res = c.create_or_update_playlist("T", _tracks(4), existing_mbid="EXIST")
assert res["updated"] is True and res["playlist_mbid"] == "EXIST"
def test_create_or_update_falls_back_to_create_when_gone(monkeypatch):
c = _client()
fake, _ = _route(existing_count=None) # remembered playlist deleted on LB
monkeypatch.setattr(c, "_make_request_with_retry", fake)
res = c.create_or_update_playlist("T", _tracks(4), existing_mbid="DEAD")
assert res["success"] is True
assert res["updated"] is False # fell back to create
assert res["playlist_mbid"] == "NEW-MBID"
def test_create_or_update_creates_when_no_existing(monkeypatch):
c = _client()
fake, _ = _route()
monkeypatch.setattr(c, "_make_request_with_retry", fake)
res = c.create_or_update_playlist("T", _tracks(3), existing_mbid=None)
assert res["playlist_mbid"] == "NEW-MBID" and res["updated"] is False
def test_delete_playlist(monkeypatch):
c = _client()
monkeypatch.setattr(c, "_make_request_with_retry", lambda *a, **k: _Resp(200, {}))
assert c.delete_playlist("MBID") is True
monkeypatch.setattr(c, "_make_request_with_retry", lambda *a, **k: _Resp(404, {}))
assert c.delete_playlist("MBID") is False

6
web_server.py
View file

@ -27200,9 +27200,13 @@ def _run_playlist_export(job_id, playlist_id, title, mode):
job['phase'] = 'pushing'
from core.listenbrainz_client import ListenBrainzClient
client = ListenBrainzClient()
res = client.create_playlist(title, jspf['playlist']['track'])
# Re-export updates the same LB playlist in place instead of duplicating it (#903).
existing = db.get_playlist_export_target(int(playlist_id), 'listenbrainz')
res = client.create_or_update_playlist(title, jspf['playlist']['track'], existing_mbid=existing)
job['push'] = res
if res.get('success'):
if res.get('playlist_mbid'):
db.set_playlist_export_target(int(playlist_id), 'listenbrainz', res['playlist_mbid'])
job['phase'] = 'done'
else:
job['phase'] = 'error'