HoneyandClaude Sonnet 4.6
06b78be199
- Log each Python stderr line separately (no 500-char truncation) so we can see exactly what's happening in the bundle scan - Try embed page (open.spotify.com/embed/...) first for GQL hashes — focused app, fewer bundles, GQL chunks directly referenced - Suppress noisy bundle-fetch errors to keep logs readable Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
373 lines
17 KiB
Python
373 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
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, 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)]
|
|
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 _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); 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 = {}
|
|
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
|
|
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')
|
|
for op in need_ops:
|
|
if op in gql_hashes: continue
|
|
h = _find_gql_hash(js, op)
|
|
if h:
|
|
gql_hashes[op] = h
|
|
eprint(f'[spotify] GQL hash {op}: {h[:16]}...')
|
|
except Exception:
|
|
pass # skip failed bundles silently
|
|
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: try embed page — focused app, GQL chunks often directly referenced
|
|
if GQL_OP not in gql_hashes:
|
|
for extra_url in [
|
|
f'https://open.spotify.com/embed/{sp_type}/{sp_id}?utm_source=generator',
|
|
f'https://open.spotify.com/{sp_type}/{sp_id}',
|
|
]:
|
|
eprint(f'[spotify] scanning {extra_url[:60]} for GQL hash...')
|
|
try:
|
|
page_html = session.get(extra_url, headers=page_headers, timeout=15).text
|
|
extra_js = _collect_js_urls(page_html, seen)
|
|
eprint(f'[spotify] found {len(extra_js)} new JS URLs on that page')
|
|
_, _, extra_hashes = _scan_bundles(session, extra_js[:30], False, [GQL_OP])
|
|
gql_hashes.update(extra_hashes)
|
|
except Exception as e:
|
|
eprint(f'[spotify] page scan error: {e}')
|
|
if GQL_OP in gql_hashes:
|
|
break
|
|
|
|
st = session.get('https://open.spotify.com/api/server-time', headers={
|
|
'Accept': 'application/json', 'Referer': 'https://open.spotify.com/',
|
|
}, timeout=10).json()
|
|
|
|
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
|
|
|
|
# ─── HELPERS ─────────────────────────────────────────────────────────────────
|
|
|
|
def _auth_headers(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 _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)
|
|
|
|
def best_art(images):
|
|
if not images: return ''
|
|
return sorted(images, key=lambda x: -(x.get('width') or x.get('maxWidth') or 0))[0].get('url', '')
|
|
|
|
# ─── 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': 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'),
|
|
}
|
|
|
|
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)
|
|
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, sp_type, sp_id)
|
|
|
|
GQL_OP = {'track': 'getTrack', 'album': 'getAlbum', 'playlist': 'fetchPlaylist'}[sp_type]
|
|
gql_hash = gql_hashes.get(GQL_OP)
|
|
|
|
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')
|
|
|
|
if result is None:
|
|
eprint('[spotify] trying REST API (api.spotify.com/v1)')
|
|
result = resolve_rest(session, token, sp_type, sp_id)
|
|
|
|
print(json.dumps(result, ensure_ascii=False))
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
main()
|
|
except SystemExit: raise
|
|
except Exception as e:
|
|
eprint(f'[spotify] fatal: {e}')
|
|
sys.exit(1)
|