Skip to content

Instantly share code, notes, and snippets.

@llimllib

llimllib/auth.py

Created Aug 22, 2018
Embed
What would you like to do?
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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.