Merge pull request #53 from ChocoMeow/Replace-internal-spotify-with-lavalink
Replace internal spotify with lavalink
This commit is contained in:
Choco
178
.gitignore
vendored
178
.gitignore
vendored
@@ -1,5 +1,179 @@
|
||||
Vocard.rar
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
|
||||
# C extensions
|
||||
*.so
|
||||
|
||||
# Distribution / packaging
|
||||
.Python
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
downloads/
|
||||
eggs/
|
||||
.eggs/
|
||||
lib/
|
||||
lib64/
|
||||
parts/
|
||||
sdist/
|
||||
var/
|
||||
wheels/
|
||||
share/python-wheels/
|
||||
*.egg-info/
|
||||
.installed.cfg
|
||||
*.egg
|
||||
MANIFEST
|
||||
**/.DS_Store
|
||||
|
||||
# PyInstaller
|
||||
# Usually these files are written by a python script from a template
|
||||
# before PyInstaller builds the exe, so as to inject date/other infos into it.
|
||||
*.manifest
|
||||
*.spec
|
||||
|
||||
# Installer logs
|
||||
pip-log.txt
|
||||
pip-delete-this-directory.txt
|
||||
|
||||
# Unit test / coverage reports
|
||||
htmlcov/
|
||||
.tox/
|
||||
.nox/
|
||||
.coverage
|
||||
.coverage.*
|
||||
.cache
|
||||
nosetests.xml
|
||||
coverage.xml
|
||||
*.cover
|
||||
*.py,cover
|
||||
.hypothesis/
|
||||
.pytest_cache/
|
||||
cover/
|
||||
|
||||
# Translations
|
||||
*.mo
|
||||
*.pot
|
||||
|
||||
# Django stuff:
|
||||
*.log
|
||||
local_settings.py
|
||||
db.sqlite3
|
||||
db.sqlite3-journal
|
||||
|
||||
# Flask stuff:
|
||||
instance/
|
||||
.webassets-cache
|
||||
|
||||
# Scrapy stuff:
|
||||
.scrapy
|
||||
|
||||
# Sphinx documentation
|
||||
docs/_build/
|
||||
|
||||
# PyBuilder
|
||||
.pybuilder/
|
||||
target/
|
||||
|
||||
# Jupyter Notebook
|
||||
.ipynb_checkpoints
|
||||
|
||||
# IPython
|
||||
profile_default/
|
||||
ipython_config.py
|
||||
|
||||
# pyenv
|
||||
# For a library or package, you might want to ignore these files since the code is
|
||||
# intended to run in multiple environments; otherwise, check them in:
|
||||
# .python-version
|
||||
|
||||
# pipenv
|
||||
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
|
||||
# However, in case of collaboration, if having platform-specific dependencies or dependencies
|
||||
# having no cross-platform support, pipenv may install dependencies that don't work, or not
|
||||
# install all needed dependencies.
|
||||
#Pipfile.lock
|
||||
|
||||
# UV
|
||||
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
|
||||
# This is especially recommended for binary packages to ensure reproducibility, and is more
|
||||
# commonly ignored for libraries.
|
||||
#uv.lock
|
||||
|
||||
# poetry
|
||||
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
|
||||
# This is especially recommended for binary packages to ensure reproducibility, and is more
|
||||
# commonly ignored for libraries.
|
||||
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
|
||||
#poetry.lock
|
||||
|
||||
# pdm
|
||||
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
|
||||
#pdm.lock
|
||||
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
|
||||
# in version control.
|
||||
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
|
||||
.pdm.toml
|
||||
.pdm-python
|
||||
.pdm-build/
|
||||
|
||||
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
|
||||
__pypackages__/
|
||||
|
||||
# Celery stuff
|
||||
celerybeat-schedule
|
||||
celerybeat.pid
|
||||
|
||||
# SageMath parsed files
|
||||
*.sage.py
|
||||
|
||||
# Environments
|
||||
.env
|
||||
*.pyc
|
||||
.venv
|
||||
env/
|
||||
venv/
|
||||
ENV/
|
||||
env.bak/
|
||||
venv.bak/
|
||||
|
||||
# Spyder project settings
|
||||
.spyderproject
|
||||
.spyproject
|
||||
|
||||
# Rope project settings
|
||||
.ropeproject
|
||||
|
||||
# mkdocs documentation
|
||||
/site
|
||||
|
||||
# mypy
|
||||
.mypy_cache/
|
||||
.dmypy.json
|
||||
dmypy.json
|
||||
|
||||
# Pyre type checker
|
||||
.pyre/
|
||||
|
||||
# pytype static type analyzer
|
||||
.pytype/
|
||||
|
||||
# Cython debug symbols
|
||||
cython_debug/
|
||||
|
||||
# PyCharm
|
||||
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
|
||||
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
|
||||
# and can be added to the global gitignore or merged into this file. For a more nuclear
|
||||
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
|
||||
#.idea/
|
||||
|
||||
# Ruff stuff:
|
||||
.ruff_cache/
|
||||
|
||||
# PyPI configuration file
|
||||
.pypirc
|
||||
|
||||
# Custom file
|
||||
settings.json
|
||||
logs
|
||||
@@ -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")
|
||||
|
||||
@@ -86,24 +86,18 @@ class Basic(commands.Cog):
|
||||
return [app_commands.Choice(name=c.capitalize(), value=c) for c in self.bot.cogs if c not in ["Nodes", "Task"] and current in c]
|
||||
|
||||
async def play_autocomplete(self, interaction: discord.Interaction, current: str) -> list:
|
||||
if voicelink.pool.URL_REGEX.match(current): return []
|
||||
if voicelink.pool.URL_REGEX.match(current):
|
||||
return []
|
||||
|
||||
if current:
|
||||
node = voicelink.NodePool.get_node()
|
||||
if node and node.spotify_client:
|
||||
try:
|
||||
tracks: list[voicelink.Track] = await node.spotifySearch(current, requester=interaction.user)
|
||||
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 []
|
||||
if not node:
|
||||
return []
|
||||
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] if tracks else []
|
||||
|
||||
history: dict[str, str] = {}
|
||||
for track_id in reversed(await get_user(interaction.user.id, "history")):
|
||||
track_dict = voicelink.decode(track_id)
|
||||
history[track_dict["identifier"]] = track_dict
|
||||
|
||||
history_tracks = [app_commands.Choice(name=truncate_string(f"🕒 {track['author']} - {track['title']}", 100), value=track['uri']) for track in history.values() if len(track['uri']) <= 100][:25]
|
||||
return history_tracks
|
||||
history = {track["identifier"]: track for track_id in reversed(await get_user(interaction.user.id, "history")) if (track := voicelink.decode(track_id))["uri"]}
|
||||
return [app_commands.Choice(name=truncate_string(f"🕒 {track['author']} - {track['title']}", 100), value=track['uri']) for track in history.values() if len(track['uri']) <= 100][:25]
|
||||
|
||||
@commands.hybrid_command(name="connect", aliases=get_aliases("connect"))
|
||||
@app_commands.describe(channel="Provide a channel to connect.")
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -14,8 +14,10 @@ plugins:
|
||||
clients:
|
||||
- MUSIC
|
||||
- ANDROID_VR
|
||||
- ANDROID_MUSIC
|
||||
- WEB
|
||||
- WEBEMBEDDED
|
||||
- WEBEMBEDDED
|
||||
- TVHTML5EMBEDDED
|
||||
# The below section of the config allows setting specific options for each client, such as the requests they will handle.
|
||||
# If an option, or client, is unspecified, then the default option value/client values will be used instead.
|
||||
# If a client is configured, but is not registered above, the options for that client will be ignored.
|
||||
@@ -32,9 +34,76 @@ plugins:
|
||||
# Example: Configuring a client to exclusively be used for video loading and playback.
|
||||
playlistLoading: false # Disables loading of playlists and mixes.
|
||||
searching: false # Disables the ability to search for videos.
|
||||
lavasrc:
|
||||
providers: # Custom providers for track loading. This is the default
|
||||
# - "dzisrc:%ISRC%" # Deezer ISRC provider
|
||||
# - "dzsearch:%QUERY%" # Deezer search provider
|
||||
- "ytsearch:\"%ISRC%\"" # Will be ignored if track does not have an ISRC. See https://en.wikipedia.org/wiki/International_Standard_Recording_Code
|
||||
- "ytsearch:%QUERY%" # Will be used if track has no ISRC or no track could be found for the ISRC
|
||||
# you can add multiple other fallback sources here
|
||||
sources:
|
||||
spotify: true # Enable Spotify source
|
||||
applemusic: false # Enable Apple Music source
|
||||
deezer: false # Enable Deezer source
|
||||
yandexmusic: false # Enable Yandex Music source
|
||||
flowerytts: false # Enable Flowery TTS source
|
||||
youtube: false # Enable YouTube search source (https://github.com/topi314/LavaSearch)
|
||||
vkmusic: false # Enable Vk Music source
|
||||
lyrics-sources:
|
||||
spotify: false # Enable Spotify lyrics source
|
||||
deezer: false # Enable Deezer lyrics source
|
||||
youtube: false # Enable YouTube lyrics source
|
||||
yandexmusic: false # Enable Yandex Music lyrics source
|
||||
vkmusic: false # Enable Vk Music lyrics source
|
||||
spotify:
|
||||
clientId: ""
|
||||
clientSecret: ""
|
||||
# spDc: "your sp dc cookie" # the sp dc cookie used for accessing the spotify lyrics api
|
||||
countryCode: "US" # the country code you want to use for filtering the artists top tracks. See https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2
|
||||
playlistLoadLimit: 6 # The number of pages at 100 tracks each
|
||||
albumLoadLimit: 6 # The number of pages at 50 tracks each
|
||||
resolveArtistsInSearch: true # Whether to resolve artists in track search results (can be slow)
|
||||
localFiles: false # Enable local files support with Spotify playlists. Please note `uri` & `isrc` will be `null` & `identifier` will be `"local"`
|
||||
applemusic:
|
||||
countryCode: "US" # the country code you want to use for filtering the artists top tracks and language. See https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2
|
||||
mediaAPIToken: "your apple music api token" # apple music api token
|
||||
# or specify an apple music key
|
||||
keyID: "your key id"
|
||||
teamID: "your team id"
|
||||
musicKitKey: |
|
||||
-----BEGIN PRIVATE KEY-----
|
||||
your key
|
||||
-----END PRIVATE KEY-----
|
||||
playlistLoadLimit: 6 # The number of pages at 300 tracks each
|
||||
albumLoadLimit: 6 # The number of pages at 300 tracks each
|
||||
deezer:
|
||||
masterDecryptionKey: "your master decryption key" # the master key used for decrypting the deezer tracks. (yes this is not here you need to get it from somewhere else)
|
||||
# arl: "your deezer arl" # the arl cookie used for accessing the deezer api this is optional but required for formats above MP3_128
|
||||
formats: [ "FLAC", "MP3_320", "MP3_256", "MP3_128", "MP3_64", "AAC_64" ] # the formats you want to use for the deezer tracks. "FLAC", "MP3_320", "MP3_256" & "AAC_64" are only available for premium users and require a valid arl
|
||||
yandexmusic:
|
||||
accessToken: "your access token" # the token used for accessing the yandex music api. See https://github.com/TopiSenpai/LavaSrc#yandex-music
|
||||
playlistLoadLimit: 1 # The number of pages at 100 tracks each
|
||||
albumLoadLimit: 1 # The number of pages at 50 tracks each
|
||||
artistLoadLimit: 1 # The number of pages at 10 tracks each
|
||||
flowerytts:
|
||||
voice: "default voice" # (case-sensitive) get default voice from here https://api.flowery.pw/v1/tts/voices
|
||||
translate: false # whether to translate the text to the native language of voice
|
||||
silence: 0 # the silence parameter is in milliseconds. Range is 0 to 10000. The default is 0.
|
||||
speed: 1.0 # the speed parameter is a float between 0.5 and 10. The default is 1.0. (0.5 is half speed, 2.0 is double speed, etc.)
|
||||
audioFormat: "mp3" # supported formats are: mp3, ogg_opus, ogg_vorbis, aac, wav, and flac. Default format is mp3
|
||||
youtube:
|
||||
countryCode: "US" # the country code you want to use for searching lyrics via ISRC. See https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2
|
||||
vkmusic:
|
||||
userToken: "your user token" # This token is needed for authorization in the api. Guide: https://github.com/topi314/LavaSrc#vk-music
|
||||
playlistLoadLimit: 1 # The number of pages at 50 tracks each
|
||||
artistLoadLimit: 1 # The number of pages at 10 tracks each
|
||||
recommendationsLoadLimit: 10 # Number of tracks
|
||||
lavalink:
|
||||
plugins:
|
||||
- dependency: "dev.lavalink.youtube:youtube-plugin:1.11.4"
|
||||
- dependency: "dev.lavalink.youtube:youtube-plugin:1.11.5"
|
||||
snapshot: false
|
||||
- dependency: "com.github.topi314.lavasrc:lavasrc-plugin:06b7cab"
|
||||
snapshot: true
|
||||
# - dependency: "com.github.example:example-plugin:1.0.0" # required, the coordinates of your plugin
|
||||
# repository: "https://maven.example.com/releases" # optional, defaults to the Lavalink releases repository by default
|
||||
# snapshot: false # optional, defaults to false, used to tell Lavalink to use the snapshot repository instead of the release repository
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -29,7 +29,7 @@ __version__ = "v2.7.0b4"
|
||||
|
||||
GITHUB_API_URL = "https://api.github.com/repos/ChocoMeow/Vocard/releases/latest"
|
||||
VOCARD_URL = "https://github.com/ChocoMeow/Vocard/archive/"
|
||||
IGNORE_FILES = ["settings.json", "logs"]
|
||||
IGNORE_FILES = ["settings.json", "logs", "last-session.json"]
|
||||
|
||||
class bcolors:
|
||||
WARNING = '\033[93m'
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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,7 @@ class Track:
|
||||
"author",
|
||||
"uri",
|
||||
"source",
|
||||
"spotify",
|
||||
"artist_id",
|
||||
"original",
|
||||
"_search_type",
|
||||
"spotify_track",
|
||||
"thumbnail",
|
||||
"emoji",
|
||||
"length",
|
||||
@@ -73,7 +68,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 +77,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 +131,9 @@ class Playlist:
|
||||
|
||||
__slots__ = (
|
||||
"playlist_info",
|
||||
"tracks_raw",
|
||||
"spotify",
|
||||
"name",
|
||||
"spotify_playlist",
|
||||
"_thumbnail",
|
||||
"_uri",
|
||||
"thumbnail",
|
||||
"uri",
|
||||
"tracks"
|
||||
)
|
||||
|
||||
@@ -158,29 +143,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 +160,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)
|
||||
|
||||
@@ -151,21 +151,19 @@ class Player(VoiceProtocol):
|
||||
@property
|
||||
def position(self) -> float:
|
||||
"""Property which returns the player's position in a track in milliseconds"""
|
||||
current = self._current.original
|
||||
|
||||
if not self.is_playing or not self._current:
|
||||
return 0
|
||||
|
||||
if self.is_paused:
|
||||
return min(self._last_position, current.length)
|
||||
return min(self._last_position, self._current.length)
|
||||
|
||||
difference = (time.time() * 1000) - self._last_update
|
||||
position = self._last_position + difference
|
||||
|
||||
if position > current.length:
|
||||
if position > self._current.length:
|
||||
return 0
|
||||
|
||||
return min(position, current.length)
|
||||
return min(position, self._current.length)
|
||||
|
||||
@property
|
||||
def is_playing(self) -> bool:
|
||||
@@ -524,10 +522,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 +573,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)
|
||||
}
|
||||
|
||||
@@ -674,7 +661,7 @@ class Player(VoiceProtocol):
|
||||
|
||||
async def seek(self, position: float, requester: Member = None) -> float:
|
||||
"""Seeks to a position in the currently playing track milliseconds"""
|
||||
if position < 0 or position > self._current.original.length:
|
||||
if position < 0 or position > self._current.length:
|
||||
raise TrackInvalidPosition("Seek position must be between 0 and the track length")
|
||||
|
||||
await self.send(method=RequestMethod.PATCH, data={"position": position})
|
||||
|
||||
@@ -31,17 +31,15 @@ import logging
|
||||
|
||||
from discord import Client, Member
|
||||
from discord.ext.commands import Bot
|
||||
from typing import Dict, Optional, TYPE_CHECKING, Union, List
|
||||
from typing import Dict, Optional, Union, List, Any, TYPE_CHECKING
|
||||
from urllib.parse import quote
|
||||
|
||||
from . import (
|
||||
__version__,
|
||||
spotify,
|
||||
__version__
|
||||
)
|
||||
|
||||
from .enums import SearchType, NodeAlgorithm
|
||||
from .exceptions import (
|
||||
InvalidSpotifyClientAuthorization,
|
||||
NodeConnectionFailure,
|
||||
NodeCreationError,
|
||||
NodeException,
|
||||
@@ -57,15 +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_.]+)+"
|
||||
)
|
||||
|
||||
URL_REGEX = re.compile(
|
||||
r"https?://(?:www\.)?.+"
|
||||
)
|
||||
@@ -74,8 +63,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 +79,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 +113,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 +127,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"""
|
||||
@@ -382,166 +355,49 @@ class Node:
|
||||
requester: Member,
|
||||
search_type: SearchType = SearchType.YOUTUBE
|
||||
) -> Union[List[Track], Playlist]:
|
||||
"""Fetches tracks from the node's REST api to parse into Lavalink.
|
||||
"""
|
||||
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.
|
||||
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:
|
||||
query = f"{search_type}:{query}"
|
||||
if not URL_REGEX.match(query) and ':' 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):
|
||||
data = await self.send(RequestMethod.GET, f"loadtracks?identifier={quote(query)}")
|
||||
|
||||
try:
|
||||
track: dict = data["data"]
|
||||
except:
|
||||
raise TrackLoadError("Not able to find the provided track.")
|
||||
|
||||
return [
|
||||
Track(
|
||||
track_id=track["encoded"],
|
||||
info=track["info"],
|
||||
requester=requester
|
||||
)
|
||||
]
|
||||
else:
|
||||
data = await self.send(RequestMethod.GET, f"loadtracks?identifier={quote(query)}")
|
||||
|
||||
load_type = data.get("loadType")
|
||||
response: dict[str, Any] = await self.send(RequestMethod.GET, f"loadtracks?identifier={quote(query)}")
|
||||
data = response.get("data")
|
||||
load_type = response.get("loadType")
|
||||
|
||||
if not load_type:
|
||||
raise TrackLoadError("There was an error while trying to load this track.")
|
||||
|
||||
elif load_type == "error":
|
||||
exception = data["data"]
|
||||
raise TrackLoadError(f"{exception['message']} [{exception['severity']}]")
|
||||
|
||||
|
||||
elif load_type == "empty":
|
||||
return None
|
||||
|
||||
elif load_type == "playlist":
|
||||
data = data.get("data")
|
||||
|
||||
return Playlist(
|
||||
playlist_info=data["info"],
|
||||
tracks=data["tracks"],
|
||||
requester=requester
|
||||
)
|
||||
elif load_type == "error":
|
||||
raise TrackLoadError(f"{data['message']} [{data['severity']}]")
|
||||
|
||||
elif load_type in ("playlist", "recommendations"):
|
||||
return Playlist(playlist_info=data["info"], tracks=data["tracks"], requester=requester)
|
||||
|
||||
elif load_type == "search":
|
||||
return [
|
||||
Track(
|
||||
track_id=track["encoded"],
|
||||
info=track["info"],
|
||||
requester=requester
|
||||
)
|
||||
for track in data["data"]
|
||||
]
|
||||
return [Track(track_id=track["encoded"], info=track["info"], requester=requester) for track in data]
|
||||
|
||||
elif load_type == "track":
|
||||
track = data["data"]
|
||||
return [
|
||||
Track(
|
||||
track_id=track["encoded"],
|
||||
info=track["info"],
|
||||
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 ]
|
||||
return [Track(track_id=data["encoded"], info=data["info"], requester=requester)]
|
||||
|
||||
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 +495,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 +510,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()
|
||||
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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}")
|
||||
Reference in New Issue
Block a user