Skip to content

tux.cogs.moderation.jail

Classes:

Name Description
Jail

Classes

Jail(bot: Tux)

Bases: ModerationCogBase

Methods:

Name Description
get_jail_role

Get the jail role for the guild.

get_jail_channel

Get the jail channel for the guild.

get_user_lock

Get or create a lock for operations on a specific user.

is_jailed

Check if a user is jailed.

clean_user_locks

Remove locks for users that are not currently in use.

jail

Jail a member in the server.

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.

Source code in tux/cogs/moderation/jail.py
Python
def __init__(self, bot: Tux) -> None:
    super().__init__(bot)
    self.jail.usage = generate_usage(self.jail, JailFlags)

Functions

get_jail_role(guild: discord.Guild) -> discord.Role | None async

Get the jail role for the guild.

Parameters:

Name Type Description Default
guild Guild

The guild to get the jail role for.

required

Returns:

Type Description
Role | None

The jail role, or None if not found.

Source code in tux/cogs/moderation/jail.py
Python
async def get_jail_role(self, guild: discord.Guild) -> discord.Role | None:
    """
    Get the jail role for the guild.

    Parameters
    ----------
    guild : discord.Guild
        The guild to get the jail role for.

    Returns
    -------
    discord.Role | None
        The jail role, or None if not found.
    """
    jail_role_id = await self.db.guild_config.get_jail_role_id(guild.id)
    return None if jail_role_id is None else guild.get_role(jail_role_id)
get_jail_channel(guild: discord.Guild) -> discord.TextChannel | None async

Get the jail channel for the guild.

Source code in tux/cogs/moderation/jail.py
Python
async def get_jail_channel(self, guild: discord.Guild) -> discord.TextChannel | None:
    """
    Get the jail channel for the guild.
    """
    jail_channel_id = await self.db.guild_config.get_jail_channel_id(guild.id)
    channel = guild.get_channel(jail_channel_id) if jail_channel_id is not None else None
    return channel if isinstance(channel, discord.TextChannel) else None
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/jail.py
Python
async def get_jail_channel(self, guild: discord.Guild) -> discord.TextChannel | None:
    """
    Get the jail channel for the guild.
    """
    jail_channel_id = await self.db.guild_config.get_jail_channel_id(guild.id)
    channel = guild.get_channel(jail_channel_id) if jail_channel_id is not None else None
    return channel if isinstance(channel, discord.TextChannel) else None

async def is_jailed(self, guild_id: int, user_id: int) -> bool:
    """
    Check if a user is jailed.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to check in.
    user_id : int
        The ID of the user to check.

    Returns
    -------
    bool
is_jailed(guild_id: int, user_id: int) -> bool async

Check if a user is jailed.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to check in.

required
user_id int

The ID of the user to check.

required

Returns:

Type Description
bool

True if the user is jailed, False otherwise.

Source code in tux/cogs/moderation/jail.py
Python
async def is_jailed(self, guild_id: int, user_id: int) -> bool:
    """
    Check if a user is jailed.

    Parameters
    ----------
    guild_id : int
        The ID of the guild to check in.
    user_id : int
        The ID of the user to check.

    Returns
    -------
    bool
        True if the user is jailed, False otherwise.
    """
    # Get latest case for this user (more efficient than counting all cases)
    latest_case = await self.db.case.get_latest_case_by_user(
        guild_id=guild_id,
        user_id=user_id,
        case_types=[CaseType.JAIL, CaseType.UNJAIL],
    )

    # If no cases exist or latest case is an unjail, user is not jailed
    return bool(latest_case and latest_case.case_type == CaseType.JAIL)
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/jail.py
Python
    # Get latest case for this user (more efficient than counting all cases)
    latest_case = await self.db.case.get_latest_case_by_user(
        guild_id=guild_id,
        user_id=user_id,
        case_types=[CaseType.JAIL, CaseType.UNJAIL],
    )

    # If no cases exist or latest case is an unjail, user is not jailed
    return bool(latest_case and latest_case.case_type == CaseType.JAIL)

@commands.hybrid_command(
    name="jail",
    aliases=["j"],
)
@commands.guild_only()
@checks.has_pl(2)
async def jail(
jail(ctx: commands.Context[Tux], member: discord.Member, *, flags: JailFlags) -> None async

Jail a member in the server.

Parameters:

Name Type Description Default
ctx Context[Tux]

The context in which the command is being invoked.

required
member Member

The member to jail.

required
flags JailFlags

The flags for the command. (reason: str, silent: bool)

required

Raises:

Type Description
Forbidden

If the bot is unable to jail the user.

HTTPException

If an error occurs while jailing the user.

Source code in tux/cogs/moderation/jail.py
Python
@commands.hybrid_command(
    name="jail",
    aliases=["j"],
)
@commands.guild_only()
@checks.has_pl(2)
async def jail(
    self,
    ctx: commands.Context[Tux],
    member: discord.Member,
    *,
    flags: JailFlags,
) -> None:
    """
    Jail a member in the server.

    Parameters
    ----------
    ctx : commands.Context[Tux]
        The context in which the command is being invoked.
    member : discord.Member
        The member to jail.
    flags : JailFlags
        The flags for the command. (reason: str, silent: bool)

    Raises
    ------
    discord.Forbidden
        If the bot is unable to jail the user.
    discord.HTTPException
        If an error occurs while jailing the user.
    """

    assert ctx.guild

    await ctx.defer(ephemeral=True)

    # Get jail role
    jail_role = await self.get_jail_role(ctx.guild)
    if not jail_role:
        await ctx.send("No jail role found.", ephemeral=True)
        return

    # Get jail channel
    jail_channel = await self.get_jail_channel(ctx.guild)
    if not jail_channel:
        await ctx.send("No jail channel found.", ephemeral=True)
        return

    # Check if user is already jailed
    if await self.is_jailed(ctx.guild.id, member.id):
        await ctx.send("User is already jailed.", ephemeral=True)
        return

    # Check if moderator has permission to jail the member
    if not await self.check_conditions(ctx, member, ctx.author, "jail"):
        return

    # Use a transaction-like pattern to ensure consistency
    try:
        # Get roles that can be managed by the bot
        user_roles = self._get_manageable_roles(member, jail_role)

        # Convert roles to IDs
        case_user_roles = [role.id for role in user_roles]

        # First create the case - if this fails, no role changes are made
        case = await self.db.case.insert_case(
            guild_id=ctx.guild.id,
            case_user_id=member.id,
            case_moderator_id=ctx.author.id,
            case_type=CaseType.JAIL,
            case_reason=flags.reason,
            case_user_roles=case_user_roles,
        )

        # Add jail role immediately - this is the most important part
        await member.add_roles(jail_role, reason=flags.reason)

        # Send DM to member
        dm_sent = await self.send_dm(ctx, flags.silent, member, flags.reason, "jailed")

        # Handle case response - send embed immediately
        await self.handle_case_response(ctx, CaseType.JAIL, case.case_number, flags.reason, member, dm_sent)

        # Remove old roles in the background after sending the response
        if user_roles:
            try:
                # Try to remove all at once for efficiency
                await member.remove_roles(*user_roles, reason=flags.reason, atomic=False)
            except Exception as e:
                logger.warning(
                    f"Failed to remove all roles at once from {member}, falling back to individual removal: {e}",
                )
                # Fall back to removing one by one
                for role in user_roles:
                    try:
                        await member.remove_roles(role, reason=flags.reason)
                    except Exception as role_e:
                        logger.error(f"Failed to remove role {role} from {member}: {role_e}")
                        # Continue with other roles even if one fails

    except Exception as e:
        logger.error(f"Failed to jail {member}: {e}")
        await ctx.send(f"Failed to jail {member}: {e}", ephemeral=True)
        return
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/jail.py
Python
    ctx: commands.Context[Tux],
    member: discord.Member,
    *,
    flags: JailFlags,
) -> None:
    """
    Jail a member in the server.

    Parameters
    ----------
    ctx : commands.Context[Tux]
        The context in which the command is being invoked.
    member : discord.Member
        The member to jail.
    flags : JailFlags
        The flags for the command. (reason: str, silent: bool)

    Raises
    ------
    discord.Forbidden
        If the bot is unable to jail the user.
    discord.HTTPException
        If an error occurs while jailing the user.
    """

    assert ctx.guild

    await ctx.defer(ephemeral=True)

    # Get jail role
_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.

Source code in tux/cogs/moderation/jail.py
Python
if not jail_role:
    await ctx.send("No jail role found.", ephemeral=True)
    return

# Get jail channel
jail_channel = await self.get_jail_channel(ctx.guild)
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/jail.py
Python
            await ctx.send("No jail channel found.", ephemeral=True)
            return

        # Check if user is already jailed
        if await self.is_jailed(ctx.guild.id, member.id):
            await ctx.send("User is already jailed.", ephemeral=True)
            return

        # Check if moderator has permission to jail the member
        if not await self.check_conditions(ctx, member, ctx.author, "jail"):
            return

        # Use a transaction-like pattern to ensure consistency
        try:
            # Get roles that can be managed by the bot
            user_roles = self._get_manageable_roles(member, jail_role)

            # Convert roles to IDs
            case_user_roles = [role.id for role in user_roles]

            # First create the case - if this fails, no role changes are made
            case = await self.db.case.insert_case(
                guild_id=ctx.guild.id,
                case_user_id=member.id,
                case_moderator_id=ctx.author.id,
                case_type=CaseType.JAIL,
                case_reason=flags.reason,
                case_user_roles=case_user_roles,
            )

            # Add jail role immediately - this is the most important part
            await member.add_roles(jail_role, reason=flags.reason)

            # Send DM to member
            dm_sent = await self.send_dm(ctx, flags.silent, member, flags.reason, "jailed")

            # Handle case response - send embed immediately
            await self.handle_case_response(ctx, CaseType.JAIL, case.case_number, flags.reason, member, dm_sent)

            # Remove old roles in the background after sending the response
            if user_roles:
                try:
                    # Try to remove all at once for efficiency
                    await member.remove_roles(*user_roles, reason=flags.reason, atomic=False)
                except Exception as e:
                    logger.warning(
                        f"Failed to remove all roles at once from {member}, falling back to individual removal: {e}",
                    )
                    # Fall back to removing one by one
                    for role in user_roles:
                        try:
                            await member.remove_roles(role, reason=flags.reason)
                        except Exception as role_e:
                            logger.error(f"Failed to remove role {role} from {member}: {role_e}")
                            # Continue with other roles even if one fails

        except Exception as e:
            logger.error(f"Failed to jail {member}: {e}")
            await ctx.send(f"Failed to jail {member}: {e}", ephemeral=True)
            return

    @staticmethod
    def _get_manageable_roles(
        member: discord.Member,
        jail_role: discord.Role,
    ) -> list[discord.Role]:
        """
        Get the roles that can be managed by the bot.

        Parameters
        ----------
        member : discord.Member
            The member to jail.
        jail_role : discord.Role
            The jail role.

        Returns
        -------
        list[discord.Role]
            A list of roles that can be managed by the bot.
        """

        return [
            role
            for role in member.roles
            if not (
                role.is_bot_managed()
                or role.is_premium_subscriber()
                or role.is_integration()
                or role.is_default()
                or role == jail_role
            )
            and role.is_assignable()
        ]


async def setup(bot: Tux) -> None:
    await bot.add_cog(Jail(bot))
_get_manageable_roles(member: discord.Member, jail_role: discord.Role) -> list[discord.Role] staticmethod

Get the roles that can be managed by the bot.

Parameters:

Name Type Description Default
member Member

The member to jail.

required
jail_role Role

The jail role.

required

Returns:

Type Description
list[Role]

A list of roles that can be managed by the bot.

Source code in tux/cogs/moderation/jail.py
Python
@staticmethod
def _get_manageable_roles(
    member: discord.Member,
    jail_role: discord.Role,
) -> list[discord.Role]:
    """
    Get the roles that can be managed by the bot.

    Parameters
    ----------
    member : discord.Member
        The member to jail.
    jail_role : discord.Role
        The jail role.

    Returns
    -------
    list[discord.Role]
        A list of roles that can be managed by the bot.
    """

    return [
        role
        for role in member.roles
        if not (
            role.is_bot_managed()
            or role.is_premium_subscriber()
            or role.is_integration()
            or role.is_default()
            or role == jail_role
        )
        and role.is_assignable()
    ]
_handle_dm_result(user: discord.Member | discord.User, dm_result: Any) -> bool

Handle the result of sending a DM.

Parameters:

Name Type Description Default
user Union[Member, User]

The user the DM was sent to.

required
dm_result Any

The result of the DM sending operation.

required

Returns:

Type Description
bool

Whether the DM was successfully sent.

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 an embed to the log channel.

Parameters:

Name Type Description Default
ctx Context[Tux]

The context of the command.

required
embed Embed

The embed to send.

required
log_type str

The type of log to send the embed to.

required
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

Format a case title.

Parameters:

Name Type Description Default
case_type CaseType

The type of case.

required
case_number Optional[int]

The case number.

required
duration Optional[str]

The duration of the case.

required

Returns:

Type Description
str

The formatted case title.

is_pollbanned(guild_id: int, user_id: int) -> bool async

Check if a user is poll banned.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to check in.

required
user_id int

The ID of the user to check.

required

Returns:

Type Description
bool

True if the user is poll banned, False otherwise.

is_snippetbanned(guild_id: int, user_id: int) -> bool async

Check if a user is snippet banned.

Parameters:

Name Type Description Default
guild_id int

The ID of the guild to check in.

required
user_id int

The ID of the user to check.

required

Returns:

Type Description
bool

True if the user is snippet banned, False otherwise.

Functions