fix: use SpotipyFree as primary resolver (SpotAPI + Chrome TLS fingerprint)
SpotipyFree wraps SpotAPI which uses tls_client to impersonate Chrome's TLS fingerprint and hits Spotify's internal endpoints. This is exactly what TzurSoffer's spotdl fork uses. Removes the complex GQL hash extraction (which couldn't find hashes in lazy-loaded webpack chunks). curl_cffi + TOTP kept as fallback. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
HoneyandClaude Sonnet 4.6 committed
1 parent
06b78be199
commit
c36dbb3c94
1 file changed
+164
-320
+164
-320
@@ -1,15 +1,86 @@
|
||||
#!/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.
|
||||
Resolves Spotify URLs.
|
||||
Approach 1: SpotipyFree (SpotAPI + tls_client Chrome fingerprint, internal endpoints)
|
||||
Approach 2: curl_cffi + TOTP anonymous token + partner GraphQL API
|
||||
Approach 3: curl_cffi + TOTP + REST API (rate-limited fallback)
|
||||
"""
|
||||
import sys, json, re, hmac, hashlib, struct, time, urllib.parse
|
||||
|
||||
def eprint(*a): print(*a, file=sys.stderr)
|
||||
|
||||
# ─── TOTP ────────────────────────────────────────────────────────────────────
|
||||
# ─── APPROACH 1: SpotipyFree ─────────────────────────────────────────────────
|
||||
|
||||
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 {}
|
||||
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'),
|
||||
}
|
||||
|
||||
def resolve_via_spotipyfree(sp_type, sp_id):
|
||||
from SpotipyFree import Spotify as Sp
|
||||
sp = Sp()
|
||||
eprint('[SpotipyFree] session created')
|
||||
|
||||
if sp_type == 'track':
|
||||
t = sp.track(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 ''
|
||||
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)
|
||||
artist = ', '.join(a['name'] for a in (alb.get('artists') or []))
|
||||
imgs = alb.get('images') or []
|
||||
art = sorted(imgs, key=lambda x: -(x.get('width') or 0))[0].get('url','') if imgs else ''
|
||||
tracks, offset = [], 0
|
||||
while True:
|
||||
page = sp.album_tracks(sp_id, limit=50, offset=offset)
|
||||
for t in (page.get('items') or []):
|
||||
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 [])
|
||||
time.sleep(0.3)
|
||||
eprint(f'[SpotipyFree] album ok — {len(tracks)} tracks')
|
||||
return {'type':'album','name':alb.get('name',''),'artist':artist,'artworkUrl':art,'tracks':tracks}
|
||||
|
||||
if sp_type == 'playlist':
|
||||
pl = sp.playlist(sp_id, fields='name,owner,images')
|
||||
imgs = pl.get('images') or []
|
||||
art = sorted(imgs, key=lambda x: -(x.get('width') or 0))[0].get('url','') if imgs else ''
|
||||
owner = (pl.get('owner') or {}).get('display_name','')
|
||||
tracks, skipped, offset = [], 0, 0
|
||||
while True:
|
||||
page = sp.playlist_items(sp_id, limit=100, offset=offset,
|
||||
additional_types=['track'])
|
||||
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
|
||||
tr = _fmt_spotipy_track(t)
|
||||
if tr: tracks.append(tr)
|
||||
if not page.get('next'): break
|
||||
offset += len(page.get('items') or [])
|
||||
time.sleep(0.3)
|
||||
eprint(f'[SpotipyFree] playlist ok — {len(tracks)} tracks')
|
||||
return {'type':'playlist','name':pl.get('name',''),'artist':owner,
|
||||
'artworkUrl':art,'skippedLocal':skipped,'tracks':tracks}
|
||||
|
||||
# ─── APPROACH 2+3: curl_cffi + TOTP ─────────────────────────────────────────
|
||||
|
||||
def _totp(secret_str, server_ts):
|
||||
secret_bytes = [ord(c) for c in secret_str]
|
||||
@@ -22,309 +93,94 @@ def _totp(secret_str, server_ts):
|
||||
(digest[offset+2] & 0xff) << 8 | (digest[offset+3] & 0xff)) % 1_000_000
|
||||
return str(code).zfill(6)
|
||||
|
||||
def _collect_js_urls(html, seen):
|
||||
urls = []
|
||||
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); 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
|
||||
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
|
||||
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
|
||||
|
||||
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(totp_secret, st['serverTime'])
|
||||
'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={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
|
||||
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']
|
||||
|
||||
# ─── HELPERS ─────────────────────────────────────────────────────────────────
|
||||
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 _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):
|
||||
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_headers(token), timeout=15)
|
||||
_check_429(r)
|
||||
if not r.ok: raise Exception(f'REST API {r.status_code}: {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 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 _best_art(imgs): return sorted(imgs, key=lambda x: -(x.get('width') or 0))[0].get('url','') if imgs else ''
|
||||
|
||||
def resolve_via_curl(sp_type, sp_id):
|
||||
from curl_cffi import requests as cf
|
||||
session = cf.Session(impersonate='chrome124')
|
||||
token = _get_token(session)
|
||||
|
||||
def resolve_rest(session, token, sp_type, sp_id):
|
||||
if sp_type == 'track':
|
||||
t = sp_rest(session, f'/tracks/{sp_id}', token)
|
||||
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_rest_track(t)]}
|
||||
art = _best_art((t.get('album') or {}).get('images') or [])
|
||||
return {'type':'track','name':t.get('name',''),'artist':artist,'artworkUrl':art,
|
||||
'tracks':[_fmt_spotipy_track(t)]}
|
||||
|
||||
if sp_type == 'album':
|
||||
alb = sp_rest(session, f'/albums/{sp_id}', token)
|
||||
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 [])
|
||||
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')]
|
||||
pg = _rest(session, path, token)
|
||||
tracks += [t for t in [_fmt_spotipy_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
|
||||
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}
|
||||
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_spotipy_track(t)
|
||||
if tr: tracks.append(tr)
|
||||
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 ────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -332,41 +188,29 @@ 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 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 SpotipyFree first (SpotAPI + tls_client Chrome fingerprint)
|
||||
try:
|
||||
main()
|
||||
result = resolve_via_spotipyfree(sp_type, sp_id)
|
||||
print(json.dumps(result, ensure_ascii=False))
|
||||
return
|
||||
except SystemExit: raise
|
||||
except Exception as e:
|
||||
eprint(f'[spotify] fatal: {e}')
|
||||
eprint(f'[SpotipyFree] failed: {type(e).__name__}: {e}')
|
||||
|
||||
# Fall back to curl_cffi + TOTP
|
||||
try:
|
||||
result = resolve_via_curl(sp_type, sp_id)
|
||||
print(json.dumps(result, ensure_ascii=False))
|
||||
except SystemExit: raise
|
||||
except Exception as e:
|
||||
eprint(f'[curl_cffi] failed: {e}')
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == '__main__':
|
||||
try: main()
|
||||
except SystemExit: raise
|
||||
except Exception as e: eprint(f'fatal: {e}'); sys.exit(1)
|
||||
Reference in new issue
Block a user