majjo/Vocard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replaced internal spotify with lavalink

Browse Source
This commit is contained in:
Choco
2025-03-22 18:42:28 +08:00
parent 7e8bfb6e33
commit 5498c60466
15 changed files with 29 additions and 601 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
addons/settings.py
View File

@@ -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")

4
cogs/basic.py
View File

@@ -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 []

4
cogs/listeners.py
View File

@@ -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
)

47
ipc/methods.py
View File

@@ -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:

2
settings Example.json
View File

@@ -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",

2
views/debug.py
View File

@@ -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
)

2
voicelink/enums.py
View File

@@ -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.

20
voicelink/exceptions.py
View File

@@ -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

54
voicelink/objects.py
View File

@@ -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"<Voicelink.playlist name={self.name!r} track_count={len(self.tracks)}>"
@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)

15
voicelink/player.py
View File

@@ -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)
}

134
voicelink/pool.py
View File

@@ -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/(?P<type>album|playlist|track|artist)/(?P<id>[a-zA-Z0-9]+)"
)
DISCORD_MP3_URL_REGEX = re.compile(
r"https?://cdn.discordapp.com/attachments/(?P<channel_id>[0-9]+)/"
r"(?P<message_id>[0-9]+)/(?P<file>[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()

26
voicelink/spotify/__init__.py
View File

@@ -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

175
voicelink/spotify/client.py
View File

@@ -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/(?P<type>album|playlist|track|artist)/(?P<id>[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()

8
voicelink/spotify/exceptions.py
View File

@@ -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

135
voicelink/spotify/objects.py
View File

@@ -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"<Voicelink.spotify.Track name={self.name} artists={self.artists} "
f"length={self.length} id={self.id}>"
)
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"<Voicelink.spotify.Album name={self.name} artists={self.artists} id={self.id} "
f"total_tracks={self.total_tracks} tracks={self.tracks}>"
)
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"<Voicelink.spotify.Artist name={self.name} owner={self.owner} id={self.id} "
f"total_tracks={self.total_tracks} tracks={self.tracks}>"
)
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"<Voicelink.spotify.Playlist name={self.name} owner={self.owner} id={self.id} "
f"total_tracks={self.total_tracks} tracks={self.tracks}>"
)
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"<Voicelink.spotify.Category name={self.name} id={self.id}")