Skip to content

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
Python
def __init__(self, bot: Tux) -> None:
    super().__init__(bot)
    self.tempban.usage = generate_usage(self.tempban, TempBanFlags)
    self._processing_tempbans = False  # Lock to prevent overlapping task runs
    self.tempban_check.start()

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

Source code in tux/cogs/moderation/tempban.py
Python
processed_count, failed_count = 0, 0
try:
    # Perform the unban
    await guild.unban(
        discord.Object(id=case.case_user_id),
        reason="Temporary ban expired.",
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
Python
        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
Python
@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

Wait for the bot to be ready before starting the loop.

Source code in tux/cogs/moderation/tempban.py
Python
@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()
cog_unload() -> None async

Cancel the tempban check loop when the cog is unloaded.

Source code in tux/cogs/moderation/tempban.py
Python
async def cog_unload(self) -> None:
    """Cancel the tempban check loop when the cog is unloaded."""
    self.tempban_check.cancel()
_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.

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

Check if a user is jailed using the optimized latest case method.

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.

Functions