Create cog disabling API (#4043)

* create cog disbale base

* Because defaults...

* lol

* announcer needs to respect this

* defaultdict mishap

* Allow None as guild

- Mostly for interop with with ctx.guild

* a whitespace issue

* Apparently, I broke this too

* Apply suggestions from code review

Co-authored-by: jack1142 <6032823+jack1142@users.noreply.github.com>

* This can probably be more optimized later, but since this is a cached value, it's not a large issue

* Report tunnel closing

* mod too

* whitespace issue

* Fix Artifact of prior method naming

* these 3 places should have the check if i understood it correctly

* Announce the closed tunnels

* tunnel oversight

* Make the player stop at next track

* added where draper said to put it

* Apply suggestions from code review

Co-authored-by: jack1142 <6032823+jack1142@users.noreply.github.com>

Co-authored-by: jack1142 <6032823+jack1142@users.noreply.github.com>
Co-authored-by: Drapersniper <27962761+drapersniper@users.noreply.github.com>
This commit is contained in:
Michael H
2020-07-28 14:52:36 -04:00
committed by GitHub
parent 97379afe6d
commit 1d80fe9aec
18 changed files with 329 additions and 5 deletions

View File

@@ -37,7 +37,12 @@ from .dev_commands import Dev
from .events import init_events
from .global_checks import init_global_checks
from .settings_caches import PrefixManager, IgnoreManager, WhitelistBlacklistManager
from .settings_caches import (
PrefixManager,
IgnoreManager,
WhitelistBlacklistManager,
DisabledCogCache,
)
from .rpc import RPCMixin
from .utils import common_filters
@@ -132,12 +137,16 @@ class RedBase(
self._config.register_channel(embeds=None, ignored=False)
self._config.register_user(embeds=None)
self._config.init_custom("COG_DISABLE_SETTINGS", 2)
self._config.register_custom("COG_DISABLE_SETTINGS", disabled=None)
self._config.init_custom(CUSTOM_GROUPS, 2)
self._config.register_custom(CUSTOM_GROUPS)
self._config.init_custom(SHARED_API_TOKENS, 2)
self._config.register_custom(SHARED_API_TOKENS)
self._prefix_cache = PrefixManager(self._config, cli_flags)
self._disabled_cog_cache = DisabledCogCache(self._config)
self._ignored_cache = IgnoreManager(self._config)
self._whiteblacklist_cache = WhitelistBlacklistManager(self._config)
@@ -217,6 +226,41 @@ class RedBase(
return_exceptions=return_exceptions,
)
async def cog_disabled_in_guild(
self, cog: commands.Cog, guild: Optional[discord.Guild]
) -> bool:
"""
Check if a cog is disabled in a guild
Parameters
----------
cog: commands.Cog
guild: Optional[discord.Guild]
Returns
-------
bool
"""
if guild is None:
return False
return await self._disabled_cog_cache.cog_disabled_in_guild(cog.qualified_name, guild.id)
async def cog_disabled_in_guild_raw(self, cog_name: str, guild_id: int) -> bool:
"""
Check if a cog is disabled in a guild without the cog or guild object
Parameters
----------
cog_name: str
This should be the cog's qualified name, not neccessarily the classname
guild_id: int
Returns
-------
bool
"""
return await self._disabled_cog_cache.cog_disabled_in_guild(cog_name, guild_id)
def remove_before_invoke_hook(self, coro: PreInvokeCoroutine) -> None:
"""
Functional method to remove a `before_invoke` hook.

View File

@@ -511,6 +511,10 @@ class Requires:
bot_user = ctx.bot.user
else:
bot_user = ctx.guild.me
cog = ctx.cog
if cog and await ctx.bot.cog_disabled_in_guild(cog, ctx.guild):
raise discord.ext.commands.DisabledCommand()
bot_perms = ctx.channel.permissions_for(bot_user)
if not (bot_perms.administrator or bot_perms >= self.bot_perms):
raise BotMissingPermissions(missing=self._missing_perms(self.bot_perms, bot_perms))

View File

@@ -2174,9 +2174,83 @@ class Core(commands.Cog, CoreLogic):
@checks.guildowner_or_permissions(administrator=True)
@commands.group(name="command")
async def command_manager(self, ctx: commands.Context):
"""Manage the bot's commands."""
"""Manage the bot's commands and cogs."""
pass
@checks.is_owner()
@command_manager.command(name="defaultdisablecog")
async def command_default_disable_cog(self, ctx: commands.Context, *, cogname: str):
"""Set the default state for a cog as disabled."""
cog = self.bot.get_cog(cogname)
if not cog:
return await ctx.send(_("Cog with the given name doesn't exist."))
if cog == self:
return await ctx.send(_("You can't disable this cog by default."))
await self.bot._disabled_cog_cache.default_disable(cogname)
await ctx.send(_("{cogname} has been set as disabled by default.").format(cogname=cogname))
@checks.is_owner()
@command_manager.command(name="defaultenablecog")
async def command_default_enable_cog(self, ctx: commands.Context, *, cogname: str):
"""Set the default state for a cog as enabled."""
cog = self.bot.get_cog(cogname)
if not cog:
return await ctx.send(_("Cog with the given name doesn't exist."))
await self.bot._disabled_cog_cache.default_enable(cogname)
await ctx.send(_("{cogname} has been set as enabled by default.").format(cogname=cogname))
@commands.guild_only()
@command_manager.command(name="disablecog")
async def command_disable_cog(self, ctx: commands.Context, *, cogname: str):
"""Disable a cog in this guild."""
cog = self.bot.get_cog(cogname)
if not cog:
return await ctx.send(_("Cog with the given name doesn't exist."))
if cog == self:
return await ctx.send(_("You can't disable this cog as you would lock yourself out."))
if await self.bot._disabled_cog_cache.disable_cog_in_guild(cogname, ctx.guild.id):
await ctx.send(_("{cogname} has been disabled in this guild.").format(cogname=cogname))
else:
await ctx.send(
_("{cogname} was already disabled (nothing to do).").format(cogname=cogname)
)
@commands.guild_only()
@command_manager.command(name="enablecog")
async def command_enable_cog(self, ctx: commands.Context, *, cogname: str):
"""Enable a cog in this guild."""
if await self.bot._disabled_cog_cache.enable_cog_in_guild(cogname, ctx.guild.id):
await ctx.send(_("{cogname} has been enabled in this guild.").format(cogname=cogname))
else:
# putting this here allows enabling a cog that isn't loaded but was disabled.
cog = self.bot.get_cog(cogname)
if not cog:
return await ctx.send(_("Cog with the given name doesn't exist."))
await ctx.send(
_("{cogname} was not disabled (nothing to do).").format(cogname=cogname)
)
@commands.guild_only()
@command_manager.command(name="listdisabledcogs")
async def command_list_disabled_cogs(self, ctx: commands.Context):
"""List the cogs which are disabled in this guild."""
disabled = [
cog.qualified_name
for cog in self.bot.cogs.values()
if await self.bot._disabled_cog_cache.cog_disabled_in_guild(
cog.qualified_name, ctx.guild.id
)
]
if disabled:
output = _("The following cogs are disabled in this guild:\n")
output += humanize_list(disabled)
for page in pagify(output):
await ctx.send(page)
else:
await ctx.send(_("There are no disabled cogs in this guild."))
@command_manager.group(name="listdisabled", invoke_without_command=True)
async def list_disabled(self, ctx: commands.Context):
"""

View File

@@ -1,7 +1,8 @@
from __future__ import annotations
from typing import Dict, List, Optional, Union, Set, Iterable
from typing import Dict, List, Optional, Union, Set, Iterable, Tuple
from argparse import Namespace
from collections import defaultdict
import discord
@@ -254,3 +255,108 @@ class WhitelistBlacklistManager:
)
self._cached_blacklist[gid].difference_update(role_or_user)
await self._config.guild_from_id(gid).blacklist.set(list(self._cached_blacklist[gid]))
class DisabledCogCache:
def __init__(self, config: Config):
self._config = config
self._disable_map: Dict[str, Dict[int, bool]] = defaultdict(dict)
async def cog_disabled_in_guild(self, cog_name: str, guild_id: int) -> bool:
"""
Check if a cog is disabled in a guild
Parameters
----------
cog_name: str
This should be the cog's qualified name, not neccessarily the classname
guild_id: int
Returns
-------
bool
"""
if guild_id in self._disable_map[cog_name]:
return self._disable_map[cog_name][guild_id]
gset = await self._config.custom("COG_DISABLE_SETTINGS", cog_name, guild_id).disabled()
if gset is None:
gset = await self._config.custom("COG_DISABLE_SETTINGS", cog_name, 0).disabled()
if gset is None:
gset = False
self._disable_map[cog_name][guild_id] = gset
return gset
async def default_disable(self, cog_name: str):
"""
Sets the default for a cog as disabled.
Parameters
----------
cog_name: str
This should be the cog's qualified name, not neccessarily the classname
"""
await self._config.custom("COG_DISABLE_SETTINGS", cog_name, 0).disabled.set(True)
del self._disable_map[cog_name]
async def default_enable(self, cog_name: str):
"""
Sets the default for a cog as enabled.
Parameters
----------
cog_name: str
This should be the cog's qualified name, not neccessarily the classname
"""
await self._config.custom("COG_DISABLE_SETTINGS", cog_name, 0).disabled.clear()
del self._disable_map[cog_name]
async def disable_cog_in_guild(self, cog_name: str, guild_id: int) -> bool:
"""
Disable a cog in a guild.
Parameters
----------
cog_name: str
This should be the cog's qualified name, not neccessarily the classname
guild_id: int
Returns
-------
bool
Whether or not any change was made.
This may be useful for settings commands.
"""
if await self.cog_disabled_in_guild(cog_name, guild_id):
return False
self._disable_map[cog_name][guild_id] = True
await self._config.custom("COG_DISABLE_SETTINGS", cog_name, guild_id).disabled.set(True)
return True
async def enable_cog_in_guild(self, cog_name: str, guild_id: int) -> bool:
"""
Enable a cog in a guild.
Parameters
----------
cog_name: str
This should be the cog's qualified name, not neccessarily the classname
guild_id: int
Returns
-------
bool
Whether or not any change was made.
This may be useful for settings commands.
"""
if not await self.cog_disabled_in_guild(cog_name, guild_id):
return False
self._disable_map[cog_name][guild_id] = False
await self._config.custom("COG_DISABLE_SETTINGS", cog_name, guild_id).disabled.set(False)
return True

View File

@@ -1,3 +1,4 @@
import asyncio
import discord
from datetime import datetime
from redbot.core.utils.chat_formatting import pagify
@@ -175,6 +176,19 @@ class Tunnel(metaclass=TunnelMeta):
# Backwards-compatible typo fix (GH-2496)
files_from_attatch = files_from_attach
async def close_because_disabled(self, close_message: str):
"""
Sends a mesage to both ends of the tunnel that the tunnel is now closed.
Parameters
----------
close_message: str
The message to send to both ends of the tunnel.
"""
tasks = [destination.send(close_message) for destination in (self.recipient, self.origin)]
await asyncio.gather(*tasks, return_exceptions=True)
async def communicate(
self, *, message: discord.Message, topic: str = None, skip_message_content: bool = False
):