diff --git a/backend/spotify_resolve.py b/backend/spotify_resolve.py index 43fd0fa..58342c4 100644 --- a/backend/spotify_resolve.py +++ b/backend/spotify_resolve.py @@ -1,18 +1,15 @@ #!/usr/bin/env python3 """ -Resolves Spotify URLs. -- SPOTIFY_PROXY set: curl_cffi + TOTP + REST API routed through proxy (bypasses IP ban) -- No proxy: SpotipyFree (SpotAPI + tls_client, internal spclient endpoints) +Resolves Spotify URLs via SpotipyFree (SpotAPI + tls_client Chrome fingerprint, +hits Spotify's internal spclient endpoints — not the rate-limited public REST API). """ -import sys, json, re, hmac, hashlib, struct, time, os, urllib.parse - -PROXY = os.environ.get('SPOTIFY_PROXY', '').strip() +import sys, json, re, time def eprint(*a): print(*a, file=sys.stderr) -# ─── FORMAT HELPERS ─────────────────────────────────────────────────────────── +# ─── APPROACH 1: SpotipyFree ───────────────────────────────────────────────── -def _fmt_track(t, default_album='', default_art='', default_artist=''): +def _fmt_spotipy_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 {} @@ -29,115 +26,6 @@ def _fmt_track(t, default_album='', default_art='', default_artist=''): 'durationMs': t.get('duration_ms'), } -def _best_art(imgs): - return sorted(imgs, key=lambda x: -(x.get('width') or 0))[0].get('url','') if imgs else '' - -# ─── APPROACH 1: curl_cffi + TOTP (used when SPOTIFY_PROXY is set) ─────────── - -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) - -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 _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'} - -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() - -def resolve_via_curl(sp_type, sp_id, proxy): - from curl_cffi import requests as cf - proxies = {'https': proxy, 'http': proxy} if proxy else {} - session = cf.Session(impersonate='chrome124', proxies=proxies) - if proxy: eprint(f'[curl_cffi] using proxy') - token = _get_token(session) - - 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)]} - - 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'); time.sleep(0.3) if path else None - eprint(f'[curl_cffi] album ok — {len(tracks)} tracks') - return {'type':'album','name':alb.get('name',''),'artist':artist,'artworkUrl':art,'tracks':tracks} - - 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) - if tr: tracks.append(tr) - path = pg.get('next'); time.sleep(0.3) if path else None - eprint(f'[curl_cffi] playlist ok — {len(tracks)} tracks') - return {'type':'playlist','name':pl.get('name',''),'artist':owner, - 'artworkUrl':art,'skippedLocal':skipped,'tracks':tracks} - -# ─── APPROACH 2: SpotipyFree (used when no proxy, non-banned IPs) ───────────── - def resolve_via_spotipyfree(sp_type, sp_id): from SpotipyFree import Spotify as Sp sp = Sp() @@ -148,8 +36,8 @@ def resolve_via_spotipyfree(sp_type, sp_id): artist = ', '.join(a['name'] for a in (t.get('artists') or [])) imgs = (t.get('album') or {}).get('images') or [] art = sorted(imgs, key=lambda x: -(x.get('width') or 0))[0].get('url','') if imgs else '' - return {'type':'track','name':t.get('name',''),'artist':artist,'artworkUrl':art, - 'tracks':[_fmt_track(t)]} + tr = _fmt_spotipy_track(t) + return {'type':'track','name':t.get('name',''),'artist':artist,'artworkUrl':art,'tracks':[tr]} if sp_type == 'album': alb = sp.album(sp_id) @@ -160,7 +48,7 @@ def resolve_via_spotipyfree(sp_type, sp_id): while True: page = sp.album_tracks(sp_id, limit=50, offset=offset) for t in (page.get('items') or []): - tr = _fmt_track(t, alb.get('name',''), art, artist) + tr = _fmt_spotipy_track(t, alb.get('name',''), art, artist) if tr: tracks.append(tr) if not page.get('next'): break offset += len(page.get('items') or []) @@ -181,7 +69,7 @@ def resolve_via_spotipyfree(sp_type, sp_id): 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) + tr = _fmt_spotipy_track(t) if tr: tracks.append(tr) if not page.get('next'): break offset += len(page.get('items') or []) @@ -200,21 +88,12 @@ def main(): sp_type, sp_id = m.group(1), m.group(2) eprint(f'[spotify] resolving {sp_type} {sp_id}') - if PROXY: - # Proxy set: route all Spotify API calls through it (bypasses server IP ban) - eprint(f'[spotify] proxy mode') - result = resolve_via_curl(sp_type, sp_id, PROXY) - print(json.dumps(result, ensure_ascii=False)) - return - - # No proxy: try SpotipyFree (uses internal Spotify endpoints, not the public REST API) try: result = resolve_via_spotipyfree(sp_type, sp_id) print(json.dumps(result, ensure_ascii=False)) except SystemExit: raise except Exception as e: eprint(f'[SpotipyFree] failed: {type(e).__name__}: {e}') - eprint('[spotify] tip: set SPOTIFY_PROXY=socks5://host:port or http://host:port to bypass IP restrictions') sys.exit(1) if __name__ == '__main__': diff --git a/docker-compose.yml b/docker-compose.yml index cebd740..8106b7b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -30,7 +30,6 @@ services: - MAX_PARALLEL=${MAX_PARALLEL:-2} - DOWNLOAD_BACKEND=${DOWNLOAD_BACKEND:-auto} - SQUID_QUALITY=${SQUID_QUALITY:-mp3} - - SPOTIFY_PROXY=${SPOTIFY_PROXY:-} command: > sh -c " apt-get update && apt-get install -y --no-install-recommends python3 python3-pip ffmpeg curl git && rm -rf /var/lib/apt/lists/* &&