auth.py

import base64
import os
import sys
from urllib.parse import urlencode
from webbrowser import open as open_url

import requests

SPOTIFY_OAUTH_URL = "https://accounts.spotify.com/authorize"
SPOTIFY_TOKEN_URL = "https://accounts.spotify.com/api/token"
SPOTIFY_API_URL = "https://api.spotify.com/v1"


class AuthException(Exception):
    def __init__(self, response):
        self.response = response
        Exception.__init__(self, f"Unable to authenticate: {response.text}")


class ApiException(Exception):
    def __init__(self, response):
        self.response = response
        Exception.__init__(self, f"API Exception: {response.text}")


def login():
    scope = " ".join(
        ["user-read-recently-played", "user-top-read", "user-library-read"])
    params = {
        "client_id": os.environ["SPOTIFY_CLIENT_ID"],
        "response_type": "code",
        "redirect_uri": os.environ["SPOTIFY_REDIRECT_URI"],
        "scope": scope
    }
    oauth_url = f"{SPOTIFY_OAUTH_URL}?{urlencode(params)}"
    open_url(oauth_url)
    code = input("What's the value of the 'code' parameter? ")
    auth = b"Basic " + base64.b64encode(
        f"{os.environ['SPOTIFY_CLIENT_ID']}:"
        f"{os.environ['SPOTIFY_CLIENT_SECRET']}".encode("utf8"))
    res = requests.post(
        SPOTIFY_TOKEN_URL,
        data={
            "grant_type": 'authorization_code',
            "code": code,
            "redirect_uri": os.environ["SPOTIFY_REDIRECT_URI"]
        },
        headers={
            "Authorization": auth
        })
    if res.status_code != 200:
        raise AuthException(res)
    return res.json()


def api_call(access, endpoint, **params):
    auth = f"Bearer {access['access_token']}"
    res = requests.get(
        f"{SPOTIFY_API_URL}{endpoint}",
        params=params,
        headers={
            "Authorization": auth
        })
    # TODO: handle token expiration
    if res.status_code != 200:
        raise AuthException(res)
    return res.json()


def api_call_get_all(access, endpoint, **params):
    limit = 50
    page = 0
    response = api_call(access, endpoint, limit=limit)
    all_items = []
    while response["items"]:
        w('.')
        all_items.extend(response["items"])
        page += 1
        response = api_call(access, endpoint, limit=limit, offset=page * limit)
    w('\n\n')
    return all_items


def w(s):
    sys.stdout.write(s)
    sys.stdout.flush()


def saved_albums_by_popularity(access):
    albums_by_popularity = {}
    for album in api_call_get_all(access, "/me/albums"):
        al = album["album"]
        artist = ", ".join(a['name'] for a in al['artists'])
        albums_by_popularity.setdefault(al['popularity'], set()).add(
            (artist, al['name'], al['uri']))

    return albums_by_popularity


def top_tracks_by_popularity(access):
    tracks_by_popularity = {}
    for track in api_call_get_all(access, "/me/top/tracks"):
        artist = ", ".join(t['name'] for t in track['artists'])
        tracks_by_popularity.setdefault(track['popularity'], set()).add(
            (artist, track['name'], track['uri']))

    return tracks_by_popularity


def main():
    access = login()
    albums = saved_albums_by_popularity(access)
    print("Your least popular albums:")
    keys = sorted(key for key in albums if key < 15)
    for key in keys:
        for album in albums[key]:
            print(f"{key}: {album}")

    print('Your least popular "top tracks":')
    tracks = top_tracks_by_popularity(access)
    keys = sorted(key for key in tracks if key < 50)
    for key in keys:
        for track in tracks[key]:
            print(f"{key}: {track}")

    # return albums


if __name__ == "__main__":
    main()