From 89c2e12489c20d70669fbfe0a90d3f22ada1a391 Mon Sep 17 00:00:00 2001 From: strNophix Date: Sat, 12 Nov 2022 22:43:03 +0100 Subject: [PATCH] Migrated settings cog --- bot.py | 12 ++- cogs/music/__init__.py | 2 +- cogs/settings/__init__.py | 156 ++-------------------------------- cogs/settings/interactions.py | 145 +++++++++++++++++++++++++++++++ context.py | 7 +- tunebot/context.py | 28 ++++++ tunebot/redis/autojoin.py | 8 +- tunebot/redis/entity.py | 4 +- utils/decorators.py | 27 ------ 9 files changed, 201 insertions(+), 188 deletions(-) create mode 100644 cogs/settings/interactions.py create mode 100644 tunebot/context.py delete mode 100644 utils/decorators.py diff --git a/bot.py b/bot.py index 4049867..18b1a33 100644 --- a/bot.py +++ b/bot.py @@ -12,7 +12,6 @@ import lavalink from aioredis import Redis from aioredis.client import Redis from discord import ActivityType -from discord import app_commands from discord import Message from discord.colour import Color from discord.ext import commands @@ -23,6 +22,8 @@ from discord.ext.commands.errors import ExtensionNotFound from discord.ext.commands.errors import NoEntryPointError from context import CustomContext +from tunebot.redis import RedisPlaylistSource +from tunebot.redis import RedisAutoJoin from tunebot.redis import GlobalRedisAutoJoin from tunebot.redis import GlobalRedisPlaylist from tunebot.redis import GlobalRedisPlaylistSource @@ -30,9 +31,12 @@ from utils.assets import process_colours from utils.log import logger if TYPE_CHECKING: + from tunebot import AutoJoin + from tunebot import PlaylistSource from tunebot import GlobalPlaylist from tunebot import GlobalPlaylistSource from tunebot import GlobalAutoJoin + from tunebot.context import ContextLike config_path = "config.json" @@ -136,6 +140,12 @@ class TuneBot(commands.Bot): async def get_context(self, message: Message, *, cls=CustomContext): return await super().get_context(message, cls=cls) + def autojoin_context(self, ctx: "ContextLike") -> "AutoJoin": + return RedisAutoJoin(ctx) + + def playlist_source_context(self, ctx: "ContextLike") -> "PlaylistSource": + return RedisPlaylistSource(ctx) + @tasks.loop(seconds=30) async def update_status(self): await self.wait_until_ready() diff --git a/cogs/music/__init__.py b/cogs/music/__init__.py index 50ff6e1..51b61d3 100644 --- a/cogs/music/__init__.py +++ b/cogs/music/__init__.py @@ -54,7 +54,7 @@ class MusicCog(BaseCog, name="Music"): embed = helper.create_track_embed(event.player.current) await channel.send(embed=embed) elif isinstance(event, lavalink.events.TrackEndEvent): - await self.fill_player_queue(event.player, 1) + await helper.fill_player_queue(self.bot, event.player, 1) async def setup(bot: "TuneBot"): diff --git a/cogs/settings/__init__.py b/cogs/settings/__init__.py index fcd49f2..77c80f2 100644 --- a/cogs/settings/__init__.py +++ b/cogs/settings/__init__.py @@ -1,163 +1,17 @@ -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.decorators import source_manager_only -from utils.EmbedGenerator import EmbedGenerator +from cogs.settings.interactions import SettingCommands +from cogs.settings.interactions import SourceCommands 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) - - @source_manager_only() - @commands.group( - name="source", - aliases=["src"], - invoke_without_command=True, - 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 \n{prefix}source remove \n{prefix}source sync```" - await ctx.send(embed=embed) - - @source_manager_only() - @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) - - @source_manager_only() - @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) - - @source_manager_only() - @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) - - @source_manager_only() - @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) + pass async def setup(bot: TuneBot): await bot.add_cog(SettingsCog(bot)) + bot.tree.add_command(SettingCommands(bot), override=True) + bot.tree.add_command(SourceCommands(bot), override=True) diff --git a/cogs/settings/interactions.py b/cogs/settings/interactions.py new file mode 100644 index 0000000..05d14ff --- /dev/null +++ b/cogs/settings/interactions.py @@ -0,0 +1,145 @@ +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) diff --git a/context.py b/context.py index 731c11b..a5aeec0 100644 --- a/context.py +++ b/context.py @@ -6,6 +6,7 @@ from discord.ext import commands from discord.ext.commands.errors import CommandInvokeError from lavalink.models import DefaultPlayer +from tunebot.context import ContextLike from tunebot.redis import RedisAutoJoin from tunebot.redis import RedisPlaylistSource @@ -43,11 +44,13 @@ class CustomContext(commands.Context): @property def playlist_source(self) -> "PlaylistSource": if not hasattr(self, "_playlist_source"): - self._playlist_source = RedisPlaylistSource(self) + ctx = ContextLike.from_discord_context(self) + self._playlist_source = self.bot.playlist_source_context(ctx) return self._playlist_source @property def autojoin(self) -> "AutoJoin": if not hasattr(self, "_autojoin"): - self._autojoin = RedisAutoJoin(self) + ctx = ContextLike.from_discord_context(self) + self._autojoin = self.bot.autojoin_context(ctx) return self._autojoin diff --git a/tunebot/context.py b/tunebot/context.py new file mode 100644 index 0000000..f21f16c --- /dev/null +++ b/tunebot/context.py @@ -0,0 +1,28 @@ +import typing +from dataclasses import dataclass + +if typing.TYPE_CHECKING: + from aioredis import Redis + from bot import TuneBot + from discord import Interaction + from context import CustomContext + + +@dataclass +class ContextLike: + redis: "Redis" + bot: "TuneBot" + guild_id: int + + @staticmethod + def from_discord_context(ctx: "CustomContext") -> "ContextLike": + return ContextLike(redis=ctx.redis, bot=ctx.bot, guild_id=ctx.guild.id) + + @staticmethod + def from_discord_interaction(ctx: "Interaction") -> "ContextLike": + return ContextLike( + redis=ctx.client._redis_client, bot=ctx.client, guild_id=ctx.guild_id + ) + + +__all__ = ("ContextLike",) diff --git a/tunebot/redis/autojoin.py b/tunebot/redis/autojoin.py index cbcb6c2..bc6e88b 100644 --- a/tunebot/redis/autojoin.py +++ b/tunebot/redis/autojoin.py @@ -22,20 +22,20 @@ class RedisAutoJoin(RedisContextEntity, AutoJoin): voice_channel_id (int): [description] text_channel_id (int): [description] """ - if not self.ctx.guild: + if not self.ctx.guild_id: raise Exception("This method can only be invoked inside of a guild.") value = f"{voice_channel_id}:{text_channel_id}" - await self.redis.hset(self.key("autojoin"), self.ctx.guild.id, value) + await self.redis.hset(self.key("autojoin"), self.ctx.guild_id, value) async def disable(self): """ Removes the AutoJoin configuration of a guild. """ - if not self.ctx.guild: + if not self.ctx.guild_id: raise Exception("This method can only be invoked inside of a guild.") - await self.redis.hdel(self.key("autojoin"), self.ctx.guild.id) + await self.redis.hdel(self.key("autojoin"), self.ctx.guild_id) __all__ = ("GlobalRedisAutoJoin", "RedisAutoJoin") diff --git a/tunebot/redis/entity.py b/tunebot/redis/entity.py index 21f15b1..30c2774 100644 --- a/tunebot/redis/entity.py +++ b/tunebot/redis/entity.py @@ -3,7 +3,7 @@ from abc import abstractmethod from typing import TYPE_CHECKING if TYPE_CHECKING: - from context import CustomContext + from tunebot.context import ContextLike from aioredis.client import Redis @@ -34,7 +34,7 @@ class RedisBotEntity(RedisEntity): class RedisContextEntity(RedisEntity): - def __init__(self, ctx: "CustomContext") -> None: + def __init__(self, ctx: "ContextLike") -> None: self.ctx = ctx super().__init__() diff --git a/utils/decorators.py b/utils/decorators.py deleted file mode 100644 index 4851dfd..0000000 --- a/utils/decorators.py +++ /dev/null @@ -1,27 +0,0 @@ -from typing import Callable -from typing import TYPE_CHECKING -from typing import TypeVar - -from discord.ext.commands import check -from discord.ext.commands.errors import NotOwner - -if TYPE_CHECKING: - from context import CustomContext - -T = TypeVar("T") - - -def source_manager_only() -> Callable[[T], T]: - """ - A :func:`.check` that checks if the person invoking this command is allowed to modify the radio sources. - """ - - async def predicate(ctx: "CustomContext") -> bool: - is_manager = ctx.author.id in ctx.bot.config["manager_ids"] - is_owner = await ctx.bot.is_owner(ctx.author) - if not is_manager and not is_owner: - raise NotOwner("You are not allowed to modify the sources.") - - return True - - return check(predicate)