Merge pull request #37 from Badgie/cleanup

Cleanup and error handling

api.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python
+import json
+import requests
+
+url_base = 'https://api.spotify.com/v1'
+
+
+def error_handle(request_domain: str, expected_code: int, request_type: str, response=None):
+    """
+    Dispatch error message depending on request type
+    :param request_domain: domain of the request, e.g. 'recommendation'
+    :param expected_code: expected status code
+    :param request_type: type of request, e.g. GET, POST, PUT
+    :param response: response object
+    """
+    if response.status_code is not expected_code:
+        print(f'{request_type} request for {request_domain} failed with status code {response.status_code} '
+              f'(expected {expected_code}). Reason: {response.reason}')
+        exit(1)
+
+
+def get_top_list(list_type: str, limit: int, headers: dict) -> json:
+    """
+    Retrieve list of top artists of tracks from user's profile.
+    :param list_type: type of list to retrieve; 'artists' or 'tracks'
+    :param limit: amount of entries to retrieve; min 1, max 50
+    :param headers: request headers
+    :return: top list as json object
+    """
+    params = {'limit': limit}
+    response = requests.get(f'{url_base}/me/top/{list_type}', headers=headers, params=params)
+    error_handle(f'top {list_type}', 200, 'GET', response=response)
+    return json.loads(response.content.decode('utf-8'))
+
+
+def get_user_id(headers: dict) -> str:
+    """
+    Retrieve user ID from API.
+    :param headers: request headers
+    :return: user ID as a string
+    """
+    response = requests.get(f'{url_base}/me', headers=headers)
+    error_handle('user info', 200, 'GET', response=response)
+    return json.loads(response.content.decode('utf-8'))['id']
+
+
+def create_playlist(playlist_name: str, playlist_description: str, headers: dict) -> str:
+    """
+    Creates playlist on user's account.
+    :param playlist_name: name of the playlist
+    :param playlist_description: description of the playlist
+    :param headers: request headers
+    :return: playlist id
+    """
+    data = {'name': playlist_name,
+            'description': playlist_description}
+    print('Creating playlist')
+    response = requests.post(f'{url_base}/users/{get_user_id(headers)}/playlists', json=data, headers=headers)
+    error_handle('playlist creation', 201, 'POST', response=response)
+    return json.loads(response.content.decode('utf-8'))['id']
+
+
+def upload_image(playlist_id: str, data: str, img_headers: dict):
+    """
+    Upload the generated image to the playlist
+    :param playlist_id: id of the playlist
+    :param data: base64 encoded jpeg image
+    :param img_headers: request headers
+    """
+    response = requests.put(f'{url_base}/playlists/{playlist_id}/images', headers=img_headers, data=data)
+    error_handle('image upload', 202, 'PUT', response=response)
+
+
+def add_to_playlist(tracks: list, playlist_id: str, headers: dict):
+    """
+    Add tracks to playlist.
+    :param tracks: list of track URIs
+    :param playlist_id: id of playlist
+    :param headers: request headers
+    """
+    data = {'uris': tracks}
+    response = requests.post(f'{url_base}/playlists/{playlist_id}/tracks', headers=headers, json=data)
+    error_handle('adding tracks', 201, 'POST', response=response)
+
+
+def get_recommendations(rec_params: dict, headers: dict) -> json:
+    """
+    Retrieve recommendations from API.
+    :param rec_params: parameters for recommendation request
+    :param headers: request headers
+    :return: recommendations as json object
+    """
+    response = requests.get(f'{url_base}/recommendations', params=rec_params, headers=headers)
+    error_handle('recommendations', 200, 'GET', response=response)
+    return json.loads(response.content.decode('utf-8'))
+
+
+def request_data(uri: str, data_type: str, headers: dict) -> json:
+    """
+    Requests data about an artist or a track.
+    :param uri: uri for the artist or track
+    :param data_type: the type of data to request; 'artists' or 'tracks'
+    :param headers: request headers
+    :return: data about artist or track as a json obj
+    """
+    response = requests.get(f'{url_base}/{data_type}/{uri.split(":")[2]}', headers=headers)
+    error_handle(f'single {data_type.strip("s")}', 200, 'GET', response=response)
+    return json.loads(response.content.decode('utf-8'))
+
+
+def get_genre_seeds(headers: dict) -> json:
+    """
+    Retrieves available genre seeds from Spotify API.
+    :param headers: request headers
+    :return: genre seeds as a json obj
+    """
+    response = requests.get(f'{url_base}/recommendations/available-genre-seeds', headers=headers)
+    error_handle('genre seeds', 200, 'GET', response=response)
+    return json.loads(response.content.decode('utf-8'))

oauth2.py

@@ -3,19 +3,23 @@
 import time
 import requests
 import base64
+import api
 from urllib import parse
+from pathlib import Path
 
 
 class SpotifyOAuth:
-    oauth_auth_url = 'https://accounts.spotify.com/authorize'
-    oauth_token_url = 'https://accounts.spotify.com/api/token'
+    OAUTH_AUTH_URL = 'https://accounts.spotify.com/authorize'
+    OAUTH_TOKEN_URL = 'https://accounts.spotify.com/api/token'
+    PORT = 8080
 
-    def __init__(self, client_id: str, client_secret: str, redirect: str, scopes: str, cache: str):
-        self.client_id = client_id
-        self.client_secret = client_secret
-        self.redirect = redirect
-        self.scopes = scopes
-        self.cache = cache
+    def __init__(self):
+        self.client_id = '466a89a53359403b82df7d714030ec5f'
+        self.client_secret = '28147de72c3549e98b1e790f3d080b85'
+        self.redirect = f'http://localhost:{self.PORT}'
+        self.scopes = 'user-top-read playlist-modify-public playlist-modify-private user-read-private ' \
+                      'user-read-email ugc-image-upload'
+        self.cache = f'{Path.home()}/.config/spotirec/spotirecoauth'
 
     def get_credentials(self) -> json:
         """
@@ -49,7 +53,8 @@ def refresh_token(self, refresh_token: str) -> json:
         """
         body = {'grant_type': 'refresh_token',
                 'refresh_token': refresh_token}
-        response = requests.post(self.oauth_token_url, data=body, headers=self.encode_header())
+        response = requests.post(self.OAUTH_TOKEN_URL, data=body, headers=self.encode_header())
+        api.error_handle('token refresh', 200, 'POST', response=response)
         token = json.loads(response.content.decode('utf-8'))
         try:
             assert token['refresh_token'] is not None
@@ -76,7 +81,8 @@ def retrieve_access_token(self, code: str) -> json:
         body = {'grant_type': 'authorization_code',
                 'code': code,
                 'redirect_uri': self.redirect}
-        response = requests.post(self.oauth_token_url, data=body, headers=self.encode_header())
+        response = requests.post(self.OAUTH_TOKEN_URL, data=body, headers=self.encode_header())
+        api.error_handle('token retrieve', 200, 'POST', response=response)
         token = json.loads(response.content.decode('utf-8'))
         self.save_token(token)
         return token
@@ -90,7 +96,7 @@ def get_authorize_url(self) -> str:
                   'response_type': 'code',
                   'redirect_uri': self.redirect,
                   'scope': self.scopes}
-        return f'{self.oauth_auth_url}?{parse.urlencode(params)}'
+        return f'{self.OAUTH_AUTH_URL}?{parse.urlencode(params)}'
 
     def parse_response_code(self, url: str) -> str:
         """

recommendation.py

@@ -8,7 +8,7 @@ class Recommendation:
     """
     def __init__(self, t=time.localtime()):
         self.limit = 20
-        self.limit_fill = 0
+        self.limit_original = self.limit
         self.created_at = time.ctime(time.time())
         self.based_on = 'top genres'
         self.seed = ''
@@ -30,21 +30,27 @@ def playlist_description(self) -> str:
             for x in self.seed_info.values())
         return f'{desc}{seeds}'
 
-    def update_limit(self, limit: int):
+    def update_limit(self, limit: int, init=False):
         """
         Update playlist limit as object field and in request parameters.
         :param limit: user-defined playlist limit
+        :param init: should only be true when updated by -l arg
         """
         self.limit = limit
         self.rec_params['limit'] = str(self.limit)
+        if init:
+            self.limit_original = limit
 
     def print_selection(self):
         """
         Print seed selection into terminal.
         """
         print('Selection:')
-        for x in self.seed_info:
-            print(f'\t{self.seed_info[x]}')
+        for x in self.seed_info.values():
+            try:
+                print(f'\t{x["type"].capitalize()}: {x["name"]} - {", ".join(str(y) for y in x["artists"])}')
+            except KeyError:
+                print(f'\t{x["type"].capitalize()}: {x["name"]}')
 
     def add_seed_info(self, data_dict=None, data_string=None):
         """