From 4ec441af57907188adbc2797259faca563ce113b Mon Sep 17 00:00:00 2001 From: Honey Date: Tue, 12 May 2026 21:02:28 +0200 Subject: [PATCH] fix: switch to spotapi (api-partner.spotify.com) to avoid rate limits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit api.spotify.com/v1/ aggressively rate-limits server IPs regardless of TLS fingerprint or token. SpotipyFree was already hitting the correct endpoint (api-partner.spotify.com) but crashed on KeyError: 'ownerV2' due to its own outdated field parsing — not a network block. spotapi uses the same internal GraphQL API (Spotify's web player API) which is on a separate rate-limit bucket from the public REST endpoint. Co-Authored-By: Claude Sonnet 4.6 --- backend/spotify_resolve.py | 226 ++++++++++++++++++------------------- docker-compose.yml | 2 +- 2 files changed, 114 insertions(+), 114 deletions(-) diff --git a/backend/spotify_resolve.py b/backend/spotify_resolve.py index ea3bb41..70c4a33 100644 --- a/backend/spotify_resolve.py +++ b/backend/spotify_resolve.py @@ -1,136 +1,136 @@ #!/usr/bin/env python3 """ -Resolves Spotify URLs via curl_cffi Chrome TLS impersonation + TOTP anonymous token. -curl_cffi presents a real Chrome TLS fingerprint, which gets past Spotify's bot filters -on open.spotify.com/api/server-time (plain Node.js/Python TLS does not). +Resolves Spotify URLs via the Spotify partner GraphQL API (api-partner.spotify.com). +This is the same internal API the Spotify web player uses — separate rate-limit bucket +from the public api.spotify.com/v1/ which gets throttled on server IPs. """ -import sys, json, re, hmac, hashlib, struct, time +import sys, json, re, time 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)] - 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 - 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) +# ─── RESPONSE PARSING ──────────────────────────────────────────────────────── +# Partner API track items look like: +# { "itemV2": { "data": { "trackUnion": { "name":…, "artists":…, "albumOfTrack":…, +# "duration":…, "trackNumber":…, "contentRating":… } } } } -def _get_token(session): - html = session.get('https://open.spotify.com/', headers={ - '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) - secret = version = None - for u in js_urls[:20]: - if secret: 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)) - except: pass - if not secret: raise Exception('TOTP secret not found') - eprint(f'[TOTP] secret v{version} found') - st = session.get('https://open.spotify.com/api/server-time', headers={ - 'Accept': 'application/json', 'Referer': 'https://open.spotify.com/'}, timeout=10).json() - totp = _totp(secret, st['serverTime']) - tok = 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 tok: raise Exception(f'No accessToken: {json.dumps(tok)[:100]}') - eprint('[TOTP] token ok') - return tok['accessToken'] +def _parse_track(item, default_album='', default_art='', default_artist=''): + try: + inner = (item.get('itemV2') or item.get('track') or {}).get('data') or {} + t = inner.get('trackUnion') or inner + if not t.get('name'): return None -def _auth_h(token): - return {'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'} + artist_items = (t.get('artists') or {}).get('items') or [] + artists = ', '.join( + a.get('profile', {}).get('name', '') for a in artist_items + ) or default_artist -def _rest(session, path, token): - url = path if path.startswith('http') else f'https://api.spotify.com/v1{path}' - r = session.get(url, headers=_auth_h(token), timeout=15) - if r.status_code == 429: - retry = int(r.headers.get('Retry-After', '30')) - eprint(f'[REST] 429 Retry-After={retry}s') - print(json.dumps({'error': 'RATE_LIMITED', 'retryAfter': retry})) - sys.exit(1) - if not r.ok: raise Exception(f'REST {r.status_code}: {path}') - return r.json() + alb = t.get('albumOfTrack') or {} + alb_name = alb.get('name', '') or default_album + alb_imgs = (alb.get('images') or {}).get('items') or [] + art = alb_imgs[0].get('url', '') if alb_imgs else default_art + release = ((alb.get('date') or {}).get('isoString') or '')[:10] -def _fmt_track(t, default_album='', default_art='', default_artist=''): - if not t or not t.get('name'): return None - artists = ', '.join(a.get('name', '') for a in (t.get('artists') or [])) or default_artist - alb = t.get('album') or {} - imgs = alb.get('images') or [] - art = sorted(imgs, key=lambda x: -(x.get('width') or 0))[0].get('url', '') if imgs else default_art - return { - 'trackName': t.get('name', ''), - 'artistName': artists, - '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'), - } + dur_ms = (t.get('duration') or {}).get('totalMilliseconds') + rating = (t.get('contentRating') or {}).get('label', '') -def _best_art(imgs): - return sorted(imgs, key=lambda x: -(x.get('width') or 0))[0].get('url', '') if imgs else '' + return { + 'trackName': t.get('name', ''), + 'artistName': artists, + 'collectionName': alb_name, + 'artworkUrl100': art, + 'releaseDate': release, + 'trackNumber': t.get('trackNumber'), + 'trackExplicitness': 'explicit' if rating == 'EXPLICIT' else 'notExplicit', + 'durationMs': dur_ms, + } + except Exception as e: + eprint(f'[parse] error on item: {e}') + return None + +def _parse_images(imgs_obj): + items = (imgs_obj or {}).get('items') or [] + return items[0].get('url', '') if items else '' + +# ─── RESOLVER ──────────────────────────────────────────────────────────────── def resolve(sp_type, sp_id): - from curl_cffi import requests as cf - session = cf.Session(impersonate='chrome124') - token = _get_token(session) + from spotapi import PublicPlaylist, PublicAlbum, PublicSong + + url = f'https://open.spotify.com/{sp_type}/{sp_id}' if sp_type == 'track': - t = _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_track(t)]} + song = PublicSong(url) + raw = song.get_song_info() + eprint(f'[spotapi] track response keys: {list(raw.keys()) if isinstance(raw, dict) else type(raw)}') + tr = _parse_track(raw) + if not tr: + # Fallback: raw might already be a flat dict + tr = { + 'trackName': raw.get('name', ''), + 'artistName': raw.get('artist', ''), + 'collectionName': raw.get('album', ''), + 'artworkUrl100': raw.get('image', ''), + 'releaseDate': raw.get('release_date', ''), + 'trackNumber': raw.get('track_number'), + 'trackExplicitness': 'notExplicit', + 'durationMs': raw.get('duration_ms'), + } + name = tr.get('trackName', '') + artist = tr.get('artistName', '') + art = tr.get('artworkUrl100', '') + return {'type': 'track', 'name': name, 'artist': artist, 'artworkUrl': art, 'tracks': [tr]} if sp_type == 'album': - alb = _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 = _rest(session, path, token) - tracks += [t for t in [_fmt_track(i, alb.get('name', ''), art, artist) - for i in (pg.get('items') or []) if i and i.get('id')] if t] - path = pg.get('next') - if path: time.sleep(0.3) - eprint(f'[curl_cffi] album ok — {len(tracks)} tracks') - return {'type': 'album', 'name': alb.get('name', ''), 'artist': artist, - 'artworkUrl': art, 'tracks': tracks} + alb = PublicAlbum(url) + alb_info = alb.get_album_info() + eprint(f'[spotapi] album response type: {type(alb_info)}, keys: {list(alb_info.keys()) if isinstance(alb_info, dict) else "?"}') - pl = _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 = _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 - tr = _fmt_track(t) + # alb_info may be a dict with metadata + items, or just items + meta = alb_info if isinstance(alb_info, dict) else {} + name = meta.get('name', '') + artist = meta.get('artist', '') or meta.get('artists', '') + art = meta.get('image', '') or meta.get('cover', '') + + raw_tracks = meta.get('tracks') or meta.get('items') or (alb_info if isinstance(alb_info, list) else []) + tracks = [t for t in (_parse_track(i) for i in raw_tracks) if t] + eprint(f'[spotapi] album ok — {len(tracks)} tracks') + return {'type': 'album', 'name': name, 'artist': artist, 'artworkUrl': art, 'tracks': tracks} + + # playlist + pl = PublicPlaylist(url) + pl_meta = pl.get_playlist_info(limit=1) # first call often returns metadata too + eprint(f'[spotapi] playlist meta type: {type(pl_meta)}') + + tracks, skipped = [], 0 + offset = 0 + while True: + page = pl.get_playlist_info(limit=100, offset=offset) + items = page if isinstance(page, list) else (page.get('items') or page.get('tracks') or []) + if not items: break + for item in items: + if (item.get('itemV2') or item.get('track', {}).get('data', {})).get('__typename') == 'LocalTrackResponseWrapper': + skipped += 1; continue + tr = _parse_track(item) if tr: tracks.append(tr) - path = pg.get('next') - if path: time.sleep(0.3) - eprint(f'[curl_cffi] playlist ok — {len(tracks)} tracks') - return {'type': 'playlist', 'name': pl.get('name', ''), 'artist': owner, + if len(items) < 100: break + offset += len(items) + time.sleep(0.3) + + # Extract playlist name/art from first page if available + name = '' + owner = '' + art = '' + if isinstance(pl_meta, dict): + name = pl_meta.get('name', '') + owner = pl_meta.get('owner', '') or pl_meta.get('owner_name', '') + art = pl_meta.get('image', '') or pl_meta.get('cover', '') + + eprint(f'[spotapi] playlist ok — {len(tracks)} tracks') + return {'type': 'playlist', 'name': 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] @@ -143,7 +143,7 @@ def main(): print(json.dumps(result, ensure_ascii=False)) except SystemExit: raise except Exception as e: - eprint(f'[curl_cffi] failed: {type(e).__name__}: {e}') + eprint(f'[spotapi] failed: {type(e).__name__}: {e}') sys.exit(1) if __name__ == '__main__': diff --git a/docker-compose.yml b/docker-compose.yml index 8106b7b..6895021 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -34,7 +34,7 @@ services: sh -c " apt-get update && apt-get install -y --no-install-recommends python3 python3-pip ffmpeg curl git && rm -rf /var/lib/apt/lists/* && curl -sL https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp -o /usr/local/bin/yt-dlp && chmod a+rx /usr/local/bin/yt-dlp && - pip install --break-system-packages SpotipyFree 'git+https://github.com/TzurSoffer/spotify-downloader' && + pip install --break-system-packages spotapi 'git+https://github.com/TzurSoffer/spotify-downloader' && mkdir -p /deps && cd /deps && npm install express && cd /app && NODE_PATH=/deps/node_modules node novasonic-server.js "