spotify: update fetching to current Web API

- OAuth: URL-encode authorize params (scope spaces were unencoded),
  bind loopback redirect server to 127.0.0.1 to match the redirect_uri
  per Spotify's 2025 loopback rules, capture refresh_token.
- Search: URL-encode the query and drop the bogus track=1 param.
- Pagination: follow the paging object's `next` field instead of manual
  offset/total bookkeeping (Spotify's recommended approach).
- Fix invalid `raise "string"` in save_image_from_url.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Шурупов Илья Викторович 2026-06-15 00:03:10 +03:00
parent 9cd2c9189b
commit 16b55b0674
2 changed files with 33 additions and 58 deletions

View file

@ -11,6 +11,7 @@ import os.path
client_id = None client_id = None
client_secret = None client_secret = None
token = None token = None
refresh_token = None
def retrieve_web_api_token(authorization_code): def retrieve_web_api_token(authorization_code):
@ -44,7 +45,10 @@ def retrieve_web_api_token(authorization_code):
print(token_response.json()) print(token_response.json())
raise ValueError raise ValueError
token = token_response.json()['access_token'] global refresh_token
response_json = token_response.json()
token = response_json['access_token']
refresh_token = response_json.get('refresh_token')
class RequestHandler(BaseHTTPRequestHandler): class RequestHandler(BaseHTTPRequestHandler):
@ -73,7 +77,7 @@ class RequestHandler(BaseHTTPRequestHandler):
def run_redirect_server(): def run_redirect_server():
httpd = HTTPServer(('localhost', 8888), RequestHandler) httpd = HTTPServer(('127.0.0.1', 8888), RequestHandler)
print('Starting HTTP redirection server') print('Starting HTTP redirection server')
httpd.serve_forever() httpd.serve_forever()
@ -88,8 +92,9 @@ def open_authenticate_page():
"scope": "user-top-read playlist-read-collaborative user-read-email user-library-read user-read-private playlist-read-private", "scope": "user-top-read playlist-read-collaborative user-read-email user-library-read user-read-private playlist-read-private",
} }
auth_url = f"{authorize_url}?client_id={params['client_id']}&response_type={params['response_type']}&redirect_uri={params['redirect_uri']}&scope={params['scope']}" auth_url = f"{authorize_url}?{urllib.parse.urlencode(params)}"
webbrowser.open(auth_url) webbrowser.open(auth_url)
print(f"If the browser did not open, visit:\n{auth_url}")
def authenticate(id, secret): def authenticate(id, secret):

View file

@ -1,4 +1,5 @@
import requests import requests
import urllib.parse
from PIL import Image from PIL import Image
from io import BytesIO from io import BytesIO
@ -11,7 +12,7 @@ token = ''
def fetch(endpoint, method='GET', body=None): def fetch(endpoint, method='GET', body=None):
url = f'https://api.spotify.com/{endpoint}' url = endpoint if endpoint.startswith('http') else f'https://api.spotify.com/{endpoint}'
headers = { headers = {
'Authorization': f'Bearer {token}', 'Authorization': f'Bearer {token}',
} }
@ -21,83 +22,52 @@ def fetch(endpoint, method='GET', body=None):
return response.json() return response.json()
def fetch_paginated(endpoint):
"""Collect every item of a paging object by following its `next` field,
which is Spotify's recommended way to page (more robust than tracking
offset/total manually)."""
items = []
page = fetch(endpoint)
while True:
items += page.get('items', [])
next_url = page.get('next')
if not next_url:
break
page = fetch(next_url)
return items
def find_song(song_pattern): def find_song(song_pattern):
return fetch(f"v1/search/?type=track&track=1&q={song_pattern}")['tracks']['items'] query = urllib.parse.quote(song_pattern)
return fetch(f"v1/search?type=track&limit=10&q={query}")['tracks']['items']
def fetch_playlists(user_id): def fetch_playlists(user_id):
print("Fetching playlists") print("Fetching playlists")
endpoint = f"v1/me/playlists" return fetch_paginated(f"v1/me/playlists?limit={FETCH_STEP}")
pl_total = fetch(f'{endpoint}?limit=1')['total']
pl_count = 0
playlists = []
while pl_count < pl_total:
res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={pl_count}')
playlists += res['items']
pl_count += FETCH_STEP
return playlists
def fetch_playlist_items(playlist_id): def fetch_playlist_items(playlist_id):
endpoint = f"v1/playlists/{playlist_id}/tracks" return fetch_paginated(f"v1/playlists/{playlist_id}/tracks?limit={FETCH_STEP}")
items = fetch(f'{endpoint}?fields=total')['total']
loaded = 0
tracks = []
while loaded < items:
res = fetch(f'{endpoint}?limit={FETCH_STEP}&offset={loaded}')
tracks += res['items']
loaded += FETCH_STEP
return tracks
def fetch_tracks(user_id): def fetch_tracks(user_id):
print("Fetching tracks") print("Fetching tracks")
endpoint = f"v1/me/tracks" tracks = fetch_paginated(f"v1/me/tracks?limit={FETCH_STEP}&market=ES")
total = fetch(f"{endpoint}?limit=1&market=ES")['total']
current = 0
tracks = []
while total > current:
res = fetch(f"{endpoint}?limit={FETCH_STEP}&offset={current}&market=ES")
tracks += res['items']
current += FETCH_STEP
save_data(tracks, "tracks") save_data(tracks, "tracks")
return tracks return tracks
def fetch_tracks_top(): def fetch_tracks_top():
print("Fetching top tracks") print("Fetching top tracks")
total = fetch(f'v1/me/top/tracks?fields=total')['total'] tracks = fetch_paginated(f"v1/me/top/tracks?limit={FETCH_STEP}&time_range=long_term")
current = 0
tracks = []
while current < total:
res = fetch(f'v1/me/top/tracks?limit={FETCH_STEP}&offset={current}&time_range=long_term')
tracks += res['items']
current += FETCH_STEP
save_data(tracks, "top_tracks") save_data(tracks, "top_tracks")
return tracks return tracks
def fetch_artists_top(): def fetch_artists_top():
print("Fetching top artists") print("Fetching top artists")
artists = fetch_paginated(f"v1/me/top/artists?limit={FETCH_STEP}&time_range=long_term")
total = fetch(f'v1/me/top/artists?fields=total')['total']
current = 0
artists = []
while current < total:
res = fetch(f'v1/me/top/artists?limit={FETCH_STEP}&offset={current}&time_range=long_term')
artists += res['items']
current += FETCH_STEP
save_data(artists, "top_artists") save_data(artists, "top_artists")
return artists return artists
@ -140,7 +110,7 @@ def save_image_from_url(url, name, image_dir="playlist_covers"):
response = requests.get(url) response = requests.get(url)
if response.status_code != 200: if response.status_code != 200:
raise "Cannot fetch the playlist cover" raise RuntimeError("Cannot fetch the playlist cover")
image = Image.open(BytesIO(response.content)) image = Image.open(BytesIO(response.content))