fix: add SPOTIFY_PROXY support to bypass server IP ban
Spotify blocks API calls from server IPs (open.spotify.com/api/* times out, api.spotify.com/v1/ returns 429). Setting SPOTIFY_PROXY routes all Spotify calls through curl_cffi via that proxy, bypassing the IP ban. Without proxy, SpotipyFree is still tried (works on non-banned IPs). A clear tip is printed if SpotipyFree fails without a proxy configured. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
1 parent
7f7bedc15e
commit
207e241600
2 files changed
+131
-9
No files matched your search
+130
-9
@@ -1,15 +1,18 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
"""
|
"""
|
||||||
Resolves Spotify URLs via SpotipyFree (SpotAPI + tls_client Chrome fingerprint,
|
Resolves Spotify URLs.
|
||||||
hits Spotify's internal spclient endpoints — not the rate-limited public REST API).
|
- SPOTIFY_PROXY set: curl_cffi + TOTP + REST API routed through proxy (bypasses IP ban)
|
||||||
|
- No proxy: SpotipyFree (SpotAPI + tls_client, internal spclient endpoints)
|
||||||
"""
|
"""
|
||||||
import sys, json, re, time
|
import sys, json, re, hmac, hashlib, struct, time, os, urllib.parse
|
||||||
|
|
||||||
|
PROXY = os.environ.get('SPOTIFY_PROXY', '').strip()
|
||||||
|
|
||||||
def eprint(*a): print(*a, file=sys.stderr)
|
def eprint(*a): print(*a, file=sys.stderr)
|
||||||
|
|
||||||
# ─── APPROACH 1: SpotipyFree ─────────────────────────────────────────────────
|
# ─── FORMAT HELPERS ───────────────────────────────────────────────────────────
|
||||||
|
|
||||||
def _fmt_spotipy_track(t, default_album='', default_art='', default_artist=''):
|
def _fmt_track(t, default_album='', default_art='', default_artist=''):
|
||||||
if not t or not t.get('name'): return None
|
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
|
artists = ', '.join(a.get('name','') for a in (t.get('artists') or [])) or default_artist
|
||||||
alb = t.get('album') or {}
|
alb = t.get('album') or {}
|
||||||
@@ -26,6 +29,115 @@ def _fmt_spotipy_track(t, default_album='', default_art='', default_artist=''):
|
|||||||
'durationMs': t.get('duration_ms'),
|
'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):
|
def resolve_via_spotipyfree(sp_type, sp_id):
|
||||||
from SpotipyFree import Spotify as Sp
|
from SpotipyFree import Spotify as Sp
|
||||||
sp = Sp()
|
sp = Sp()
|
||||||
@@ -36,8 +148,8 @@ def resolve_via_spotipyfree(sp_type, sp_id):
|
|||||||
artist = ', '.join(a['name'] for a in (t.get('artists') or []))
|
artist = ', '.join(a['name'] for a in (t.get('artists') or []))
|
||||||
imgs = (t.get('album') or {}).get('images') 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 ''
|
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,
|
||||||
return {'type':'track','name':t.get('name',''),'artist':artist,'artworkUrl':art,'tracks':[tr]}
|
'tracks':[_fmt_track(t)]}
|
||||||
|
|
||||||
if sp_type == 'album':
|
if sp_type == 'album':
|
||||||
alb = sp.album(sp_id)
|
alb = sp.album(sp_id)
|
||||||
@@ -48,7 +160,7 @@ def resolve_via_spotipyfree(sp_type, sp_id):
|
|||||||
while True:
|
while True:
|
||||||
page = sp.album_tracks(sp_id, limit=50, offset=offset)
|
page = sp.album_tracks(sp_id, limit=50, offset=offset)
|
||||||
for t in (page.get('items') or []):
|
for t in (page.get('items') or []):
|
||||||
tr = _fmt_spotipy_track(t, alb.get('name',''), art, artist)
|
tr = _fmt_track(t, alb.get('name',''), art, artist)
|
||||||
if tr: tracks.append(tr)
|
if tr: tracks.append(tr)
|
||||||
if not page.get('next'): break
|
if not page.get('next'): break
|
||||||
offset += len(page.get('items') or [])
|
offset += len(page.get('items') or [])
|
||||||
@@ -69,7 +181,7 @@ def resolve_via_spotipyfree(sp_type, sp_id):
|
|||||||
t = item.get('track')
|
t = item.get('track')
|
||||||
if not t or not t.get('id'): continue
|
if not t or not t.get('id'): continue
|
||||||
if t.get('is_local'): skipped += 1; continue
|
if t.get('is_local'): skipped += 1; continue
|
||||||
tr = _fmt_spotipy_track(t)
|
tr = _fmt_track(t)
|
||||||
if tr: tracks.append(tr)
|
if tr: tracks.append(tr)
|
||||||
if not page.get('next'): break
|
if not page.get('next'): break
|
||||||
offset += len(page.get('items') or [])
|
offset += len(page.get('items') or [])
|
||||||
@@ -88,12 +200,21 @@ def main():
|
|||||||
sp_type, sp_id = m.group(1), m.group(2)
|
sp_type, sp_id = m.group(1), m.group(2)
|
||||||
eprint(f'[spotify] resolving {sp_type} {sp_id}')
|
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:
|
try:
|
||||||
result = resolve_via_spotipyfree(sp_type, sp_id)
|
result = resolve_via_spotipyfree(sp_type, sp_id)
|
||||||
print(json.dumps(result, ensure_ascii=False))
|
print(json.dumps(result, ensure_ascii=False))
|
||||||
except SystemExit: raise
|
except SystemExit: raise
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
eprint(f'[SpotipyFree] failed: {type(e).__name__}: {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)
|
sys.exit(1)
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
|
|||||||
@@ -30,6 +30,7 @@ services:
|
|||||||
- MAX_PARALLEL=${MAX_PARALLEL:-2}
|
- MAX_PARALLEL=${MAX_PARALLEL:-2}
|
||||||
- DOWNLOAD_BACKEND=${DOWNLOAD_BACKEND:-auto}
|
- DOWNLOAD_BACKEND=${DOWNLOAD_BACKEND:-auto}
|
||||||
- SQUID_QUALITY=${SQUID_QUALITY:-mp3}
|
- SQUID_QUALITY=${SQUID_QUALITY:-mp3}
|
||||||
|
- SPOTIFY_PROXY=${SPOTIFY_PROXY:-}
|
||||||
command: >
|
command: >
|
||||||
sh -c "
|
sh -c "
|
||||||
apt-get update && apt-get install -y --no-install-recommends python3 python3-pip ffmpeg curl git && rm -rf /var/lib/apt/lists/* &&
|
apt-get update && apt-get install -y --no-install-recommends python3 python3-pip ffmpeg curl git && rm -rf /var/lib/apt/lists/* &&
|
||||||
|
|||||||
Reference in new issue
Block a user