MusicIndexer/SpotifyWebAPI.py

import requests
from PIL import Image
from io import BytesIO
from SpotifyAuthenticator import authenticate
from DataBase 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 find_song(song_pattern):
    return fetch(f"v1/search/?type=track&track=1&q={song_pattern}")['tracks']['items']


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 save_image_from_url(url, name, image_dir="playlist_covers"):
    response = requests.get(url)

    if response.status_code != 200:
        raise "Cannot fetch the playlist cover"

    image = Image.open(BytesIO(response.content))

    directory = os.path.join(data_dir, image_dir)
    os.makedirs(directory, exist_ok=True)
    file_path = os.path.join(directory, f"{name}.jpg")
    print(f"Image saved {file_path}")
    image.save(file_path)


def load_playlist_covers():
    playlists = get_data('playlists')

    for pl in playlists:
        if len(pl['images']):
            url = pl['images'][0]['url']
            save_image_from_url(url, pl['name'])


def load_user_cover():
    user_data = get_data("user_data")
    url = user_data['images'][1]['url']
    save_image_from_url(url, "user", ".")


def load_artworks():
    load_user_cover()
    load_playlist_covers()


def update_access_token():
    global token
    token = authenticate()


def fetch_data():
    user_id = get_user_data()

    fetch_tracks(user_id)
    fetch_all_playlist_data(user_id)
    fetch_tracks_top()
    fetch_artists_top()

    load_artworks()


if __name__ == "__main__":
    update_access_token()
    fetch_data()