diff --git a/backend/spotify_resolve.py b/backend/spotify_resolve.py index cc51a85..84cfdcf 100644 --- a/backend/spotify_resolve.py +++ b/backend/spotify_resolve.py @@ -22,49 +22,86 @@ 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_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,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.9', - 'Accept-Encoding': 'identity', - }, timeout=20).text - - js_urls, seen = [], set() +def _collect_js_urls(html, seen): + urls = [] 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 u not in seen: seen.add(u); urls.append(u) + return urls +def _find_gql_hash(js, op): + """Try multiple patterns to find the persisted-query hash for an operation.""" + for p in [ + rf'operationName:"{op}"[^{{}}]{{0,1000}}sha256Hash:"([a-f0-9]{{64}})"', + rf'sha256Hash:"([a-f0-9]{{64}})"[^{{}}]{{0,500}}operationName:"{op}"', + rf'"{op}"[^;]{{0,2000}}sha256Hash:"([a-f0-9]{{64}})"', + ]: + m = re.search(p, js) + if m: return m.group(1) + # Fallback: any 64-hex string within 3000 chars of the op name + pos = js.find(f'"{op}"') + if pos >= 0: + window = js[max(0, pos - 3000):pos + 3000] + m = re.search(r'sha256Hash:"([a-f0-9]{64})"', window) + if m: return m.group(1) + return None + +def _scan_bundles(session, js_urls, need_totp, need_ops): + """Scan JS bundles for TOTP secret and/or GQL hashes.""" 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): + gql_hashes = {} + for u in js_urls: + if (not need_totp or totp_secret) and all(op in gql_hashes for op in need_ops): break try: js = session.get(u, headers={'Accept': '*/*', 'Accept-Encoding': 'identity'}, timeout=8).text - # TOTP secret - if not totp_secret: + if need_totp and 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: + for op in need_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]}...') + h = _find_gql_hash(js, op) + if h: + gql_hashes[op] = h + eprint(f'[spotify] GQL hash {op}: {h[:16]}...') except Exception as e: - eprint(f'[spotify] bundle error: {e}') + eprint(f'[spotify] bundle error ({u[-40:]}): {e}') + return totp_secret, totp_version, gql_hashes + +def get_token_and_hashes(session, sp_type, sp_id): + """Fetch TOTP token + extract GQL hashes. + Scans homepage bundles for TOTP, then the specific resource page for GQL hashes + (Spotify lazy-loads feature chunks only on the relevant page).""" + page_headers = { + 'Accept': 'text/html,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.9', + 'Accept-Encoding': 'identity', + } + GQL_OP = {'track': 'getTrack', 'album': 'getAlbum', 'playlist': 'fetchPlaylist'}[sp_type] + + # Step 1: homepage — find TOTP secret + seen = set() + home_html = session.get('https://open.spotify.com/', headers=page_headers, timeout=20).text + home_js = _collect_js_urls(home_html, seen) + totp_secret, totp_version, gql_hashes = _scan_bundles(session, home_js[:20], True, [GQL_OP]) if not totp_secret: raise Exception('TOTP secret not found') + # Step 2: resource page — find GQL hash (lazy-loaded playlist/album/track chunk) + if GQL_OP not in gql_hashes: + eprint(f'[spotify] fetching {sp_type} page for GQL hash...') + try: + page_url = f'https://open.spotify.com/{sp_type}/{sp_id}' + page_html = session.get(page_url, headers=page_headers, timeout=15).text + extra_js = _collect_js_urls(page_html, seen) + _, _, extra_hashes = _scan_bundles(session, extra_js[:25], False, [GQL_OP]) + gql_hashes.update(extra_hashes) + except Exception as e: + eprint(f'[spotify] page scan error: {e}') + st = session.get('https://open.spotify.com/api/server-time', headers={ 'Accept': 'application/json', 'Referer': 'https://open.spotify.com/', }, timeout=10).json() @@ -295,7 +332,7 @@ def main(): from curl_cffi import requests as cf session = cf.Session(impersonate='chrome124') - token, gql_hashes = get_token_and_hashes(session) + token, gql_hashes = get_token_and_hashes(session, sp_type, sp_id) GQL_OP = {'track': 'getTrack', 'album': 'getAlbum', 'playlist': 'fetchPlaylist'}[sp_type] gql_hash = gql_hashes.get(GQL_OP)