Adding new classes to upgrade the music code organization

vulkanbot/ErrorHandler.py

@@ -0,0 +1,73 @@
+import sys
+import datetime
+import traceback
+
+MAXIMUM_TRIES = 10
+
+
+class ErrorHandler():
+    """Receive errors to write into a log file
+
+        Arg:
+        folder_path = Relative path from root to where the logs should be create
+    """
+
+    def __init__(self, folder_path='') -> None:
+        self.__folder_path = folder_path
+
+    def write(self, error) -> bool:
+        """Write the error to a log file"""
+        if not isinstance(error, Exception):
+            return False
+
+        full_path = self.__open_file()
+        if isinstance(full_path, bool):
+            return False
+
+        error_str = self.__prepare_error(error)
+
+        try:
+            with open(file=full_path, mode='w+') as log_file:
+                log_file.write(error_str)
+            return True
+        except Exception as e:
+            print(e)
+            return False
+
+    def __prepare_error(self, error: Exception) -> str:
+        """Receive the error and return a good str"""
+        time_now = datetime.datetime.now()
+        tipo = sys.exc_info()[0]
+        msg = sys.exc_info()[1]
+        tb = traceback.format_tb(sys.exc_info()[2])
+        error_str = f'Error Time: {time_now}\nTipo: {tipo}\nMsg: {msg}\nTraceback: {tb}'
+        return error_str
+
+    def __open_file(self) -> str:
+        """Open a file to write the error"""
+        successfull = False
+        tries = 0
+
+        while True:
+            now = datetime.datetime.now()
+            date_now = now.strftime(r'%d.%b.%Y-%Hh%Mm%Ss')
+
+            full_path = f'{self.__folder_path}/{date_now}({tries}).txt'
+
+            try:
+                log_file = open(file=full_path, mode='w+')
+                log_file.close()
+            except Exception as e:
+                print(e)
+                tries += 1
+                if tries > MAXIMUM_TRIES:
+                    successfull = False
+                    break
+            else:
+                successfull = True
+                break
+
+        if successfull:
+            return full_path
+        else:
+            return False

vulkanbot/commands/Phrases.py

@@ -0,0 +1,59 @@
+import requests
+import json
+import discord
+from config import config
+from discord.ext import commands
+from random import random as rand
+
+
+class Phrases(commands.Cog):
+    """Deal with the generation of motivational phrases"""
+
+    def __init__(self, bot):
+        self.__bot = bot
+
+    @property
+    def bot(self):
+        return self.__bot
+
+    @bot.setter
+    def bot(self, newBot):
+        self.__bot = newBot
+
+    @commands.command(name='frase', help='Envia uma frase legal no seu PV')
+    async def send_phrase(self, ctx):
+        # There is a chance that the phrase will be send for the dev
+        sended = await self.calculate_rgn(ctx)
+        if sended:
+            return
+
+        while True:
+            try:
+                response = requests.get(
+                    'http://api.forismatic.com/api/1.0/?method=getQuote&key=457653&format=json&lang=en')
+                data = json.loads(response.content)
+
+                phrase = data['quoteText']
+                author = data['quoteAuthor']
+
+                text = f'{phrase} \nBy: {author}'
+                await ctx.send(text)
+                break
+            except json.decoder.JSONDecodeError:
+                continue
+            except Exception as e:
+                print(e)
+                await ctx.channel.send('Houve um erro inesperado :/')
+                break
+
+    async def calculate_rgn(self, ctx):
+        x = rand()
+        if x < 0.15:
+            await ctx.send(config.SECRET_MESSAGE)
+            return True
+        else:
+            return False
+
+
+def setup(bot):
+    bot.add_cog(Phrases(bot))

vulkanbot/commands/Warframe.py

@@ -0,0 +1,51 @@
+import requests
+import json
+import discord
+from dotenv import dotenv_values
+from discord.ext import commands
+from config import config
+
+
+class Warframe(commands.Cog):
+    """Deal with the generation of warframe data"""
+
+    def __init__(self, bot):
+        self.__bot = bot
+
+    @property
+    def bot(self):
+        return self.__bot
+
+    @bot.setter
+    def bot(self, newBot):
+        self.__bot = newBot
+
+    @commands.command(name='cetus', help='Informa o tempo atual de Cetus - Warframe')
+    async def get_cetus(self, ctx):
+        try:
+            response = requests.get(config.CETUS_API)
+            data = json.loads(response.content)
+            short = data['shortString']
+
+            responseText = f'{short}'
+
+            embed = discord.Embed(
+                title='Warframe Cetus Timing',
+                description=responseText,
+                colour=0xFF0000
+            )
+            await ctx.send(embed=embed)
+
+        except Exception as e:
+            print(e)
+            responseText = f'Houve um erro inesperado :/'
+            embed = discord.Embed(
+                title='Warframe Cetus Timing',
+                description=responseText,
+                colour=0xFF0000
+            )
+            await ctx.send(embed=embed)
+
+
+def setup(bot):
+    bot.add_cog(Warframe(bot))

vulkanbot/general/Control.py

@@ -0,0 +1,38 @@
+import discord
+from discord.ext.commands.errors import CommandNotFound, MissingRequiredArgument
+from discord.ext import commands
+from config import config
+
+
+class Control(commands.Cog):
+    """Control the flow of the Bot"""
+
+    def __init__(self, bot):
+        self.__bot = bot
+
+    @property
+    def bot(self):
+        return self.__bot
+
+    @bot.setter
+    def bot(self, newBot):
+        self.__bot = newBot
+
+    @commands.Cog.listener()
+    async def on_ready(self):
+        print(config.STARTUP_MESSAGE)
+        await self.__bot.change_presence(status=discord.Status.online, activity=discord.Game(name=f"Vulkan | type {config.BOT_PREFIX}help"))
+        print(config.STARTUP_COMPLETE_MESSAGE)
+
+    @commands.Cog.listener()
+    async def on_command_error(self, ctx, error):
+        if isinstance(error, MissingRequiredArgument):
+            await ctx.channel.send(f'Falta argumentos. Digite {self.__bot.prefix}help para ver os comandos')
+        elif isinstance(error, CommandNotFound):
+            await ctx.channel.send(f'O comando não existe')
+        else:
+            raise error
+
+
+def setup(bot):
+    bot.add_cog(Control(bot))

vulkanbot/general/Filter.py

@@ -0,0 +1,28 @@
+from discord.ext import commands
+
+
+class Filter(commands.Cog):
+    """Deal with filtering of discord messages"""
+
+    def __init__(self, bot):
+        self.__bot = bot
+
+    @property
+    def bot(self):
+        return self.__bot
+
+    @bot.setter
+    def bot(self, newBot):
+        self.__bot = newBot
+
+    @commands.Cog.listener()
+    async def on_message(self, message):
+        if message.author == self.__bot.user:
+            return
+
+        if 'elx' in message.content:
+            await message.channel.send(f'Coé {message.author.name}, tu é gay memo hein bicho')
+
+
+def setup(bot):
+    bot.add_cog(Filter(bot))

vulkanbot/music/Downloader.py

@@ -0,0 +1,13 @@
+class Downloader():
+    """Download music source from Youtube with a music name"""
+
+    def __init__(self) -> None:
+        pass
+
+    def download(self, track_name: str) -> str:
+        if type(track_name) != str:
+            return
+
+    def download_many(self, track_list: list) -> list:
+        if type(track_list) != list:
+            return

vulkanbot/music/Music.py

@@ -0,0 +1,242 @@
+import discord
+from discord.ext import commands
+from youtube_dl import YoutubeDL
+
+from vulkanbot.music.Downloader import Downloader
+from vulkanbot.music.Searcher import Searcher
+
+colours = {
+    'red': 0xDC143C,
+    'green': 0x00FF7F,
+    'grey': 0x708090,
+    'blue': 0x0000CD
+}
+
+
+class Music(commands.Cog):
+    def __init__(self, client):
+        self.__searcher = Searcher()
+        self.__downloader = Downloader()
+
+        self.client = client
+        self.is_playing = False
+        self.repetingOne = False
+        self.repetingAll = False
+        self.current = ()
+        # 2d array containing [song, channel]
+        # self.music_queue vai conter as buscas recebidas feitas no youtube em ordem
+        # Caminho do executável para rodar na minha máquina
+        self.ffmpeg = 'C:/ffmpeg/bin/ffmpeg.exe'
+        # Segue o padrão de [[{'source', 'title'}, canal], [musica, canal]]
+        self.music_queue = []
+        self.vc = ""  # Objeto voice_client do discord
+        self.YDL_OPTIONS = {'format': 'bestaudio', 'noplaylist': 'True'}
+        self.FFMPEG_OPTIONS = {'executable': self.ffmpeg,
+                               'before_options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5', 'options': '-vn'}
+
+    def search_yt(self, item):
+        with YoutubeDL(self.YDL_OPTIONS) as ydl:
+            try:  # Busca um video no youtube e traz o titulo e a fonte dele em formato de dict
+                info = ydl.extract_info("ytsearch:%s" %
+                                        item, download=False)['entries'][0]
+
+            except Exception:
+                return False
+        # Retorna a fonte e o titulo buscado
+        return {'source': info['formats'][0]['url'], 'title': info['title']}
+
+    def play_next(self):
+        if len(self.music_queue) > 0:
+            if self.repetingOne:
+                # Coloca a musica atual no topo da fila
+                self.music_queue.insert(0, self.current)
+            elif self.repetingAll:
+                # Joga a musica atual para o final da fila
+                self.music_queue.append(self.current)
+
+            self.is_playing = True
+            source = self.music_queue[0][0]['source']
+            self.current = self.music_queue[0]  # Update current music
+            self.music_queue.pop(0)  # Remove from the queue
+            player = discord.FFmpegPCMAudio(source, **self.FFMPEG_OPTIONS)
+            self.vc.play(player, after=lambda e: self.play_next())  # Play
+        else:
+            self.is_playing = False
+            self.repetingAll = False
+            self.repetingOne = False
+
+    # infinite loop checking
+    async def play_music(self):
+        if len(self.music_queue) > 0:
+            self.is_playing = True
+            source = self.music_queue[0][0]['source']
+
+            # Try to connect to voice channel if you are not already connected
+            if self.vc == "" or not self.vc.is_connected() or self.vc == None:
+                # Conecta o voice_client no channel da primeira música da lista
+                self.vc = await self.music_queue[0][1].connect()
+            else:
+                await self.vc.move_to(self.music_queue[0][1])
+
+            self.current = self.music_queue[0]  # Update current music
+            self.music_queue.pop(0)  # Remove from the queue
+            player = discord.FFmpegPCMAudio(source, **self.FFMPEG_OPTIONS)
+            # Start the player
+            self.vc.play(player, after=lambda e: self.play_next())
+        else:
+            self.is_playing = False
+            await self.vc.disconnect()
+
+    @commands.command(name="help", alisases=['ajuda'], help="Comando de ajuda")
+    async def ajuda(self, ctx):
+        helptxt = ''
+        for command in self.client.commands:
+            helptxt += f'**{command}** - {command.help}\n'
+        embedhelp = discord.Embed(
+            colour=1646116,  # grey
+            title=f'Comandos do {self.client.user.name}',
+            description=helptxt
+        )
+        embedhelp.set_thumbnail(url=self.client.user.avatar_url)
+        await ctx.send(embed=embedhelp)
+
+    @commands.command(name="play", help="Toca uma música do YouTube", aliases=['p', 'tocar'])
+    async def p(self, ctx, *args):
+        query = " ".join(args)
+
+        try:
+            # Nome do canal de voz que vai entrar
+            voice_channel = ctx.author.voice.channel
+        except:
+            # If voice_channel is None:
+            await self.send_embed(ctx, title='Para tocar música, primeiro se conecte a um canal de voz.', colour_name='grey')
+            return
+        else:
+            song = self.search_yt(query)
+            if type(song) == type(True):  # Caso seja retornado um booleano da busca
+                await self.send_embed(ctx, description='Algo deu errado! Tente escrever o nome da música novamente!', colour_name='red')
+                return
+            else:
+                await self.send_embed(ctx, description=f"Você adicionou a música **{song['title']}** à fila!", colour_name='green')
+                self.music_queue.append([song, voice_channel])
+
+                if self.is_playing == False:
+                    await self.play_music()
+
+    @commands.command(name="queue", help="Mostra as atuais músicas da fila.", aliases=['q', 'fila'])
+    async def q(self, ctx):
+        fila = ""
+        for x in range(len(self.music_queue)):
+            fila += f"**{x+1} - ** {self.music_queue[x][0]['title']}\n"
+
+        if self.repetingOne:  # If Repeting one
+            await self.send_embed(ctx, title='Repeting One Music',
+                                  description=f'Música: **{self.current[0]["title"]}**', colour_name='green')
+        elif fila != "":
+            if self.repetingAll:  # If repeting all
+                await self.send_embed(ctx, title='Repetindo todas', description=fila, colour_name='green')
+            else:  # Repeting off
+                await self.send_embed(ctx, description=fila, colour_name='green')
+        else:  # No music
+            await self.send_embed(ctx, description='Não existem músicas na fila.', colour_name='red')
+
+    @commands.command(name="skip", help="Pula a atual música que está tocando.", aliases=['pular'])
+    async def skip(self, ctx):
+        if self.vc != "" and self.vc:
+            self.vc.stop()
+            await self.send_embed(ctx, description=f'Você pulou a música\nRepetindo Uma: {self.repetingOne} \
+                                \nRepetindo Todas: {self.repetingAll}', colour_name='green')
+
+    @commands.command(name='stop', help='Para de tocar músicas')
+    async def stop(self, ctx):
+        if self.vc == '':
+            return
+        if self.vc.is_connected():
+            # Remove todas as músicas da lista
+            self.music_queue = []
+            self.current = ()
+            self.repetingOne = False
+            self.repetingAll = False
+            self.is_playing = False
+            self.vc.stop()
+            await self.vc.disconnect()
+
+    @commands.command(name='pause', help='Pausa a música')
+    async def pause(self, ctx):
+        if self.vc == '':
+            return
+        if self.vc.is_playing():
+            self.vc.pause()
+            await self.send_embed(ctx, description='Música pausada', colour_name='green')
+
+    @commands.command(name='resume', help='Despausa a música atual')
+    async def resume(self, ctx):
+        if self.vc == '':
+            return
+        if self.vc.is_paused():
+            self.vc.resume()
+            await self.send_embed(ctx, description='Música tocando', colour_name='green')
+
+    @commands.command(name='repeat_one', help='Repete a música atual')
+    async def repeat_one(self, ctx):
+        if not self.is_playing:  # Garante que o Bot está tocando
+            await self.send_embed(ctx, title='Vulkan não está tocando agora', colour_name='red')
+            return
+
+        if self.repetingAll:  # Verifica se o repeting all não está ligado
+            await self.send_embed(ctx, title='Já está repetindo todas', colour_name='red')
+            return
+        else:  # Liga o repeting one
+            self.repetingOne = True
+            await self.send_embed(ctx, description='Repetir uma música ligado', colour_name='green')
+
+    @commands.command(name='repeat_all', help='Repete toda a fila')
+    async def repeat_all(self, ctx):
+        if not self.is_playing:  # Garante que o Bot está tocando
+            await self.send_embed(ctx, title='Vulkan não está tocando agora', colour_name='red')
+            return
+
+        if self.repetingOne:  # Verifica se o repeting all não está ligado
+            await self.send_embed(ctx, title='Já está repetindo uma música', colour_name='red')
+            return
+        else:  # Liga o repeting one
+            self.repetingAll = True
+            await self.send_embed(ctx, description='Repetir todas as músicas ligado', colour_name='green')
+
+    @commands.command(name='repeat_off', help='Desativa o repetir músicas')
+    async def repeat_off(self, ctx):
+        if not self.is_playing:  # Garante que o Bot está tocando
+            await self.send_embed(ctx, title='Vulkan não está tocando agora', colour_name='red')
+            return
+        else:
+            self.repetingOne = False
+            self.repetingAll = False
+            await self.send_embed(ctx, description='Repetir músicas desligado', colour_name='green')
+
+    @skip.error  # Erros para kick
+    async def skip_error(self, ctx, error):
+        if isinstance(error, commands.MissingPermissions):
+            embedvc = discord.Embed(
+                colour=12255232,
+                description=f"Você precisa da permissão **Gerenciar canais** para pular músicas."
+            )
+            await ctx.send(embed=embedvc)
+        else:
+            raise error
+
+    async def send_embed(self, ctx, title='', description='', colour_name='red'):
+        try:
+            colour = colours[colour_name]
+        except Exception as e:
+            colour = colours['red']
+
+        embedvc = discord.Embed(
+            title=title,
+            description=description,
+            colour=colour
+        )
+        await ctx.send(embed=embedvc)
+
+
+def setup(client):
+    client.add_cog(Music(client))

vulkanbot/music/Playlist.py

@@ -0,0 +1,9 @@
+class Playlist():
+    """Class to manage and control the musics to play"""
+
+    def __init__(self) -> None:
+        self.queue = []
+        self.history = []
+
+    def next_music():
+        pass

vulkanbot/music/Searcher.py

@@ -0,0 +1,43 @@
+from vulkanbot.music.Types import Provider
+from vulkanbot.music.Spotify import SpotifySearch
+from vulkanbot.music.Youtube import YoutubeSearch
+
+
+class Searcher():
+    """Turn the user input into list of musics names, support youtube and spotify"""
+
+    def __init__(self) -> None:
+        self.__Youtube = YoutubeSearch()
+        self.__Spotify = SpotifySearch()
+        print(f'Spotify Connected: {self.__Spotify.connect()}')
+
+    def search(self, music: str) -> list:
+        """Return a list with the track name of a music or playlist"""
+        url_type = self.__identify_source(music)
+
+        if url_type == Provider.Name:
+            return [music]
+
+        elif url_type == Provider.YouTube:
+            musics = self.__Youtube.search(music)
+            return musics
+
+        elif url_type == Provider.Spotify:
+            musics = self.__Spotify.search(music)
+            return musics
+
+    def __identify_source(self, music):
+        if 'http' not in music:
+            return Provider.Name
+
+        if "https://www.youtu" in music or "https://youtu.be" in music:
+            return Provider.YouTube
+
+        if "https://open.spotify.com/track" in music:
+            return Provider.Spotify
+
+        if "https://open.spotify.com/playlist" in music or "https://open.spotify.com/album" in music:
+            return Provider.Spotify_Playlist
+
+        # If no match
+        return Provider.Unknown

vulkanbot/music/Song.py

@@ -0,0 +1,12 @@
+class Song():
+    """Deal with information of a song"""
+
+    def __init__(self, source) -> None:
+        self.__source = source
+        self.__get_info()
+
+    def __get_info(self):
+        pass
+
+    def info():
+        """Return the compiled info of this song"""

vulkanbot/music/Spotify.py

@@ -0,0 +1,22 @@
+import spotipy
+from spotipy.oauth2 import SpotifyClientCredentials
+from config import config
+
+
+class SpotifySearch():
+    """Search and return musics names from Spotify"""
+
+    def __init__(self) -> None:
+        pass
+
+    def connect(self) -> bool:
+        try:
+            # Initialize the connection with Spotify API
+            self.__sp_api = spotipy.Spotify(auth_manager=SpotifyClientCredentials(
+                client_id=config.SPOTIFY_ID, client_secret=config.SPOTIFY_SECRET))
+            return True
+        except:
+            return False
+
+    def search(self, music) -> list:
+        pass

vulkanbot/music/Types.py

@@ -0,0 +1,21 @@
+from enum import Enum
+
+
+class Provider(Enum):
+    """Store Enum Types of the Providers"""
+    Spotify = "Spotify"
+    Spotify_Playlist = "Spotify Playlist"
+    YouTube = "YouTube"
+    Name = 'Track Name'
+    Unknown = "Unknown"
+
+
+class Playlist_Types(Enum):
+    Spotify_Playlist = "Spotify Playlist"
+    YouTube_Playlist = "YouTube Playlist"
+    Unknown = "Unknown"
+
+
+class Origins(Enum):
+    Default = "Default"
+    Playlist = "Playlist"

vulkanbot/music/Youtube.py

@@ -0,0 +1,8 @@
+class YoutubeSearch():
+    """Search for tracks in youtube"""
+
+    def __init__(self) -> None:
+        pass
+
+    def search(self, track):
+        pass