TuneBot/cogs/settings.py

from typing import Any

from discord.ext import commands
from discord.message import Message

from bot import TuneBot
from context import CustomContext
from utils.classes import BaseCog
from utils.EmbedGenerator import EmbedGenerator


class SettingsCog(BaseCog, name="Settings"):
    @commands.group(aliases=["aj"], invoke_without_command=True)
    @commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
    async def autojoin(self, ctx: CustomContext):
        """Enable/Disable the bot automatically joining"""
        await EmbedGenerator.Message(
            ctx,
            "Autojoin",
            f"Usage:\n\n`{ctx.prefix}autojoin enable`\n`{ctx.prefix}autojoin disable`",
        )

    @autojoin.command(name="enable", aliases=["set"])
    @commands.has_permissions(manage_channels=True)
    @commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
    async def autojoin_set(self, ctx: CustomContext):
        """Enable the bot automatically joining"""
        voice_state = ctx.author.voice
        if not voice_state:
            embed = ctx.create_embed()
            embed.title = "Please join a voice channel before running this command."
            await ctx.send(embed=embed)
            return

        await ctx.autojoin.update(voice_state.channel.id, ctx.message.channel.id)
        embed = ctx.create_embed()
        embed.title = f"AutoJoin enabled for #{voice_state.channel.name}"
        await ctx.send(embed=embed)

    @autojoin.command(name="disable", aliases=["unset"])
    @commands.has_permissions(manage_channels=True)
    @commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
    async def autojoin_del(self, ctx: CustomContext):
        """Disable the bot automatically joining"""
        await ctx.autojoin.disable()
        embed = ctx.create_embed()
        embed.title = f"AutoJoin disabled"
        await ctx.send(embed=embed)

    @commands.is_owner()
    @commands.group(
        name="source",
        aliases=["src"],
        invoke_without_command=True,
        slash_command=False,
        hidden=True,
    )
    async def source(self, ctx: CustomContext):
        """Displays all possible options for the `source` command"""
        prefix = self.bot.config["prefixes"][0]
        embed = ctx.create_embed()
        embed.title = "All options:"
        embed.description = f"```{prefix}source list\n{prefix}source add <url>\n{prefix}source remove <url>\n{prefix}source sync```"
        await ctx.send(embed=embed)

    @commands.is_owner()
    @source.command(name="remove")
    async def source_remove(self, ctx: CustomContext, source_url: str):
        """Removes a source from the bot"""
        if await ctx.playlist_source.remove(source_url):
            prefix = self.bot.config["prefixes"][0]
            embed = ctx.create_embed()
            embed.title = "Removed source succesfully"
            embed.description = (
                f"Please use `{prefix}source sync` to persist these changes."
            )
            await ctx.send(embed=embed)
            return

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

    @commands.is_owner()
    @source.command(name="add")
    async def source_add(self, ctx: CustomContext, source_url: str):
        """
        Add's a source to the bot

        Supported sources: YouTube, SoundCloud, Bandcamp, Vimeo, Twitch and HTTP(S) URL's
        """
        embed = ctx.create_embed()
        embed.title = "Started processing source"
        message = await ctx.send(embed=embed)

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

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

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

        embed = ctx.create_embed()
        embed.title = "Finished processing source"
        embed.description = f"Added {len(track_urls)} tracks"
        if isinstance(message, Message):
            await message.edit(embed=embed)
        else:
            await ctx.send(embed=embed)

    @commands.is_owner()
    @source.command(name="list", aliases=["ls"])
    async def source_list(self, ctx: CustomContext):
        """Display a list of sources"""
        # TODO: Implement pagination for sources
        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 = ctx.create_embed()
        embed.title = "All sources:"
        embed.description = description
        await ctx.send(embed=embed)

    @commands.is_owner()
    @source.command(name="sync")
    async def source_sync(self, ctx: CustomContext):
        """Forcefully resyncs all sources"""
        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: 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 = ctx.create_embed()
        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.send(embed=embed)


def setup(bot: TuneBot):
    bot.add_cog(SettingsCog(bot))