feat: Add minimal Deezer search + download script
This commit is contained in:
abdus committed
1 parent
f4764699e8
commit
6a1ce2ad18
1 file changed
+282
Executable
+282
@@ -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(
|
||||
"<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()
|
||||
Reference in new issue
Block a user