weyne/Vulkan
1
0
Fork 0
mirror of https://github.com/RafaelSolVargas/Vulkan.git synced 2025-10-29 16:57:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reformulating Music module to manage players instead of playing the song

Browse Source
This commit is contained in:
Rafael Vargas
2022-01-06 13:22:29 -04:00
parent 06a00e227d
commit efc1293ba0
1 changed files with 100 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

293
vulkanbot/music/Music.py
View File

@@ -2,239 +2,140 @@ import discord
from discord.ext import commands from discord.ext import commands
from config import config from config import config
from vulkanbot.music.Downloader import Downloader from vulkanbot.music.Player import Player
from vulkanbot.music.Playlist import Playlist
from vulkanbot.music.Searcher import Searcher
from vulkanbot.music.Types import Provider
from vulkanbot.music.utils import * from vulkanbot.music.utils import *
class Music(commands.Cog): class Music(commands.Cog):
def __init__(self, bot): def __init__(self, bot):
self.__searcher: Searcher = Searcher() self.__guilds = {}
self.__downloader: Downloader = Downloader()
self.__playlist: Playlist = Playlist()
self.__bot: discord.Client = bot self.__bot: discord.Client = bot
self.__playing = False
self.__vc = ""
self.YDL_OPTIONS = {'format': 'bestaudio', 'noplaylist': 'True'}
self.FFMPEG_OPTIONS = {'before_options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5',
'options': '-vn'}
def __play_next(self, error, ctx):
source = self.__playlist.next_song()
if source != None: # If there is not a source for the song
coro = self.__play_music(ctx, source)
self.__bot.loop.create_task(coro)
else:
self.__playing = False
async def __play_music(self, ctx, song):
self.__playing = True
player = discord.FFmpegPCMAudio(song.source, **self.FFMPEG_OPTIONS)
self.__vc.play(player, after=lambda e: self.__play_next(e, ctx))
await ctx.invoke(self.__bot.get_command('np'))
songs = self.__playlist.songs_to_preload
await self.__downloader.preload(songs)
@commands.command(name="play", help=config.HELP_PLAY, aliases=['p', 'tocar']) @commands.command(name="play", help=config.HELP_PLAY, aliases=['p', 'tocar'])
async def play(self, ctx, *args): async def play(self, ctx, *args):
user_input = " ".join(args) user_input = " ".join(args)
try: player = self.__get_player(ctx)
if len(self.__bot.voice_clients) == 0: if player == None:
voice_channel = ctx.author.voice.channel player = Player(self.__bot, ctx.guild)
self.__vc = await voice_channel.connect() self.__guilds[ctx.guild] = player
except Exception as e:
print(e)
await self.__send_embed(ctx, title='Para tocar música, primeiro se conecte a um canal de voz.', colour_name='grey')
return
else:
songs_quant = 0
musics_identifiers, provider = self.__searcher.search(user_input)
if provider == Provider.Unknown: # If type not identified if is_connected(ctx) == None:
await self.__send_embed(ctx, description='Entrada inválida, tente algo melhor', colour_name='blue') result = await player.connect(ctx)
if result['success'] == False:
await self.__send_embed(ctx, description=result['reason'], colour_name='red')
return return
if provider == Provider.YouTube: # If youtube source await player.play(ctx, user_input)
musics_identifiers = self.__downloader.extract_youtube_link(
musics_identifiers[0])
for identifier in musics_identifiers: # Creating songs
last_song = self.__playlist.add_song(identifier)
songs_quant += 1
songs_preload = self.__playlist.songs_to_preload
await self.__downloader.preload(songs_preload)
if songs_quant == 1: # If only one music downloaded
song = self.__downloader.download_one(
last_song) # Download the new music
if song == None: # If song not downloaded
await self.__send_embed(ctx, description='Houve um problema no download dessa música, tente novamente', colour_name='blue')
elif not self.__playing: # If not playing
await self.__send_embed(ctx, description=f'Você adicionou a música **{song.title}** à playlist', colour_name='blue')
else: # If playing
await ctx.send(embed=song.embed(title='Song added to Queue'))
else:
await self.__send_embed(ctx, description=f"Você adicionou {songs_quant} músicas à fila!", colour_name='blue')
if not self.__playing:
try_another = True
while try_another:
first = self.__playlist.next_song()
if first == None:
await self.__send_embed(ctx, description='Houve um problema no download dessa música, tente novamente', colour_name='blue')
break
while True:
if first.source != None: # If song got downloaded
try_another = False
break
if first.problematic: # If song got any error, try another one
break
else: # The song is downloading, checking another time
continue
if first != None:
await self.__play_music(ctx, first)
@commands.command(name="queue", help=config.HELP_QUEUE, aliases=['q', 'fila']) @commands.command(name="queue", help=config.HELP_QUEUE, aliases=['q', 'fila'])
async def queue(self, ctx): async def queue(self, ctx):
if self.__playlist.looping_one: # If Repeating one player = self.__get_player(ctx)
await self.now_playing(ctx) if player == None:
return return
songs_preload = self.__playlist.songs_to_preload embed = await player.queue()
await self.__downloader.preload(songs_preload) await ctx.send(embed=embed)
total_time = format_time(sum([int(song.duration if song.duration else 0)
for song in songs_preload])) # Sum the duration
total_songs = len(self.__playlist)
text = f'Total musics: {total_songs} | Duration: `{total_time}` downloaded \n\n'
for pos, song in enumerate(songs_preload, start=1):
title = song.title if song.title else 'Downloading...'
text += f"**`{pos}` - ** {title} - `{format_time(song.duration)}`\n"
if len(songs_preload) > 0:
if self.__playlist.looping_all: # If repeating all
await self.__send_embed(ctx, title='Repeating all', description=text, colour_name='blue')
else: # Repeating off
await self.__send_embed(ctx, title='Songs in Queue', description=text, colour_name='blue')
else: # No music
await self.__send_embed(ctx, description='There is not musics in queue.', colour_name='red')
@commands.command(name="skip", help=config.HELP_SKIP, aliases=['pular']) @commands.command(name="skip", help=config.HELP_SKIP, aliases=['pular'])
async def skip(self, ctx): async def skip(self, ctx):
if len(self.__bot.voice_clients) > 0: player = self.__get_player(ctx)
self.__vc.stop() if player == None:
return
else:
await player.skip()
@commands.command(name='stop', help=config.HELP_STOP) @commands.command(name='stop', help=config.HELP_STOP, aliases=['parar'])
async def stop(self, ctx): async def stop(self, ctx):
if self.__vc == '': player = self.__get_player(ctx)
if player == None:
return return
if self.__vc.is_connected(): else:
self.__playlist.clear() await player.stop()
self.__vc.stop()
await self.__vc.disconnect()
@commands.command(name='pause', help=config.HELP_PAUSE) @commands.command(name='pause', help=config.HELP_PAUSE, aliases=['pausar'])
async def pause(self, ctx): async def pause(self, ctx):
if self.__vc == '': player = self.__get_player(ctx)
if player == None:
print('No player')
return return
if self.__vc.is_playing(): else:
self.__vc.pause() success = await player.pause()
await self.__send_embed(ctx, description='Música pausada', colour_name='green') if success:
await self.__send_embed(ctx, description='Song paused', colour_name='blue')
@commands.command(name='resume', help=config.HELP_RESUME) @commands.command(name='resume', help=config.HELP_RESUME, aliases=['soltar'])
async def resume(self, ctx): async def resume(self, ctx):
if self.__vc == '': player = self.__get_player(ctx)
if player == None:
return return
if self.__vc.is_paused(): else:
self.__vc.resume() success = await player.resume()
await self.__send_embed(ctx, description='Música tocando', colour_name='green') if success:
await self.__send_embed(ctx, description='Song Playing', colour_name='blue')
@commands.command(name='loop', help=config.HELP_LOOP) @commands.command(name='loop', help=config.HELP_LOOP, aliases=['repeat'])
async def loop(self, ctx, args: str): async def loop(self, ctx, args: str):
args = args.lower() player = self.__get_player(ctx)
if args == 'one': if player == None:
description = self.__playlist.loop_one()
elif args == 'all':
description = self.__playlist.loop_all()
elif args == 'off':
description = self.__playlist.loop_off()
else:
description = 'Comando Loop\nOne - Repete a música atual\nAll - Repete as músicas atuais\nOff - Desativa o loop'
await self.__send_embed(ctx, description=description, colour_name='grey')
@commands.command(name='clear', help=config.HELP_CLEAR)
async def clear(self, ctx):
self.__playlist.clear()
@commands.command(name='np', help=config.HELP_NP)
async def now_playing(self, ctx):
if self.__playlist.looping_one:
title = 'Song Looping Now'
else:
title = 'Song Playing Now'
current_song = self.__playlist.current
await self.__clean_messages(ctx)
await ctx.send(embed=current_song.embed(title=title))
@commands.command(name='shuffle', help=config.HELP_SHUFFLE)
async def shuffle(self, ctx):
self.__playlist.shuffle()
songs = self.__playlist.songs_to_preload
await self.__downloader.preload(songs)
await self.__send_embed(ctx, description='Musicas embaralhadas', colour_name='blue')
@commands.command(name='move', help=config.HELP_MOVE)
async def move(self, ctx, pos1, pos2='1'):
try:
pos1 = int(pos1)
pos2 = int(pos2)
except Exception as e:
print(e)
await ctx.send('This command require a number')
return return
else:
result = await player.loop(args)
await self.__send_embed(ctx, description=result, colour_name='blue')
result = self.__playlist.move_songs(pos1, pos2) @commands.command(name='clear', help=config.HELP_CLEAR, aliases=['limpar'])
async def clear(self, ctx):
player = self.__get_player(ctx)
if player == None:
return
else:
await player.clear()
songs = self.__playlist.songs_to_preload @commands.command(name='np', help=config.HELP_NP, aliases=['playing', 'now'])
await self.__downloader.preload(songs) async def now_playing(self, ctx):
await ctx.send(result) player = self.__get_player(ctx)
if player == None:
return
else:
embed = await player.now_playing()
await self.__clean_messages(ctx)
await ctx.send(embed=embed)
@commands.command(name='remove', help=config.HELP_MOVE) @commands.command(name='shuffle', help=config.HELP_SHUFFLE, aliases=['aleatorio'])
async def shuffle(self, ctx):
player = self.__get_player(ctx)
if player == None:
return
else:
result = await player.shuffle()
await self.__send_embed(ctx, description=result, colour_name='blue')
@commands.command(name='move', help=config.HELP_MOVE, aliases=['mover'])
async def move(self, ctx, pos1, pos2='1'):
player = self.__get_player(ctx)
if player == None:
return
else:
result = await player.move(pos1, pos2)
await self.__send_embed(ctx, description=result, colour_name='blue')
@commands.command(name='remove', help=config.HELP_REMOVE, aliases=['remover'])
async def remove(self, ctx, position): async def remove(self, ctx, position):
"""Remove a song from the queue in the position""" """Remove a song from the queue in the position"""
try: player = self.__get_player(ctx)
position = int(position) if player == None:
except Exception as e:
print(e)
await ctx.send('This command require a number')
return return
else:
result = await player.remove(position)
await self.__send_embed(ctx, description=result, colour_name='blue')
result = self.__playlist.remove_song(position) @commands.command(name='reset', help=config.HELP_RESET, aliases=['resetar'])
await ctx.send(result) async def reset(self, ctx):
player = self.__get_player(ctx)
if player != None:
await player.stop()
self.__guilds[ctx.guild] = Player(self.__bot, ctx.guild)
async def __send_embed(self, ctx, title='', description='', colour_name='grey'): async def __send_embed(self, ctx, title='', description='', colour_name='grey'):
try: try:
@@ -258,12 +159,18 @@ class Music(commands.Cog):
if message.author == self.__bot.user: if message.author == self.__bot.user:
if len(message.embeds) > 0: if len(message.embeds) > 0:
embed = message.embeds[0] embed = message.embeds[0]
if embed.title == 'Song Playing Now': if embed.title == 'Song Playing Now' or embed.title == 'Song Looping Now':
await message.delete() await message.delete()
except Exception as e: except Exception as e:
print(e) print(e)
continue continue
def __get_player(self, ctx):
try:
return self.__guilds[ctx.guild]
except:
return None
def setup(bot): def setup(bot):
bot.add_cog(Music(bot)) bot.add_cog(Music(bot))