diff --git a/pomice/spotify/__init__.py b/pomice/spotify/__init__.py index fa31d18..d0bf1a9 100644 --- a/pomice/spotify/__init__.py +++ b/pomice/spotify/__init__.py @@ -1,6 +1,6 @@ """Spotify module for Pomice, made possible by cloudwithax 2021""" -from .exceptions import SpotifyRequestException +from .exceptions import InvalidSpotifyURL, SpotifyRequestException from .track import Track from .playlist import Playlist from .album import Album diff --git a/pomice/spotify/album.py b/pomice/spotify/album.py index ec2270e..9b914ec 100644 --- a/pomice/spotify/album.py +++ b/pomice/spotify/album.py @@ -1,15 +1,20 @@ from .track import Track + class Album: """The base class for a Spotify album""" + def __init__(self, data: dict) -> None: - self.name = data['name'] - self.artists = ", ".join(artist["name"] for artist in data['artists']) - self.tracks = [Track(track) for track in data['tracks']['items']] - self.total_tracks = data['total_tracks'] - self.id = data['id'] - self.image = data['images'][0]['url'] - self.uri = data['external_urls']['spotify'] + self.name = data["name"] + self.artists = ", ".join(artist["name"] for artist in data["artists"]) + self.tracks = [Track(track) for track in data["tracks"]["items"]] + self.total_tracks = data["total_tracks"] + self.id = data["id"] + self.image = data["images"][0]["url"] + self.uri = data["external_urls"]["spotify"] def __repr__(self) -> str: - return f"" \ No newline at end of file + return ( + f"" + ) diff --git a/pomice/spotify/client.py b/pomice/spotify/client.py index 4861f57..a19061b 100644 --- a/pomice/spotify/client.py +++ b/pomice/spotify/client.py @@ -1,50 +1,55 @@ -import base64 import re import time +from base64 import b64encode import aiohttp from .album import Album -from .exceptions import SpotifyRequestException +from .exceptions import InvalidSpotifyURL, SpotifyRequestException from .playlist import Playlist from .track import Track GRANT_URL = "https://accounts.spotify.com/api/token" +REQUEST_URL = "https://api.spotify.com/v1/{type}s/{id}" SPOTIFY_URL_REGEX = re.compile( r"https?://open.spotify.com/(?Palbum|playlist|track)/(?P[a-zA-Z0-9]+)" ) + class Client: """The base client for the Spotify module of Pomice. This class will do all the heavy lifting of getting all the metadata for any Spotify URL you throw at it. """ + def __init__(self, client_id: str, client_secret: str) -> None: - self._client_id: str = client_id - self._client_secret: str = client_secret + self._client_id = client_id + self._client_secret = client_secret self.session = aiohttp.ClientSession() self._bearer_token: str = None - self._expiry: int = 0 - self._auth_token = base64.b64encode(":".join((self._client_id, self._client_secret)).encode()) + self._expiry = 0 + self._auth_token = b64encode(f"{self._client_id}:{self._client_secret}".encode()) self._grant_headers = {"Authorization": f"Basic {self._auth_token.decode()}"} self._bearer_headers = None async def _fetch_bearer_token(self) -> None: - data = {"grant_type": "client_credentials"} - async with self.session.post(GRANT_URL, data=data, headers=self._grant_headers) as resp: + _data = {"grant_type": "client_credentials"} + + async with self.session.post(GRANT_URL, data=_data, headers=self._grant_headers) as resp: if resp.status != 200: - raise SpotifyRequestException(f"Error: {resp.status} {resp.reason}") + raise SpotifyRequestException( + f"Error fetching bearer token: {resp.status} {resp.reason}" + ) - data = await resp.json() - self._bearer_token = data["access_token"] - self._expiry = time.time() + (int(data["expires_in"]) - 10) - self._bearer_headers = {"Authorization": f"Bearer {self._bearer_token}"} + data: dict = await resp.json() + self._bearer_token = data["access_token"] + self._expiry = time.time() + (int(data["expires_in"]) - 10) + self._bearer_headers = {"Authorization": f"Bearer {self._bearer_token}"} async def search(self, *, query: str): - if not self._bearer_token or time.time() >= self._expiry: await self._fetch_bearer_token() @@ -53,60 +58,37 @@ class Client: spotify_id = result.group("id") if not result: - return SpotifyRequestException("The Spotify link provided is not valid.") + return InvalidSpotifyURL("The Spotify link provided is not valid.") + + request_url = REQUEST_URL.format(type=spotify_type, id=spotify_id) + + async with self.session.get(request_url, headers=self._bearer_headers) as resp: + if resp.status != 200: + raise SpotifyRequestException( + f"Error while fetching results: {resp.status} {resp.reason}" + ) + + data: dict = await resp.json() if spotify_type == "track": - request_url = f"https://api.spotify.com/v1/tracks/{spotify_id}" - async with self.session.get(request_url, headers=self._bearer_headers) as resp: - if resp.status != 200: - raise SpotifyRequestException(resp.status, resp.reason) - - data: dict = await resp.json() - return Track(data) - elif spotify_type == "album": - request_url = f"https://api.spotify.com/v1/albums/{spotify_id}" + return Album(data) - async with self.session.get(request_url, headers=self._bearer_headers) as resp: + # processing a playlist result + tracks = [Track(track["track"]) for track in data["tracks"]["items"]] + next_page_url = data["tracks"]["next"] + + while next_page_url is not None: + async with self.session.get(next_page_url, headers=self._bearer_headers) as resp: if resp.status != 200: - raise SpotifyRequestException(resp.status, resp.reason) - - album_data: dict = await resp.json() - - return Album(album_data) - - elif spotify_type == "playlist": - request_url = f"https://api.spotify.com/v1/playlists/{spotify_id}" - tracks = [] - async with self.session.get(request_url, headers=self._bearer_headers) as resp: - if resp.status != 200: - raise SpotifyRequestException(resp.status, resp.reason) - - playlist_data: dict = await resp.json() - - tracks += [Track(track["track"]) for track in playlist_data["tracks"]["items"]] - - next_page_url = playlist_data["tracks"]["next"] - - while next_page_url != None: - async with self.session.get(next_page_url, headers=self._bearer_headers) as resp: - if resp.status != 200: - raise SpotifyRequestException(resp.status, resp.reason) - - next_page_data: dict = await resp.json() - - tracks += [Track(track["track"]) for track in next_page_data["items"]] - next_page_url = next_page_data["next"] - - return Playlist(playlist_data, tracks) - - - - - - - + raise SpotifyRequestException( + f"Error while fetching results: {resp.status} {resp.reason}" + ) + next_data: dict = await resp.json() + tracks += [Track(track["track"]) for track in next_data["items"]] + next_page_url = next_data["next"] + return Playlist(data, tracks) diff --git a/pomice/spotify/exceptions.py b/pomice/spotify/exceptions.py index f1b5310..bd67ec5 100644 --- a/pomice/spotify/exceptions.py +++ b/pomice/spotify/exceptions.py @@ -1,3 +1,7 @@ class SpotifyRequestException(Exception): """An error occurred when making a request to the Spotify API""" - pass \ No newline at end of file + pass + + +class InvalidSpotifyURL(Exception): + pass diff --git a/pomice/spotify/playlist.py b/pomice/spotify/playlist.py index 20aaac3..6f59806 100644 --- a/pomice/spotify/playlist.py +++ b/pomice/spotify/playlist.py @@ -1,16 +1,21 @@ from .track import Track from typing import List + class Playlist: """The base class for a Spotify playlist""" + def __init__(self, data: dict, tracks: List[Track]) -> None: - self.name = data['name'] + self.name = data["name"] self.tracks = tracks - self.owner = data['owner']['display_name'] - self.total_tracks = data['tracks']['total'] - self.id = data['id'] - self.image = data['images'][0]['url'] - self.uri = data['external_urls']['spotify'] + self.owner = data["owner"]["display_name"] + self.total_tracks = data["tracks"]["total"] + self.id = data["id"] + self.image = data["images"][0]["url"] + self.uri = data["external_urls"]["spotify"] def __repr__(self) -> str: - return f"" \ No newline at end of file + return ( + f"" + ) diff --git a/pomice/spotify/track.py b/pomice/spotify/track.py index 96af859..328836c 100644 --- a/pomice/spotify/track.py +++ b/pomice/spotify/track.py @@ -2,17 +2,20 @@ class Track: """The base class for a Spotify Track""" def __init__(self, data: dict) -> None: - self.name = data['name'] - self.artists = ", ".join(artist["name"] for artist in data['artists']) - self.length = data['duration_ms'] - self.id = data['id'] + self.name = data["name"] + self.artists = ", ".join(artist["name"] for artist in data["artists"]) + self.length = data["duration_ms"] + self.id = data["id"] if data.get("album") and data["album"]["images"]: - self.image = data['album']['images'][0]['url'] + self.image = data["album"]["images"][0]["url"] else: self.image = None - self.uri = data['external_urls']['spotify'] + self.uri = data["external_urls"]["spotify"] def __repr__(self) -> str: - return f"" + return ( + f"" + )