<filename>cogs/music_cog.py<gh_stars>0
﻿
import asyncio
import json
import nextcord
import youtube_dl
from nextcord.ext import commands
from utils import http
import typing


get = nextcord.utils.get
youtube_dl.utils.bug_reports_message = lambda: ''


def get_config():
    with open('config.json',"r") as f:
        config=json.load(f)
        if config[str('music_cog')] == True and config[str('maintenance')] == False:
            return True
        else:
            return False



  

           
        

youtube_dl.utils.bug_reports_message = lambda: ''


ytdl_format_options = {
    'format': 'bestaudio/best',
    'outtmpl': '%(extractor)s-%(id)s-%(title)s.%(ext)s',
    'restrictfilenames': True,
    #'noplaylist': True,
    'nocheckcertificate': True,
    'ignoreerrors': False,
    'logtostderr': False,
    'quiet': True,
    'no_warnings': True,
    'default_search': 'auto',
    'source_address': '0.0.0.0' 
}

ffmpeg_options = {
    'options': '-vn'
}

ytdl = youtube_dl.YoutubeDL(ytdl_format_options)




class YTDLSource(nextcord.PCMVolumeTransformer):
    def __init__(self, source, *, data, volume = 0.5):
        super().__init__(source, volume)

        self.data = data

        self.title = data.get('title')
        self.url = data.get('url')
        

    @classmethod
    async def from_url(cls, url, *, loop=None, stream=False):
        loop = loop or asyncio.get_event_loop()

        data = await loop.run_in_executor(None, lambda: ytdl.extract_info(url, download=not stream))
        
        if 'entries' in data:
            data = data['entries'][0]
            

        filename = str(data['url'] if stream else ytdl.prepare_filename(data))
        
        return cls(nextcord.FFmpegPCMAudio(filename, **ffmpeg_options), data=data)

    



    
queues = {}
    


    
class Music(commands.Cog):
    def __init__(self,bot):
        self.bot : commands.Bot = bot
        
        self.persistent_views_added = False
        def check_queue(ctx,id_ : int): 
            if id_ in queues:
                if queues[id_] != []:
                    player_ = queues[id_].pop(0)
                    ctx.voice_client.play(player_, after=lambda e: print('Player error: %s' % e) if e else self.check_queue(ctx,ctx.guild.id))

            else: del queues[id_]
                
        self.check_queue = check_queue

    @commands.command(aliases=['p'])
    @commands.guild_only()
    async def play(self, ctx : commands.Context, *, url : str = 'never gona'):
        async with ctx.typing():
         
            if ctx.voice_client is None:
                if ctx.author.voice:
                    await ctx.author.voice.channel.connect(timeout=None, reconnect=True)
                else:
                    await ctx.send("You are not connected to a voice channel.")
                    raise commands.CommandError("Author not connected to a voice channel.")
                    
            
            #elif ctx.voice_client.is_playing(): 
                #ctx.voice_client.stop()


            elif ctx.voice_client.channel is not None and  ctx.voice_client.channel != ctx.voice_client.channel: 
                await ctx.voice_client.move_to(channel=ctx.author.voice.channel)           

            

            player_ = await YTDLSource.from_url(url, loop=self.bot.loop, stream=True)
            
            if ctx.guild.id in queues:
                if queues[ctx.guild.id] != [] or queues[ctx.guild.id] == [] and ctx.voice_client.is_playing():
                    await ctx.send("Song added to queue")
                    return queues[ctx.guild.id].append(player_)

                elif queues[ctx.guild.id] == [] and not ctx.voice_client.is_playing():
                    queues[ctx.guild.id] = player_


                

            else:
                queues[ctx.guild.id] = [player_]
            player = queues[ctx.guild.id].pop(0)
            ctx.voice_client.play(player, after=lambda e: print('Player error: %s' % e) if e else self.check_queue(ctx,ctx.guild.id))

            embed=nextcord.Embed(title=f'Now playing: __**`{player.title}`**__',description=' ',color=nextcord.Color.red())
            embed.set_author(name=player.title,url=f"https://www.youtube.com/watch?v={player.data['id']}" ,icon_url="https://icones.pro/wp-content/uploads/2021/02/youtube-logo-icone.png")
            
            class View(nextcord.ui.View):
                def __init__(self):
                    def check_queue(ctx,id_ : int): 
                        if id_ in queues:
                            if queues[id_] != []:
                                player_ = queues[id_].pop(0)
                                ctx.voice_client.play(player_, after=lambda e: print('Player error: %s' % e) if e else self.check_queue(ctx,ctx.guild.id))

                        else: del queues[id_]
                    self.check_queue = check_queue
                    super().__init__(timeout=None)
               
                @nextcord.ui.button(label="Stop",style=nextcord.ButtonStyle.red)
                async def stop__(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
                    if interaction.user != ctx.author:
                        return
                    if ctx.author.voice is not None:
                        if (ctx.voice_client.is_playing() or ctx.voice_client.is_paused()):
                            ctx.voice_client.stop()
                            del queues[ctx.guild.id]
                        else:
                 
                            await interaction.response.send_message("put something to stop ", ephemeral=True)
                
                    else:
                        await interaction.response.send_message("You aren't in a voice channel", ephemeral=True)

                @nextcord.ui.button(label="Pause",style=nextcord.ButtonStyle.blurple)
                async def pause__(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
                    if interaction.user != ctx.author:
                        return
                    if ctx.author.voice is not None:
                        if ctx.voice_client.is_playing():
                            ctx.voice_client.pause()  
                        else:
                            None
                    else:
                        None
                    
                @nextcord.ui.button(label="Resume",style=nextcord.ButtonStyle.blurple)
                async def resume__(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
                    if interaction.user != ctx.author:
                        return
                    if ctx.author.voice is not None:
                        if ctx.voice_client.is_paused():
                            ctx.voice_client.resume()
                          


                @nextcord.ui.button(label="Leave",style=nextcord.ButtonStyle.red)
                async def leave__(self, button: nextcord.ui.Button, interaction: nextcord.Interaction): 
                    if interaction.user != ctx.author:
                        return
                    if ctx.author.voice is not None:
                        #ctx.voice_client.cleanup()
                        ctx.voice_client.stop()
                        del queues[ctx.guild.id]
                        await ctx.voice_client.disconnect()
                    

                @nextcord.ui.button(label="Skip", style=nextcord.ButtonStyle.blurple)
                async def skip__(self, button : nextcord.ui.Button, interaction : nextcord.Interaction):
                    if interaction.user != ctx.author:
                        return
                    if ctx.voice_client.is_playing():
                        ctx.voice_client.stop()

                    self.check_queue(ctx,ctx.guild.id)

                    

                    

                    

                
        
        
        
        try: await ctx.send(embed=embed, view=View())
        except: await ctx.send(embed=embed)
        

    @commands.command(aliases=["vol"])   
    @commands.guild_only()
    async def volume(self,ctx : commands.Context,Vol : typing.Union[int,str,float]=None):  
        if Vol is None:
            Vol = ctx.voice_client.source.volume * 100
        else:
            Vol=float(str(Vol).replace("%",""))
        
            if Vol > 150:
                return await ctx.send("Volume too high")

            ctx.voice_client.source.volume = Vol / 100
                
        class View(nextcord.ui.View):
            def __init__(self):
                super().__init__(timeout=None)

            @nextcord.ui.button(label="+10",style=nextcord.ButtonStyle.blurple)
            async def add__(self,button : nextcord.Button , interaction : nextcord.Interaction):
                Vol = float(ctx.voice_client.source.volume)
                
                if (Vol * 100 + 10) > 151:
                    return await interaction.response.send_message("Volume too high", ephemeral=True)
                Vol += float(10 / 100)
  
                ctx.voice_client.source.volume = Vol
      
                await interaction.message.edit(content=f"Changed volume to {round(Vol * 100)}%")
             
                    
            @nextcord.ui.button(label="-10",style=nextcord.ButtonStyle.blurple)
            async def rest__(self,button : nextcord.Button , interaction : nextcord.Interaction):
                Vol = float(ctx.voice_client.source.volume)
                
                if (Vol * 100 - 10) < -1:
                    return await interaction.response.send_message("Volume too low")
                Vol -= float(10 / 100)
    
                ctx.voice_client.source.volume = Vol

                await interaction.message.edit(content=f"Changed volume to {round(Vol * 100)}%")

      
  
        try: await ctx.send(f"Changed volume to {round(Vol)}%",view=View())
        except: await ctx.send(f"Changed volume to {round(Vol)}%")

    @commands.command(aliases=['l'])
    @commands.guild_only()
    async def leave(self, ctx : commands.Context):
        if ctx.author.voice is not None:
            #ctx.voice_client.cleanup()
            ctx.voice_client.stop()
            del queues[ctx.guild.id]
            await ctx.voice_client.disconnect()
  

    @commands.command(aliases=['pa'])
    @commands.guild_only()
    async def pause(self,ctx : commands.Context):
        if ctx.author.voice is not None:
            if ctx.voice_client.is_playing():
                ctx.voice_client.pause()  
            else:
                await ctx.send("Currently no audio is playing.")
        else:
            await ctx.send("im not connected in a voice channel")


            
    @commands.command(aliases=['unpa','resume','r'])
    @commands.guild_only()
    async def unpause(self,ctx : commands.Context):
        if ctx.author.voice is not None:
            if ctx.voice_client.is_paused():
                ctx.voice_client.resume()
  
            else:
                await ctx.send("The audio is not paused.")



    @commands.command(aliases=["s"])
    @commands.guild_only()
    async def stop(self,ctx : commands.Context):
        if ctx.author.voice is not None:
            if (ctx.voice_client.is_playing() or ctx.voice_client.is_paused()):
                ctx.voice_client.stop()
                del queues[ctx.guild.id]
            else:
                 
                await ctx.send("put something to stop ")
                
        else:
            await ctx.send("You aren't in a voice channel")


    @commands.command(aliases=['letra','ly','le'])
    async def lyrics(self,ctx : commands.Context,*,Song_name : str = '<NAME> - Never Gonna Give You Up'):
        async with ctx.typing():
            url = await http.get(
                    f'https://some-random-api.ml/lyrics/?title={Song_name}', res_method="json"
                )
            if 'error' in url:
                if url['error'] == "Sorry I couldn't find that song's lyrics":
                    await ctx.send("lyrics not found")

            title=url['title']
            author=url['author']
            lyrics=url['lyrics']        
            thumbnail=url['thumbnail']['genius']

            embed = nextcord.Embed(title=f"{title} : {author}",description=lyrics)
            embed.set_thumbnail(url=thumbnail)
        await ctx.send(embed=embed)
        
        

    @commands.command(aliases=['mv'])
    async def move_to(self,ctx : commands.Context):
        if ctx.author.voice.channel != ctx.voice_client.channel:
            _ = False
            if ctx.voice_client.is_playing():
                _ = True
                ctx.voice_client.pause()

            await ctx.voice_client.move_to(channel=ctx.author.voice.channel)

            if _ == True:
                ctx.voice_client.resume()


        else:  
            await ctx.send("you have to be in a diferent voice channel")


    @commands.command()
    async def skip(self,ctx : commands.Context):
        if ctx.voice_client.is_playing():
             ctx.voice_client.stop()

        self.check_queue(ctx,ctx.guild.id)

    @commands.command()
    async def queue(self,ctx : commands.Context):
        if ctx.guild.id in queues:
            if queues[ctx.guild.id] != []:
                queue = queues[ctx.guild.id]
            else:
                queue : list = ['"Error" : "Guild not have a queue"']
                
        else:
            queue : list = ['"Error" : "Guild not have a queue"'] 
            

        qlist = ""
        songs = 1
        
        for song in queue:
            if hasattr(song, "data"):
                qlist += f"{songs}. {song.title} \n"
                songs += 1
                

            else:
                 qlist += f"{songs}. {song} \n"

        embed = nextcord.Embed(title="Music Queue", description=f"```{qlist}```",color=nextcord.Color.dark_blue())
        
        await ctx.send(embed=embed)


     

    
        
        


def setup(bot):
    if get_config() == True:
        bot.add_cog(Music(bot))
        

   

<filename>cogs/music_cog.py<gh_stars>0
﻿
import asyncio
import json
import nextcord
import youtube_dl
from nextcord.ext import commands
from utils import http
import typing


get = nextcord.utils.get
youtube_dl.utils.bug_reports_message = lambda: ''


def get_config():
    with open('config.json',"r") as f:
        config=json.load(f)
        if config[str('music_cog')] == True and config[str('maintenance')] == False:
            return True
        else:
            return False



  

           
        

youtube_dl.utils.bug_reports_message = lambda: ''


ytdl_format_options = {
    'format': 'bestaudio/best',
    'outtmpl': '%(extractor)s-%(id)s-%(title)s.%(ext)s',
    'restrictfilenames': True,
    #'noplaylist': True,
    'nocheckcertificate': True,
    'ignoreerrors': False,
    'logtostderr': False,
    'quiet': True,
    'no_warnings': True,
    'default_search': 'auto',
    'source_address': '0.0.0.0' 
}

ffmpeg_options = {
    'options': '-vn'
}

ytdl = youtube_dl.YoutubeDL(ytdl_format_options)




class YTDLSource(nextcord.PCMVolumeTransformer):
    def __init__(self, source, *, data, volume = 0.5):
        super().__init__(source, volume)

        self.data = data

        self.title = data.get('title')
        self.url = data.get('url')
        

    @classmethod
    async def from_url(cls, url, *, loop=None, stream=False):
        loop = loop or asyncio.get_event_loop()

        data = await loop.run_in_executor(None, lambda: ytdl.extract_info(url, download=not stream))
        
        if 'entries' in data:
            data = data['entries'][0]
            

        filename = str(data['url'] if stream else ytdl.prepare_filename(data))
        
        return cls(nextcord.FFmpegPCMAudio(filename, **ffmpeg_options), data=data)

    



    
queues = {}
    


    
class Music(commands.Cog):
    def __init__(self,bot):
        self.bot : commands.Bot = bot
        
        self.persistent_views_added = False
        def check_queue(ctx,id_ : int): 
            if id_ in queues:
                if queues[id_] != []:
                    player_ = queues[id_].pop(0)
                    ctx.voice_client.play(player_, after=lambda e: print('Player error: %s' % e) if e else self.check_queue(ctx,ctx.guild.id))

            else: del queues[id_]
                
        self.check_queue = check_queue

    @commands.command(aliases=['p'])
    @commands.guild_only()
    async def play(self, ctx : commands.Context, *, url : str = 'never gona'):
        async with ctx.typing():
         
            if ctx.voice_client is None:
                if ctx.author.voice:
                    await ctx.author.voice.channel.connect(timeout=None, reconnect=True)
                else:
                    await ctx.send("You are not connected to a voice channel.")
                    raise commands.CommandError("Author not connected to a voice channel.")
                    
            
            #elif ctx.voice_client.is_playing(): 
                #ctx.voice_client.stop()


            elif ctx.voice_client.channel is not None and  ctx.voice_client.channel != ctx.voice_client.channel: 
                await ctx.voice_client.move_to(channel=ctx.author.voice.channel)           

            

            player_ = await YTDLSource.from_url(url, loop=self.bot.loop, stream=True)
            
            if ctx.guild.id in queues:
                if queues[ctx.guild.id] != [] or queues[ctx.guild.id] == [] and ctx.voice_client.is_playing():
                    await ctx.send("Song added to queue")
                    return queues[ctx.guild.id].append(player_)

                elif queues[ctx.guild.id] == [] and not ctx.voice_client.is_playing():
                    queues[ctx.guild.id] = player_


                

            else:
                queues[ctx.guild.id] = [player_]
            player = queues[ctx.guild.id].pop(0)
            ctx.voice_client.play(player, after=lambda e: print('Player error: %s' % e) if e else self.check_queue(ctx,ctx.guild.id))

            embed=nextcord.Embed(title=f'Now playing: __**`{player.title}`**__',description=' ',color=nextcord.Color.red())
            embed.set_author(name=player.title,url=f"https://www.youtube.com/watch?v={player.data['id']}" ,icon_url="https://icones.pro/wp-content/uploads/2021/02/youtube-logo-icone.png")
            
            class View(nextcord.ui.View):
                def __init__(self):
                    def check_queue(ctx,id_ : int): 
                        if id_ in queues:
                            if queues[id_] != []:
                                player_ = queues[id_].pop(0)
                                ctx.voice_client.play(player_, after=lambda e: print('Player error: %s' % e) if e else self.check_queue(ctx,ctx.guild.id))

                        else: del queues[id_]
                    self.check_queue = check_queue
                    super().__init__(timeout=None)
               
                @nextcord.ui.button(label="Stop",style=nextcord.ButtonStyle.red)
                async def stop__(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
                    if interaction.user != ctx.author:
                        return
                    if ctx.author.voice is not None:
                        if (ctx.voice_client.is_playing() or ctx.voice_client.is_paused()):
                            ctx.voice_client.stop()
                            del queues[ctx.guild.id]
                        else:
                 
                            await interaction.response.send_message("put something to stop ", ephemeral=True)
                
                    else:
                        await interaction.response.send_message("You aren't in a voice channel", ephemeral=True)

                @nextcord.ui.button(label="Pause",style=nextcord.ButtonStyle.blurple)
                async def pause__(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
                    if interaction.user != ctx.author:
                        return
                    if ctx.author.voice is not None:
                        if ctx.voice_client.is_playing():
                            ctx.voice_client.pause()  
                        else:
                            None
                    else:
                        None
                    
                @nextcord.ui.button(label="Resume",style=nextcord.ButtonStyle.blurple)
                async def resume__(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
                    if interaction.user != ctx.author:
                        return
                    if ctx.author.voice is not None:
                        if ctx.voice_client.is_paused():
                            ctx.voice_client.resume()
                          


                @nextcord.ui.button(label="Leave",style=nextcord.ButtonStyle.red)
                async def leave__(self, button: nextcord.ui.Button, interaction: nextcord.Interaction): 
                    if interaction.user != ctx.author:
                        return
                    if ctx.author.voice is not None:
                        #ctx.voice_client.cleanup()
                        ctx.voice_client.stop()
                        del queues[ctx.guild.id]
                        await ctx.voice_client.disconnect()
                    

                @nextcord.ui.button(label="Skip", style=nextcord.ButtonStyle.blurple)
                async def skip__(self, button : nextcord.ui.Button, interaction : nextcord.Interaction):
                    if interaction.user != ctx.author:
                        return
                    if ctx.voice_client.is_playing():
                        ctx.voice_client.stop()

                    self.check_queue(ctx,ctx.guild.id)

                    

                    

                    

                
        
        
        
        try: await ctx.send(embed=embed, view=View())
        except: await ctx.send(embed=embed)
        

    @commands.command(aliases=["vol"])   
    @commands.guild_only()
    async def volume(self,ctx : commands.Context,Vol : typing.Union[int,str,float]=None):  
        if Vol is None:
            Vol = ctx.voice_client.source.volume * 100
        else:
            Vol=float(str(Vol).replace("%",""))
        
            if Vol > 150:
                return await ctx.send("Volume too high")

            ctx.voice_client.source.volume = Vol / 100
                
        class View(nextcord.ui.View):
            def __init__(self):
                super().__init__(timeout=None)

            @nextcord.ui.button(label="+10",style=nextcord.ButtonStyle.blurple)
            async def add__(self,button : nextcord.Button , interaction : nextcord.Interaction):
                Vol = float(ctx.voice_client.source.volume)
                
                if (Vol * 100 + 10) > 151:
                    return await interaction.response.send_message("Volume too high", ephemeral=True)
                Vol += float(10 / 100)
  
                ctx.voice_client.source.volume = Vol
      
                await interaction.message.edit(content=f"Changed volume to {round(Vol * 100)}%")
             
                    
            @nextcord.ui.button(label="-10",style=nextcord.ButtonStyle.blurple)
            async def rest__(self,button : nextcord.Button , interaction : nextcord.Interaction):
                Vol = float(ctx.voice_client.source.volume)
                
                if (Vol * 100 - 10) < -1:
                    return await interaction.response.send_message("Volume too low")
                Vol -= float(10 / 100)
    
                ctx.voice_client.source.volume = Vol

                await interaction.message.edit(content=f"Changed volume to {round(Vol * 100)}%")

      
  
        try: await ctx.send(f"Changed volume to {round(Vol)}%",view=View())
        except: await ctx.send(f"Changed volume to {round(Vol)}%")

    @commands.command(aliases=['l'])
    @commands.guild_only()
    async def leave(self, ctx : commands.Context):
        if ctx.author.voice is not None:
            #ctx.voice_client.cleanup()
            ctx.voice_client.stop()
            del queues[ctx.guild.id]
            await ctx.voice_client.disconnect()
  

    @commands.command(aliases=['pa'])
    @commands.guild_only()
    async def pause(self,ctx : commands.Context):
        if ctx.author.voice is not None:
            if ctx.voice_client.is_playing():
                ctx.voice_client.pause()  
            else:
                await ctx.send("Currently no audio is playing.")
        else:
            await ctx.send("im not connected in a voice channel")


            
    @commands.command(aliases=['unpa','resume','r'])
    @commands.guild_only()
    async def unpause(self,ctx : commands.Context):
        if ctx.author.voice is not None:
            if ctx.voice_client.is_paused():
                ctx.voice_client.resume()
  
            else:
                await ctx.send("The audio is not paused.")



    @commands.command(aliases=["s"])
    @commands.guild_only()
    async def stop(self,ctx : commands.Context):
        if ctx.author.voice is not None:
            if (ctx.voice_client.is_playing() or ctx.voice_client.is_paused()):
                ctx.voice_client.stop()
                del queues[ctx.guild.id]
            else:
                 
                await ctx.send("put something to stop ")
                
        else:
            await ctx.send("You aren't in a voice channel")


    @commands.command(aliases=['letra','ly','le'])
    async def lyrics(self,ctx : commands.Context,*,Song_name : str = '<NAME> - Never Gonna Give You Up'):
        async with ctx.typing():
            url = await http.get(
                    f'https://some-random-api.ml/lyrics/?title={Song_name}', res_method="json"
                )
            if 'error' in url:
                if url['error'] == "Sorry I couldn't find that song's lyrics":
                    await ctx.send("lyrics not found")

            title=url['title']
            author=url['author']
            lyrics=url['lyrics']        
            thumbnail=url['thumbnail']['genius']

            embed = nextcord.Embed(title=f"{title} : {author}",description=lyrics)
            embed.set_thumbnail(url=thumbnail)
        await ctx.send(embed=embed)
        
        

    @commands.command(aliases=['mv'])
    async def move_to(self,ctx : commands.Context):
        if ctx.author.voice.channel != ctx.voice_client.channel:
            _ = False
            if ctx.voice_client.is_playing():
                _ = True
                ctx.voice_client.pause()

            await ctx.voice_client.move_to(channel=ctx.author.voice.channel)

            if _ == True:
                ctx.voice_client.resume()


        else:  
            await ctx.send("you have to be in a diferent voice channel")


    @commands.command()
    async def skip(self,ctx : commands.Context):
        if ctx.voice_client.is_playing():
             ctx.voice_client.stop()

        self.check_queue(ctx,ctx.guild.id)

    @commands.command()
    async def queue(self,ctx : commands.Context):
        if ctx.guild.id in queues:
            if queues[ctx.guild.id] != []:
                queue = queues[ctx.guild.id]
            else:
                queue : list = ['"Error" : "Guild not have a queue"']
                
        else:
            queue : list = ['"Error" : "Guild not have a queue"'] 
            

        qlist = ""
        songs = 1
        
        for song in queue:
            if hasattr(song, "data"):
                qlist += f"{songs}. {song.title} \n"
                songs += 1
                

            else:
                 qlist += f"{songs}. {song} \n"

        embed = nextcord.Embed(title="Music Queue", description=f"```{qlist}```",color=nextcord.Color.dark_blue())
        
        await ctx.send(embed=embed)


     

    
        
        


def setup(bot):
    if get_config() == True:
        bot.add_cog(Music(bot))
        

   

#'noplaylist': True, #elif ctx.voice_client.is_playing(): #ctx.voice_client.stop() #ctx.voice_client.cleanup() #ctx.voice_client.cleanup()