diff --git a/deezer_cli.py b/deezer_cli.py new file mode 100755 index 0000000..b8acec8 --- /dev/null +++ b/deezer_cli.py @@ -0,0 +1,282 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.9" +# dependencies = [ +# "bottle", +# "httpx", +# "pycryptodomex", +# ] +# /// +"""Small local Deezer search and download server. + +Examples: + ./deezer_cli.py --arl "$DEEZER_ARL" + DEEZER_ARL=... ./deezer_cli.py --port 8080 +""" + +import argparse +import hashlib +import html +import html.parser +import json +import os +import re +import threading +import webbrowser +from binascii import a2b_hex, b2a_hex + +from bottle import Bottle, request, response +import httpx +from Cryptodome.Cipher import AES, Blowfish + + +API_URL = "https://api.deezer.com" +WEB_URL = "https://www.deezer.com" +MEDIA_URL = "https://media.deezer.com/v1/get_url" +HEADERS = { + "Origin": WEB_URL, + "Accept-Encoding": "utf-8", + "Referer": f"{WEB_URL}/login", +} + + +class ScriptExtractor(html.parser.HTMLParser): + def __init__(self): + super().__init__() + self.scripts = [] + self.in_script = False + + def handle_starttag(self, tag, attrs): + self.in_script = tag.lower() == "script" + + def handle_data(self, data): + if self.in_script: + self.scripts.append(data) + + def handle_endtag(self, tag): + if tag.lower() == "script": + self.in_script = False + + +def api_get(client, path, **params): + response = client.get(f"{API_URL}/{path.lstrip('/')}", params=params) + response.raise_for_status() + data = response.json() + if "error" in data: + raise RuntimeError(data["error"].get("message", "Deezer API request failed.")) + return data + + +def search_tracks(client, query): + return api_get(client, "search/track", q=query, limit=25).get("data", []) + + +def track_metadata(client, track_id): + data = api_get(client, f"track/{track_id}") + if not data.get("readable"): + raise RuntimeError(f"Track {track_id} is unavailable or not playable.") + return data + + +def website_song_data(client, track_id): + response = client.get(f"{WEB_URL}/us/track/{track_id}") + response.raise_for_status() + parser = ScriptExtractor() + parser.feed(response.text) + for script in parser.scripts: + match = re.search(r'{"DATA":.*', script) + if match: + state = json.loads(match.group(0)) + return state["DATA"] + raise RuntimeError("Could not find Deezer track data in the web page.") + + +def md5hex(data): + return b2a_hex(hashlib.md5(data).digest()) + + +def encrypted_url(song_id, md5_origin, media_version, quality): + joined = b"\xa4".join([ + md5_origin.encode(), + str(quality).encode(), + str(song_id).encode(), + str(media_version).encode(), + ]) + data = b"\xa4".join([md5hex(joined), joined]) + b"\xa4" + data += b"\0" * (-len(data) % 16) + cipher = AES.new(b"jo6aey6haid2Teih", AES.MODE_ECB) + return b2a_hex(cipher.encrypt(data)).decode() + + +def decrypt_stream(data, song_id): + song_id_md5 = md5hex(str(song_id).encode()) + key = bytes(song_id_md5[i] ^ song_id_md5[i + 16] ^ b"g4el58wc0zvf9na1"[i] + for i in range(16)) + cipher = Blowfish.new(key, Blowfish.MODE_CBC, iv=a2b_hex("0001020304050607")) + output = bytearray() + for block_number, start in enumerate(range(0, len(data), 2048)): + block = data[start:start + 2048] + if block_number % 3 == 0 and len(block) == 2048: + block = cipher.decrypt(block) + output.extend(block) + return bytes(output) + + +def safe_filename(value): + value = str(value or "") + value = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "-", value) + return value.rstrip(" .") or "unknown" + + +def audio_data(client, token, track_id): + metadata = track_metadata(client, track_id) + song = website_song_data(client, track_id) + + choices = ( + ("MP3_320", 3, "FILESIZE_MP3_320"), + ("MP3_256", 5, "FILESIZE_MP3_256"), + ("MP3_128", 1, "FILESIZE_MP3_128"), + ) + song_format, quality, _ = choices[-1] + for choice in choices: + if int(song.get(choice[2]) or 0) > 0: + song_format, quality, _ = choice + break + + response = client.post( + MEDIA_URL, + headers=HEADERS, + json={ + "license_token": token, + "media": [{"type": "FULL", "formats": [{"cipher": "BF_CBC_STRIPE", "format": song_format}]}], + "track_tokens": [song["TRACK_TOKEN"]], + }, + ) + response.raise_for_status() + media_items = response.json().get("data") or [] + media = (media_items[0].get("media") or []) if media_items else [] + sources = (media[0].get("sources") or []) if media else [] + if sources: + media_url = sources[0]["url"] + else: + media_url = ( + f"https://e-cdns-proxy-{song['MD5_ORIGIN'][0]}.dzcdn.net/mobile/1/" + f"{encrypted_url(song['SNG_ID'], song['MD5_ORIGIN'], song['MEDIA_VERSION'], 1)}" + ) + + media_response = client.get(media_url) + media_response.raise_for_status() + filename = f"{safe_filename(metadata.get('artist', {}).get('name'))} - {safe_filename(metadata.get('title'))}.mp3" + return filename, decrypt_stream(media_response.content, song["SNG_ID"]) + + +def render_page(query="", results=(), error=""): + rows = "".join( + "" + f"\"\"" + f"{html.escape(result.get('title', ''))}" + f"{html.escape(result.get('artist', {}).get('name', ''))}" + f"{html.escape(result.get('album', {}).get('title', ''))}" + f"Download" + "" + for result in results + ) + table = ( + "" + f"{rows}
TitleArtistAlbum
" + if results else ("

No results.

" if query and not error else "") + ) + error_box = f"

{html.escape(error)}

" if error else "" + return f""" + + + + + Deezer Search + + + +

Deezer Search

+
+ + +
+ {error_box} + {table} + +""" + + +def create_app(client, token): + app = Bottle() + + @app.get("/") + def index(): + query = request.query.q.strip() + if not query: + return render_page() + try: + return render_page(query, search_tracks(client, query)) + except (httpx.HTTPError, KeyError, ValueError, RuntimeError) as exc: + return render_page(query, error=str(exc)) + + @app.get("/download/") + def download_route(track_id): + try: + filename, data = audio_data(client, token, track_id) + except (httpx.HTTPError, KeyError, IndexError, ValueError, RuntimeError) as exc: + response.status = 502 + return f"Download failed: {html.escape(str(exc))}" + response.content_type = "audio/mpeg" + response.headers["Content-Disposition"] = f'attachment; filename="{filename}"' + response.headers["Content-Length"] = str(len(data)) + return data + + return app + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument("--arl", default=os.environ.get("DEEZER_ARL"), + help="Deezer ARL cookie (defaults to DEEZER_ARL).") + parser.add_argument("--host", default="127.0.0.1", help="Bind address (default: 127.0.0.1).") + parser.add_argument("--port", default=8080, type=int, help="Bind port (default: 8080).") + args = parser.parse_args(argv) + if not args.arl: + parser.error("--arl is required unless DEEZER_ARL is set.") + + try: + client = httpx.Client(headers=HEADERS, cookies={"arl": args.arl, "comeback": "1"}, + follow_redirects=True, timeout=30) + user_data = client.post( + f"{WEB_URL}/ajax/gw-light.php", + params={"api_version": "1.0", "api_token": "null", "input": "3", + "method": "deezer.getUserData"}, + ) + user_data.raise_for_status() + token = user_data.json()["results"]["USER"]["OPTIONS"]["license_token"] + except (httpx.HTTPError, KeyError, IndexError, ValueError, RuntimeError, json.JSONDecodeError) as exc: + parser.error(str(exc)) + + app = create_app(client, token) + browser_host = "127.0.0.1" if args.host in ("0.0.0.0", "::") else args.host + browser_url = f"http://{browser_host}:{args.port}/" + print(f"Serving {browser_url}") + threading.Timer(0.5, webbrowser.open, args=(browser_url,)).start() + app.run(host=args.host, port=args.port) + + +if __name__ == "__main__": + main()