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:
parent
b6d88d435a
commit
08e739a768
2 changed files with 33 additions and 58 deletions
|
|
@ -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):
|
||||||
|
|
|
||||||
|
|
@ -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))
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue