nambrogi
/
pomice
mirror of https://github.com/cloudwithax/pomice.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove all Spotify client code in preparation for 1.1.8

This commit is contained in:
cloudwithax 2022-09-03 10:36:25 -04:00
parent 1f7f85df93
commit ab708a1cfb
9 changed files with 33 additions and 392 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
pomice/__init__.py
View File

@ -18,15 +18,15 @@ if not discord.__version__.startswith("2.0"):
"using 'pip install discord.py'"
)
__version__ = "1.1.7"
__version__ = "1.1.8a"
__title__ = "pomice"
__author__ = "cloudwithax"
from .enums import SearchType
from .enums import *
from .events import *
from .exceptions import *
from .filters import *
from .objects import *
from .player import Player
from .player import *
from .pool import *
from .queue import *

19
pomice/exceptions.py
View File

@ -52,25 +52,6 @@ class FilterTagAlreadyInUse(PomiceException):
pass
class SpotifyAlbumLoadFailed(PomiceException):
"""The pomice Spotify client was unable to load an album."""
pass
class SpotifyTrackLoadFailed(PomiceException):
"""The pomice Spotify client was unable to load a track."""
pass
class SpotifyPlaylistLoadFailed(PomiceException):
"""The pomice Spotify client was unable to load a playlist."""
pass
class InvalidSpotifyClientAuthorization(PomiceException):
"""No Spotify client authorization was provided for track searching."""
pass
class QueueException(Exception):
"""Base Pomice queue exception."""
pass

29
pomice/objects.py
View File

@ -21,17 +21,12 @@ class Track:
track_id: str,
info: dict,
ctx: Optional[commands.Context] = None,
spotify: bool = False,
search_type: SearchType = SearchType.ytsearch,
spotify_track = None,
):
self.track_id = track_id
self.info = info
self.spotify = spotify
self.original: Optional[Track] = None if spotify else self
self._search_type = search_type
self.spotify_track = spotify_track
self.title = info.get("title")
self.author = info.get("author")
@ -84,29 +79,21 @@ class Playlist:
playlist_info: dict,
tracks: list,
ctx: Optional[commands.Context] = None,
spotify: bool = False,
spotify_playlist = None
):
self.playlist_info = playlist_info
self.tracks_raw = tracks
self.spotify = spotify
self.name = playlist_info.get("name")
self.spotify_playlist = spotify_playlist
self._thumbnail = None
self._uri = 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["track"], info=track["info"], ctx=ctx)
for track in self.tracks_raw
]
self._thumbnail = None
self._uri = None
self.tracks = [
Track(track_id=track["track"], info=track["info"], ctx=ctx)
for track in self.tracks_raw
]
self._thumbnail = None
self._uri = None
if (index := playlist_info.get("selectedTrack")) == -1:
self.selected_track = None

49
pomice/player.py
View File

@ -17,7 +17,7 @@ from discord.ext import commands
from . import events
from .enums import SearchType
from .events import PomiceEvent, TrackEndEvent, TrackStartEvent
from .exceptions import FilterInvalidArgument, FilterTagAlreadyInUse, FilterTagInvalid, TrackInvalidPosition, TrackLoadError
from .exceptions import FilterInvalidArgument, FilterTagAlreadyInUse, FilterTagInvalid, TrackInvalidPosition
from .filters import Filter
from .objects import Track
from .pool import Node, NodePool
@ -290,44 +290,15 @@ 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."""
# Make sure we've never searched the track before
if track.original is None:
# First lets try using the tracks ISRC, every track has one (hopefully)
try:
if not track.isrc:
# We have to bare raise here because theres no other way to skip this block feasibly
raise
search: Track = (await self._node.get_tracks(
f"{track._search_type}:{track.isrc}", ctx=track.ctx))[0]
except Exception:
# First method didn't work, lets try just searching it up
try:
search: Track = (await self._node.get_tracks(
f"{track._search_type}:{track.title} - {track.author}", ctx=track.ctx))[0]
except:
# The song wasn't able to be found, raise error
raise TrackLoadError (
"No equivalent track was able to be found."
)
data = {
"op": "play",
"guildId": str(self.guild.id),
"track": search.track_id,
"startTime": str(start),
"noReplace": ignore_if_playing
}
track.original = search
track.track_id = search.track_id
# Set track_id for later lavalink searches
else:
data = {
"op": "play",
"guildId": str(self.guild.id),
"track": track.track_id,
"startTime": str(start),
"noReplace": ignore_if_playing
}
"""Plays a track."""
data = {
"op": "play",
"guildId": str(self.guild.id),
"track": track.track_id,
"startTime": str(start),
"noReplace": ignore_if_playing
}
if end > 0:
data["endTime"] = str(end)

105
pomice/pool.py
View File

@ -14,15 +14,12 @@ from discord.ext import commands
from . import (
__version__,
spotify,
)
from .enums import SearchType, NodeAlgorithm
from .exceptions import (
InvalidSpotifyClientAuthorization,
NodeConnectionFailure,
NodeCreationError,
NodeException,
NodeNotAvailable,
NoNodesAvailable,
TrackLoadError
@ -51,14 +48,13 @@ URL_REGEX = re.compile(
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
"""
def __init__(
self,
*,
pool,
bot: Client,
bot: commands.Bot,
host: str,
port: int,
password: str,
@ -66,18 +62,15 @@ class Node:
secure: bool = False,
heartbeat: int = 30,
session: Optional[aiohttp.ClientSession] = None,
spotify_client_id: Optional[str] = None,
spotify_client_secret: Optional[str] = None,
):
self._bot = bot
self._host = host
self._port = port
self._bot: commands.Bot = bot
self._host: str = host
self._port: int = port
self._pool = pool
self._password = password
self._identifier = identifier
self._heartbeat = heartbeat
self._secure = secure
self._password: str = password
self._identifier: str = identifier
self._heartbeat: str = heartbeat
self._secure: bool = secure
self._websocket_uri = f"{'wss' if self._secure else 'ws'}://{self._host}:{self._port}"
@ -99,14 +92,6 @@ class Node:
self._players: Dict[int, Player] = {}
self._spotify_client_id = spotify_client_id
self._spotify_client_secret = spotify_client_secret
if self._spotify_client_id and self._spotify_client_secret:
self._spotify_client = spotify.Client(
self._spotify_client_id, self._spotify_client_secret
)
self._bot.add_listener(self._update_handler, "on_socket_response")
def __repr__(self):
@ -133,7 +118,7 @@ class Node:
@property
def bot(self) -> Client:
def bot(self) -> commands.Bot:
"""Property which returns the discord.py client linked to this node"""
return self._bot
@ -290,9 +275,6 @@ class Node:
):
"""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.
"""
@ -300,70 +282,8 @@ class Node:
if not URL_REGEX.match(query) and not re.match(r"(?:ytm?|sc)search:.", query):
query = f"{search_type}:{query}"
if SPOTIFY_URL_REGEX.match(query):
if not self._spotify_client_id and not self._spotify_client_secret:
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/"
)
spotify_results = await self._spotify_client.search(query=query)
if isinstance(spotify_results, spotify.Track):
return [
Track(
track_id=spotify_results.id,
ctx=ctx,
search_type=search_type,
spotify=True,
spotify_track=spotify_results,
info={
"title": spotify_results.name,
"author": spotify_results.artists,
"length": spotify_results.length,
"identifier": spotify_results.id,
"uri": spotify_results.uri,
"isStream": False,
"isSeekable": True,
"position": 0,
"thumbnail": spotify_results.image,
"isrc": spotify_results.isrc
}
)
]
tracks = [
Track(
track_id=track.id,
ctx=ctx,
search_type=search_type,
spotify=True,
spotify_track=track,
info={
"title": track.name,
"author": track.artists,
"length": track.length,
"identifier": track.id,
"uri": track.uri,
"isStream": False,
"isSeekable": True,
"position": 0,
"thumbnail": track.image,
"isrc": track.isrc
}
) for track in spotify_results.tracks
]
return Playlist(
playlist_info={"name": spotify_results.name, "selectedTrack": 0},
tracks=tracks,
ctx=ctx,
spotify=True,
spotify_playlist=spotify_results
)
elif discord_url := DISCORD_MP3_URL_REGEX.match(query):
if discord_url := DISCORD_MP3_URL_REGEX.match(query):
async with self._session.get(
url=f"{self._rest_uri}/loadtracks?identifier={quote(query)}",
headers={"Authorization": self._password}
@ -508,8 +428,6 @@ class NodePool:
identifier: str,
secure: bool = False,
heartbeat: int = 30,
spotify_client_id: Optional[str] = None,
spotify_client_secret: Optional[str] = None,
session: Optional[aiohttp.ClientSession] = None,
) -> Node:
@ -522,8 +440,7 @@ class NodePool:
node = Node(
pool=cls, bot=bot, host=host, port=port, password=password,
identifier=identifier, secure=secure, heartbeat=heartbeat,
spotify_client_id=spotify_client_id,
session=session, spotify_client_secret=spotify_client_secret
session=session
)
await node.connect()

5
pomice/spotify/__init__.py
View File

@ -1,5 +0,0 @@
"""Spotify module for Pomice, made possible by cloudwithax 2021"""
from .exceptions import *
from .objects import *
from .client import Client

113
pomice/spotify/client.py
View File

@ -1,113 +0,0 @@
import re
import time
from base64 import b64encode
import aiohttp
import orjson as json
from .exceptions import InvalidSpotifyURL, SpotifyRequestException
from .objects import *
GRANT_URL = "https://accounts.spotify.com/api/token"
REQUEST_URL = "https://api.spotify.com/v1/{type}s/{id}"
SPOTIFY_URL_REGEX = re.compile(
r"https?://open.spotify.com/(?P<type>album|playlist|track|artist)/(?P<id>[a-zA-Z0-9]+)"
)
class Client:
"""The base client for the Spotify module of Pomice.
This class will do all the heavy lifting of getting all the metadata
for any Spotify URL you throw at it.
"""
def __init__(self, client_id: str, client_secret: str) -> None:
self._client_id = client_id
self._client_secret = client_secret
self.session = aiohttp.ClientSession()
self._bearer_token: str = None
self._expiry = 0
self._auth_token = b64encode(f"{self._client_id}:{self._client_secret}".encode())
self._grant_headers = {"Authorization": f"Basic {self._auth_token.decode()}"}
self._bearer_headers = None
async def _fetch_bearer_token(self) -> None:
_data = {"grant_type": "client_credentials"}
async with self.session.post(GRANT_URL, data=_data, headers=self._grant_headers) as resp:
if resp.status != 200:
raise SpotifyRequestException(
f"Error fetching bearer token: {resp.status} {resp.reason}"
)
data: dict = await resp.json(loads=json.loads)
self._bearer_token = data["access_token"]
self._expiry = time.time() + (int(data["expires_in"]) - 10)
self._bearer_headers = {"Authorization": f"Bearer {self._bearer_token}"}
async def search(self, *, query: str):
if not self._bearer_token or time.time() >= self._expiry:
await self._fetch_bearer_token()
result = SPOTIFY_URL_REGEX.match(query)
spotify_type = result.group("type")
spotify_id = result.group("id")
if not result:
raise InvalidSpotifyURL("The Spotify link provided is not valid.")
request_url = REQUEST_URL.format(type=spotify_type, id=spotify_id)
async with self.session.get(request_url, headers=self._bearer_headers) as resp:
if resp.status != 200:
raise SpotifyRequestException(
f"Error while fetching results: {resp.status} {resp.reason}"
)
data: dict = await resp.json(loads=json.loads)
if spotify_type == "track":
return Track(data)
elif spotify_type == "album":
return Album(data)
elif spotify_type == "artist":
async with self.session.get(f"{request_url}/top-tracks?market=US", headers=self._bearer_headers) as resp:
if resp.status != 200:
raise SpotifyRequestException(
f"Error while fetching results: {resp.status} {resp.reason}"
)
track_data: dict = await resp.json(loads=json.loads)
tracks = track_data['tracks']
return Artist(data, tracks)
else:
tracks = [
Track(track["track"])
for track in data["tracks"]["items"] if track["track"] is not None
]
if not len(tracks):
raise SpotifyRequestException("This playlist is empty and therefore cannot be queued.")
next_page_url = data["tracks"]["next"]
while next_page_url is not None:
async with self.session.get(next_page_url, headers=self._bearer_headers) as resp:
if resp.status != 200:
raise SpotifyRequestException(
f"Error while fetching results: {resp.status} {resp.reason}"
)
next_data: dict = await resp.json(loads=json.loads)
tracks += [
Track(track["track"])
for track in next_data["items"] if track["track"] is not None
]
next_page_url = next_data["next"]
return Playlist(data, tracks)

8
pomice/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

89
pomice/spotify/objects.py
View File

@ -1,89 +0,0 @@
from typing import List
class Track:
"""The base class for a Spotify Track"""
def __init__(self, data: dict, image = None) -> None:
self.name = data["name"]
self.artists = ", ".join(artist["name"] for artist in data["artists"])
self.length = data["duration_ms"]
self.id = data["id"]
if data.get("external_ids"):
self.isrc = data["external_ids"]["isrc"]
else:
self.isrc = None
if data.get("album") and data["album"].get("images"):
self.image = data["album"]["images"][0]["url"]
else:
self.image = image
if data["is_local"]:
self.uri = None
else:
self.uri = data["external_urls"]["spotify"]
def __repr__(self) -> str:
return (
f"<Pomice.spotify.Track name={self.name} artists={self.artists} "
f"length={self.length} id={self.id} isrc={self.isrc}>"
)
class Playlist:
"""The base class for a Spotify playlist"""
def __init__(self, data: dict, tracks: List[Track]) -> None:
self.name = data["name"]
self.tracks = tracks
self.owner = data["owner"]["display_name"]
self.total_tracks = data["tracks"]["total"]
self.id = data["id"]
if data.get("images") and len(data["images"]):
self.image = data["images"][0]["url"]
else:
self.image = None
self.uri = data["external_urls"]["spotify"]
def __repr__(self) -> str:
return (
f"<Pomice.spotify.Playlist name={self.name} owner={self.owner} id={self.id} "
f"total_tracks={self.total_tracks} tracks={self.tracks}>"
)
class Album:
"""The base class for a Spotify album"""
def __init__(self, data: dict) -> None:
self.name = data["name"]
self.artists = ", ".join(artist["name"] for artist in data["artists"])
self.image = data["images"][0]["url"]
self.tracks = [Track(track, image=self.image) for track in data["tracks"]["items"]]
self.total_tracks = data["total_tracks"]
self.id = data["id"]
self.uri = data["external_urls"]["spotify"]
def __repr__(self) -> str:
return (
f"<Pomice.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 artist"""
def __init__(self, data: dict, tracks: dict) -> None:
self.name = f"Top tracks for {data['name']}" # Setting that because its only playing top tracks
self.genres = ", ".join(genre for genre in data["genres"])
self.followers = data["followers"]["total"]
self.image = data["images"][0]["url"]
self.tracks = [Track(track, image=self.image) for track in tracks]
self.id = data["id"]
self.uri = data["external_urls"]["spotify"]
def __repr__(self) -> str:
return (
f"<Pomice.spotify.Artist name={self.name} id={self.id} "
f"tracks={self.tracks}>"
)