From 5498c604661a20742e0371b80b87a94a78951efa Mon Sep 17 00:00:00 2001 From: Choco <94597336+ChocoMeow@users.noreply.github.com> Date: Sat, 22 Mar 2025 18:42:28 +0800 Subject: [PATCH] Replaced internal spotify with lavalink --- addons/settings.py | 2 - cogs/basic.py | 4 +- cogs/listeners.py | 4 +- ipc/methods.py | 47 +-------- settings Example.json | 2 - views/debug.py | 2 - voicelink/enums.py | 2 - voicelink/exceptions.py | 20 ---- voicelink/objects.py | 54 ++-------- voicelink/player.py | 15 +-- voicelink/pool.py | 134 +++--------------------- voicelink/spotify/__init__.py | 26 ----- voicelink/spotify/client.py | 175 -------------------------------- voicelink/spotify/exceptions.py | 8 -- voicelink/spotify/objects.py | 135 ------------------------ 15 files changed, 29 insertions(+), 601 deletions(-) delete mode 100644 voicelink/spotify/__init__.py delete mode 100644 voicelink/spotify/client.py delete mode 100644 voicelink/spotify/exceptions.py delete mode 100644 voicelink/spotify/objects.py diff --git a/addons/settings.py b/addons/settings.py index c5b8c5c..5e8e4a9 100644 --- a/addons/settings.py +++ b/addons/settings.py @@ -32,8 +32,6 @@ class Settings: def __init__(self, settings: Dict) -> None: self.token: str = settings.get("token") self.client_id: int = int(settings.get("client_id", 0)) - self.spotify_client_id: str = settings.get("spotify_client_id") - self.spotify_client_secret: str = settings.get("spotify_client_secret") self.genius_token: str = settings.get("genius_token") self.mongodb_url: str = settings.get("mongodb_url") self.mongodb_name: str = settings.get("mongodb_name") diff --git a/cogs/basic.py b/cogs/basic.py index 49740cd..90d1eb9 100644 --- a/cogs/basic.py +++ b/cogs/basic.py @@ -90,9 +90,9 @@ class Basic(commands.Cog): if current: node = voicelink.NodePool.get_node() - if node and node.spotify_client: + if node: try: - tracks: list[voicelink.Track] = await node.spotifySearch(current, requester=interaction.user) + tracks: list[voicelink.Track] = await node.get_tracks(current, requester=interaction.user, search_type=SearchType.SPOTIFY) return [app_commands.Choice(name=truncate_string(f"🎵 {track.author} - {track.title}", 100), value=truncate_string(f"{track.author} - {track.title}", 100)) for track in tracks] except voicelink.TrackLoadError: return [] diff --git a/cogs/listeners.py b/cogs/listeners.py index 90d0e90..f8a5583 100644 --- a/cogs/listeners.py +++ b/cogs/listeners.py @@ -44,9 +44,7 @@ class Listeners(commands.Cog): for n in func.settings.nodes.values(): try: await self.voicelink.create_node( - bot=self.bot, - spotify_client_id=func.settings.spotify_client_id, - spotify_client_secret=func.settings.spotify_client_secret, + bot=self.bot, logger=func.logger, **n ) diff --git a/ipc/methods.py b/ipc/methods.py index 6d05a4b..1771ead 100644 --- a/ipc/methods.py +++ b/ipc/methods.py @@ -697,49 +697,6 @@ async def updateSettings(bot: commands.Bot, data: Dict) -> None: await func.update_settings(guild.id, {"$set": data}) -async def getFeaturedPlaylists(bot: commands.Bot, data: Dict) -> Dict: - locale = data.get("locale", "sv_SE") - limit = data.get("limit", 20) - offset = data.get("offset", 0) - - request_url = f"https://api.spotify.com/v1/browse/featured-playlists?locale={locale}&limit={max(1, min(limit, 50))}&offset={max(0, offset)}" - - node = NodePool.get_node() - result = await node.spotify_client.get_request(request_url) - - return { - "op": "getFeaturedPlaylists", - "userId": data.get("userId"), - "callback": data.get("callback"), - "playlists": [ - { - "id": item.get("id"), - "title": item.get("name"), - "description": item.get("description"), - "imageUrl": item.get("images", [{}])[0].get("url"), - "href": item.get("external_urls", {}).get("spotify") - } - for item in result.get("playlists", {}).get("items", []) - ] - } - -async def getCategoryPlaylists(bot: commands.Bot, data: Dict) -> Dict: - node = NodePool.get_node() - - return { - "op": "getCategoryPlaylists", - "userId": data.get("userId"), - "callback": data.get("callback"), - "playlists": [ - { - "id": category.id, - "title": category.name, - "imageUrl": category.icon, - } - for category in await node.spotify_client.get_categories() - ] - } - METHODS: Dict[str, Union[SystemMethod, PlayerMethod]] = { "initBot": SystemMethod(initBot, credit=0), "initUser": SystemMethod(initUser, credit=2), @@ -766,9 +723,7 @@ METHODS: Dict[str, Union[SystemMethod, PlayerMethod]] = { "updatePosition": PlayerMethod(updatePosition), "toggleAutoplay": PlayerMethod(toggleAutoplay), "updateFilter": PlayerMethod(updateFilter), - "searchAndPlay": PlayerMethod(searchAndPlay, credit=5, auto_connect=True), - "getFeaturedPlaylists": SystemMethod(getFeaturedPlaylists, credit=5), - "getCategoryPlaylists": SystemMethod(getCategoryPlaylists, credit=2) + "searchAndPlay": PlayerMethod(searchAndPlay, credit=5, auto_connect=True) } async def process_methods(ipc_client, bot: commands.Bot, data: Dict) -> None: diff --git a/settings Example.json b/settings Example.json index 3c195a5..4322aa5 100644 --- a/settings Example.json +++ b/settings Example.json @@ -1,8 +1,6 @@ { "token": "YOUR_BOT_TOKEN", "client_id": "YOUR_BOT_CLIENT_ID", - "spotify_client_id": "YOUR_SPOTIFY_CLIENT_ID", - "spotify_client_secret": "YOUR_SPOTIFY_CLIENT_SECRET", "genius_token": "YOUR_GENIUS_TOKEN", "mongodb_url": "YOUR_MONGODB_URL", "mongodb_name": "YOUR_MONGODB_DB_NAME", diff --git a/views/debug.py b/views/debug.py index 8ec1e80..0b4c2dd 100644 --- a/views/debug.py +++ b/views/debug.py @@ -105,8 +105,6 @@ class AddNodeModal(discord.ui.Modal): try: await voicelink.NodePool.create_node( bot=interaction.client, - spotify_client_id=func.settings.spotify_client_id, - spotify_client_secret=func.settings.spotify_client_secret, logger=func.logger, **config ) diff --git a/voicelink/enums.py b/voicelink/enums.py index 58a0bc8..ef74c58 100644 --- a/voicelink/enums.py +++ b/voicelink/enums.py @@ -38,8 +38,6 @@ class LoopType(Enum): class SearchType(Enum): """The enum for the different search types for Voicelink. - This feature is exclusively for the Spotify search feature of Voicelink. - If you are not using this feature, this class is not necessary. SearchType.YOUTUBE searches using regular Youtube, which is best for all scenarios. diff --git a/voicelink/exceptions.py b/voicelink/exceptions.py index 0c7fca8..59c0570 100644 --- a/voicelink/exceptions.py +++ b/voicelink/exceptions.py @@ -74,26 +74,6 @@ class FilterTagInvalid(VoicelinkException): """An invalid tag was passed or Voicelink was unable to find a filter tag""" pass -class SpotifyAlbumLoadFailed(VoicelinkException): - """The voicelink Spotify client was unable to load an album.""" - pass - - -class SpotifyTrackLoadFailed(VoicelinkException): - """The voicelink Spotify client was unable to load a track.""" - pass - - -class SpotifyPlaylistLoadFailed(VoicelinkException): - """The voicelink Spotify client was unable to load a playlist.""" - pass - - -class InvalidSpotifyClientAuthorization(VoicelinkException): - """No Spotify client authorization was provided for track searching.""" - pass - - class QueueFull(VoicelinkException): pass diff --git a/voicelink/objects.py b/voicelink/objects.py index 7c8bdcf..3f9ad78 100644 --- a/voicelink/objects.py +++ b/voicelink/objects.py @@ -33,7 +33,6 @@ from function import ( time as ctime ) -from .spotify import Playlist as spPlaylist from .formatter import encode YOUTUBE_REGEX = re.compile(r'(https?://)?(www\.)?youtube\.(com|nl)/watch\?v=([-\w]+)') @@ -51,11 +50,9 @@ class Track: "author", "uri", "source", - "spotify", "artist_id", "original", "_search_type", - "spotify_track", "thumbnail", "emoji", "length", @@ -73,7 +70,6 @@ class Track: info: dict, requester: Member, search_type: SearchType = SearchType.YOUTUBE, - spotify_track = None, ): self._track_id: Optional[str] = track_id self.info: dict = info @@ -83,13 +79,7 @@ class Track: self.author: str = info.get("author", "Unknown") self.uri: str = info.get("uri", "https://discord.com/application-directory/605618911471468554") self.source: str = info.get("sourceName", extract(self.uri).domain) - self.spotify: bool = self.source == "spotify" - if self.spotify: - self.artist_id: Optional[list] = info.get("artist_id") - - self.original: Optional[Track] = None if self.spotify else self - self._search_type: SearchType = SearchType.YOUTUBE if self.spotify else search_type - self.spotify_track: Track = spotify_track + self._search_type: SearchType = search_type self.thumbnail: str = info.get("artworkUrl") if not self.thumbnail and YOUTUBE_REGEX.match(self.uri): @@ -143,12 +133,9 @@ class Playlist: __slots__ = ( "playlist_info", - "tracks_raw", - "spotify", "name", - "spotify_playlist", - "_thumbnail", - "_uri", + "thumbnail", + "uri", "tracks" ) @@ -158,29 +145,16 @@ class Playlist: playlist_info: dict, tracks: list, requester: Member = None, - spotify: bool = False, - spotify_playlist: Optional[spPlaylist] = None ): self.playlist_info: dict = playlist_info - self.tracks_raw: list[Track] = tracks - self.spotify: bool = spotify self.name: str = playlist_info.get("name") - self.spotify_playlist: Optional[spPlaylist] = spotify_playlist - - self._thumbnail: str = None - self._uri: str = None + self.thumbnail: str = None + self.uri: str = None - if self.spotify: - self.tracks = tracks - self._thumbnail = self.spotify_playlist.image - self._uri = self.spotify_playlist.uri - else: - self.tracks = [ - Track(track_id=track["encoded"], info=track["info"], requester=requester) - for track in self.tracks_raw - ] - self._thumbnail = None - self._uri = None + self.tracks = [ + Track(track_id=track["encoded"], info=track["info"], requester=requester) + for track in tracks + ] def __str__(self) -> str: return self.name @@ -188,16 +162,6 @@ class Playlist: def __repr__(self) -> str: return f"" - @property - def uri(self) -> Optional[str]: - """Spotify album/playlist URI, or None if not a Spotify object.""" - return self._uri - - @property - def thumbnail(self) -> Optional[str]: - """Spotify album/playlist thumbnail, or None if not a Spotify object.""" - return self._thumbnail - @property def track_count(self) -> int: return len(self.tracks) diff --git a/voicelink/player.py b/voicelink/player.py index 4480b79..a24c07b 100644 --- a/voicelink/player.py +++ b/voicelink/player.py @@ -524,10 +524,6 @@ class Player(VoiceProtocol): ) -> Union[List[Track], Playlist]: """Fetches tracks from the node's REST api to parse into Lavalink. - If you passed in Spotify API credentials when you created the node, - you can also pass in a Spotify URL of a playlist, album or track and it will be parsed - accordingly. - You can also pass in a discord.py Context object to get a Context object on any track you search. """ @@ -579,19 +575,12 @@ class Player(VoiceProtocol): end: int = 0, ignore_if_playing: bool = False ) -> Track: - """Plays a track. If a Spotify track is passed in, it will be handled accordingly.""" + """Plays a track.""" if not self._node: return track - if track.spotify: - if not track.original: - search_results = await self._node.get_tracks(f"{track.author} - {track.title}", requester=track.requester) - if not search_results: - raise TrackLoadError("Can't find a playable source!") - track.original = search_results[0] - data = { - "encodedTrack": track.original.track_id if track.original else track.track_id, + "encodedTrack": track.track_id, "position": str(start if start else track.position) } diff --git a/voicelink/pool.py b/voicelink/pool.py index 593695e..4965289 100644 --- a/voicelink/pool.py +++ b/voicelink/pool.py @@ -35,13 +35,11 @@ from typing import Dict, Optional, TYPE_CHECKING, Union, List from urllib.parse import quote from . import ( - __version__, - spotify, + __version__ ) from .enums import SearchType, NodeAlgorithm from .exceptions import ( - InvalidSpotifyClientAuthorization, NodeConnectionFailure, NodeCreationError, NodeException, @@ -57,10 +55,6 @@ from .ratelimit import YTRatelimit, YTToken, STRATEGY if TYPE_CHECKING: from .player import Player -SPOTIFY_URL_REGEX = re.compile( - r"https?://open.spotify.com/(?Palbum|playlist|track|artist)/(?P[a-zA-Z0-9]+)" -) - DISCORD_MP3_URL_REGEX = re.compile( r"https?://cdn.discordapp.com/attachments/(?P[0-9]+)/" r"(?P[0-9]+)/(?P[a-zA-Z0-9_.]+)+" @@ -74,8 +68,7 @@ NODE_VERSION = "v4" class Node: """The base class for a node. - This node object represents a Lavalink node. - To enable Spotify searching, pass in a proper Spotify Client ID and Spotify Client Secret + This node object represents a Lavalink node. """ def __init__( @@ -91,8 +84,6 @@ class Node: heartbeat: int = 30, yt_ratelimit: dict = None, session: Optional[aiohttp.ClientSession] = None, - spotify_client_id: Optional[str] = None, - spotify_client_secret: Optional[str] = None, resume_key: Optional[str] = None, logger: Optional[logging.Logger] = None ): @@ -127,10 +118,6 @@ class Node: self._players: Dict[int, Player] = {} self._info: Optional[NodeInfo] = None - self._spotify_client_id: Optional[str] = spotify_client_id - self._spotify_client_secret: Optional[str] = spotify_client_secret - self._spotify_client: Optional[spotify.Client] = None - self.yt_ratelimit: Optional[YTRatelimit] = STRATEGY.get(yt_ratelimit.get("strategy"))(self, yt_ratelimit) if yt_ratelimit else None self._bot.add_listener(self._update_handler, "on_socket_response") @@ -145,15 +132,6 @@ class Node: """Takes a guild ID as a parameter. Returns a voicelink Player object.""" return self._players.get(guild_id, None) - @property - def spotify_client(self) -> Optional[spotify.Client]: - if not self._spotify_client: - self._spotify_client = spotify.Client( - self._spotify_client_id, self._spotify_client_secret - ) - - return self._spotify_client - @property def is_connected(self) -> bool: """"Property which returns whether this node is connected or not""" @@ -384,56 +362,15 @@ class Node: ) -> Union[List[Track], Playlist]: """Fetches tracks from the node's REST api to parse into Lavalink. - If you passed in Spotify API credentials, you can also pass in a - Spotify URL of a playlist, album or track and it will be parsed accordingly. - You can also pass in a discord.py Context object to get a Context object on any track you search. """ if not URL_REGEX.match(query): - if search_type == SearchType.SPOTIFY: - return await self.spotifySearch(query=query, requester=requester) - - else: + if ':' not in query: query = f"{search_type}:{query}" - if SPOTIFY_URL_REGEX.match(query): - try: - spotify_results = await self.spotify_client.search(query=query) - except Exception as _: - raise TrackLoadError("Not able to find the provided Spotify entity, is it private?") - - if isinstance(spotify_results, spotify.Track): - return [ - Track( - track_id=None, - info=spotify_results.to_dict(), - requester=requester, - search_type=search_type, - spotify_track=spotify_results, - ) - ] - - tracks = [ - Track( - track_id=None, - info=track.to_dict(), - requester=requester, - search_type=search_type, - spotify_track=track, - ) for track in spotify_results.tracks if track.uri - ] - - return Playlist( - playlist_info={"name": spotify_results.name, "selectedTrack": 0}, - tracks=tracks, - requester=requester, - spotify=True, - spotify_playlist=spotify_results - ) - - elif DISCORD_MP3_URL_REGEX.match(query): + if DISCORD_MP3_URL_REGEX.match(query): data = await self.send(RequestMethod.GET, f"loadtracks?identifier={quote(query)}") try: @@ -463,7 +400,7 @@ class Node: elif load_type == "empty": return None - elif load_type == "playlist": + elif load_type in ("playlist", "recommendations"): data = data.get("data") return Playlist( @@ -491,57 +428,18 @@ class Node: requester=requester ) ] - - async def spotifySearch(self, query: str, *, requester: Member) -> Optional[List[Track]]: - try: - if not self.spotify_client: - raise InvalidSpotifyClientAuthorization( - "You did not provide proper Spotify client authorization credentials. " - "If you would like to use the Spotify searching feature, " - "please obtain Spotify API credentials here: https://developer.spotify.com/" - ) - - tracks = await self._spotify_client.track_search(query=query) - except Exception as _: - raise TrackLoadError("Not able to find the provided Spotify entity, is it private?") - - return [ - Track( - track_id=None, - requester=requester, - search_type=SearchType.YOUTUBE, - spotify_track=track, - info=track.to_dict() - ) - for track in tracks ] async def get_recommendations(self, track: Track, limit: int = 20) -> List[Optional[Track]]: - if track.spotify: - if not self.spotify_client: - return [] - - spotify_tracks = await self.spotify_client.similar_track(seed_tracks=track.identifier, limit=limit) - - tracks = [ - Track( - track_id=None, - search_type=SearchType.YOUTUBE, - spotify_track=track, - info=track.to_dict(), - requester=self.bot.user - ) - for track in spotify_tracks - ] + if track.source == "youtube": + query = f"https://www.youtube.com/watch?v={track.identifier}&list=RD{track.identifier}" - else: - if track.source != 'youtube': - return [] - - tracks = await self.get_tracks( - f"https://www.youtube.com/watch?v={track.identifier}&list=RD{track.identifier}", - requester=self.bot.user - ) + elif track.source == "spotify": + query = f"sprec:seed_tracks={track.identifier}" + if not query: + return [] + + tracks = await self.get_tracks(query=query, requester=self.bot.user) if isinstance(tracks, Playlist): tracks = tracks.tracks @@ -639,14 +537,11 @@ class NodePool: secure: bool = False, heartbeat: int = 30, yt_ratelimit: dict = None, - spotify_client_id: Optional[str] = None, - spotify_client_secret: Optional[str] = None, session: Optional[aiohttp.ClientSession] = None, resume_key: Optional[str] = None, logger: Optional[logging.Logger] = None ) -> Node: """Creates a Node object to be then added into the node pool. - For Spotify searching capabilites, pass in valid Spotify API credentials. """ if identifier in cls._nodes.keys(): raise NodeCreationError(f"A node with identifier '{identifier}' already exists.") @@ -657,8 +552,7 @@ class NodePool: node = Node( pool=cls, bot=bot, host=host, port=port, password=password, identifier=identifier, secure=secure, heartbeat=heartbeat, yt_ratelimit=yt_ratelimit, - session=session, spotify_client_id=spotify_client_id, spotify_client_secret=spotify_client_secret, - resume_key=resume_key, logger=logger + session=session, resume_key=resume_key, logger=logger ) await node.connect() diff --git a/voicelink/spotify/__init__.py b/voicelink/spotify/__init__.py deleted file mode 100644 index a7a4a10..0000000 --- a/voicelink/spotify/__init__.py +++ /dev/null @@ -1,26 +0,0 @@ -"""MIT License - -Copyright (c) 2022 Vocard Development - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -SOFTWARE. -""" - -from .exceptions import InvalidSpotifyURL, SpotifyRequestException -from .objects import * -from .client import Client diff --git a/voicelink/spotify/client.py b/voicelink/spotify/client.py deleted file mode 100644 index c247b9c..0000000 --- a/voicelink/spotify/client.py +++ /dev/null @@ -1,175 +0,0 @@ -"""MIT License - -Copyright (c) 2022 Vocard Development - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -SOFTWARE. -""" - -import re -import time -import aiohttp - -from base64 import b64encode -from typing import ( - List, - Dict, - Union, - Optional -) - -from .objects import Track, Album, Artist, Playlist, Category -from .exceptions import InvalidSpotifyURL, SpotifyRequestException - -BASE_URL = "https://api.spotify.com/v1/" -GRANT_URL = "https://accounts.spotify.com/api/token" -ANONYMOUS_GRANT_URL = "https://open.spotify.com/get_access_token" -REQUEST_URL = BASE_URL + "{type}s/{id}" -SEARCH_URL = BASE_URL + "search?q={query}&type={type}&limit={limit}" -SUGGESTION_URL = BASE_URL + "recommendations?limit={limit}&seed_tracks={seed_tracks}" -SPOTIFY_URL_REGEX = re.compile( - r"https?://open.spotify.com/(?Palbum|playlist|track|artist)/(?P[a-zA-Z0-9]+)" -) - -class Client: - """The base client for the Spotify module of Voicelink. - 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: Optional[str] = client_id - self._client_secret: Optional[str] = client_secret - - self.session: aiohttp.ClientSession = aiohttp.ClientSession() - - self._bearer_token: str = None - self._expiry: int = 0 - self._auth_token: bytes = b64encode(f"{self._client_id}:{self._client_secret}".encode()) - self._grant_headers: Dict[str, str] = {"Authorization": f"Basic {self._auth_token.decode()}"} - self._bearer_headers: Dict[str, str] = None - - self._categories: List[Category] = [] - - async def _fetch_bearer_token(self) -> None: - """Fetches and stores a bearer token for API authentication.""" - if self._client_id and self._client_secret: - url, data = GRANT_URL, {"grant_type": "client_credentials"} - else: - url, data = ANONYMOUS_GRANT_URL, None - - async with self.session.post(url, data=data, headers=self._grant_headers) if data else self.session.get(url) as resp: - if resp.status != 200: - raise SpotifyRequestException( - f"Error fetching bearer token: {resp.status} {resp.reason}" - ) - - response_data: Dict = await resp.json() - - if self._client_id and self._client_secret: - self._bearer_token = response_data["access_token"] - self._expiry = time.time() + int(response_data["expires_in"]) - 10 - else: - self._bearer_token = response_data["accessToken"] - self._expiry = response_data["accessTokenExpirationTimestampMs"] / 1000 - - self._bearer_headers = {"Authorization": f"Bearer {self._bearer_token}"} - - async def get_request(self, url: str) -> Dict: - """Performs a GET request to the specified URL with authorization headers.""" - if not self._bearer_token or time.time() >= self._expiry: - await self._fetch_bearer_token() - - async with self.session.get(url, headers=self._bearer_headers) as resp: - if resp.status != 200: - raise SpotifyRequestException( - f"Error while fetching results: {resp.status} {resp.reason}" - ) - - return await resp.json() - - async def track_search(self, query: str, track: str = "track", limit: int = 10) -> List[Track]: - """Searches for tracks based on the provided query and returns a list of Track objects.""" - request_url = SEARCH_URL.format(query=query, type=track, limit=limit) - data = await self.get_request(request_url) - return [ Track(track) for track in data['tracks']['items'] ] - - async def similar_track(self, seed_tracks: str, *, limit: int = 10) -> List[Track]: - """Retrieves tracks similar to the provided seed tracks and returns them as Track objects.""" - request_url = SUGGESTION_URL.format(limit=limit, seed_tracks=seed_tracks) - data = await self.get_request(request_url) - return [ Track(track) for track in data['tracks'] ] - - async def search(self, *, query: str) -> Union[Track, Album, Playlist]: - """Searches for an item (track, album, artist, or playlist) by query and returns the corresponding object.""" - result = SPOTIFY_URL_REGEX.match(query) - if not result: - raise InvalidSpotifyURL("The Spotify link provided is not valid.") - - spotify_type = result.group("type") - spotify_id = result.group("id") - request_url = REQUEST_URL.format(type=spotify_type, id=spotify_id) - - if isArtist := (spotify_type == "artist"): - request_url += "/top-tracks?market=US" - - data = await self.get_request(request_url) - - if spotify_type == "track": - return Track(data) - elif spotify_type == "album": - return Album(data) - elif isArtist: - return Artist(data) - - tracks = [ - Track(track["track"]) - for track in data["tracks"]["items"] if track.get("track") is not None - ] - - if not tracks: - raise SpotifyRequestException("This playlist is empty and therefore cannot be queued.") - - next_page_url = data["tracks"].get("next") - - while next_page_url: - next_data = await self.get_request(next_page_url) - tracks.extend([ - Track(track["track"]) - for track in next_data.get("items", []) if track.get("track") is not None - ]) - next_page_url = next_data.get("next") - - return Playlist(data, tracks) - - async def get_categories(self) -> List[Category]: - """Fetches and returns available music categories from the Spotify API.""" - if not self._categories: - request_url = f"{BASE_URL}browse/categories" - - while request_url: - data = await self.get_request(request_url) - items = data.get("categories", {}).get("items", []) - self._categories.extend(Category(item) for item in items) - request_url = data.get("categories", {}).get("next") - - return self._categories - - async def close(self) -> None: - """Closes the HTTP session used for making API requests.""" - await self.session.close() \ No newline at end of file diff --git a/voicelink/spotify/exceptions.py b/voicelink/spotify/exceptions.py deleted file mode 100644 index e421fbf..0000000 --- a/voicelink/spotify/exceptions.py +++ /dev/null @@ -1,8 +0,0 @@ -class SpotifyRequestException(Exception): - """An error occurred when making a request to the Spotify API""" - pass - - -class InvalidSpotifyURL(Exception): - """An invalid Spotify URL was passed""" - pass diff --git a/voicelink/spotify/objects.py b/voicelink/spotify/objects.py deleted file mode 100644 index 59a645d..0000000 --- a/voicelink/spotify/objects.py +++ /dev/null @@ -1,135 +0,0 @@ -class Track: - """The base class for a Spotify Track""" - - __slots__ = ( - "name", - "artists", - "artist_id", - "length", - "id", - "image", - "uri" - ) - - def __init__(self, data: dict, image = None) -> None: - self.name: str = data.get('name', 'Unknown') - self.artists: str = ", ".join(filter(None, (artist["name"] for artist in data.get('artists')))) - self.artist_id: list[str] = [artist["id"] for artist in data.get('artists')] - self.length: int = data.get('duration_ms') - self.id: str = data.get('id') - self.image: str = images[0]["url"] if (images := data.get("album", {}).get("images")) else image - self.uri: str = None if data["is_local"] else data["external_urls"]["spotify"] - - def to_dict(self) -> dict: - return { - "title": self.name, - "author": self.artists, - "length": self.length, - "identifier": self.id, - "artist_id": self.artist_id, - "uri": self.uri, - "isStream": False, - "isSeekable": True, - "position": 0, - "artworkUrl": self.image - } - - def __repr__(self) -> str: - return ( - f"" - ) - -class Album: - """The base class for a Spotify album""" - - __slots__ = ( - "name", - "artists", - "image", - "tracks", - "total_tracks", - "id", - "uri" - ) - - def __init__(self, data: dict) -> None: - self.name: str = data.get('name', 'Unknown') - self.artists: str = ", ".join(filter(None, (artist["name"] for artist in data.get('artists')))) - self.image: str = data["images"][0]["url"] - self.tracks: list[Track] = [Track(track, image=self.image) for track in data["tracks"]["items"]] - self.total_tracks: int = data["total_tracks"] - self.id: str = data.get('id') - self.uri: str = data["external_urls"]["spotify"] - - def __repr__(self) -> str: - return ( - f"" - ) - -class Artist: - """The base class for a Spotify playlist""" - - __slots__ = ( - "tracks", - "image", - "total_tracks", - "owner", - "id", - "uri", - "name" - ) - def __init__(self, data: dict) -> None: - self.tracks: list[Track] = [Track(track) for track in data['tracks']] - if self.tracks: - self.image: str = self.tracks[0].image - self.total_tracks: int = len(self.tracks) - self.owner: str = self.tracks[0].artists - self.id: str = self.tracks[0].artist_id - self.uri: str = data['tracks'][0]['album']['artists'][0]['external_urls']['spotify'] - self.name: str = f"Top tracks - {self.owner}" - - def __repr__(self) -> str: - return ( - f"" - ) - -class Playlist: - """The base class for a Spotify playlist""" - - __slots__ = ( - "name", - "tracks", - "owner", - "total_tracks", - "id", - "image", - "uri" - ) - - def __init__(self, data: dict, tracks: list[Track]) -> None: - self.name: str = data.get('name', 'Unknown') - self.tracks: list[Track] = tracks - self.owner: str = data["owner"]["display_name"] - self.total_tracks: int = data["tracks"]["total"] - self.id: str = data.get('id') - self.image: str = data["images"][0]["url"] if len(data.get("images", [])) else None - self.uri: str = data["external_urls"]["spotify"] - - def __repr__(self) -> str: - return ( - f"" - ) - -class Category: - def __init__(self, data: dict) -> None: - self.href: str = data.get("href") - self.id: str = data.get("id") - self.name: str = data.get("name") - self.icon: str = data.get("icons", [{}])[0].get("url") - - def __repr__(self) -> str: - return (f"