Merge remote-tracking branch 'upstream/dev' into database-rewrite

cogs/settings.py

@@ -1,5 +1,7 @@
-from discord import Message
+from typing import Any
+
 from discord.ext import commands
+from discord.message import Message
 
 from bot import TuneBot
 from context import CustomContext
@@ -156,6 +158,117 @@ class SettingsCog(BaseCog, name="Settings"):
 
         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 PlaylistSource.remove(ctx.redis, 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 PlaylistSource.add(ctx.redis, source_url)
+        track_urls = [str(track["info"]["uri"]) for track in query_result["tracks"]]
+        await Playlist.add_bulk(ctx.redis, 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 PlaylistSource.get_all(ctx.redis)
+        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 Playlist.clear(ctx.redis)
+        sources = await PlaylistSource.get_all(ctx.redis)
+        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 Playlist.add_bulk(ctx.redis, 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))