TuneBot/cogs/settings/interactions.py

import typing

from discord import app_commands
from discord import Interaction

from tunebot.context import ContextLike
from utils.embed import create_embed

if typing.TYPE_CHECKING:
    from discord import VoiceState
    from bot import TuneBot


async def is_source_owner(ctx: Interaction) -> bool:
    is_manager = ctx.user.id in ctx.client.config["manager_ids"]
    is_owner: bool = await ctx.client.is_owner(ctx.user)
    return is_manager or is_owner


class SourceCommands(app_commands.Group):
    def __init__(self, bot: "TuneBot"):
        super().__init__(name="source", description="Manage the radio sources")
        self.bot: "TuneBot" = bot

    @app_commands.command(name="list", description="List all radio sources")
    @app_commands.check(is_source_owner)
    async def list(self, ctx: Interaction):
        sources = await self.bot.global_playlist_source.fetch_sources()
        if len(sources) > 0:
            description = "\n".join([f"[{source}]({source})" for source in sources])
        else:
            description = "This bot has no sources yet"

        embed = create_embed(ctx.user)
        embed.title = "All sources:"
        embed.description = description
        await ctx.response.send_message(embed=embed)

    @app_commands.command(name="add", description="Add a new radio source")
    @app_commands.describe(url="The URL of the to be added source")
    @app_commands.check(is_source_owner)
    async def add(self, ctx: Interaction, url: str):
        embed = create_embed(ctx.user)
        clike = ContextLike.from_discord_interaction(ctx)
        playlist_source = self.bot.playlist_source_context(clike)

        query_result: typing.Any = await self.bot.lavalink.get_tracks(url)

        if query_result["loadType"] == "LOAD_FAILED":
            embed = create_embed(ctx.user)
            embed.title = "The specified URL is not a valid source"
            embed.description = "Supported sources: YouTube, SoundCloud, Bandcamp, Vimeo, Twitch and HTTP(S) URL's"
            await ctx.response.send_message(embed=embed)
            return

        await playlist_source.add(url)

        track_urls = [str(track["info"]["uri"]) for track in query_result["tracks"]]
        await self.bot.global_playlist.add_tracks(track_urls)

        embed.title = "Finished processing source"
        embed.description = f"Added {len(track_urls)} tracks"
        await ctx.response.send_message(embed=embed)

    @app_commands.command(name="remove", description="Remove a radio source")
    @app_commands.describe(url="The URL of the to be removed source")
    @app_commands.check(is_source_owner)
    async def remove(self, ctx: Interaction, url: str):
        embed = create_embed(ctx.user)
        clike = ContextLike.from_discord_interaction(ctx)
        playlist_source = self.bot.playlist_source_context(clike)
        if await playlist_source.remove(url):
            prefix = self.bot.config["prefixes"][0]
            embed.title = "Removed source succesfully"
            embed.description = (
                f"Please use `{prefix}source sync` to persist these changes."
            )
            await ctx.response.send_message(embed=embed)
            return

        embed.title = "Could not remove source, the specified source might not exist"
        await ctx.response.send_message(embed=embed)

    @app_commands.command(name="sync", description="Synchronize all sources")
    @app_commands.check(is_source_owner)
    async def sync(self, ctx: Interaction):
        embed = create_embed(ctx.user)
        failed_sources: list[str] = []
        await self.bot.global_playlist.clear()
        sources = await self.bot.global_playlist_source.fetch_sources()
        for source_url in sources:
            query_result: typing.Any = await self.bot.lavalink.get_tracks(source_url)
            if query_result["loadType"] == "LOAD_FAILED":
                failed_sources.append(source_url)
                continue

            track_urls = [str(track["info"]["uri"]) for track in query_result["tracks"]]
            await self.bot.global_playlist.add_tracks(track_urls)

        embed.title = f"Finished sync ({len(failed_sources)} issues)"
        if len(failed_sources) > 0:
            embed.description = "\n".join(
                [f"[{source_url}]({source_url})" for source_url in failed_sources]
            )

        await ctx.response.send_message(embed=embed)

class SettingCommands(app_commands.Group):
    def __init__(self, bot: "TuneBot"):
        super().__init__(name="settings", description="Tweak and customize")
        self.bot: "TuneBot" = bot

    @app_commands.command(
        name="autojoin",
        description="Let the bot automatically join a channel after occassional maintenance",
    )
    @app_commands.describe(state="enabled or disabled")
    @app_commands.choices(
        state=[
            app_commands.Choice(name="enable", value=0),
            app_commands.Choice(name="disable", value=1),
        ]
    )
    @app_commands.check(is_source_owner)
    async def autojoin(self, ctx: Interaction, state: app_commands.Choice[int]):
        voice_state: "VoiceState" = ctx.user.voice
        embed = create_embed(ctx.user)
        if not voice_state:
            embed.title = "Please join a voice channel before running this command."
            await ctx.response.send_message(embed=embed)
            return

        clike = ContextLike.from_discord_interaction(ctx)
        autojoin = self.bot.autojoin_context(clike)

        enabled = state.value
        if not enabled:
            await autojoin.update(voice_state.channel.id, ctx.channel_id)
            embed.title = f"AutoJoin enabled for #{voice_state.channel.name}"
            await ctx.response.send_message(embed=embed)
            return

        await autojoin.disable()
        embed.title = "AutoJoin disabled"
        await ctx.response.send_message(embed=embed)