fix: replace spotdl CLI with direct curl_cffi Chrome TLS impersonation
spotdl save hangs for 60s initializing its full pipeline. Rewrite spotify_resolve.py to use curl_cffi (already installed) which impersonates Chrome's TLS fingerprint directly. TOTP token fetched in Python too. RATE_LIMITED propagates without TOTP fallback. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
HoneyandClaude Sonnet 4.6 committed
1 parent
b53910ed45
commit
8bf481a2da
2 files changed
+163
-99
No files matched your search
+160
-98
@@ -1,109 +1,171 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Resolves a Spotify URL to track metadata using spotdl (TzurSoffer fork).
|
||||
Usage: python3 spotify_resolve.py <spotify_url>
|
||||
Outputs JSON to stdout. Errors go to stderr for clean separation.
|
||||
Resolves Spotify URLs using curl_cffi Chrome TLS impersonation + TOTP token.
|
||||
No spotdl pipeline, no API credentials needed.
|
||||
"""
|
||||
import sys
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import subprocess
|
||||
import sys, json, re, hmac, hashlib, struct, time
|
||||
|
||||
def eprint(*args):
|
||||
print(*args, file=sys.stderr)
|
||||
def eprint(*a): print(*a, file=sys.stderr)
|
||||
|
||||
def _totp(secret_str, server_ts):
|
||||
secret_bytes = [ord(c) for c in secret_str]
|
||||
transformed = [b ^ ((i % 33) + 9) for i, b in enumerate(secret_bytes)]
|
||||
# Mirrors Node.js: join as number string → utf8 bytes → hex → parse hex as key
|
||||
nums_str = ''.join(str(b) for b in transformed)
|
||||
key = bytes.fromhex(nums_str.encode('utf-8').hex())
|
||||
counter_b = struct.pack('>Q', server_ts // 30)
|
||||
digest = hmac.new(key, counter_b, hashlib.sha1).digest()
|
||||
offset = digest[-1] & 0xf
|
||||
code = ((digest[offset] & 0x7f) << 24 | (digest[offset+1] & 0xff) << 16 |
|
||||
(digest[offset+2] & 0xff) << 8 | (digest[offset+3] & 0xff)) % 1_000_000
|
||||
return str(code).zfill(6)
|
||||
|
||||
def get_token(session):
|
||||
html = session.get('https://open.spotify.com/', headers={
|
||||
'Accept': 'text/html,application/xhtml+xml,*/*;q=0.8',
|
||||
'Accept-Language': 'en-US,en;q=0.9',
|
||||
'Accept-Encoding': 'identity',
|
||||
}, timeout=20).text
|
||||
|
||||
js_urls, seen = [], set()
|
||||
for m in re.finditer(r'src="((?:https?://[^"]*|/[^"]*?)\.js(?:\?[^"]*)?)"', html):
|
||||
u = m.group(1)
|
||||
if not u.startswith('http'):
|
||||
u = 'https://open.spotify.com' + u
|
||||
if u not in seen:
|
||||
seen.add(u); js_urls.append(u)
|
||||
|
||||
secret = version = None
|
||||
for u in js_urls[:15]:
|
||||
try:
|
||||
js = session.get(u, headers={'Accept': '*/*', 'Accept-Encoding': 'identity'}, timeout=8).text
|
||||
m = re.search(r'\{secret:\s*["\']([^"\']+)["\']\s*,\s*version:\s*(\d+)\s*\}', js)
|
||||
if m:
|
||||
secret, version = m.group(1), int(m.group(2))
|
||||
eprint(f'[spotify] TOTP secret v{version} found')
|
||||
break
|
||||
except Exception as e:
|
||||
eprint(f'[spotify] bundle fetch error: {e}')
|
||||
if not secret:
|
||||
raise Exception('TOTP secret not found in JS bundles')
|
||||
|
||||
st = session.get('https://open.spotify.com/api/server-time', headers={
|
||||
'Accept': 'application/json', 'Referer': 'https://open.spotify.com/',
|
||||
}, timeout=10).json()
|
||||
server_time = st['serverTime']
|
||||
|
||||
totp = _totp(secret, server_time)
|
||||
token_res = session.get(
|
||||
f'https://open.spotify.com/api/token?reason=init&productType=web-player&totp={totp}&totpVer={version}&totpServer={totp}',
|
||||
headers={'Accept': 'application/json', 'Referer': 'https://open.spotify.com/'},
|
||||
timeout=10
|
||||
).json()
|
||||
|
||||
if 'accessToken' not in token_res:
|
||||
raise Exception(f'No accessToken: {json.dumps(token_res)[:200]}')
|
||||
eprint('[spotify] token ok (Chrome/TOTP)')
|
||||
return token_res['accessToken']
|
||||
|
||||
def sp_get(session, path, token):
|
||||
url = path if path.startswith('http') else f'https://api.spotify.com/v1{path}'
|
||||
r = session.get(url, headers={
|
||||
'Authorization': f'Bearer {token}',
|
||||
'app-platform': 'WebPlayer',
|
||||
'spotify-app-version': '1.2.46.25.g7f189073',
|
||||
'Origin': 'https://open.spotify.com',
|
||||
'Referer': 'https://open.spotify.com/',
|
||||
'Accept': 'application/json',
|
||||
'Accept-Language': 'en-US,en;q=0.9',
|
||||
}, timeout=15)
|
||||
if r.status_code == 429:
|
||||
retry = int(r.headers.get('Retry-After', '30'))
|
||||
eprint(f'[spotify] 429 Retry-After={retry}s')
|
||||
print(json.dumps({'error': 'RATE_LIMITED', 'retryAfter': retry}))
|
||||
sys.exit(1)
|
||||
if not r.ok:
|
||||
raise Exception(f'Spotify API {r.status_code}: {path}')
|
||||
return r.json()
|
||||
|
||||
def best_art(images):
|
||||
if not images: return ''
|
||||
return sorted(images, key=lambda x: -(x.get('width') or 0))[0].get('url', '')
|
||||
|
||||
def fmt_track(t, default_album='', default_art='', default_artist=''):
|
||||
artists = t.get('artists') or []
|
||||
artist_str = ', '.join(a.get('name', '') for a in artists) or default_artist
|
||||
alb = t.get('album') or {}
|
||||
art = best_art(alb.get('images') or []) or default_art
|
||||
return {
|
||||
'trackName': t.get('name', ''),
|
||||
'artistName': artist_str,
|
||||
'collectionName': alb.get('name', '') or default_album,
|
||||
'artworkUrl100': art,
|
||||
'releaseDate': alb.get('release_date', ''),
|
||||
'trackNumber': t.get('track_number'),
|
||||
'trackExplicitness': 'explicit' if t.get('explicit') else 'notExplicit',
|
||||
'durationMs': t.get('duration_ms'),
|
||||
}
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
eprint("No URL provided")
|
||||
sys.exit(1)
|
||||
eprint('No URL'); sys.exit(1)
|
||||
|
||||
url = sys.argv[1]
|
||||
tmp = tempfile.mktemp(suffix=".spotdl")
|
||||
m = re.search(r'open\.spotify\.com/(?:.*/)?(track|album|playlist)/([A-Za-z0-9]+)', url)
|
||||
if not m:
|
||||
eprint('Invalid Spotify URL'); sys.exit(1)
|
||||
|
||||
sp_type, sp_id = m.group(1), m.group(2)
|
||||
eprint(f'[spotify] resolving {sp_type} {sp_id}')
|
||||
|
||||
from curl_cffi import requests as cf
|
||||
session = cf.Session(impersonate='chrome124')
|
||||
|
||||
token = get_token(session)
|
||||
|
||||
if sp_type == 'track':
|
||||
t = sp_get(session, f'/tracks/{sp_id}', token)
|
||||
artist = ', '.join(a['name'] for a in (t.get('artists') or []))
|
||||
art = best_art((t.get('album') or {}).get('images') or [])
|
||||
print(json.dumps({'type': 'track', 'name': t.get('name', ''), 'artist': artist,
|
||||
'artworkUrl': art, 'tracks': [fmt_track(t)]}))
|
||||
|
||||
elif sp_type == 'album':
|
||||
alb = sp_get(session, f'/albums/{sp_id}', token)
|
||||
artist = ', '.join(a['name'] for a in (alb.get('artists') or []))
|
||||
art = best_art(alb.get('images') or [])
|
||||
tracks, path = [], f'/albums/{sp_id}/tracks?limit=50'
|
||||
while path:
|
||||
page = sp_get(session, path, token)
|
||||
tracks += [fmt_track(t, alb.get('name', ''), art, artist)
|
||||
for t in (page.get('items') or []) if t and t.get('id')]
|
||||
path = page.get('next')
|
||||
if path: time.sleep(0.3)
|
||||
print(json.dumps({'type': 'album', 'name': alb.get('name', ''), 'artist': artist,
|
||||
'artworkUrl': art, 'tracks': tracks}))
|
||||
|
||||
elif sp_type == 'playlist':
|
||||
pl = sp_get(session, f'/playlists/{sp_id}?fields=name,description,owner,images', token)
|
||||
owner = (pl.get('owner') or {}).get('display_name', '')
|
||||
art = best_art(pl.get('images') or [])
|
||||
tracks, skipped, path = [], 0, f'/playlists/{sp_id}/tracks?limit=100'
|
||||
while path:
|
||||
page = sp_get(session, path, token)
|
||||
for item in (page.get('items') or []):
|
||||
t = item.get('track')
|
||||
if not t or not t.get('id'): continue
|
||||
if t.get('is_local'): skipped += 1; continue
|
||||
tracks.append(fmt_track(t))
|
||||
path = page.get('next')
|
||||
if path: time.sleep(0.3)
|
||||
print(json.dumps({'type': 'playlist', 'name': pl.get('name', ''), 'artist': owner,
|
||||
'artworkUrl': art, 'skippedLocal': skipped, 'tracks': tracks}))
|
||||
|
||||
if __name__ == '__main__':
|
||||
try:
|
||||
eprint(f"[spotdl] running: spotdl save {url}")
|
||||
result = subprocess.run(
|
||||
["spotdl", "save", url, "--save-file", tmp, "--log-level", "ERROR"],
|
||||
capture_output=True, text=True, timeout=60
|
||||
)
|
||||
eprint(f"[spotdl] exit={result.returncode}")
|
||||
if result.stdout.strip():
|
||||
eprint(f"[spotdl] stdout: {result.stdout.strip()[:300]}")
|
||||
if result.stderr.strip():
|
||||
eprint(f"[spotdl] stderr: {result.stderr.strip()[:300]}")
|
||||
|
||||
if result.returncode != 0:
|
||||
err = result.stderr.strip() or result.stdout.strip() or "spotdl exited non-zero"
|
||||
eprint(f"[spotdl] FAILED: {err}")
|
||||
sys.exit(1)
|
||||
|
||||
if not os.path.exists(tmp):
|
||||
eprint("[spotdl] no output file produced")
|
||||
sys.exit(1)
|
||||
|
||||
with open(tmp, "r", encoding="utf-8") as f:
|
||||
songs = json.load(f)
|
||||
|
||||
if not songs:
|
||||
eprint("[spotdl] empty track list")
|
||||
sys.exit(1)
|
||||
|
||||
eprint(f"[spotdl] got {len(songs)} tracks")
|
||||
|
||||
def fmt_track(s):
|
||||
artists = s.get("artists") or []
|
||||
artist_str = ", ".join(artists) if artists else s.get("artist", "")
|
||||
date = s.get("date") or (str(s["year"]) if s.get("year") else "")
|
||||
return {
|
||||
"trackName": s.get("name", ""),
|
||||
"artistName": artist_str,
|
||||
"collectionName": s.get("album_name", ""),
|
||||
"artworkUrl100": s.get("cover_url", ""),
|
||||
"releaseDate": date,
|
||||
"trackNumber": s.get("track_number"),
|
||||
"trackExplicitness": "explicit" if s.get("explicit") else "notExplicit",
|
||||
"durationMs": int(s["duration"] * 1000) if s.get("duration") else None,
|
||||
}
|
||||
|
||||
tracks = [fmt_track(s) for s in songs]
|
||||
first = songs[0]
|
||||
artists0 = first.get("artists") or []
|
||||
artist0 = ", ".join(artists0) if artists0 else first.get("artist", "")
|
||||
|
||||
if "/track/" in url:
|
||||
out = {
|
||||
"type": "track",
|
||||
"name": first.get("name", ""),
|
||||
"artist": artist0,
|
||||
"artworkUrl": first.get("cover_url", ""),
|
||||
"tracks": tracks,
|
||||
}
|
||||
elif "/album/" in url:
|
||||
out = {
|
||||
"type": "album",
|
||||
"name": first.get("album_name", ""),
|
||||
"artist": first.get("album_artist", artist0),
|
||||
"artworkUrl": first.get("cover_url", ""),
|
||||
"tracks": tracks,
|
||||
}
|
||||
else:
|
||||
out = {
|
||||
"type": "playlist",
|
||||
"name": first.get("list_name", ""),
|
||||
"artist": first.get("album_artist", ""),
|
||||
"artworkUrl": first.get("cover_url", ""),
|
||||
"skippedLocal": 0,
|
||||
"tracks": tracks,
|
||||
}
|
||||
|
||||
print(json.dumps(out, ensure_ascii=False))
|
||||
|
||||
finally:
|
||||
try:
|
||||
os.unlink(tmp)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
main()
|
||||
except SystemExit:
|
||||
raise
|
||||
except Exception as e:
|
||||
eprint(f'[spotify] fatal: {e}')
|
||||
sys.exit(1)
|
||||
Reference in new issue
Block a user