import logging from discord.ext import commands import discord # trunk-ignore(bandit/B404) import subprocess import requests import uuid import asyncio from datetime import datetime from conjurer.backup_old_docker.constants import ( RADIO_HARBOR_ADDRESS, SKIP_TRACK, FILE_SERVICE_ADDRESS, ADD_TO_PRIO_PLAYLIST, REQUEST_MUSIC, CREATE_PRIO_PLAYLIST, CLEAR_PRIO, service_headers, ) SERVICE_HEADERS = service_headers() class RadioModule(commands.Cog): def __init__(self, bot, logger_name): self.bot = bot self.logger = logging.getLogger(logger_name) @commands.hybrid_command( name="skip_track", description="Przeskocz kawałek w radiu", guild=discord.Object(id=664789470779932693), ) @commands.has_any_role('Legenda', 'Jarl', 'Thane' , 'Bartender') async def skip_track(self, ctx): async with ctx.typing(): allowed = False for role in ctx.author.roles: if role.name in ("Thane", "Jarl"): allowed = True if not allowed: await ctx.send("Łapy precz od radia") else: coroutine = asyncio.to_thread( requests.get, f"{RADIO_HARBOR_ADDRESS}{SKIP_TRACK}", timeout=360, ) result = await coroutine self.logger.info("Done %s", result) await ctx.send("Zrobione szefie!") @commands.hybrid_command( name="zrestartuj_radio", description="Komenda ktora uruchamia radio ponownie jakby się zawiesiło", guild=discord.Object(id=664789470779932693), ) @commands.has_any_role('Legenda', 'Jarl', 'Thane' , 'Bartender') async def zrestartuj_radio(self, ctx): async with ctx.typing(): allowed = False for role in ctx.author.roles: if role.name in ("Thane", "Jarl"): allowed = True if not allowed: await ctx.send("Łapy precz od radia") else: retcode = subprocess.run( "/home/pi/Conjurer/restart_radio.sh", # trunk-ignore(bandit/B603) shell=False, check=False, capture_output=True, ) self.logger.info("Wynik: %s", retcode) @commands.hybrid_command( name="dodaj_do_ulubionych", description="Dodaje do listy ulubionych w radiu", guild=discord.Object(id=664789470779932693), ) @commands.has_any_role('Legenda', 'Jarl', 'Thane' , 'Bartender') async def dodaj_do_ulubionych(self, ctx): """ Generate a playlist in queue. First word in this command shall be int defining length of the playlist. Rest of the line are search terms. :param ctx: ctx stands for "context" and is a parameter commonly used in Discord.py commands. It represents the context in which the command was invoked, including information such as the message, the channel, the server, and the user who invoked the command. This parameter is required in all Discord.py commands """ async with ctx.typing(): self.logger.info("Zaczynam szukać timestamp %s", datetime.now()) dlugosc_playlisty = ctx.message.content.split()[1] word_list = ctx.message.content.split() jrequest = { "lista_slow": word_list, "dlugosc_plejlisty": dlugosc_playlisty, "UUID": str(uuid.uuid4()), } coroutine = asyncio.to_thread( requests.post, f"{FILE_SERVICE_ADDRESS}{ADD_TO_PRIO_PLAYLIST}", json=jrequest, headers=SERVICE_HEADERS, timeout=360, ) result = await coroutine self.logger.info("Done %s", result) await ctx.send("Zrobione szefie!") @commands.hybrid_command( name="ja_chciol", description="Dodaje do listy ulubionych w radiu", guild=discord.Object(id=664789470779932693), ) @commands.has_any_role('Nocna Zmiana', 'Jarl', 'Thane' , 'Bartender') async def request_radio(self, ctx): """ Generate a playlist in queue. First word in this command shall be int defining length of the playlist. Rest of the line are search terms. :param ctx: ctx stands for "context" and is a parameter commonly used in Discord.py commands. It represents the context in which the command was invoked, including information such as the message, the channel, the server, and the user who invoked the command. This parameter is required in all Discord.py commands """ async with ctx.typing(): self.logger.info("Zaczynam szukać timestamp %s", datetime.now()) word_list = ctx.message.content.split() jrequest = { "lista_slow": word_list, "UUID": str(uuid.uuid4()), } coroutine = asyncio.to_thread( requests.post, f"{FILE_SERVICE_ADDRESS}{REQUEST_MUSIC}", json=jrequest, headers=SERVICE_HEADERS, timeout=360, ) result = await coroutine self.logger.info("Done %s", result) await ctx.send("Zrobione szefie!") @commands.hybrid_command( name="stworz_audycje", description="Dodaje do listy ulubionych w radiu", guild=discord.Object(id=664789470779932693), ) @commands.has_any_role('Legenda', 'Jarl', 'Thane' , 'Bartender') async def stworz_audycje(self, ctx): """ Generate a playlist in queue. First word in this command shall be int defining length of the playlist. Rest of the line are search terms. :param ctx: ctx stands for "context" and is a parameter commonly used in Discord.py commands. It represents the context in which the command was invoked, including information such as the message, the channel, the server, and the user who invoked the command. This parameter is required in all Discord.py commands """ async with ctx.typing(): self.logger.info("Zaczynam szukać timestamp %s", datetime.now()) dlugosc_playlisty = ctx.message.content.split()[1] word_list = ctx.message.content.split() jrequest = { "lista_slow": word_list, "dlugosc_plejlisty": dlugosc_playlisty, "UUID": str(uuid.uuid4()), } coroutine = asyncio.to_thread( requests.post, f"{FILE_SERVICE_ADDRESS}{CREATE_PRIO_PLAYLIST}", json=jrequest, headers=SERVICE_HEADERS, timeout=360, ) result = await coroutine self.logger.info("Done %s", result) await ctx.send("Zrobione szefie!") @commands.hybrid_command( name="wyczysc_ulubione", description="Czysci liste ulubionych w radiu", guild=discord.Object(id=664789470779932693), ) @commands.has_any_role('Legenda', 'Jarl', 'Thane' , 'Bartender') async def wyczysc_ulubione(self, ctx): """ Generate a playlist in queue. First word in this command shall be int defining length of the playlist. Rest of the line are search terms. :param ctx: ctx stands for "context" and is a parameter commonly used in Discord.py commands. It represents the context in which the command was invoked, including information such as the message, the channel, the server, and the user who invoked the command. This parameter is required in all Discord.py commands """ async with ctx.typing(): coroutine = asyncio.to_thread( requests.get, f"{FILE_SERVICE_ADDRESS}{CLEAR_PRIO}", headers=SERVICE_HEADERS, timeout=360, ) result = await coroutine self.logger.info("Done %s", result) await ctx.send("Zrobione szefie!") async def setup(bot): logger = logging.getLogger("discord") await bot.add_cog(RadioModule(bot, "discord")) logger.info("Loading raadio commands module done")