mirror of
https://github.com/RafaelSolVargas/Vulkan.git
synced 2025-10-29 16:57:23 +00:00
Adding Cleaner and fixing bugs
This commit is contained in:
@@ -8,7 +8,6 @@ from Utils.Utils import is_url, run_async
|
||||
|
||||
|
||||
class Downloader():
|
||||
"""Download musics direct URL and title or Source from Youtube using a music name or Youtube URL"""
|
||||
config = Configs()
|
||||
__YDL_OPTIONS = {'format': 'bestaudio/best',
|
||||
'default_search': 'auto',
|
||||
@@ -38,7 +37,6 @@ class Downloader():
|
||||
self.__playlist_keys = ['entries']
|
||||
|
||||
def finish_one_song(self, song: Song) -> Song:
|
||||
"""Receives a song object, finish his download and return it"""
|
||||
if song.identifier == None:
|
||||
return None
|
||||
|
||||
@@ -51,17 +49,11 @@ class Downloader():
|
||||
return song
|
||||
|
||||
async def preload(self, songs: List[Song]) -> None:
|
||||
"""Download the full info of the songs objects"""
|
||||
for song in songs:
|
||||
asyncio.ensure_future(self.__download_song(song))
|
||||
|
||||
@run_async
|
||||
def extract_info(self, url: str) -> List[dict]:
|
||||
"""Extract all songs direct URL from a Youtube Link
|
||||
|
||||
Arg: Url String
|
||||
Return: List with the direct youtube URL of each song
|
||||
"""
|
||||
if is_url(url): # If Url
|
||||
options = Downloader.__YDL_OPTIONS_EXTRACT
|
||||
with YoutubeDL(options) as ydl:
|
||||
@@ -100,11 +92,6 @@ class Downloader():
|
||||
return []
|
||||
|
||||
def __download_url(self, url) -> dict:
|
||||
"""Download musics full info and source from Music URL
|
||||
|
||||
Arg: URL from Youtube
|
||||
Return: Dict with the full youtube information of the music, including source to play it
|
||||
"""
|
||||
options = Downloader.__YDL_OPTIONS
|
||||
with YoutubeDL(options) as ydl:
|
||||
try:
|
||||
@@ -116,7 +103,6 @@ class Downloader():
|
||||
return None
|
||||
|
||||
async def __download_song(self, song: Song) -> None:
|
||||
"""Download a music object asynchronously"""
|
||||
if song.source is not None: # If Music already preloaded
|
||||
return None
|
||||
|
||||
@@ -134,11 +120,6 @@ class Downloader():
|
||||
await asyncio.wait(fs=fs, return_when=asyncio.ALL_COMPLETED)
|
||||
|
||||
def __download_title(self, title: str) -> dict:
|
||||
"""Download a music full information using his name.
|
||||
|
||||
Arg: Music Name
|
||||
Return: A dict containing the song information
|
||||
"""
|
||||
options = Downloader.__YDL_OPTIONS
|
||||
with YoutubeDL(options) as ydl:
|
||||
try:
|
||||
|
||||
@@ -2,8 +2,6 @@ from abc import ABC, abstractproperty, abstractmethod
|
||||
|
||||
|
||||
class IPlaylist(ABC):
|
||||
"""Class to manage and control the songs to play and played"""
|
||||
|
||||
@abstractproperty
|
||||
def looping_one(self):
|
||||
pass
|
||||
@@ -58,8 +56,6 @@ class IPlaylist(ABC):
|
||||
|
||||
|
||||
class ISong(ABC):
|
||||
"""Store the usefull information about a Song"""
|
||||
|
||||
@abstractmethod
|
||||
def finish_down(self, info: dict) -> None:
|
||||
pass
|
||||
|
||||
@@ -36,41 +36,6 @@ class Player(commands.Cog):
|
||||
def playlist(self) -> Playlist:
|
||||
return self.__playlist
|
||||
|
||||
def __play_next(self, error, ctx: Context) -> None:
|
||||
if self.__force_stop: # If it's forced to stop player
|
||||
self.__force_stop = False
|
||||
return None
|
||||
|
||||
song = self.__playlist.next_song()
|
||||
|
||||
if song != None:
|
||||
coro = self.__play_music(ctx, song)
|
||||
self.__bot.loop.create_task(coro)
|
||||
else:
|
||||
self.__playing = False
|
||||
|
||||
async def __play_music(self, ctx: Context, song: Song) -> None:
|
||||
try:
|
||||
source = self.__ensure_source(song)
|
||||
if source == None:
|
||||
self.__play_next(None, ctx)
|
||||
|
||||
self.__playing = True
|
||||
|
||||
player = FFmpegPCMAudio(song.source, **self.FFMPEG_OPTIONS)
|
||||
self.__guild.voice_client.play(
|
||||
player, after=lambda e: self.__play_next(e, ctx))
|
||||
|
||||
self.__timer.cancel()
|
||||
self.__timer = Timer(self.__timeout_handler)
|
||||
|
||||
await ctx.invoke(self.__bot.get_command('np'))
|
||||
|
||||
songs = self.__playlist.songs_to_preload
|
||||
await self.__down.preload(songs)
|
||||
except:
|
||||
self.__play_next(None, ctx)
|
||||
|
||||
async def play(self, ctx: Context, track: str, requester: str) -> str:
|
||||
try:
|
||||
links, provider = self.__searcher.search(track)
|
||||
@@ -141,7 +106,6 @@ class Player(commands.Cog):
|
||||
await self.__play_music(ctx, first_song)
|
||||
|
||||
async def play_prev(self, ctx: Context) -> None:
|
||||
"""Stop the currently playing cycle, load the previous song and play"""
|
||||
if self.__playlist.looping_one or self.__playlist.looping_all: # Do not allow play if loop
|
||||
embed = Embed(
|
||||
title=self.__config.SONG_PLAYER,
|
||||
@@ -168,17 +132,6 @@ class Player(commands.Cog):
|
||||
|
||||
await self.__play_music(ctx, song)
|
||||
|
||||
async def stop(self) -> bool:
|
||||
if self.__guild.voice_client is None:
|
||||
return False
|
||||
|
||||
if self.__guild.voice_client.is_connected():
|
||||
self.__playlist.clear()
|
||||
self.__playlist.loop_off()
|
||||
self.__guild.voice_client.stop()
|
||||
await self.__guild.voice_client.disconnect()
|
||||
return True
|
||||
|
||||
async def force_stop(self) -> None:
|
||||
try:
|
||||
if self.__guild.voice_client is None:
|
||||
@@ -191,8 +144,42 @@ class Player(commands.Cog):
|
||||
except Exception as e:
|
||||
print(f'DEVELOPER NOTE -> Force Stop Error: {e}')
|
||||
|
||||
def __play_next(self, error, ctx: Context) -> None:
|
||||
if self.__force_stop: # If it's forced to stop player
|
||||
self.__force_stop = False
|
||||
return None
|
||||
|
||||
song = self.__playlist.next_song()
|
||||
|
||||
if song != None:
|
||||
coro = self.__play_music(ctx, song)
|
||||
self.__bot.loop.create_task(coro)
|
||||
else:
|
||||
self.__playing = False
|
||||
|
||||
async def __play_music(self, ctx: Context, song: Song) -> None:
|
||||
try:
|
||||
source = self.__ensure_source(song)
|
||||
if source == None:
|
||||
self.__play_next(None, ctx)
|
||||
|
||||
self.__playing = True
|
||||
|
||||
player = FFmpegPCMAudio(song.source, **self.FFMPEG_OPTIONS)
|
||||
self.__guild.voice_client.play(
|
||||
player, after=lambda e: self.__play_next(e, ctx))
|
||||
|
||||
self.__timer.cancel()
|
||||
self.__timer = Timer(self.__timeout_handler)
|
||||
|
||||
await ctx.invoke(self.__bot.get_command('np'))
|
||||
|
||||
songs = self.__playlist.songs_to_preload
|
||||
await self.__down.preload(songs)
|
||||
except:
|
||||
self.__play_next(None, ctx)
|
||||
|
||||
def __format_embed(self, info: dict, title='', position='Playing Now') -> Embed:
|
||||
"""Configure the embed to show the song information"""
|
||||
embedvc = Embed(
|
||||
title=title,
|
||||
description=f"[{info['title']}]({info['original_url']})",
|
||||
|
||||
@@ -4,17 +4,10 @@ from Utils.Utils import is_url
|
||||
|
||||
|
||||
class Searcher():
|
||||
"""Turn the user input into list of musics names, support youtube and spotify"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.__Spotify = SpotifySearch()
|
||||
|
||||
def search(self, music: str) -> list:
|
||||
"""Return a list with the song names or an URL
|
||||
|
||||
Arg -> User Input, a string with the
|
||||
Return -> A list of musics names and Provider Type
|
||||
"""
|
||||
provider = self.__identify_source(music)
|
||||
|
||||
if provider == Provider.YouTube:
|
||||
@@ -35,7 +28,6 @@ class Searcher():
|
||||
return None, Provider.Unknown
|
||||
|
||||
def __identify_source(self, music) -> Provider:
|
||||
"""Identify the provider of a music"""
|
||||
if not is_url(music):
|
||||
return Provider.Name
|
||||
|
||||
|
||||
@@ -2,17 +2,14 @@ from Music.Interfaces import ISong, IPlaylist
|
||||
|
||||
|
||||
class Song(ISong):
|
||||
"""Store the usefull information about a Song"""
|
||||
|
||||
def __init__(self, identifier: str, playlist: IPlaylist, requester: str) -> None:
|
||||
"""Create a song with only the URL to the youtube song"""
|
||||
self.__identifier = identifier
|
||||
self.__info = {'requester': requester}
|
||||
self.__problematic = False
|
||||
self.__playlist: IPlaylist = playlist
|
||||
|
||||
def finish_down(self, info: dict) -> None:
|
||||
"""Get and store the full information of the song"""
|
||||
self.__usefull_keys = ['duration',
|
||||
'title', 'webpage_url',
|
||||
'channel', 'id', 'uploader',
|
||||
@@ -34,7 +31,6 @@ class Song(ISong):
|
||||
|
||||
@property
|
||||
def source(self) -> str:
|
||||
"""Return the Song Source URL to play"""
|
||||
if 'url' in self.__info.keys():
|
||||
return self.__info['url']
|
||||
else:
|
||||
@@ -42,7 +38,6 @@ class Song(ISong):
|
||||
|
||||
@property
|
||||
def title(self) -> str:
|
||||
"""Return the Song Title"""
|
||||
if 'title' in self.__info.keys():
|
||||
return self.__info['title']
|
||||
else:
|
||||
@@ -50,7 +45,6 @@ class Song(ISong):
|
||||
|
||||
@property
|
||||
def duration(self) -> str:
|
||||
"""Return the Song Title"""
|
||||
if 'duration' in self.__info.keys():
|
||||
return self.__info['duration']
|
||||
else:
|
||||
@@ -65,7 +59,6 @@ class Song(ISong):
|
||||
return self.__problematic
|
||||
|
||||
def destroy(self) -> None:
|
||||
"""Mark this song with problems and removed from the playlist due to any type of error"""
|
||||
print(f'DEVELOPER NOTE -> Music self destroying {self.__identifier}')
|
||||
self.__problematic = True
|
||||
self.__playlist.destroy_song(self)
|
||||
|
||||
@@ -4,8 +4,6 @@ from Config.Config import Configs
|
||||
|
||||
|
||||
class SpotifySearch():
|
||||
"""Search a Spotify music or playlist and return the musics names"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.__config = Configs()
|
||||
self.__connected = False
|
||||
@@ -27,7 +25,6 @@ class SpotifySearch():
|
||||
return False
|
||||
|
||||
def search(self, music=str) -> list:
|
||||
"""Search and return the title of musics on Spotify"""
|
||||
type = music.split('/')[3].split('?')[0]
|
||||
code = music.split('/')[4].split('?')[0]
|
||||
if type == 'album':
|
||||
@@ -44,10 +41,6 @@ class SpotifySearch():
|
||||
return musics
|
||||
|
||||
def __get_album(self, code=str) -> list:
|
||||
"""Convert a album ID to list of songs names
|
||||
|
||||
ARG: Spotify Code of the Album
|
||||
"""
|
||||
if self.__connected == True:
|
||||
try:
|
||||
results = self.__api.album_tracks(code)
|
||||
@@ -70,10 +63,6 @@ class SpotifySearch():
|
||||
raise e
|
||||
|
||||
def __get_playlist(self, code=str) -> list:
|
||||
"""Convert a playlist ID to list of songs names
|
||||
|
||||
Arg: Spotify Code of the Playlist
|
||||
"""
|
||||
try:
|
||||
results = self.__api.playlist_items(code)
|
||||
itens = results['items']
|
||||
@@ -100,10 +89,6 @@ class SpotifySearch():
|
||||
raise e
|
||||
|
||||
def __get_track(self, code=str) -> list:
|
||||
"""Convert a track ID to the title of the music
|
||||
|
||||
ARG: Spotify Code of the Track
|
||||
"""
|
||||
results = self.__api.track(code)
|
||||
name = results['name']
|
||||
artists = ''
|
||||
@@ -113,10 +98,6 @@ class SpotifySearch():
|
||||
return [f'{name} {artists}']
|
||||
|
||||
def __get_artist(self, code=str) -> list:
|
||||
"""Convert a external_url track to the title of the music
|
||||
|
||||
ARG: Spotify Code of the Music
|
||||
"""
|
||||
results = self.__api.artist_top_tracks(code, country='BR')
|
||||
|
||||
musics_titles = []
|
||||
@@ -127,10 +108,6 @@ class SpotifySearch():
|
||||
return musics_titles
|
||||
|
||||
def __extract_title(self, music: dict) -> str:
|
||||
"""Receive a spotify music object and return his title
|
||||
|
||||
ARG: music dict returned by Spotify
|
||||
"""
|
||||
title = f'{music["name"]} '
|
||||
for artist in music['artists']:
|
||||
title += f'{artist["name"]} '
|
||||
|
||||
@@ -2,8 +2,7 @@ from enum import Enum
|
||||
|
||||
|
||||
class Provider(Enum):
|
||||
"""Store Enum Types of the Providers"""
|
||||
Spotify = "Spotify"
|
||||
YouTube = "YouTube"
|
||||
Spotify = 'Spotify'
|
||||
YouTube = 'YouTube'
|
||||
Name = 'Track Name'
|
||||
Unknown = "Unknown"
|
||||
Unknown = 'Unknown'
|
||||
|
||||
Reference in New Issue
Block a user