diff --git a/backend/spotify_resolve.py b/backend/spotify_resolve.py index 62bdf32..cc51a85 100644 --- a/backend/spotify_resolve.py +++ b/backend/spotify_resolve.py @@ -1,18 +1,20 @@ #!/usr/bin/env python3 """ -Resolves Spotify URLs using curl_cffi Chrome TLS impersonation + TOTP token. -No spotdl pipeline, no API credentials needed. +Resolves Spotify URLs using: +1. api-partner.spotify.com (GraphQL, internal web-player endpoint, separate rate limits) +2. api.spotify.com/v1 fallback (REST, rate-limited by IP) +Both use TOTP anonymous token + curl_cffi Chrome TLS impersonation. """ -import sys, json, re, hmac, hashlib, struct, time +import sys, json, re, hmac, hashlib, struct, time, urllib.parse def eprint(*a): print(*a, file=sys.stderr) +# ─── TOTP ──────────────────────────────────────────────────────────────────── + 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()) + key = bytes.fromhex(''.join(str(b) for b in transformed).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 @@ -20,55 +22,68 @@ def _totp(secret_str, server_ts): (digest[offset+2] & 0xff) << 8 | (digest[offset+3] & 0xff)) % 1_000_000 return str(code).zfill(6) -def get_token(session): +def get_token_and_hashes(session): + """Fetch TOTP token + extract GraphQL persisted-query hashes from JS bundles.""" 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': 'text/html,*/*;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) + 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]: + totp_secret = totp_version = None + gql_hashes = {} # operationName -> sha256Hash + GQL_OPS = ['fetchPlaylist', 'getAlbum', 'getTrack'] + + for u in js_urls[:20]: + if totp_secret and all(op in gql_hashes for op in GQL_OPS): + break 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 + # TOTP secret + if not totp_secret: + m = re.search(r'\{secret:\s*["\']([^"\']+)["\']\s*,\s*version:\s*(\d+)\s*\}', js) + if m: + totp_secret, totp_version = m.group(1), int(m.group(2)) + eprint(f'[spotify] TOTP secret v{totp_version} found') + # GraphQL hashes + for op in GQL_OPS: + if op in gql_hashes: continue + m = re.search(rf'operationName:"{op}"[^;]{{0,600}}sha256Hash:"([a-f0-9]{{64}})"', js) + if not m: + m = re.search(rf'sha256Hash:"([a-f0-9]{{64}})"[^;]{{0,300}}operationName:"{op}"', js) + if m: + gql_hashes[op] = m.group(1) + eprint(f'[spotify] GQL hash {op}: {m.group(1)[:16]}...') except Exception as e: - eprint(f'[spotify] bundle fetch error: {e}') - if not secret: - raise Exception('TOTP secret not found in JS bundles') + eprint(f'[spotify] bundle error: {e}') + + if not totp_secret: + raise Exception('TOTP secret not found') 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}', + totp = _totp(totp_secret, st['serverTime']) + tok = session.get( + f'https://open.spotify.com/api/token?reason=init&productType=web-player&totp={totp}&totpVer={totp_version}&totpServer={totp}', headers={'Accept': 'application/json', 'Referer': 'https://open.spotify.com/'}, timeout=10 ).json() + if 'accessToken' not in tok: + raise Exception(f'No accessToken: {json.dumps(tok)[:200]}') + eprint('[spotify] token ok') + return tok['accessToken'], gql_hashes - 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'] +# ─── HELPERS ───────────────────────────────────────────────────────────────── -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={ +def _auth_headers(token): + return { 'Authorization': f'Bearer {token}', 'app-platform': 'WebPlayer', 'spotify-app-version': '1.2.46.25.g7f189073', @@ -76,28 +91,157 @@ def sp_get(session, path, token): 'Referer': 'https://open.spotify.com/', 'Accept': 'application/json', 'Accept-Language': 'en-US,en;q=0.9', - }, timeout=15) + } + +def _check_429(r): 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', '') + return sorted(images, key=lambda x: -(x.get('width') or x.get('maxWidth') 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 +# ─── PARTNER API (GraphQL) ─────────────────────────────────────────────────── + +def gql_get(session, token, operation, variables, gql_hash): + var_str = json.dumps(variables, separators=(',', ':')) + ext_str = json.dumps({'persistedQuery': {'version': 1, 'sha256Hash': gql_hash}}, separators=(',', ':')) + url = (f'https://api-partner.spotify.com/pathfinder/v1/query' + f'?operationName={operation}' + f'&variables={urllib.parse.quote(var_str)}' + f'&extensions={urllib.parse.quote(ext_str)}') + r = session.get(url, headers=_auth_headers(token), timeout=15) + _check_429(r) + if not r.ok: + raise Exception(f'Partner API {r.status_code}: {r.text[:200]}') + data = r.json() + if 'errors' in data: + raise Exception(f'GQL errors: {json.dumps(data["errors"])[:200]}') + return data.get('data', {}) + +def _gql_art(obj): + """Extract best artwork URL from GQL image objects (various structures).""" + sources = (obj or {}).get('sources') or (obj or {}).get('items') or [] + if not sources: return '' + return sorted(sources, key=lambda x: -(x.get('width') or x.get('maxWidth') or 0))[0].get('url', '') + +def _gql_artists(items): + return ', '.join( + (i.get('profile') or i.get('data') or {}).get('name', '') or i.get('name', '') + for i in (items or []) + ).strip(', ') + +def fmt_gql_track(item_data, default_album='', default_art='', default_artist=''): + """Parse a track from the GraphQL trackData/itemV2.data structure.""" + artists = _gql_artists((item_data.get('artists') or {}).get('items') or []) or default_artist + alb = item_data.get('albumOfTrack') or {} + art = _gql_art(alb.get('coverArt')) or default_art + date = (alb.get('date') or {}).get('isoString', '') or '' + dur = (item_data.get('duration') or {}).get('totalMilliseconds') + explicit = (item_data.get('contentRating') or {}).get('label', '') == 'EXPLICIT' + return { + 'trackName': item_data.get('name', ''), + 'artistName': artists, + 'collectionName': alb.get('name', '') or default_album, + 'artworkUrl100': art, + 'releaseDate': date, + 'trackNumber': item_data.get('trackNumber'), + 'trackExplicitness': 'explicit' if explicit else 'notExplicit', + 'durationMs': dur, + } + +def resolve_playlist_gql(session, token, sp_id, gql_hash): + tracks, skipped, offset = [], 0, 0 + pl_name = pl_owner = pl_art = None + while True: + d = gql_get(session, token, 'fetchPlaylist', { + 'uri': f'spotify:playlist:{sp_id}', 'offset': offset, 'limit': 100, 'locale': '', + }, gql_hash) + pl = d.get('playlistV2') or {} + if pl_name is None: + pl_name = pl.get('name', '') + pl_owner = ((pl.get('ownerV2') or {}).get('data') or {}).get('name', '') + pl_art = _gql_art(pl.get('images') or {}) + content = pl.get('content') or {} + for item in content.get('items') or []: + iv2 = (item.get('itemV2') or {}) + if iv2.get('__typename') == 'RestrictedTrackResponse': skipped += 1; continue + td = iv2.get('data') or {} + if not td.get('name'): skipped += 1; continue + tracks.append(fmt_gql_track(td)) + paging = content.get('pagingInfo') or {} + next_off = paging.get('nextOffset') + if next_off is None or next_off <= offset: break + offset = next_off + time.sleep(0.3) + return {'type': 'playlist', 'name': pl_name or '', 'artist': pl_owner or '', + 'artworkUrl': pl_art or '', 'skippedLocal': skipped, 'tracks': tracks} + +def resolve_album_gql(session, token, sp_id, gql_hash): + tracks, offset = [], 0 + alb_name = alb_artist = alb_art = None + while True: + d = gql_get(session, token, 'getAlbum', { + 'uri': f'spotify:album:{sp_id}', 'offset': offset, 'limit': 50, 'locale': '', + }, gql_hash) + alb = d.get('albumUnion') or {} + if alb_name is None: + alb_name = alb.get('name', '') + alb_artist = _gql_artists((alb.get('artists') or {}).get('items') or []) + alb_art = _gql_art(alb.get('coverArt') or {}) + disc_items = (alb.get('tracks') or {}).get('items') or [] + for item in disc_items: + td = item.get('track') or item + if not td.get('name'): continue + tracks.append(fmt_gql_track(td, alb_name, alb_art, alb_artist)) + paging = (alb.get('tracks') or {}).get('pagingInfo') or {} + next_off = paging.get('nextOffset') + if next_off is None or next_off <= offset: break + offset = next_off + time.sleep(0.3) + return {'type': 'album', 'name': alb_name or '', 'artist': alb_artist or '', + 'artworkUrl': alb_art or '', 'tracks': tracks} + +def resolve_track_gql(session, token, sp_id, gql_hash): + d = gql_get(session, token, 'getTrack', {'uri': f'spotify:track:{sp_id}'}, gql_hash) + t = d.get('trackUnion') or {} + artist = _gql_artists((t.get('firstArtist') or t.get('artists') or {}).get('items') or []) + alb = t.get('albumOfTrack') or {} + art = _gql_art(alb.get('coverArt') or {}) + date = (alb.get('date') or {}).get('isoString', '') + explicit = (t.get('contentRating') or {}).get('label', '') == 'EXPLICIT' + track = { + 'trackName': t.get('name', ''), + 'artistName': artist, + 'collectionName': alb.get('name', ''), + 'artworkUrl100': art, + 'releaseDate': date, + 'trackNumber': t.get('trackNumber'), + 'trackExplicitness': 'explicit' if explicit else 'notExplicit', + 'durationMs': (t.get('duration') or {}).get('totalMilliseconds'), + } + return {'type': 'track', 'name': t.get('name', ''), 'artist': artist, + 'artworkUrl': art, 'tracks': [track]} + +# ─── REST FALLBACK (api.spotify.com/v1) ───────────────────────────────────── + +def sp_rest(session, path, token): + url = path if path.startswith('http') else f'https://api.spotify.com/v1{path}' + r = session.get(url, headers=_auth_headers(token), timeout=15) + _check_429(r) + if not r.ok: raise Exception(f'REST API {r.status_code}: {path}') + return r.json() + +def fmt_rest_track(t, default_album='', default_art='', default_artist=''): + artists = ', '.join(a.get('name', '') for a in (t.get('artists') or [])) 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, + 'artistName': artists, 'collectionName': alb.get('name', '') or default_album, 'artworkUrl100': art, 'releaseDate': alb.get('release_date', ''), @@ -106,66 +250,80 @@ def fmt_track(t, default_album='', default_art='', default_artist=''): 'durationMs': t.get('duration_ms'), } -def main(): - if len(sys.argv) < 2: - eprint('No URL'); sys.exit(1) +def resolve_rest(session, token, sp_type, sp_id): + if sp_type == 'track': + t = sp_rest(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 []) + return {'type': 'track', 'name': t.get('name', ''), 'artist': artist, + 'artworkUrl': art, 'tracks': [fmt_rest_track(t)]} + if sp_type == 'album': + alb = sp_rest(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: + pg = sp_rest(session, path, token) + tracks += [fmt_rest_track(t, alb.get('name',''), art, artist) for t in (pg.get('items') or []) if t and t.get('id')] + path = pg.get('next'); time.sleep(0.3) if path else None + return {'type': 'album', 'name': alb.get('name',''), 'artist': artist, 'artworkUrl': art, 'tracks': tracks} + if sp_type == 'playlist': + pl = sp_rest(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: + pg = sp_rest(session, path, token) + for item in (pg.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_rest_track(t)) + path = pg.get('next'); time.sleep(0.3) if path else None + return {'type': 'playlist', 'name': pl.get('name',''), 'artist': owner, + 'artworkUrl': art, 'skippedLocal': skipped, 'tracks': tracks} +# ─── MAIN ──────────────────────────────────────────────────────────────────── + +def main(): + if len(sys.argv) < 2: eprint('No URL'); sys.exit(1) url = sys.argv[1] 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) - + 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, gql_hashes = get_token_and_hashes(session) - token = get_token(session) + GQL_OP = {'track': 'getTrack', 'album': 'getAlbum', 'playlist': 'fetchPlaylist'}[sp_type] + gql_hash = gql_hashes.get(GQL_OP) - 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)]})) + result = None + if gql_hash: + eprint(f'[spotify] trying partner API ({GQL_OP})') + try: + if sp_type == 'playlist': result = resolve_playlist_gql(session, token, sp_id, gql_hash) + elif sp_type == 'album': result = resolve_album_gql(session, token, sp_id, gql_hash) + elif sp_type == 'track': result = resolve_track_gql(session, token, sp_id, gql_hash) + eprint(f'[spotify] partner API ok — {len(result.get("tracks", []))} tracks') + except SystemExit: raise + except Exception as e: + eprint(f'[spotify] partner API failed: {e} — falling back to REST') + else: + eprint(f'[spotify] no GQL hash for {GQL_OP}, using REST directly') - 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})) + if result is None: + eprint('[spotify] trying REST API (api.spotify.com/v1)') + result = resolve_rest(session, token, sp_type, sp_id) - 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})) + print(json.dumps(result, ensure_ascii=False)) if __name__ == '__main__': try: main() - except SystemExit: - raise + except SystemExit: raise except Exception as e: eprint(f'[spotify] fatal: {e}') sys.exit(1)