From 0db72bf48d61d9ceded6aaff0acb2eda596f4783 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Sun, 28 Jun 2026 21:15:04 -0700 Subject: [PATCH] deezer: playlist-write via the ARL gw-light gateway (#945, increment 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- core/deezer_download_client.py | 49 ++++++++++++++++++++++++ tests/test_deezer_playlist_export.py | 56 ++++++++++++++++++++++++++++ 2 files changed, 105 insertions(+) create mode 100644 tests/test_deezer_playlist_export.py diff --git a/core/deezer_download_client.py b/core/deezer_download_client.py index e1830f64..d08a9ef1 100644 --- a/core/deezer_download_client.py +++ b/core/deezer_download_client.py @@ -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: diff --git a/tests/test_deezer_playlist_export.py b/tests/test_deezer_playlist_export.py new file mode 100644 index 00000000..5e63cee4 --- /dev/null +++ b/tests/test_deezer_playlist_export.py @@ -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']