173 lines
4 KiB
Python
173 lines
4 KiB
Python
import requests
|
|
from loadArtwork import load
|
|
from spotifyAuth import authenticate
|
|
from common import *
|
|
|
|
FETCH_STEP = 50
|
|
token = ''
|
|
|
|
|
|
def fetch(endpoint, method='GET', body=None):
|
|
url = f'https://api.spotify.com/{endpoint}'
|
|
headers = {
|
|
'Authorization': f'Bearer {token}',
|
|
}
|
|
|
|
response = requests.request(method, url, headers=headers, json=body)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def fetch_playlists(user_id):
|
|
print("Fetching playlists")
|
|
endpoint = f"v1/me/playlists"
|
|
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):
|
|
endpoint = f"v1/playlists/{playlist_id}/tracks"
|
|
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):
|
|
print("Fetching tracks")
|
|
endpoint = f"v1/me/tracks"
|
|
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")
|
|
return tracks
|
|
|
|
|
|
def fetch_tracks_top():
|
|
print("Fetching top tracks")
|
|
total = fetch(f'v1/me/top/tracks?fields=total')['total']
|
|
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")
|
|
return tracks
|
|
|
|
|
|
def fetch_artists_top():
|
|
print("Fetching top artists")
|
|
|
|
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")
|
|
return artists
|
|
|
|
|
|
def fetch_all_playlist_data(user_id):
|
|
playlists = fetch_playlists(user_id)
|
|
|
|
save_data(playlists, 'playlists')
|
|
|
|
print(len(playlists))
|
|
|
|
playlist_items = {}
|
|
|
|
print("Fetching playlists items")
|
|
|
|
index = 0
|
|
for pl in playlists:
|
|
pl_id = pl['id']
|
|
name = pl['name']
|
|
playlist_items[name] = fetch_playlist_items(pl_id)
|
|
print(f"{index} - {len(playlist_items[name])}")
|
|
index += 1
|
|
|
|
save_data(playlist_items, "playlistTracks")
|
|
|
|
|
|
def get_user_data():
|
|
print("Fetching user data")
|
|
|
|
user_profile = fetch(f'v1/me')
|
|
user_id = user_profile['id']
|
|
user_data = fetch(f'v1/users/{user_id}')
|
|
|
|
save_data(user_profile, "user_profile")
|
|
save_data(user_data, "user_data")
|
|
return user_id
|
|
|
|
|
|
def find_song(song_pattern):
|
|
tracks = fetch(f"v1/search/?type=track&track=1&q={song_pattern}")['tracks']['items']
|
|
return tracks
|
|
|
|
|
|
def find_local_library():
|
|
with open('prefetched/local_library.json', 'r') as file:
|
|
local_library = json.load(file)
|
|
|
|
from tqdm import tqdm
|
|
|
|
pattern_map = {}
|
|
|
|
total = len(local_library)
|
|
with tqdm(total=total, desc='Searching local songs on spotify') as pbar:
|
|
found_tracks = {}
|
|
for track, path in local_library.items():
|
|
found_tracks[track] = find_song(track)
|
|
pbar.update(1)
|
|
|
|
with open('prefetched/local_library_on_spotify.json', 'w') as file:
|
|
json.dump(found_tracks, file, indent=2)
|
|
|
|
|
|
def fetch_data():
|
|
global token
|
|
token = authenticate()
|
|
|
|
find_local_library()
|
|
|
|
user_id = get_user_data()
|
|
|
|
fetch_tracks(user_id)
|
|
fetch_all_playlist_data(user_id)
|
|
fetch_tracks_top()
|
|
fetch_artists_top()
|
|
|
|
load()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
fetch_data()
|