import datetime import random import discord import discord.opus import functools import sys import db import errors import youtube_dl import os # Init the event loop import asyncio loop = asyncio.get_event_loop() # Init the config reader import configparser config = configparser.ConfigParser() config.read("config.ini") # Find the latest git tag import subprocess import os old_wd = os.getcwd() try: os.chdir(os.path.dirname(__file__)) version = str(subprocess.check_output(["git", "describe", "--tags"]), encoding="utf8").strip() except: version = "v???" finally: os.chdir(old_wd) # Init the discord bot client = discord.Client() discord.opus.load_opus("libopus-0.dll") voice_client = None voice_player = None voice_queue = [] voice_playing = None class Video: def __init__(self): self.user = None self.info = None self.enqueued = None self.channel = None @staticmethod async def init(author, info, enqueued, channel): self = Video() discord_user = await find_user(author) self.user = discord_user.royal if discord_user is not None else None self.info = info self.enqueued = enqueued self.channel = channel return self def create_embed(self): embed = discord.Embed(type="rich", title=self.info.get("title"), url=self.info.get("webpage_url"), colour=discord.Colour(13375518)) # Uploader if self.info.get("uploader"): embed.set_author(name=self.info["uploader"], url=self.info.get("uploader_url")) # Thumbnail if "thumbnail" in self.info: embed.set_thumbnail(url=self.info["thumbnail"]) # Duration embed.add_field(name="Durata", value=str(datetime.timedelta(seconds=self.info["duration"]))) # Views if "view_count" in self.info and self.info["view_count"] is not None: embed.add_field(name="Visualizzazioni", value="{:_}".format(self.info["view_count"]).replace("_", " ")) # Likes if "like_count" in self.info and self.info["like_count"] is not None: embed.add_field(name="Mi piace", value="{:_}".format(self.info["like_count"]).replace("_", " ")) # Dislikes if "dislike_count" in self.info and self.info["dislike_count"] is not None: embed.add_field(name="Non mi piace", value="{:_}".format(self.info["dislike_count"]).replace("_", " ")) return embed async def download(self): try: with youtube_dl.YoutubeDL({"noplaylist": True, "format": "bestaudio", "postprocessors": [{ "key": 'FFmpegExtractAudio', "preferredcodec": 'opus' }], "outtmpl": "music.%(ext)s", "quiet": True}) as ytdl: info = await loop.run_in_executor(None, functools.partial(ytdl.extract_info, self.info["webpage_url"])) except Exception as e: client.send_message(self.channel, f"⚠ Errore durante il download del video:\n" f"```" f"{e}" f"```", embed=self.create_embed()) async def find_user(user: discord.User): session = await loop.run_in_executor(None, db.Session) user = await loop.run_in_executor(None, session.query(db.Discord).filter_by(discord_id=user.id).join(db.Royal).first) return user async def on_error(event, *args, **kwargs): type, exception, traceback = sys.exc_info() try: await client.send_message(client.get_channel("368447084518572034"), f"☢️ ERRORE CRITICO NELL'EVENTO `{event}`\n" f"Il bot si è chiuso e si dovrebbe riavviare entro qualche minuto.\n\n" f"Dettagli dell'errore:\n" f"```python\n" f"{repr(exception)}\n" f"```") await voice_client.disconnect() await client.change_presence(status=discord.Status.invisible) await client.close() except Exception as e: print("ERRORE CRITICO PIU' CRITICO:\n" + repr(e) + "\n" + repr(sys.exc_info())) loop.stop() os._exit(1) pass @client.event async def on_ready(): await client.send_message(client.get_channel("368447084518572034"), f"ℹ Royal Bot {version} avviato e pronto a ricevere comandi!") await client.change_presence(game=None, status=discord.Status.online) @client.event async def on_message(message: discord.Message): global voice_queue global voice_player if message.content.startswith("!register"): await client.send_typing(message.channel) session = await loop.run_in_executor(None, db.Session()) try: username = message.content.split(" ", 1)[1] except IndexError: await client.send_message(message.channel, "⚠️ Non hai specificato un username!\n" "Sintassi corretta: `!register `") return try: d = db.Discord.create(session, royal_username=username, discord_user=message.author) except errors.AlreadyExistingError: await client.send_message(message.channel, "⚠ Il tuo account Discord è già collegato a un account RYG o l'account RYG che hai specificato è già collegato a un account Discord.") return session.add(d) session.commit() session.close() await client.send_message(message.channel, "✅ Sincronizzazione completata!") elif message.content.startswith("!cv"): await client.send_typing(message.channel) if message.author.voice.voice_channel is None: await client.send_message(message.channel, "⚠ Non sei in nessun canale!") return global voice_client if voice_client is not None and voice_client.is_connected(): await voice_client.move_to(message.author.voice.voice_channel) else: voice_client = await client.join_voice_channel(message.author.voice.voice_channel) await client.send_message(message.channel, f"✅ Mi sono connesso in <#{message.author.voice.voice_channel.id}>.") elif message.content.startswith("!play"): await client.send_typing(message.channel) # The bot should be in voice chat if voice_client is None: await client.send_message(message.channel, "⚠️ Non sono connesso alla cv!\n" "Fammi entrare scrivendo `!cv` mentre sei in chat vocale.") return # Find the sent url try: url = message.content.split(" ", 1)[1] except IndexError: await client.send_message(message.channel, "⚠️ Non hai specificato un url!\n" "Sintassi corretta: `!play