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
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
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
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
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
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
# 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
@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
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.
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
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
@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
¶
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 |