deezer: playlist-write via the ARL gw-light gateway (#945, increment 3)

The Deezer write half of "export a mirrored playlist back to Deezer". Rides the private
gw-light gateway with the ARL session already authenticated for downloads (Deezer shut their
public developer API, so this is the only write path — unofficial and fragile by nature).

DeezerDownloadClient.create_or_update_playlist(name, track_ids, existing_id=None): playlist.create
with the songs (new), or playlist.addSongs to the stored target (re-export). track_ids are the
stored deezer_id per library track. Returns the same {success, playlist_id, url, added, error}
shape as the Spotify writer so the export job can treat both uniformly.

5 tests (mocked gateway): create sends playlist.create with positional songs, update sends
playlist.addSongs (no create), empty → error with no gw call, not-authed → error, gw rejection →
error. ruff clean. Additive — download paths untouched.

Both write clients done. Next: the export-job branch + endpoint (reusing get/set_playlist_export_
target for idempotency), then the modal options.
This commit is contained in:
BoulderBadgeDad 2026-06-28 21:15:04 -07:00
parent 37c2b9b569
commit 0db72bf48d
2 changed files with 105 additions and 0 deletions

View file

@ -233,6 +233,55 @@ class DeezerDownloadClient(DownloadSourcePlugin):
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, self.is_available)
# ─── Playlist export (#945) ──────────────────────────────────
#
# UNOFFICIAL: rides the private gw-light gateway with the ARL session already used
# for downloads. Deezer shut their public developer API, so this is the only write
# path — and it's fragile by nature (breaks when Deezer changes internals).
def create_or_update_playlist(self, name, track_ids, *, existing_id=None,
public=False, description=""):
"""Create a Deezer playlist (or append to an existing one) from a mirrored
playlist's tracks. ``track_ids`` are stored ``deezer_id`` values per library track.
``existing_id`` set add to that playlist (idempotent re-export reuses the stored
target); unset create a new one. Returns
``{success, playlist_id, url, added, error}``."""
if not self._authenticated:
return {"success": False, "error": "Deezer is not connected (ARL)"}
song_ids = [str(t) for t in (track_ids or []) if t]
if not song_ids:
return {"success": False, "error": "No matching Deezer tracks to export"}
try:
songs = [[sid, i] for i, sid in enumerate(song_ids)]
if existing_id:
res = self._gw_call("playlist.addSongs",
{"playlist_id": int(existing_id), "songs": songs})
if res is None:
return {"success": False, "error": "Deezer rejected the playlist update"}
playlist_id = existing_id
else:
res = self._gw_call("playlist.create", {
"title": name, "description": description,
"is_public": bool(public), "songs": songs,
})
if res is None:
return {"success": False, "error": "Deezer rejected the playlist create"}
# gw 'playlist.create' returns the new playlist id (int) as `results`.
if isinstance(res, dict):
playlist_id = res.get("PLAYLIST_ID") or res.get("id")
else:
playlist_id = res
if not playlist_id:
return {"success": False, "error": "Deezer did not return a playlist id"}
return {
"success": True,
"playlist_id": str(playlist_id),
"url": f"https://www.deezer.com/playlist/{playlist_id}",
"added": len(song_ids),
}
except Exception as e:
return {"success": False, "error": str(e)}
def reconnect(self, arl: str = None) -> bool:
"""Re-authenticate with a new or existing ARL."""
if arl is None:

View file

@ -0,0 +1,56 @@
"""Deezer playlist export via the gw-light gateway (#945) — the write half of
'sync a mirrored playlist back to Deezer'. Mocks the gateway call (the live API is the
unofficial ARL gw-light path; we test the wiring, not Deezer)."""
from core.deezer_download_client import DeezerDownloadClient
def _client(gw, authed=True):
c = DeezerDownloadClient.__new__(DeezerDownloadClient)
c._authenticated = authed
c._gw_call = gw
return c
def test_create_new_playlist():
calls = []
def gw(method, params):
calls.append((method, params))
return 12345 # gw returns the new playlist id
res = _client(gw).create_or_update_playlist('My Mix', ['100', '200'])
assert res['success'] and res['playlist_id'] == '12345'
assert res['url'] == 'https://www.deezer.com/playlist/12345'
assert res['added'] == 2
method, params = calls[0]
assert method == 'playlist.create'
assert params['title'] == 'My Mix'
assert params['songs'] == [['100', 0], ['200', 1]]
def test_update_existing_appends_no_create():
calls = []
def gw(method, params):
calls.append((method, params))
return {}
res = _client(gw).create_or_update_playlist('My Mix', ['100'], existing_id='999')
assert res['success'] and res['playlist_id'] == '999'
assert calls[0][0] == 'playlist.addSongs'
assert calls[0][1]['playlist_id'] == 999
assert not any(m == 'playlist.create' for m, _ in calls)
def test_empty_tracks_errors_no_gw_call():
calls = []
res = _client(lambda *a: calls.append(a)).create_or_update_playlist('X', [])
assert not res['success'] and 'No matching' in res['error']
assert calls == []
def test_not_authed_errors():
res = _client(lambda *a: None, authed=False).create_or_update_playlist('X', ['1'])
assert not res['success'] and 'not connected' in res['error']
def test_gw_rejection_is_an_error():
res = _client(lambda *a: None).create_or_update_playlist('X', ['1'])
assert not res['success'] and 'rejected' in res['error']