tux.cogs.moderation.command_meta
¶
Metaclass + base class for declarative moderation commands.
Subclass ModerationCommand
once per moderation action. Example::
class Ban(ModerationCommand):
name = "ban"
aliases = ["b"]
case_type = CaseType.BAN
required_pl = 3
flags = {
"purge": dict(type=int, aliases=["p"], default=0, desc="Days to delete"),
"silent": dict(type=bool, aliases=["s", "quiet"], default=False, desc="No DM"),
}
async def _action(self, guild: discord.Guild, member: discord.Member, *, flags, reason):
await guild.ban(member, reason=reason, delete_message_seconds=flags.purge * 86400)
Classes:
Name | Description |
---|---|
ModerationCommandMeta | Metaclass that turns class attributes into a real command. |
ModerationCommand | Base class to inherit for each moderation action. |
ModerationCommandsCog | |
Classes¶
ModerationCommand
¶
Base class to inherit for each moderation action.
ModerationCommandsCog(bot: commands.Bot)
¶
Bases: ModerationCogBase
Methods:
Name | Description |
---|---|
get_user_lock | Get or create a lock for operations on a specific user. |
clean_user_locks | Remove locks for users that are not currently in use. |
execute_user_action_with_lock | Execute an action on a user with a lock to prevent race conditions. |
execute_mod_action | Execute a moderation action with case creation, DM sending, and additional actions. |
send_error_response | Send a standardized error response. |
create_embed | Create an embed for moderation actions. |
send_embed | Send an embed to the log channel. |
send_dm | Send a DM to the target user. |
check_conditions | Check if the conditions for the moderation action are met. |
handle_case_response | Handle the response for a case. |
is_pollbanned | Check if a user is poll banned. |
is_snippetbanned | Check if a user is snippet banned. |
is_jailed | Check if a user is jailed using the optimized latest case method. |
execute_mixed_mod_action | Parse mixed_args according to config and execute the moderation flow. |
execute_flag_mod_action | Execute moderation flow based on flags parsed by FlagConverter. |
Source code in tux/cogs/moderation/command_meta.py
Functions¶
get_user_lock(user_id: int) -> Lock
async
¶
Get or create a lock for operations on a specific user. If the number of stored locks exceeds the cleanup threshold, unused locks are removed.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
user_id | int | The ID of the user to get a lock for. | required |
Returns:
Type | Description |
---|---|
Lock | The lock for the user. |
Source code in tux/cogs/moderation/command_meta.py
def __new__(mcls, name: str, bases: tuple[type, ...], ns: dict[str, Any]):
cls = super().__new__(mcls, name, bases, ns)
if cls.__name__ == "ModerationCommand":
return cls
# Extract required attributes
cmd_name: str = getattr(cls, "name") # type: ignore[arg-type]
aliases: list[str] = getattr(cls, "aliases", []) # type: ignore[arg-type]
case_type = getattr(cls, "case_type")
required_pl: int = getattr(cls, "required_pl", 0)
flags_spec: Dict[str, Dict[str, Any]] = getattr(cls, "flags", {}) # type: ignore[arg-type]
description: str = getattr(cls, "description", cmd_name.title())
# Build FlagConverter
FlagsCls = build_flag_converter(
cmd_name,
duration="duration" in flags_spec,
purge="purge" in flags_spec,
silent="silent" in flags_spec,
)
# Make FlagsCls resolvable for annotation eval
clean_user_locks() -> None
async
¶
Remove locks for users that are not currently in use. Iterates through the locks and removes any that are not currently locked.
Source code in tux/cogs/moderation/command_meta.py
# Expose under a stable alias so eval("FlagsCls") always succeeds
globals()['FlagsCls'] = FlagsCls
setattr(builtins, 'FlagsCls', FlagsCls)
# --------------------------------------------------
# Shared executor
# --------------------------------------------------
async def _run(self: ModerationCogBase, ctx, target: MemberOrUser, flags, reason: str):
if not await self.check_conditions(ctx, target, ctx.author, cmd_name):
return
silent = getattr(flags, "silent", False)
duration = getattr(flags, "duration", None)
action_coro = cls._action(self, ctx.guild, target, flags=flags, reason=reason) # type: ignore[arg-type]
actions = [(action_coro, type(None))]
execute_user_action_with_lock(user_id: int, action_func: Callable[..., Coroutine[Any, Any, R]], *args: Any, **kwargs: Any) -> R
async
¶
Execute an action on a user with a lock to prevent race conditions.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
user_id | int | The ID of the user to lock. | required |
action_func | Callable[..., Coroutine[Any, Any, R]] | The coroutine function to execute. | required |
*args | Any | Arguments to pass to the function. | () |
**kwargs | Any | Keyword arguments to pass to the function. | {} |
Returns:
Type | Description |
---|---|
R | The result of the action function. |
Source code in tux/cogs/moderation/command_meta.py
ctx=ctx,
case_type=case_type,
user=target,
reason=reason,
silent=silent,
dm_action=getattr(cls, "dm_action", cmd_name),
actions=actions,
duration=duration,
)
# --------------------------------------------------
# Text command (prefix)
# --------------------------------------------------
async def _text(self: ModerationCogBase, ctx: commands.Context, target: MemberOrUser, *, flags: FlagsCls | None = None, reason: str = "") -> None: # type: ignore[arg-type]
if flags is None:
flags = FlagsCls() # type: ignore[assignment]
await _run(self, ctx, target, flags, reason)
if FlagsCls is not None:
_text.__globals__[FlagsCls.__name__] = FlagsCls
_text.__globals__['FlagsCls'] = FlagsCls # also as generic alias
from typing import Dict as _Dict # noqa: WPS433
_text.__globals__.setdefault('Dict', _Dict)
_text.__name__ = cmd_name
_text.__doc__ = description
text_cmd = commands.command(name=cmd_name, aliases=aliases, help=description)(_text)
# Override usage string to exclude internal ctx parameter
text_cmd.usage = f"{cmd_name} <target> <flags> <reason>"
_dummy_action() -> None
async
¶
Dummy coroutine for moderation actions that only create a case without performing Discord API actions. Used by commands like warn, pollban, snippetban etc. that only need case creation.
execute_mod_action(ctx: commands.Context[Tux], case_type: CaseType, user: discord.Member | discord.User, reason: str, silent: bool, dm_action: str, actions: Sequence[tuple[Any, type[R]]] = (), duration: str | None = None, expires_at: datetime | None = None) -> None
async
¶
Execute a moderation action with case creation, DM sending, and additional actions.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
ctx | Context[Tux] | The context of the command. | required |
case_type | CaseType | The type of case to create. | required |
user | Union[Member, User] | The target user of the moderation action. | required |
reason | str | The reason for the moderation action. | required |
silent | bool | Whether to send a DM to the user. | required |
dm_action | str | The action description for the DM. | required |
actions | Sequence[tuple[Any, type[R]]] | Additional actions to execute and their expected return types. | () |
duration | Optional[str] | The duration of the action, if applicable (for display/logging). | None |
expires_at | Optional[datetime] | The specific expiration time, if applicable. | None |
Source code in tux/cogs/moderation/command_meta.py
duration: str | None = None,
purge: int = 0,
silent: bool = False,
reason: str = "",
) -> None: # type: ignore[arg-type]
"""App command callback (no bound self) using explicit options."""
from types import SimpleNamespace # noqa: WPS433
flags_obj = SimpleNamespace(duration=duration, purge=purge, silent=silent)
bot = interaction.client # type: ignore[attr-defined]
cog: ModerationCogBase | None = bot.get_cog("ModerationCommandsCog") # type: ignore[attr-defined]
if cog is None:
return
ctx = await cog.bot.get_context(interaction) # type: ignore[attr-defined]
await _run(cog, ctx, target, flags_obj, reason)
if FlagsCls is not None:
_slash.__globals__[FlagsCls.__name__] = FlagsCls
_slash.__globals__['FlagsCls'] = FlagsCls
slash_cmd = discord.app_commands.command(name=cmd_name, description=description)(_slash)
# store on cls
cls.text_command = text_cmd # type: ignore[attr-defined]
cls.slash_command = slash_cmd # type: ignore[attr-defined]
# register class
_REGISTRY.append(cls)
return cls
class ModerationCommand(metaclass=ModerationCommandMeta):
"""Base class to inherit for each moderation action."""
name: ClassVar[str]
aliases: ClassVar[list[str]] = []
case_type: ClassVar[Any]
required_pl: ClassVar[int] = 0
flags: ClassVar[Dict[str, Dict[str, Any]]] = {}
description: ClassVar[str] = ""
# Child classes must implement _action
async def _action(self, guild: discord.Guild, member: discord.Member | discord.User, *, flags: Any, reason: str) -> None: # noqa: D401
raise NotImplementedError
# Cog that loads all ModerationCommand subclasses
class ModerationCommandsCog(ModerationCogBase):
def __init__(self, bot: commands.Bot):
super().__init__(bot) # type: ignore[arg-type]
for cls in _REGISTRY:
self.bot.add_command(cls.text_command) # type: ignore[attr-defined]
self.bot.tree.add_command(cls.slash_command) # type: ignore[attr-defined]
async def setup(bot: commands.Bot):
# Ensure all command modules are imported so subclasses register
import importlib
importlib.import_module("tux.cogs.moderation.commands")
await bot.add_cog(ModerationCommandsCog(bot))
_handle_dm_result(user: discord.Member | discord.User, dm_result: Any) -> bool
¶
send_error_response(ctx: commands.Context[Tux], error_message: str, error_detail: Exception | None = None, ephemeral: bool = True) -> None
async
¶
Send a standardized error response.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
ctx | Context[Tux] | The context of the command. | required |
error_message | str | The error message to display. | required |
error_detail | Optional[Exception] | The exception details, if available. | None |
ephemeral | bool | Whether the message should be ephemeral. | True |
create_embed(ctx: commands.Context[Tux], title: str, fields: list[tuple[str, str, bool]], color: int, icon_url: str, timestamp: datetime | None = None, thumbnail_url: str | None = None) -> discord.Embed
¶
Create an embed for moderation actions.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
ctx | Context[Tux] | The context of the command. | required |
title | str | The title of the embed. | required |
fields | list[tuple[str, str, bool]] | The fields to add to the embed. | required |
color | int | The color of the embed. | required |
icon_url | str | The icon URL for the embed. | required |
timestamp | Optional[datetime] | The timestamp for the embed. | None |
thumbnail_url | Optional[str] | The thumbnail URL for the embed. | None |
Returns:
Type | Description |
---|---|
Embed | The embed for the moderation action. |
send_embed(ctx: commands.Context[Tux], embed: discord.Embed, log_type: str) -> None
async
¶
send_dm(ctx: commands.Context[Tux], silent: bool, user: discord.Member | discord.User, reason: str, action: str) -> bool
async
¶
Send a DM to the target user.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
ctx | Context[Tux] | The context of the command. | required |
silent | bool | Whether the command is silent. | required |
user | Union[Member, User] | The target of the moderation action. | required |
reason | str | The reason for the moderation action. | required |
action | str | The action being performed. | required |
Returns:
Type | Description |
---|---|
bool | Whether the DM was successfully sent. |
check_conditions(ctx: commands.Context[Tux], user: discord.Member | discord.User, moderator: discord.Member | discord.User, action: str) -> bool
async
¶
Check if the conditions for the moderation action are met.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
ctx | Context[Tux] | The context of the command. | required |
user | Union[Member, User] | The target of the moderation action. | required |
moderator | Union[Member, User] | The moderator of the moderation action. | required |
action | str | The action being performed. | required |
Returns:
Type | Description |
---|---|
bool | Whether the conditions are met. |
handle_case_response(ctx: commands.Context[Tux], case_type: CaseType, case_number: int | None, reason: str, user: discord.Member | discord.User, dm_sent: bool, duration: str | None = None) -> None
async
¶
Handle the response for a case.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
ctx | Context[Tux] | The context of the command. | required |
case_type | CaseType | The type of case. | required |
case_number | Optional[int] | The case number. | required |
reason | str | The reason for the case. | required |
user | Union[Member, User] | The target of the case. | required |
dm_sent | bool | Whether the DM was sent. | required |
duration | Optional[str] | The duration of the case. | None |
_format_case_title(case_type: CaseType, case_number: int | None, duration: str | None) -> str
¶
is_pollbanned(guild_id: int, user_id: int) -> bool
async
¶
is_snippetbanned(guild_id: int, user_id: int) -> bool
async
¶
is_jailed(guild_id: int, user_id: int) -> bool
async
¶
execute_mixed_mod_action(ctx: commands.Context[Tux], config: ModerationCommandConfig, user: discord.Member | discord.User, mixed_args: str) -> None
async
¶
Parse mixed_args according to config and execute the moderation flow.
This serves as the single entry-point for all dynamically generated moderation commands. It handles: 1. Mixed-argument parsing (positional + flags). 2. Validation based on config (duration required?, purge range?, etc.). 3. Permission / sanity checks via check_conditions. 4. Building the actions list and delegating to :py:meth:execute_mod_action
.
_validate_args(config: ModerationCommandConfig, parsed: dict[str, Any]) -> tuple[bool, dict[str, Any]]
¶
Validate parsed arguments against config rules.
Returns (is_valid, validated_dict). On failure sends the error message via the ctx stored in validated_dict["ctx"] and returns False.
execute_flag_mod_action(ctx: commands.Context[Tux], config: ModerationCommandConfig, user: discord.Member | discord.User, flags: Any, reason: str) -> None
async
¶
Execute moderation flow based on flags parsed by FlagConverter.
This is the preferred pathway for dynamically generated moderation commands that rely on discord.py's native FlagConverter parsing.