tux.cogs.moderation.tempban
¶
Classes:
Name | Description |
---|---|
TempBan | |
Classes¶
TempBan(bot: Tux)
¶
Bases: ModerationCogBase
Methods:
Name | Description |
---|---|
tempban | Temporarily ban a member from the server. |
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. |
tempban_check | Check for expired tempbans at a set interval and unban the user if the ban has expired. |
before_tempban_check | Wait for the bot to be ready before starting the loop. |
cog_unload | Cancel the tempban check loop when the cog is unloaded. |
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. |
Source code in tux/cogs/moderation/tempban.py
Functions¶
tempban(ctx: commands.Context[Tux], member: discord.Member, *, flags: TempBanFlags) -> None
async
¶
Temporarily ban a member from 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 ban. | required |
flags | TempBanFlags | The flags for the command. (duration: float (via converter), purge: int (< 7), silent: bool) | required |
Raises:
Type | Description |
---|---|
Forbidden | If the bot is unable to ban the user. |
HTTPException | If an error occurs while banning the user. |
Source code in tux/cogs/moderation/tempban.py
@commands.hybrid_command(name="tempban", aliases=["tb"])
@commands.guild_only()
@checks.has_pl(3)
async def tempban(
self,
ctx: commands.Context[Tux],
member: discord.Member,
*,
flags: TempBanFlags,
) -> None:
"""
Temporarily ban a member from the server.
Parameters
----------
ctx : commands.Context[Tux]
The context in which the command is being invoked.
member : discord.Member
The member to ban.
flags : TempBanFlags
The flags for the command. (duration: float (via converter), purge: int (< 7), silent: bool)
Raises
------
discord.Forbidden
If the bot is unable to ban the user.
discord.HTTPException
If an error occurs while banning the user.
"""
assert ctx.guild
# Check if moderator has permission to temp ban the member
if not await self.check_conditions(ctx, member, ctx.author, "temp ban"):
return
# Calculate expiration datetime from duration in seconds
expires_at = datetime.now(UTC) + timedelta(seconds=flags.duration)
# Create a simple duration string for logging/display
# TODO: Implement a more robust human-readable duration formatter
duration_display_str = str(timedelta(seconds=int(flags.duration))) # Simple representation
# Execute tempban with case creation and DM
await self.execute_mod_action(
ctx=ctx,
case_type=CaseType.TEMPBAN,
user=member,
reason=flags.reason,
silent=flags.silent,
dm_action="temp banned",
actions=[
(ctx.guild.ban(member, reason=flags.reason, delete_message_seconds=flags.purge * 86400), type(None)),
],
duration=duration_display_str, # Pass readable string for logging
expires_at=expires_at, # Pass calculated expiration datetime
)
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/tempban.py
Parameters
----------
ctx : commands.Context[Tux]
The context in which the command is being invoked.
member : discord.Member
The member to ban.
flags : TempBanFlags
The flags for the command. (duration: float (via converter), purge: int (< 7), silent: bool)
Raises
------
discord.Forbidden
If the bot is unable to ban the user.
discord.HTTPException
If an error occurs while banning the user.
"""
assert ctx.guild
# Check if moderator has permission to temp ban the member
if not await self.check_conditions(ctx, member, ctx.author, "temp ban"):
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/tempban.py
# Calculate expiration datetime from duration in seconds
expires_at = datetime.now(UTC) + timedelta(seconds=flags.duration)
# Create a simple duration string for logging/display
# TODO: Implement a more robust human-readable duration formatter
duration_display_str = str(timedelta(seconds=int(flags.duration))) # Simple representation
# Execute tempban with case creation and DM
await self.execute_mod_action(
ctx=ctx,
case_type=CaseType.TEMPBAN,
user=member,
reason=flags.reason,
silent=flags.silent,
dm_action="temp banned",
actions=[
(ctx.guild.ban(member, reason=flags.reason, delete_message_seconds=flags.purge * 86400), 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/tempban.py
duration=duration_display_str, # Pass readable string for logging
expires_at=expires_at, # Pass calculated expiration datetime
)
async def _process_tempban_case(self, case: Case) -> tuple[int, int]:
"""Process an individual tempban case. Returns (processed_cases, failed_cases)."""
# Check for essential data first
if not (case.guild_id and case.case_user_id and case.case_id):
logger.error(f"Invalid case data: {case}")
return 0, 0
guild = self.bot.get_guild(case.guild_id)
if not guild:
logger.warning(f"Guild {case.guild_id} not found for case {case.case_id}")
return 0, 0
# Check ban status
try:
await guild.fetch_ban(discord.Object(id=case.case_user_id))
# If fetch_ban succeeds without error, the user IS banned.
except discord.NotFound:
# User is not banned. Mark expired and consider processed.
await self.db.case.set_tempban_expired(case.case_id, case.guild_id)
return 1, 0
except Exception as e:
# Log error during ban check, but proceed to attempt unban anyway
# This matches the original logic's behavior.
logger.warning(f"Error checking ban status for {case.case_user_id} in {guild.id}: {e}")
_process_tempban_case(case: Case) -> tuple[int, int]
async
¶
Process an individual tempban case. Returns (processed_cases, failed_cases).
Source code in tux/cogs/moderation/tempban.py
async def _process_tempban_case(self, case: Case) -> tuple[int, int]:
"""Process an individual tempban case. Returns (processed_cases, failed_cases)."""
# Check for essential data first
if not (case.guild_id and case.case_user_id and case.case_id):
logger.error(f"Invalid case data: {case}")
return 0, 0
guild = self.bot.get_guild(case.guild_id)
if not guild:
logger.warning(f"Guild {case.guild_id} not found for case {case.case_id}")
return 0, 0
# Check ban status
try:
await guild.fetch_ban(discord.Object(id=case.case_user_id))
# If fetch_ban succeeds without error, the user IS banned.
except discord.NotFound:
# User is not banned. Mark expired and consider processed.
await self.db.case.set_tempban_expired(case.case_id, case.guild_id)
return 1, 0
except Exception as e:
# Log error during ban check, but proceed to attempt unban anyway
# This matches the original logic's behavior.
logger.warning(f"Error checking ban status for {case.case_user_id} in {guild.id}: {e}")
# Attempt to unban (runs if user was found banned or if ban check failed)
processed_count, failed_count = 0, 0
try:
# Perform the unban
await guild.unban(
discord.Object(id=case.case_user_id),
reason="Temporary ban expired.",
)
except (discord.Forbidden, discord.HTTPException) as e:
# Discord API unban failed
logger.error(f"Failed to unban {case.case_user_id} in {guild.id}: {e}")
failed_count = 1
except Exception as e:
# Catch other potential errors during unban
logger.error(
f"Unexpected error during unban attempt for tempban {case.case_id} (user {case.case_user_id}, guild {guild.id}): {e}",
)
failed_count = 1
else:
# Unban successful, now update the database
try:
update_result = await self.db.case.set_tempban_expired(case.case_id, case.guild_id)
if update_result == 1:
logger.info(
f"Successfully unbanned user {case.case_user_id} and marked case {case.case_id} as expired in guild {guild.id}.",
)
processed_count = 1
elif update_result is None:
logger.info(
f"Successfully unbanned user {case.case_user_id} in guild {guild.id} (case {case.case_id} was already marked expired).",
)
processed_count = 1 # Still count as success
else:
logger.error(
f"Unexpected update result ({update_result}) when marking case {case.case_id} as expired for user {case.case_user_id} in guild {guild.id}.",
)
failed_count = 1
except Exception as e:
# Catch errors during DB update
logger.error(
f"Unexpected error during DB update for tempban {case.case_id} (user {case.case_user_id}, guild {guild.id}): {e}",
)
failed_count = 1
return processed_count, failed_count
_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/tempban.py
except (discord.Forbidden, discord.HTTPException) as e:
# Discord API unban failed
logger.error(f"Failed to unban {case.case_user_id} in {guild.id}: {e}")
failed_count = 1
except Exception as e:
# Catch other potential errors during unban
logger.error(
f"Unexpected error during unban attempt for tempban {case.case_id} (user {case.case_user_id}, guild {guild.id}): {e}",
)
failed_count = 1
else:
# Unban successful, now update the database
try:
update_result = await self.db.case.set_tempban_expired(case.case_id, case.guild_id)
if update_result == 1:
logger.info(
f"Successfully unbanned user {case.case_user_id} and marked case {case.case_id} as expired in guild {guild.id}.",
)
processed_count = 1
elif update_result is None:
logger.info(
f"Successfully unbanned user {case.case_user_id} in guild {guild.id} (case {case.case_id} was already marked expired).",
)
processed_count = 1 # Still count as success
else:
logger.error(
f"Unexpected update result ({update_result}) when marking case {case.case_id} as expired for user {case.case_user_id} in guild {guild.id}.",
)
failed_count = 1
except Exception as e:
# Catch errors during DB update
logger.error(
f"Unexpected error during DB update for tempban {case.case_id} (user {case.case_user_id}, guild {guild.id}): {e}",
)
failed_count = 1
return processed_count, failed_count
@tasks.loop(minutes=1)
async def tempban_check(self) -> None:
"""
Check for expired tempbans at a set interval and unban the user if the ban has expired.
Uses a simple locking mechanism to prevent overlapping executions.
Processes bans in smaller batches to prevent timeout issues.
Raises
------
Exception
If an error occurs while checking for expired tempbans.
"""
# Skip if already processing
if self._processing_tempbans:
return
try:
self._processing_tempbans = True
# Get expired tempbans
expired_cases = await self.db.case.get_expired_tempbans()
processed_cases = 0
failed_cases = 0
for case in expired_cases:
# Process each case using the helper method
processed, failed = await self._process_tempban_case(case)
processed_cases += processed
failed_cases += failed
if processed_cases > 0 or failed_cases > 0:
logger.info(f"Tempban check: processed {processed_cases} cases, {failed_cases} failures")
except Exception as e:
logger.error(f"Failed to check tempbans: {e}")
finally:
self._processing_tempbans = False
@tempban_check.before_loop
async def before_tempban_check(self) -> None:
"""Wait for the bot to be ready before starting the loop."""
await self.bot.wait_until_ready()
async def cog_unload(self) -> None:
"""Cancel the tempban check loop when the cog is unloaded."""
self.tempban_check.cancel()
async def setup(bot: Tux) -> None:
await bot.add_cog(TempBan(bot))
tempban_check() -> None
async
¶
Check for expired tempbans at a set interval and unban the user if the ban has expired.
Uses a simple locking mechanism to prevent overlapping executions. Processes bans in smaller batches to prevent timeout issues.
Raises:
Type | Description |
---|---|
Exception | If an error occurs while checking for expired tempbans. |
Source code in tux/cogs/moderation/tempban.py
@tasks.loop(minutes=1)
async def tempban_check(self) -> None:
"""
Check for expired tempbans at a set interval and unban the user if the ban has expired.
Uses a simple locking mechanism to prevent overlapping executions.
Processes bans in smaller batches to prevent timeout issues.
Raises
------
Exception
If an error occurs while checking for expired tempbans.
"""
# Skip if already processing
if self._processing_tempbans:
return
try:
self._processing_tempbans = True
# Get expired tempbans
expired_cases = await self.db.case.get_expired_tempbans()
processed_cases = 0
failed_cases = 0
for case in expired_cases:
# Process each case using the helper method
processed, failed = await self._process_tempban_case(case)
processed_cases += processed
failed_cases += failed
if processed_cases > 0 or failed_cases > 0:
logger.info(f"Tempban check: processed {processed_cases} cases, {failed_cases} failures")
except Exception as e:
logger.error(f"Failed to check tempbans: {e}")
finally:
self._processing_tempbans = False
before_tempban_check() -> None
async
¶
cog_unload() -> None
async
¶
_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 |