fix: remove rate-limited REST fallback, explicitly install SpotipyFree

The curl_cffi fallback in spotify_resolve.py was silently catching
SpotipyFree ImportErrors and hitting api.spotify.com/v1/ directly —
the same rate-limited endpoint. Removed it entirely.

Also explicitly pip-install SpotipyFree before TzurSoffer's package
so the import is always available regardless of dependency resolution.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
HoneyandClaude Sonnet 4.6 committed 2026-05-12 18:38:23 +02:00
1 parent 0f1e6ff531
commit 7f7bedc15e
2 files changed
+4 -118

No files matched your search

+3 -117
View File
@@ -1,11 +1,9 @@
#!/usr/bin/env python3
"""
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)
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, urllib.parse
import sys, json, re, time
def eprint(*a): print(*a, file=sys.stderr)
@@ -80,108 +78,6 @@ def resolve_via_spotipyfree(sp_type, sp_id):
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]
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 _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)
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_spotipy_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_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}
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 ────────────────────────────────────────────────────────────────────
def main():
@@ -192,22 +88,12 @@ def main():
sp_type, sp_id = m.group(1), m.group(2)
eprint(f'[spotify] resolving {sp_type} {sp_id}')
# Try SpotipyFree first (SpotAPI + tls_client Chrome fingerprint)
try:
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'[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__':
+1 -1
View File
@@ -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 'git+https://github.com/TzurSoffer/spotify-downloader' &&
pip install --break-system-packages SpotipyFree '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
"