playground/deezer_cli.py

#!/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(
        "<tr>"
        f"<td class=art><img src=\"{html.escape(result.get('album', {}).get('cover_medium', ''), quote=True)}\" alt=\"\" loading=lazy></td>"
        f"<td>{html.escape(result.get('title', ''))}</td>"
        f"<td>{html.escape(result.get('artist', {}).get('name', ''))}</td>"
        f"<td>{html.escape(result.get('album', {}).get('title', ''))}</td>"
        f"<td><a class=button href=/download/{result['id']}>Download</a></td>"
        "</tr>"
        for result in results
    )
    table = (
        "<table><thead><tr><th></th><th>Title</th><th>Artist</th><th>Album</th><th></th></tr></thead>"
        f"<tbody>{rows}</tbody></table>"
        if results else ("<p class=muted>No results.</p>" if query and not error else "")
    )
    error_box = f"<p class=error>{html.escape(error)}</p>" if error else ""
    return f"""<!doctype html>
<html lang=en>
<head>
  <meta charset=utf-8>
  <meta name=viewport content="width=device-width,initial-scale=1">
  <title>Deezer Search</title>
  <style>
    body {{ max-width: 900px; margin: 40px auto; padding: 0 16px; font: 15px system-ui, sans-serif; color: #222; }}
    form {{ display: flex; gap: 8px; margin-bottom: 24px; }}
    input {{ flex: 1; padding: 9px; font: inherit; }}
    button, .button {{ border: 0; border-radius: 4px; padding: 9px 13px; background: #a238ff; color: white; text-decoration: none; cursor: pointer; }}
    table {{ width: 100%; border-collapse: collapse; }}
    th, td {{ border-bottom: 1px solid #ddd; padding: 14px 10px; text-align: left; }}
    th:last-child, td:last-child {{ text-align: right; }}
    .art {{ width: 56px; padding-right: 0; }}
    .art img {{ display: block; width: 56px; height: 56px; object-fit: cover; border-radius: 4px; background: #eee; }}
    .error {{ color: #b00020; }}
    .muted {{ color: #777; }}
  </style>
</head>
<body>
  <h1>Deezer Search</h1>
  <form method=get>
    <input name=q value="{html.escape(query, quote=True)}" placeholder="Artist or track" autofocus>
    <button type=submit>Search</button>
  </form>
  {error_box}
  {table}
</body>
</html>"""


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/<track_id:re:[0-9]+>")
    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()