diff options
Diffstat (limited to 'code/custom_source.py')
| -rw-r--r-- | code/custom_source.py | 74 |
1 files changed, 74 insertions, 0 deletions
diff --git a/code/custom_source.py b/code/custom_source.py new file mode 100644 index 0000000..cf1510e --- /dev/null +++ b/code/custom_source.py @@ -0,0 +1,74 @@ +from lavalink import LoadResult, LoadType, Source, DeferredAudioTrack, PlaylistInfo + + +class LoadError(Exception): # We'll raise this if we have trouble loading our track. + pass + + +class CustomAudioTrack(DeferredAudioTrack): + # A DeferredAudioTrack allows us to load metadata now, and a playback URL later. + # This makes the DeferredAudioTrack highly efficient, particularly in cases + # where large playlists are loaded. + + async def load(self, client): # Load our 'actual' playback track using the metadata from this one. + result: LoadResult = await client.get_tracks('ytsearch:{0.title} {0.author} audio'.format(self)) # Search for our track on YouTube. + + if result.load_type != LoadType.SEARCH or not result.tracks: # We're expecting a 'SEARCH' due to our 'ytsearch' prefix above. + raise LoadError + + first_track = result.tracks[0] # Grab the first track from the results. + base64 = first_track.track # Extract the base64 string from the track. + self.track = base64 # We'll store this for later, as it allows us to save making network requests + # if this track is re-used (e.g. repeat). + + return base64 + + +class CustomSource(Source): + def __init__(self): + super().__init__(name='custom') # Initialising our custom source with the name 'custom'. + + async def load_item(self, user, metadata): + track = CustomAudioTrack({ # Create an instance of our CustomAudioTrack. + 'identifier': metadata['id'], # Fill it with metadata that we've obtained from our source's provider. + 'isSeekable': True, + 'author': metadata['artists'][0]['name'], + 'length': metadata['duration_ms'], + 'isStream': False, + 'title': metadata['name'], + 'uri': metadata['external_urls']['spotify'], + 'duration': metadata['duration_ms'], + }, requester=user, extra=metadata['album']['images'][0]['url']) + return LoadResult(LoadType.TRACK, [track], playlist_info=PlaylistInfo.none()) + + async def load_album(self, user, metadata): + tracks = [] + for track in metadata['tracks']['items']: # Loop through each track in the album. + tracks.append(CustomAudioTrack({ # Create an instance of our CustomAudioTrack. + 'identifier': track['id'], # Fill it with metadata that we've obtained from our source's provider. + 'isSeekable': True, + 'author': track['artists'][0]['name'], + 'length': track['duration_ms'], + 'isStream': False, + 'title': track['name'], + 'uri': track['external_urls']['spotify'], + 'duration': track['duration_ms'], + }, requester=user, extra=metadata['images'][0]['url'])) + + return LoadResult(LoadType.PLAYLIST, tracks, playlist_info=PlaylistInfo.none()) + + async def load_playlist(self, user, metadata): + tracks = [] + for track in metadata['tracks']['items']: # Loop through each track in the playlist. + tracks.append(CustomAudioTrack({ # Create an instance of our CustomAudioTrack. + 'identifier': track['track']['id'], # Fill it with metadata that we've obtained from our source's provider. + 'isSeekable': True, + 'author': track['track']['artists'][0]['name'], + 'length': track['track']['duration_ms'], + 'isStream': False, + 'title': track['track']['name'], + 'uri': track['track']['external_urls']['spotify'], + 'duration': track['track']['duration_ms'], + }, requster=user, extra=track['track']['album']['images'][0]['url'])) + + return LoadResult(LoadType.PLAYLIST, tracks, playlist_info=PlaylistInfo.none())
\ No newline at end of file |