conjurer/radio_commands.py

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 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}",
                    headers=SERVICE_HEADERS,
                    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")