From 7be83811b58581d0be6d51b741681262f3ead258 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Thu, 21 Aug 2025 14:57:24 +0100 Subject: [PATCH 01/13] feat(command): add ticket system --- prisma/schema/commands/moderation.prisma | 25 + prisma/schema/guild/guild.prisma | 1 + tux/cogs/moderation/tickets.py | 1038 ++++++++++++++++++++++ tux/cogs/moderation/timeout.py | 2 +- tux/cogs/utility/ticket_log.py | 189 ++++ tux/cogs/utility/ticket_log_config.py | 51 ++ tux/database/controllers/__init__.py | 5 + tux/database/controllers/base.py | 3 + tux/database/controllers/ticket.py | 244 +++++ tux/ui/views/tickets.py | 300 +++++++ 10 files changed, 1857 insertions(+), 1 deletion(-) create mode 100644 tux/cogs/moderation/tickets.py create mode 100644 tux/cogs/utility/ticket_log.py create mode 100644 tux/cogs/utility/ticket_log_config.py create mode 100644 tux/database/controllers/ticket.py create mode 100644 tux/ui/views/tickets.py diff --git a/prisma/schema/commands/moderation.prisma b/prisma/schema/commands/moderation.prisma index 251f7f440..cd3c820e5 100644 --- a/prisma/schema/commands/moderation.prisma +++ b/prisma/schema/commands/moderation.prisma @@ -58,3 +58,28 @@ enum CaseType { POLLBAN POLLUNBAN } + +model Ticket { + ticket_id Int @id @default(autoincrement()) + guild_id BigInt + channel_id BigInt @unique + author_id BigInt + claimed_by BigInt? + title String + status TicketStatus @default(OPEN) + created_at DateTime @default(now()) + closed_at DateTime? + + guild Guild @relation(fields: [guild_id], references: [guild_id], onDelete: Cascade) + + @@index([guild_id]) + @@index([author_id]) + @@index([claimed_by]) + @@index([status]) +} + +enum TicketStatus { + OPEN + CLAIMED + CLOSED +} diff --git a/prisma/schema/guild/guild.prisma b/prisma/schema/guild/guild.prisma index e22408795..a16b34b82 100644 --- a/prisma/schema/guild/guild.prisma +++ b/prisma/schema/guild/guild.prisma @@ -11,6 +11,7 @@ model Guild { StarboardMessage StarboardMessage[] case_count BigInt @default(0) levels Levels[] + tickets Ticket[] @@index([guild_id]) } diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py new file mode 100644 index 000000000..452a3db6a --- /dev/null +++ b/tux/cogs/moderation/tickets.py @@ -0,0 +1,1038 @@ +import asyncio +import re +import traceback +from collections import defaultdict +from contextlib import suppress + +import discord +from discord import app_commands +from discord.ext import commands +from loguru import logger + +from prisma.enums import TicketStatus +from tux.bot import Tux +from tux.database.controllers import DatabaseController, guild_config +from tux.ui.embeds import EmbedCreator, EmbedType +from tux.ui.views.tickets import RequestCloseView, TicketManagementView +from tux.utils import checks + + +class TicketCreationError(Exception): + """Raised when a ticket cannot be created or initialized properly.""" + + +class Tickets(commands.Cog): + # Central small helpers to unify patterns + @staticmethod + def _ticket_error(message: str) -> TicketCreationError: + return TicketCreationError(message) + + def _raise_creation_error(self, message: str) -> None: + """Raise ticket creation error with proper message handling.""" + msg = message + raise TicketCreationError(msg) + + def _raise_persist_error(self, message: str) -> None: + """Raise ticket persistence error with proper message handling.""" + msg = message + raise TicketCreationError(msg) + + async def show_transcript(self, interaction: discord.Interaction, ticket_id: int): + """Show the transcript for a ticket as a .txt file, even if the channel is deleted.""" + await interaction.response.defer(ephemeral=True) + ticket = await self.db.ticket.get_ticket_by_id(ticket_id) + if not ticket or ticket.guild_id != interaction.guild.id: + await interaction.followup.send("Ticket not found.", ephemeral=True) + return + + ticket_log_cog = self.bot.get_cog("TicketLog") + if not ticket_log_cog: + await interaction.followup.send("Transcript system not available.", ephemeral=True) + return + + channel = interaction.guild.get_channel(ticket.channel_id) + if channel: + try: + messages = [msg async for msg in channel.history(limit=100, oldest_first=True)] + success = await ticket_log_cog.log_transcript(interaction.guild, channel, ticket, messages) + if success: + transcript_file = await ticket_log_cog.get_transcript_from_log( + interaction.guild.id, + ticket.channel_id, + ) + if transcript_file: + await interaction.followup.send( + content="Here is the ticket transcript.", + file=transcript_file, + ephemeral=True, + ) + return + await interaction.followup.send("Failed to generate transcript.", ephemeral=True) + except Exception as e: + logger.error(f"Error creating live transcript: {e}") + await interaction.followup.send("Error generating transcript.", ephemeral=True) + else: + transcript_file = await ticket_log_cog.get_transcript_from_log( + interaction.guild.id, + ticket.channel_id, + ) + if transcript_file: + await interaction.followup.send( + content="Here is the ticket transcript from logs.", + file=transcript_file, + ephemeral=True, + ) + return + await interaction.followup.send("Transcript not found.", ephemeral=True) + + def __init__(self, bot: Tux) -> None: + self.bot = bot + self.db = DatabaseController() + self.guild_config_controller = guild_config.GuildConfigController() + self.cleanup_tasks: dict[int, asyncio.Task] = {} # Store cleanup tasks by channel_id + + async def close_ticket(self, ctx, ticket, channel, author, respond_to_interaction=None): + """Close a ticket, log transcript, update permissions, and remove resources.""" + self._cancel_cleanup_task(channel.id) + self._log_ticket_closed_event(ctx, channel.id) + await self._mark_ticket_closed(ticket, channel.id) + overwrites = await self._build_closure_overwrites(ctx, author, ticket) + await channel.edit(overwrites=overwrites, reason="Ticket closed (restricted access)") + await self._send_closure_embed(channel) + await self._attempt_transcript_log(ctx, channel, ticket) + await self._delete_channel_and_category(channel) + + def _log_ticket_closed_event(self, ctx, channel_id: int) -> None: + ticket_log_cog = self.bot.get_cog("TicketLog") + if not ticket_log_cog: + return + closer_id = getattr(getattr(ctx, "author", None), "id", None) or getattr(getattr(ctx, "user", None), "id", None) + if closer_id: + ticket_log_cog.add_ticket_event( + channel_id, + "TICKET_CLOSED", + closer_id, + "Ticket closed by staff member", + ) + + async def _mark_ticket_closed(self, ticket, channel_id: int) -> None: + try: + if hasattr(self.db.ticket, "close_ticket"): + await self.db.ticket.close_ticket(channel_id) + elif hasattr(self.db.ticket, "update_ticket"): + await self.db.ticket.update_ticket(ticket.ticket_id, status=TicketStatus.CLOSED) + except Exception as e: + logger.error(f"Failed updating ticket status {ticket.ticket_id}: {e}") + + async def _build_closure_overwrites(self, ctx, author, ticket) -> dict: + overwrites = { + ctx.guild.default_role: discord.PermissionOverwrite(view_channel=False), + author: discord.PermissionOverwrite(view_channel=True, send_messages=False, read_messages=True), + ctx.guild.me: discord.PermissionOverwrite( + view_channel=True, + send_messages=True, + read_messages=True, + manage_messages=True, + manage_channels=True, + ), + } + claimed_by = getattr(ticket, "claimed_by", None) + if not claimed_by: + return overwrites + claimed_member = ctx.guild.get_member(claimed_by) + if claimed_member: + overwrites[claimed_member] = discord.PermissionOverwrite( + view_channel=True, + send_messages=False, + read_messages=True, + ) + # Fetch Jr. Mod / Mod roles (levels 2 & 3) for view-only access + try: + cfg = await checks.fetch_guild_config(ctx.guild.id) + desired_ids = {cfg.get("perm_level_2_role_id"), cfg.get("perm_level_3_role_id")} - {None} + for role in ctx.guild.roles: + if role.id in desired_ids: + overwrites[role] = discord.PermissionOverwrite( + view_channel=True, + send_messages=False, + read_messages=True, + ) + except Exception as e: + logger.warning(f"Failed building closure overwrites for ticket {ticket.ticket_id}: {e}") + return overwrites + + async def _send_closure_embed(self, channel: discord.TextChannel) -> None: + try: + embed = EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name="System", + user_display_avatar=self.bot.user.display_avatar.url, + title="Ticket Closed", + description="This ticket has been closed. The channel will be deleted.", + ) + await channel.send(embed=embed) + except Exception as e: + logger.error(f"Failed sending closure embed in {channel.id}: {e}") + + async def _attempt_transcript_log(self, ctx, channel: discord.TextChannel, ticket) -> None: + ticket_log_cog = self.bot.get_cog("TicketLog") + if not ticket_log_cog: + return + try: + messages = [msg async for msg in channel.history(oldest_first=True)] + success = await ticket_log_cog.log_transcript(ctx.guild, channel, ticket, messages) + log_func = logger.info if success else logger.warning + log_func( + f"{'Successfully' if success else 'Failed to'} log transcript for ticket {ticket.ticket_id}", + ) + except Exception as e: + logger.error(f"Transcript logging error for ticket {ticket.ticket_id}: {e}") + + async def _delete_channel_and_category(self, channel: discord.TextChannel) -> None: + category = getattr(channel, "category", None) + with suppress(Exception): + await channel.delete(reason="Ticket closed and auto-removed.") + if not category or not isinstance(category, discord.CategoryChannel): + return + if len(category.channels) == 0: + with suppress(Exception): + await category.delete(reason="Temporary ticket category auto-removed.") + + async def _ephemeral_msg(self, ctx, content): + # If content is an embed, send as embed + if isinstance(content, discord.Embed): + msg = await ctx.send(embed=content, ephemeral=True) + else: + msg = await ctx.send(content, ephemeral=True) + await asyncio.sleep(10) + with suppress(Exception): + await msg.delete() + + def _resolve_target(self, ctx, target): + """Resolve a user or role from a string or object. + + Returns discord.Member | discord.Role | None + """ + if isinstance(target, discord.Member | discord.Role): # UP038 + return target + + s = str(target).strip() + guild = ctx.guild + id_candidates: set[int] = set() + if s.isdigit(): + id_candidates.add(int(s)) + for pattern in (r"<@!?([0-9]+)>", r"<@&([0-9]+)>"): + match = re.fullmatch(pattern, s) + if match: + id_candidates.add(int(match.group(1))) + + for cid in id_candidates: + member = guild.get_member(cid) + if member: + return member + role = guild.get_role(cid) + if role: + return role + + lowered = s.lower() + member = discord.utils.find( + lambda m: m.name.lower() == lowered or m.display_name.lower() == lowered, + guild.members, + ) + if member: + return member + return discord.utils.find(lambda r: r.name.lower() == lowered, guild.roles) + + @app_commands.command(name="ticket_request_close", description="Request to close the ticket (staff only)") + @app_commands.guild_only() + @checks.ac_has_pl(2) + async def ticket_request_close(self, interaction: discord.Interaction): + """Slash command to request close (staff members).""" + if not isinstance(interaction.channel, discord.TextChannel): + await interaction.response.send_message( + "This command can only be used in a ticket channel.", + ephemeral=True, + ) + return + ticket = await self.db.ticket.get_ticket_by_channel(interaction.channel.id) + if not ticket: + await interaction.response.send_message("This is not a valid ticket channel.", ephemeral=True) + return + + author = interaction.guild.get_member(ticket.author_id) + if not author: + await interaction.response.send_message("Ticket author not found.", ephemeral=True) + return + + view = RequestCloseView(self, interaction, ticket, author) + await interaction.channel.send( + f"{author.mention}, a staff member has requested to close this ticket. Please confirm:", + view=view, + ) + await interaction.response.send_message("Confirmation sent to ticket author.", ephemeral=True) + + @commands.hybrid_group(name="ticket", aliases=["t"], invoke_without_command=True) + async def ticket(self, ctx: commands.Context): + """ + Ticket command group. Use subcommands to manage tickets. + """ + help_msg = ( + "Use `$ticket create [title]` or `/ticket create [title]` to open a ticket. " + "Use `$ticket list` to list tickets.\n" + "See the wiki for more info." + ) + if isinstance(ctx, commands.Context): + await ctx.send(help_msg) + else: + await ctx.response.send_message(help_msg, ephemeral=True) + + @ticket.command(name="create") + @app_commands.describe(title="The title/subject of your ticket (optional)") + async def create(self, ctx: commands.Context, *, title: str | None = None): + """Create a new support ticket (hybrid command).""" + user, guild, send = self._extract_ctx(ctx) + self._maybe_delete_invocation(ctx) + # Validate context & title + error = self._validate_creation_prereqs(guild, user, title) + if error: + await send(error, ephemeral=True) + return + title = self._normalize_title(title) + + if hasattr(ctx, "response") and hasattr(ctx.response, "defer"): + await ctx.response.defer(ephemeral=True) + + try: + await self._ensure_guild_config(guild.id) + if await self._user_has_open_ticket(guild.id, user.id): + await send("You already have an open ticket. Please use that one or close it first.", ephemeral=True) + return + ticket = await self.db.ticket.create_ticket( + guild_id=guild.id, + channel_id=0, + author_id=user.id, + title=title, + ) + if not ticket: + self._raise_creation_error("Failed to create ticket database entry") + + overwrites = await self._build_initial_overwrites(guild, user) + ticket_channel, category = await self._make_ticket_channel(guild, ticket, user, overwrites) + await self._persist_channel_id(ticket, ticket_channel, category) + await self._post_ticket_intro(ticket_channel, user, ticket, title) + cleanup_task = asyncio.create_task( + self._schedule_unclaimed_cleanup( + ticket_channel.id, + category.id if category else None, + ), + ) + self.cleanup_tasks[ticket_channel.id] = cleanup_task + await self._ephemeral_success(send, ticket, ticket_channel) + except TicketCreationError as e: + logger.warning(f"Ticket creation aborted: {e}") + await send(str(e), ephemeral=True) + except Exception as e: + logger.error(f"Error creating ticket: {e}") + logger.error(f"Traceback: {traceback.format_exc()}") + await send("An error occurred while creating your ticket. Please try again later.", ephemeral=True) + + # ---- Creation helpers ---- + def _extract_ctx(self, ctx): + if isinstance(ctx, commands.Context): + return ctx.author, ctx.guild, ctx.send + return ctx.user, ctx.guild, ctx.response.send_message + + def _maybe_delete_invocation(self, ctx): + if isinstance(ctx, commands.Context): + with suppress(discord.NotFound, discord.Forbidden): + task = asyncio.create_task(ctx.message.delete()) + ctx._delete_invocation_task = task + + def _validate_creation_prereqs(self, guild, user, title): + if not guild or not isinstance(user, discord.Member): + return "This command can only be used in a server." + if not title or not title.strip(): + return None + if len(title) > 100: + return "Ticket title must be 100 characters or less." + if len(title.strip()) < 3: + return "Ticket title must be at least 3 characters long." + return None + + def _normalize_title(self, title): + return "No subject provided" if not title or not title.strip() else title.strip() + + async def _ensure_guild_config(self, guild_id: int): + if not await self.guild_config_controller.get_guild_config(guild_id): + await self.guild_config_controller.insert_guild_config(guild_id) + + async def _user_has_open_ticket(self, guild_id: int, user_id: int) -> bool: + existing = await self.db.ticket.get_user_tickets(guild_id, user_id, TicketStatus.OPEN) + return bool(existing) + + async def _build_initial_overwrites(self, guild: discord.Guild, user: discord.Member): + overwrites = { + guild.default_role: discord.PermissionOverwrite(view_channel=False), + user: discord.PermissionOverwrite(view_channel=True, send_messages=True, read_messages=True), + guild.me: discord.PermissionOverwrite( + view_channel=True, + read_messages=True, + manage_messages=True, + manage_channels=True, + ), + } + cfg = await checks.fetch_guild_config(guild.id) + staff_role_ids = { + cfg.get("perm_level_2_role_id"), + cfg.get("perm_level_3_role_id"), + } - {None} + for role in guild.roles: + if role.id in staff_role_ids: + overwrites[role] = discord.PermissionOverwrite( + view_channel=True, + send_messages=True, + read_messages=True, + manage_messages=True, + ) + return overwrites + + async def _make_ticket_channel(self, guild: discord.Guild, ticket, user, overwrites): + channel_name = f"ticket-{ticket.ticket_id}" + category = None + try: + category = await guild.create_category(name=channel_name, reason=f"Temporary category for {channel_name}") + ticket_channel = await guild.create_text_channel( + channel_name, + overwrites=overwrites, + category=category, + reason=f"Ticket #{ticket.ticket_id} created by {user}", + ) + except Exception as e: + if category: + with suppress(Exception): + await category.delete(reason="Cleanup after failed ticket channel creation") + await self.db.ticket.delete_ticket(ticket.ticket_id) + msg = "Failed to create ticket channel. Please try again later." + raise TicketCreationError(msg) from e + else: + return ticket_channel, category + + async def _persist_channel_id(self, ticket, ticket_channel, category): + try: + updated = await self.db.ticket.update( + where={"ticket_id": ticket.ticket_id}, + data={"channel_id": ticket_channel.id}, + ) + if not updated: + self._raise_persist_error("Failed to update ticket with channel ID") + except Exception as e: + with suppress(Exception): + await ticket_channel.delete(reason="Failed to update ticket in database") + if category: + with suppress(Exception): + await category.delete(reason="Cleanup after failed ticket DB update") + await self.db.ticket.delete_ticket(ticket.ticket_id) + if isinstance(e, TicketCreationError): + raise + emsg = "Failed to create ticket. Please try again later." + raise TicketCreationError(emsg) from e + + async def _post_ticket_intro(self, channel: discord.TextChannel, user: discord.Member, ticket, title: str): + embed = EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name=user.display_name, + user_display_avatar=user.display_avatar.url, + title=f"Ticket #{ticket.ticket_id}", + description=( + f"Hello {user.mention}! Thank you for creating a support ticket.\n\n" + f"**Reason:** {title}\n\n" + f"**Please describe your issue in detail below.**\n\nA staff member will be with you shortly." + ), + ) + embed.add_field(name="Ticket ID", value=f"#{ticket.ticket_id}", inline=True) + embed.add_field(name="Status", value="Open", inline=True) + embed.add_field(name="Created", value=f"", inline=True) + view = TicketManagementView(self.bot, ticket.ticket_id) + await channel.send(content=f"{user.mention}", embed=embed, view=view) + + async def _schedule_unclaimed_cleanup(self, channel_id: int, category_id: int | None, delay_seconds: int = 14400): + try: + await asyncio.sleep(delay_seconds) + ticket = await self.db.ticket.get_ticket_by_channel(channel_id) + if not ticket or ticket.claimed_by: + return + guild = self.bot.get_guild(ticket.guild_id) + if not guild: + return + channel = guild.get_channel(channel_id) + category = guild.get_channel(category_id) if category_id else None + if channel: + with suppress(Exception): + await channel.delete(reason="Ticket unclaimed for 4h, auto-removed.") + if category and isinstance(category, discord.CategoryChannel) and len(category.channels) == 0: + with suppress(Exception): + await category.delete(reason="Temporary ticket category auto-removed.") + finally: + # Clean up task reference when done + self.cleanup_tasks.pop(channel_id, None) + + def _cancel_cleanup_task(self, channel_id: int) -> None: + """Cancel and remove cleanup task for a channel.""" + if (task := self.cleanup_tasks.pop(channel_id, None)) and not task.done(): + task.cancel() + + async def _ephemeral_success(self, send, ticket, ticket_channel): + msg = await send( + f"Ticket #{ticket.ticket_id} created successfully! Please check {ticket_channel.mention}", + ephemeral=True, + ) + if hasattr(msg, "delete"): + await asyncio.sleep(10) + with suppress(Exception): + await msg.delete() + + @ticket.command(name="list") + @app_commands.describe(user="User to list tickets for") + @checks.has_pl(2) + async def list(self, ctx: commands.Context, user: discord.Member): + """List tickets involving a user.""" + # Support both prefix and slash (Interaction) usage + if isinstance(ctx, commands.Context): + guild = ctx.guild + send = ctx.send + author = ctx.author + else: + guild = ctx.guild + send = ctx.response.send_message + author = ctx.user + + await ( + ctx.response.defer(ephemeral=True) + if hasattr(ctx, "response") and hasattr(ctx.response, "defer") + else asyncio.sleep(0) + ) + try: + # Ensure guild config exists + existing_config = await self.guild_config_controller.get_guild_config(guild.id) + if not existing_config: + await self.guild_config_controller.insert_guild_config(guild.id) + + target_user = user + # Fetch all tickets for the guild + all_tickets = await self.db.ticket.get_guild_tickets(guild.id) + # Filter tickets where user is author + user_tickets = [ + t + for t in all_tickets + if t.author_id == target_user.id or (hasattr(t, "participants") and target_user.id in t.participants) + ] + if not user_tickets: + embed = EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name=author.display_name, + user_display_avatar=author.display_avatar.url, + title="No Tickets Found", + description=f"No tickets found for {target_user.display_name}.", + ) + await send(embed=embed, ephemeral=True) + return + + # Pagination logic + page_size = 10 + total_pages = (len(user_tickets) + page_size - 1) // page_size + current_page = 0 + + async def send_page(page): + embed = EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name=author.display_name, + user_display_avatar=author.display_avatar.url, + title=f"Tickets for {target_user.display_name}", + description=f"Page {page + 1}/{total_pages}", + ) + for ticket in user_tickets[page * page_size : (page + 1) * page_size]: + ticket_author = guild.get_member(ticket.author_id) + author_name = ticket_author.display_name if ticket_author else f"Unknown ({ticket.author_id})" + claimed_info = "" + if ticket.claimed_by: + claimed_user = guild.get_member(ticket.claimed_by) + claimed_name = claimed_user.display_name if claimed_user else f"Unknown ({ticket.claimed_by})" + claimed_info = f" - Claimed by {claimed_name}" + status_emoji = {TicketStatus.OPEN: "šŸŸ¢", TicketStatus.CLAIMED: "šŸŸ”", TicketStatus.CLOSED: "šŸ”“"}.get( + ticket.status, + "ā“", + ) + channel = guild.get_channel(ticket.channel_id) + channel_info = channel.mention if channel else "Channel deleted" + embed.add_field( + name=f"{status_emoji} #{ticket.ticket_id} - {ticket.title}", + value=f"**Author:** {author_name}\n**Channel:** {channel_info}\n**Created:** {claimed_info}", + inline=False, + ) + + # Add navigation buttons if needed + class TicketPaginator(discord.ui.View): + def __init__(self, parent, page): + super().__init__(timeout=60) + self.parent = parent + self.page = page + + @discord.ui.button(label="Previous", style=discord.ButtonStyle.secondary, disabled=page == 0) + async def previous(self, interaction2: discord.Interaction, button: discord.ui.Button): + await interaction2.response.defer() + await send_page(self.page - 1) + + @discord.ui.button( + label="Next", + style=discord.ButtonStyle.secondary, + disabled=page == total_pages - 1, + ) + async def next(self, interaction2: discord.Interaction, button: discord.ui.Button): + await interaction2.response.defer() + await send_page(self.page + 1) + + view = TicketPaginator(self, page) if total_pages > 1 else discord.ui.View() + if hasattr(ctx, "response") and ctx.response.is_done(): + await ctx.followup.send(embed=embed, ephemeral=True, view=view) + else: + await send(embed=embed, ephemeral=True, view=view) + + await send_page(current_page) + except Exception as e: + logger.error(f"Error listing tickets: {e}") + await send("An error occurred while fetching tickets.", ephemeral=True) + + @commands.hybrid_command(name="ticket_list") + @app_commands.describe(user="User to list tickets for") + @checks.has_pl(2) + async def ticket_list(self, ctx: commands.Context, user: discord.Member): + """List tickets involving a user.""" + # Support both prefix and slash (Interaction) usage + if isinstance(ctx, commands.Context): + guild = ctx.guild + send = ctx.send + author = ctx.author + else: + guild = ctx.guild + send = ctx.response.send_message + author = ctx.user + + await ( + ctx.response.defer(ephemeral=True) + if hasattr(ctx, "response") and hasattr(ctx.response, "defer") + else asyncio.sleep(0) + ) + try: + # Ensure guild config exists + existing_config = await self.guild_config_controller.get_guild_config(guild.id) + if not existing_config: + await self.guild_config_controller.insert_guild_config(guild.id) + + target_user = user + # Fetch all tickets for the guild + all_tickets = await self.db.ticket.get_guild_tickets(guild.id) + # Filter tickets where user is author + user_tickets = [ + t + for t in all_tickets + if t.author_id == target_user.id or (hasattr(t, "participants") and target_user.id in t.participants) + ] + if not user_tickets: + embed = EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name=author.display_name, + user_display_avatar=author.display_avatar.url, + title="No Tickets Found", + description=f"No tickets found for {target_user.display_name}.", + ) + await send(embed=embed, ephemeral=True) + return + + # Pagination logic + page_size = 10 + total_pages = (len(user_tickets) + page_size - 1) // page_size + current_page = 0 + + async def send_page(page): + embed = EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name=author.display_name, + user_display_avatar=author.display_avatar.url, + title=f"Tickets for {target_user.display_name}", + description=f"Page {page + 1}/{total_pages}", + ) + for ticket in user_tickets[page * page_size : (page + 1) * page_size]: + ticket_author = guild.get_member(ticket.author_id) + author_name = ticket_author.display_name if ticket_author else f"Unknown ({ticket.author_id})" + claimed_info = "" + if ticket.claimed_by: + claimed_user = guild.get_member(ticket.claimed_by) + claimed_name = claimed_user.display_name if claimed_user else f"Unknown ({ticket.claimed_by})" + claimed_info = f" - Claimed by {claimed_name}" + status_emoji = {TicketStatus.OPEN: "šŸŸ¢", TicketStatus.CLAIMED: "šŸŸ”", TicketStatus.CLOSED: "šŸ”“"}.get( + ticket.status, + "ā“", + ) + channel = guild.get_channel(ticket.channel_id) + channel_info = channel.mention if channel else "Channel deleted" + embed.add_field( + name=f"{status_emoji} #{ticket.ticket_id} - {ticket.title}", + value=f"**Author:** {author_name}\n**Channel:** {channel_info}\n**Created:** {claimed_info}", + inline=False, + ) + + # Add navigation buttons if needed + class TicketPaginator(discord.ui.View): + def __init__(self, parent, page): + super().__init__(timeout=60) + self.parent = parent + self.page = page + + @discord.ui.button(label="Previous", style=discord.ButtonStyle.secondary, disabled=page == 0) + async def previous(self, interaction2: discord.Interaction, button: discord.ui.Button): + await interaction2.response.defer() + await send_page(self.page - 1) + + @discord.ui.button( + label="Next", + style=discord.ButtonStyle.secondary, + disabled=page == total_pages - 1, + ) + async def next(self, interaction2: discord.Interaction, button: discord.ui.Button): + await interaction2.response.defer() + await send_page(self.page + 1) + + view = TicketPaginator(self, page) if total_pages > 1 else discord.ui.View() + if hasattr(ctx, "response") and ctx.response.is_done(): + await ctx.followup.send(embed=embed, ephemeral=True, view=view) + else: + await send(embed=embed, ephemeral=True, view=view) + + await send_page(current_page) + except Exception as e: + logger.error(f"Error listing tickets: {e}") + await send("An error occurred while fetching tickets.", ephemeral=True) + + @ticket.command(name="stats") + @app_commands.describe(user="Specific user to show stats for (optional)") + @checks.has_pl(2) + async def stats(self, ctx: commands.Context, user: discord.Member = None): + """Show ticket statistics for staff members.""" + + if isinstance(ctx, commands.Context): + with suppress(discord.NotFound, discord.Forbidden, AttributeError): + _ = ctx.message.delete() + guild, author, is_prefix, send = ctx.guild, ctx.author, True, ctx.send + else: + guild, author, is_prefix = ctx.guild, ctx.user, False + send = ctx.response.send_message if hasattr(ctx, "response") else (lambda *a, **k: None) + if hasattr(ctx, "response") and hasattr(ctx.response, "defer"): + await ctx.response.defer(ephemeral=True) + + try: + all_tickets, claims_count, closed_count = await self._gather_guild_ticket_metrics(guild.id) + embed = self._build_stats_embed(author, user, guild, all_tickets, claims_count, closed_count) + + if is_prefix: + await self._ephemeral_msg(ctx, embed) + elif hasattr(ctx, "response") and ctx.response.is_done(): + await ctx.followup.send(embed=embed, ephemeral=True) + else: + await ctx.response.send_message(embed=embed, ephemeral=True) + except Exception as e: + logger.error(f"Error showing ticket stats: {e}") + msg = "An error occurred while fetching ticket statistics." + if is_prefix: + await self._ephemeral_msg(ctx, msg) + elif hasattr(ctx, "followup") and hasattr(ctx, "response") and ctx.response.is_done(): + await ctx.followup.send(msg, ephemeral=True) + else: + await send(msg, ephemeral=True) + + @ticket.command(name="leaderboard") + @app_commands.describe( + metric="What to rank by: 'claims', 'closed', or 'rate' (default: claims)", + limit="Number of users to show (default: 10)", + ) + @checks.has_pl(2) + async def leaderboard(self, ctx: commands.Context, metric: str = "claims", limit: int = 10): + """Show ticket activity leaderboard for staff.""" + + if isinstance(ctx, commands.Context): + guild, author, send = ctx.guild, ctx.author, ctx.send + else: + guild, author, send = ctx.guild, ctx.user, ctx.response.send_message + + if hasattr(ctx, "response") and hasattr(ctx.response, "defer"): + await ctx.response.defer(ephemeral=True) + + try: + tickets = await self.db.ticket.get_guild_tickets(guild.id, limit=1000) + embed = self._build_leaderboard_embed(author, guild, tickets, metric, limit) + if hasattr(ctx, "response") and ctx.response.is_done(): + await ctx.followup.send(embed=embed, ephemeral=True) + else: + await send(embed=embed, ephemeral=True) + except Exception as e: + logger.error(f"Error showing leaderboard: {e}") + await send("An error occurred while generating the leaderboard.", ephemeral=True) + + @ticket.command(name="add") + @app_commands.describe(target="User or role to add") + @checks.has_pl(2) + async def add(self, ctx: commands.Context, target: str): + """Add a user or role to the current ticket channel (Staff only).""" + if not isinstance(ctx.channel, discord.TextChannel): + await self._ephemeral_msg(ctx, "This command can only be used in a ticket channel.") + # Privacy: delete user's command message + if isinstance(ctx, commands.Context): + with suppress(discord.NotFound, discord.Forbidden): + await ctx.message.delete() + return + + ticket = await self.db.ticket.get_ticket_by_channel(ctx.channel.id) + if not ticket: + await self._ephemeral_msg(ctx, "This is not a valid ticket channel.") + # Privacy: delete user's command message + if isinstance(ctx, commands.Context): + with suppress(discord.NotFound, discord.Forbidden): + await ctx.message.delete() + return + + # Support @role mention + if target.startswith("<@&") and target.endswith(">"): + role_id = target[3:-1] + target_obj = ctx.guild.get_role(int(role_id)) if role_id.isdigit() else None # SIM108 + else: + target_obj = self._resolve_target(ctx, target) + + if not target_obj: + await self._ephemeral_msg(ctx, f"Could not find user or role for '{target}'.") + # Privacy: delete user's command message + if isinstance(ctx, commands.Context): + with suppress(discord.NotFound, discord.Forbidden): + await ctx.message.delete() + return + + await ctx.channel.set_permissions( + target_obj, + view_channel=True, + send_messages=True, + read_messages=True, + reason="Added to ticket by staff", + ) + + # Track the event + ticket_log_cog = self.bot.get_cog("TicketLog") + if ticket_log_cog: + actor_id = ctx.author.id if hasattr(ctx, "author") else ctx.user.id + event_type = "ROLE_ADDED" if isinstance(target_obj, discord.Role) else "USER_ADDED" + target_name = getattr(target_obj, "name", str(target_obj)) + ticket_log_cog.add_ticket_event(ctx.channel.id, event_type, actor_id, f"Added {target_name} to ticket") + await self._ephemeral_msg( + ctx, + f"{getattr(target_obj, 'mention', str(target_obj))} has been added to this ticket.", + ) + # Privacy: delete user's command message + if isinstance(ctx, commands.Context): + with suppress(discord.NotFound, discord.Forbidden): + await ctx.message.delete() + + @ticket.command(name="remove") + @app_commands.describe(target="User or role to remove (@mention or name)") + @checks.has_pl(2) + async def remove(self, ctx: commands.Context, target: str): + """Remove a user or role from the ticket (Staff only).""" + valid, ticket = await self._validate_ticket_channel(ctx) + if not valid: + return + target_obj = self._parse_target(ctx, target) + if not target_obj: + await self._ephemeral_msg(ctx, f"Could not find user or role for '{target}'.") + await self._maybe_delete_command_message(ctx) + return + + await self._log_removal_event(ctx, target_obj) + await self._perform_removal(ctx, target_obj) + await self._maybe_delete_command_message(ctx) + + async def _log_removal_event(self, ctx, target_obj): + """Log the removal event to ticket log.""" + ticket_log_cog = self.bot.get_cog("TicketLog") + if ticket_log_cog: + actor_id = ctx.author.id if hasattr(ctx, "author") else ctx.user.id + event_type = "ROLE_REMOVED" if isinstance(target_obj, discord.Role) else "USER_REMOVED" + target_name = getattr(target_obj, "name", str(target_obj)) + ticket_log_cog.add_ticket_event(ctx.channel.id, event_type, actor_id, f"Removed {target_name} from ticket") + + async def _perform_removal(self, ctx, target_obj): + """Remove the target object from the ticket channel.""" + if isinstance(target_obj, discord.Role): + await ctx.channel.set_permissions(target_obj, overwrite=None, reason="Removed role from ticket by staff") + # Remove all members of the role from the channel + for member in target_obj.members: + await ctx.channel.set_permissions( + member, + overwrite=None, + reason="Removed from ticket (role removal by staff)", + ) + await self._ephemeral_msg(ctx, f"{target_obj.mention} and its members have been removed from this ticket.") + else: + await ctx.channel.set_permissions(target_obj, overwrite=None, reason="Removed from ticket by staff") + await self._ephemeral_msg( + ctx, + f"{getattr(target_obj, 'mention', str(target_obj))} has been removed from this ticket.", + ) + + async def _gather_guild_ticket_metrics(self, guild_id: int): + tickets = await self.db.ticket.get_guild_tickets(guild_id, limit=1000) + claims_count: dict[int, int] = defaultdict(int) + closed_count: dict[int, int] = defaultdict(int) + for t in tickets: + cid = t.claimed_by + if not cid: + continue + claims_count[cid] += 1 + if t.status == TicketStatus.CLOSED: + closed_count[cid] += 1 + return tickets, claims_count, closed_count + + def _build_stats_embed(self, author, user, guild, tickets, claims_count, closed_count): + def _base(title: str): + return EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name=author.display_name, + user_display_avatar=author.display_avatar.url, + title=title, + description="All-time statistics", + ) + + if user: + embed = _base(f"Ticket Stats for {user.display_name}") + claims = claims_count.get(user.id, 0) + closed = closed_count.get(user.id, 0) + rate = f"{(closed / claims * 100):.1f}%" if claims else "N/A" + for name, val in (("šŸŽ« Tickets Claimed", claims), ("āœ… Tickets Closed", closed), ("šŸ“Š Close Rate", rate)): + embed.add_field(name=name, value=str(val), inline=True) + else: + embed = _base("Staff Ticket Statistics") + if not claims_count: + embed.add_field(name="No Data", value="No ticket claims found for this period.", inline=False) + else: + staff_activity = [ + (member := guild.get_member(uid), claim_count, closed_count.get(uid, 0)) + for uid, claim_count in claims_count.items() + if (guild.get_member(uid) is not None) + ] + staff_activity.sort(key=lambda x: x[1], reverse=True) + for i, (member, claims, closed) in enumerate(staff_activity[:10]): + rate = (closed / claims * 100) if claims else 0 + embed.add_field( + name=f"#{i + 1} {member.display_name}", + value=f"šŸŽ« {claims} claimed\nāœ… {closed} closed\nšŸ“Š {rate:.1f}% rate", + inline=True, + ) + total_tickets = len(tickets) + total_claimed = sum(1 for t in tickets if t.claimed_by) + total_closed = sum(1 for t in tickets if t.status == TicketStatus.CLOSED) + embed.add_field( + name="šŸ“ˆ Server Summary", + value=f"Total: {total_tickets}\nClaimed: {total_claimed}\nClosed: {total_closed}", + inline=True, + ) + return embed + + def _build_leaderboard_embed(self, author, guild, tickets, metric: str, limit: int): + claims_count: dict[int, int] = defaultdict(int) + closed_count: dict[int, int] = defaultdict(int) + for t in tickets: + cid = t.claimed_by + if not cid: + continue + claims_count[cid] += 1 + if t.status == TicketStatus.CLOSED: + closed_count[cid] += 1 + if not claims_count: + return EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name=author.display_name, + user_display_avatar=author.display_avatar.url, + title="Ticket Leaderboard", + description="No ticket activity found for this period.", + ) + + def score(c, cl): + rate = (cl / c * 100) if c else 0 + return {"claims": c, "closed": cl, "rate": rate}.get(metric, c), rate + + staff = [] + for uid, c in claims_count.items(): + member = guild.get_member(uid) + if not member: + continue + cl = closed_count.get(uid, 0) + sc, rate = score(c, cl) + staff.append((member, c, cl, rate, sc)) + staff.sort(key=lambda x: x[4], reverse=True) + embed = EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name=author.display_name, + user_display_avatar=author.display_avatar.url, + title=f"Ticket Leaderboard - {metric.title()}", + description=f"Top {min(limit, len(staff))} staff members (all-time)", + ) + medals = ["šŸ„‡", "šŸ„ˆ", "šŸ„‰"] + for i, (member, c, cl, rate, _sc) in enumerate(staff[:limit]): + medal = medals[i] if i < 3 else f"#{i + 1}" + match metric: + case "claims": + value = f"šŸŽ« {c} tickets claimed" + case "closed": + value = f"āœ… {cl} tickets closed" + case "rate": + value = f"šŸ“Š {rate:.1f}% close rate ({cl}/{c})" + case _: + value = f"šŸŽ« {c} claims ā€¢ āœ… {cl} closed" + embed.add_field(name=f"{medal} {member.display_name}", value=value, inline=False) + return embed + + async def _validate_ticket_channel(self, ctx): + if not isinstance(ctx.channel, discord.TextChannel): + await self._ephemeral_msg(ctx, "This command can only be used in a ticket channel.") + await self._maybe_delete_command_message(ctx) + return False, None + ticket = await self.db.ticket.get_ticket_by_channel(ctx.channel.id) + if not ticket: + await self._ephemeral_msg(ctx, "This is not a valid ticket channel.") + await self._maybe_delete_command_message(ctx) + return False, None + return True, ticket + + def _parse_target(self, ctx, target: str): + if target.startswith("<@&") and target.endswith(">"): + role_id = target.strip("<@&>") + return ctx.guild.get_role(int(role_id)) if role_id.isdigit() else None + return self._resolve_target(ctx, target) + + async def _maybe_delete_command_message(self, ctx): + if isinstance(ctx, commands.Context): + with suppress(discord.NotFound, discord.Forbidden): + await ctx.message.delete() + + @commands.Cog.listener() + async def on_ready(self) -> None: + self.bot.add_view(TicketManagementView(self.bot)) + + +async def setup(bot: Tux) -> None: + """Setup the Tickets cog.""" + await bot.add_cog(Tickets(bot)) diff --git a/tux/cogs/moderation/timeout.py b/tux/cogs/moderation/timeout.py index d47b1d145..5d4e847a1 100644 --- a/tux/cogs/moderation/timeout.py +++ b/tux/cogs/moderation/timeout.py @@ -19,7 +19,7 @@ def __init__(self, bot: Tux) -> None: @commands.hybrid_command( name="timeout", - aliases=["t", "to", "mute", "m"], + aliases=["to", "mute", "m"], ) @commands.guild_only() @checks.has_pl(2) diff --git a/tux/cogs/utility/ticket_log.py b/tux/cogs/utility/ticket_log.py new file mode 100644 index 000000000..534e493eb --- /dev/null +++ b/tux/cogs/utility/ticket_log.py @@ -0,0 +1,189 @@ +import io +from datetime import UTC, datetime + +import discord +from discord.ext import commands + +from prisma.enums import TicketStatus +from tux.bot import Tux + + +class TicketLog(commands.Cog): + """Cog to handle ticket transcript logging with enhanced event tracking.""" + + def __init__(self, bot: Tux): + self.bot = bot + # Track ticket events that happen outside of message content + self.ticket_events = {} # {channel_id: [events]} + + def add_ticket_event(self, channel_id: int, event_type: str, user_id: int, details: str | None = None): + """Add a ticket event to track for logging.""" + if channel_id not in self.ticket_events: + self.ticket_events[channel_id] = [] + + event = { + "timestamp": datetime.now(UTC), + "event_type": event_type, + "user_id": user_id, + "details": details, + } + self.ticket_events[channel_id].append(event) + + def get_ticket_events(self, channel_id: int): + """Get all events for a ticket channel.""" + return self.ticket_events.get(channel_id, []) + + def clear_ticket_events(self, channel_id: int): + """Clear events for a ticket channel after logging.""" + if channel_id in self.ticket_events: + del self.ticket_events[channel_id] + + async def log_transcript(self, guild, channel, ticket, messages): + """Log the transcript as a plain text file with enhanced event detection.""" + log_cog = self.bot.get_cog("TicketLogConfig") + log_channel_id = None + if log_cog: + log_channel_id = log_cog.get_log_channel_id(guild.id) + + if log_channel_id: + log_channel = guild.get_channel(log_channel_id) + if log_channel: + # Get tracked events for this channel + tracked_events = self.get_ticket_events(channel.id) + + # Combine messages and tracked events, then sort by timestamp + all_entries = [{"timestamp": msg.created_at, "type": "message", "data": msg} for msg in messages] + [ + {"timestamp": event["timestamp"], "type": "event", "data": event} for event in tracked_events + ] + + # Sort by timestamp + all_entries.sort(key=lambda x: x["timestamp"]) + + lines = [] + for entry in all_entries: + timestamp_str = entry["timestamp"].strftime("%Y-%m-%d %H:%M") + + if entry["type"] == "message": + msg = entry["data"] + author = f"{msg.author.display_name} ({msg.author.id})" + content = msg.content + + # Enhanced message content analysis + tags = self._analyze_message_content(content, msg) + tag_str = "".join(f"[{tag}] " for tag in tags) + + lines.append(f"[{timestamp_str}] {author}: {tag_str}{content}") + + elif entry["type"] == "event": + event = entry["data"] + user = guild.get_member(event["user_id"]) + user_name = user.display_name if user else f"Unknown ({event['user_id']})" + + event_msg = self._format_event_message(event, user_name) + lines.append(f"[{timestamp_str}] SYSTEM: [EVENT] {event_msg}") + + # Add ticket metadata at the beginning + metadata_lines = [ + f"=== TICKET #{ticket.ticket_id} TRANSCRIPT ===", + f"Title: {ticket.title}", + f"Author: {guild.get_member(ticket.author_id).display_name if guild.get_member(ticket.author_id) else f'Unknown ({ticket.author_id})'}", + f"Created: {ticket.created_at.strftime('%Y-%m-%d %H:%M:%S')} UTC", + f"Channel: #{channel.name} ({channel.id})", + f"Status: {ticket.status.value if hasattr(ticket.status, 'value') else ticket.status}", + f"Claimed by: {guild.get_member(ticket.claimed_by).display_name if ticket.claimed_by and guild.get_member(ticket.claimed_by) else 'None'}", + "=" * 50, + "", + ] + + transcript_text = "\n".join(metadata_lines + lines) + file = discord.File(io.BytesIO(transcript_text.encode()), filename=f"transcript_{channel.id}.txt") + + # Create summary embed + embed = discord.Embed( + title=f"Ticket #{ticket.ticket_id} Transcript", + description=f"**Title:** {ticket.title}\n**Author:** {guild.get_member(ticket.author_id).mention if guild.get_member(ticket.author_id) else 'Unknown User'}", + color=0x2ECC71 if ticket.status == TicketStatus.CLOSED else 0xE74C3C, + timestamp=datetime.now(UTC), + ) + embed.add_field(name="Messages", value=str(len(messages)), inline=True) + embed.add_field(name="Events", value=str(len(tracked_events)), inline=True) + embed.add_field( + name="Status", + value=ticket.status.value if hasattr(ticket.status, "value") else str(ticket.status), + inline=True, + ) + + if ticket.claimed_by: + claimed_user = guild.get_member(ticket.claimed_by) + embed.add_field( + name="Claimed By", + value=claimed_user.mention if claimed_user else f"Unknown ({ticket.claimed_by})", + inline=True, + ) + + await log_channel.send(embed=embed, file=file) + + # Clear tracked events after logging + self.clear_ticket_events(channel.id) + return True + return False + + def _analyze_message_content(self, content: str, msg: discord.Message) -> list: + """Analyze message content for ticket-related events.""" + tags = [] + content_lower = content.lower() + + # Bot/system messages + if msg.author.bot: + if any( + phrase in content_lower for phrase in ["ticket closed", "has been closed", "ticket has been closed"] + ): + tags.append("TICKET CLOSED") + elif any( + phrase in content_lower for phrase in ["ticket created", "ticket opened", "thank you for creating"] + ): + tags.append("TICKET CREATED") + elif "claimed this ticket" in content_lower: + tags.append("TICKET CLAIMED") + elif "unclaimed this ticket" in content_lower: + tags.append("TICKET UNCLAIMED") + + # User messages + elif any(phrase in content_lower for phrase in ["requested to close", "request close", "please close"]): + tags.append("CLOSE REQUESTED") + elif any(phrase in content_lower for phrase in ["/close", "!close", "$close"]): + tags.append("CLOSE COMMAND") + + # Check for embeds that might contain ticket info + if msg.embeds: + for embed in msg.embeds: + if embed.title: + title_lower = embed.title.lower() + if "ticket" in title_lower and "closed" in title_lower: + tags.append("TICKET CLOSED") + elif "ticket" in title_lower and "claimed" in title_lower: + tags.append("TICKET CLAIMED") + + return tags + + def _format_event_message(self, event: dict, user_name: str) -> str: + """Format an event into a readable message.""" + event_type = event["event_type"] + details = event.get("details", "") + mapping = { + "TICKET_CLAIMED": f"{user_name} claimed the ticket. {details}", + "TICKET_UNCLAIMED": f"{user_name} unclaimed the ticket. {details}", + "TICKET_CLOSED": f"{user_name} closed the ticket. {details}", + "TICKET_REOPENED": f"{user_name} reopened the ticket. {details}", + "USER_ADDED": f"{user_name} was added to the ticket. {details}", + "USER_REMOVED": f"{user_name} was removed from the ticket. {details}", + "ROLE_ADDED": f"Role was added to the ticket by {user_name}. {details}", + "ROLE_REMOVED": f"Role was removed from the ticket by {user_name}. {details}", + "PERMISSIONS_CHANGED": f"{user_name} changed ticket permissions. {details}", + "CLOSE_REQUESTED": f"{user_name} requested to close the ticket. {details}", + } + return mapping.get(event_type, f"{user_name} performed action: {event_type}. {details}").strip() + + +async def setup(bot: Tux): + await bot.add_cog(TicketLog(bot)) diff --git a/tux/cogs/utility/ticket_log_config.py b/tux/cogs/utility/ticket_log_config.py new file mode 100644 index 000000000..4edbdc8dc --- /dev/null +++ b/tux/cogs/utility/ticket_log_config.py @@ -0,0 +1,51 @@ +from pathlib import Path + +import discord +from discord.ext import commands + +from tux.bot import Tux + +CONFIG_PATH = (Path(__file__).parent / "../../assets/embeds/ticket_log_channel.txt").resolve() + + +class TicketLogConfig(commands.Cog): + """Cog to configure the ticket log channel.""" + + def __init__(self, bot: Tux): + self.bot = bot + self._load_config() + + def _load_config(self): + self.config = {} + if CONFIG_PATH.exists(): + with CONFIG_PATH.open() as f: + for raw_line in f: + line = raw_line.strip() + if not line or ":" not in line: + continue + gid, cid = line.split(":", 1) + self.config[gid] = int(cid) + + def _save_config(self): + CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) + with CONFIG_PATH.open("w") as f: + f.writelines(f"{gid}:{cid}\n" for gid, cid in self.config.items()) + + @discord.app_commands.command( + name="set_ticket_log_channel", + description="Set the channel where ticket logs will be sent.", + ) + @discord.app_commands.describe(channel="Channel to send ticket logs to") + @discord.app_commands.checks.has_permissions(administrator=True) + async def set_ticket_log_channel(self, interaction: discord.Interaction, channel: discord.TextChannel): + """Set the channel where ticket logs will be sent (slash command).""" + self.config[str(interaction.guild.id)] = channel.id + self._save_config() + await interaction.response.send_message(f"Ticket log channel set to {channel.mention}", ephemeral=True) + + def get_log_channel_id(self, guild_id: int): + return self.config.get(str(guild_id)) + + +async def setup(bot: Tux): + await bot.add_cog(TicketLogConfig(bot)) diff --git a/tux/database/controllers/__init__.py b/tux/database/controllers/__init__.py index 445c4c84f..204e20d2a 100644 --- a/tux/database/controllers/__init__.py +++ b/tux/database/controllers/__init__.py @@ -15,6 +15,7 @@ from tux.database.controllers.reminder import ReminderController from tux.database.controllers.snippet import SnippetController from tux.database.controllers.starboard import StarboardController, StarboardMessageController +from tux.database.controllers.ticket import TicketController # Define a TypeVar that can be any BaseController subclass ControllerType = TypeVar("ControllerType") @@ -49,6 +50,8 @@ class DatabaseController: The starboard controller instance. _starboard_message : StarboardMessageController, optional The starboard message controller instance. + _ticket : TicketController, optional + The ticket controller instance. """ def __init__(self) -> None: @@ -64,6 +67,7 @@ def __init__(self) -> None: self._snippet: SnippetController | None = None self._starboard: StarboardController | None = None self._starboard_message: StarboardMessageController | None = None + self._ticket: TicketController | None = None def _get_controller(self, controller_type: type[ControllerType]) -> ControllerType: """ @@ -166,6 +170,7 @@ def sync_wrapped_method(*args: Any, **kwargs: Any) -> Any: "snippet": SnippetController, "starboard": StarboardController, "starboard_message": StarboardMessageController, + "ticket": TicketController, } def __getattr__(self, name: str) -> Any: diff --git a/tux/database/controllers/base.py b/tux/database/controllers/base.py index f407e480d..53ee34c0c 100644 --- a/tux/database/controllers/base.py +++ b/tux/database/controllers/base.py @@ -17,6 +17,7 @@ Snippet, Starboard, StarboardMessage, + Ticket, ) from tux.database.client import db @@ -33,6 +34,7 @@ GuildConfig, AFKModel, Levels, + Ticket, ) RelationType = TypeVar("RelationType") @@ -50,6 +52,7 @@ class BaseController[ GuildConfig, AFKModel, Levels, + Ticket, ), ]: """Provides a base interface for database table controllers. diff --git a/tux/database/controllers/ticket.py b/tux/database/controllers/ticket.py new file mode 100644 index 000000000..0d2dde20b --- /dev/null +++ b/tux/database/controllers/ticket.py @@ -0,0 +1,244 @@ +"""Ticket database controller for managing support tickets.""" + +from datetime import UTC, datetime + +from loguru import logger + +from prisma.enums import TicketStatus +from prisma.models import Ticket +from tux.database.controllers.base import BaseController + + +class TicketController(BaseController[Ticket]): + """Controller for managing ticket database operations.""" + + def __init__(self) -> None: + super().__init__("ticket") + + async def create_ticket( + self, + guild_id: int, + channel_id: int, + author_id: int, + title: str, + ) -> Ticket | None: + """ + Create a new ticket in the database. + + Parameters + ---------- + guild_id : int + The ID of the guild where the ticket was created. + channel_id : int + The ID of the ticket channel. + author_id : int + The ID of the user who created the ticket. + title : str + The title/subject of the ticket. + + Returns + ------- + Ticket | None + The created ticket object or None if creation failed. + """ + try: + return await self.create( + data={ + "guild_id": guild_id, + "channel_id": channel_id, + "author_id": author_id, + "title": title, + "status": TicketStatus.OPEN, + }, + ) + except Exception as e: + logger.error(f"Failed to create ticket: {e}") + return None + + async def claim_ticket(self, channel_id: int, moderator_id: int) -> Ticket | None: + """ + Claim a ticket for a moderator. + + Parameters + ---------- + channel_id : int + The ID of the ticket channel. + moderator_id : int + The ID of the moderator claiming the ticket. + + Returns + ------- + Ticket | None + The updated ticket object or None if update failed. + """ + try: + return await self.update( + where={"channel_id": channel_id}, + data={ + "claimed_by": moderator_id, + "status": TicketStatus.CLAIMED, + }, + ) + except Exception as e: + logger.error(f"Failed to claim ticket {channel_id}: {e}") + return None + + async def unclaim_ticket(self, channel_id: int) -> Ticket | None: + """ + Unclaim a ticket, resetting it to open status. + + Parameters + ---------- + channel_id : int + The ID of the ticket channel. + + Returns + ------- + Ticket | None + The updated ticket object or None if update failed. + """ + try: + return await self.update( + where={"channel_id": channel_id}, + data={ + "claimed_by": None, + "status": TicketStatus.OPEN, + }, + ) + except Exception as e: + logger.error(f"Failed to unclaim ticket {channel_id}: {e}") + return None + + async def close_ticket(self, channel_id: int) -> Ticket | None: + """ + Close a ticket. + + Parameters + ---------- + channel_id : int + The ID of the ticket channel. + + Returns + ------- + Ticket | None + The updated ticket object or None if update failed. + """ + try: + return await self.update( + where={"channel_id": channel_id}, + data={ + "status": TicketStatus.CLOSED, + "closed_at": datetime.now(UTC), + }, + ) + except Exception as e: + logger.error(f"Failed to close ticket {channel_id}: {e}") + return None + + async def get_ticket_by_channel(self, channel_id: int) -> Ticket | None: + """ + Get a ticket by its channel ID. + + Parameters + ---------- + channel_id : int + The ID of the ticket channel. + + Returns + ------- + Ticket | None + The ticket object or None if not found. + """ + try: + return await self.find_one(where={"channel_id": channel_id}) + except Exception as e: + logger.error(f"Failed to get ticket by channel {channel_id}: {e}") + return None + + async def get_guild_tickets( + self, + guild_id: int, + status: TicketStatus | None = None, + limit: int = 50, + ) -> list[Ticket]: + """ + Get tickets for a guild, optionally filtered by status. + + Parameters + ---------- + guild_id : int + The ID of the guild. + status : TicketStatus | None, optional + Filter by ticket status, by default None (all statuses). + limit : int, optional + Maximum number of tickets to return, by default 50. + + Returns + ------- + list[Ticket] + List of ticket objects. + """ + try: + where_clause = {"guild_id": guild_id} + if status: + where_clause["status"] = status + + return await self.find_many(where=where_clause, take=limit, order={"created_at": "desc"}) or [] + except Exception as e: + logger.error(f"Failed to get guild tickets for {guild_id}: {e}") + return [] + + async def get_user_tickets( + self, + guild_id: int, + user_id: int, + status: TicketStatus | None = None, + ) -> list[Ticket]: + """ + Get tickets created by a specific user. + + Parameters + ---------- + guild_id : int + The ID of the guild. + user_id : int + The ID of the user. + status : TicketStatus | None, optional + Filter by ticket status, by default None (all statuses). + + Returns + ------- + list[Ticket] + List of ticket objects. + """ + try: + where_clause = {"guild_id": guild_id, "author_id": user_id} + if status: + where_clause["status"] = status + + return await self.find_many(where=where_clause, order={"created_at": "desc"}) or [] + except Exception as e: + logger.error(f"Failed to get user tickets for {user_id} in {guild_id}: {e}") + return [] + + async def delete_ticket(self, channel_id: int) -> bool: + """ + Delete a ticket from the database. + + Parameters + ---------- + channel_id : int + The ID of the ticket channel. + + Returns + ------- + bool + True if deletion was successful, False otherwise. + """ + try: + await self.delete(where={"channel_id": channel_id}) + except Exception as e: + logger.error(f"Failed to delete ticket {channel_id}: {e}") + return False + else: + return True diff --git a/tux/ui/views/tickets.py b/tux/ui/views/tickets.py new file mode 100644 index 000000000..e2b4784b7 --- /dev/null +++ b/tux/ui/views/tickets.py @@ -0,0 +1,300 @@ +from contextlib import suppress + +import discord +from loguru import logger + +from tux.bot import Tux +from tux.database.controllers import DatabaseController +from tux.ui.embeds import EmbedCreator, EmbedType +from tux.utils import checks + + +class RequestCloseView(discord.ui.View): + def __init__(self, cog, interaction: discord.Interaction, ticket, author): + super().__init__(timeout=300) + self.cog = cog + self.original_interaction = interaction + self.ticket = ticket + self.author = author + self.message = None + + @discord.ui.button(label="Confirm Close", style=discord.ButtonStyle.danger) + async def confirm_close(self, interaction: discord.Interaction, button: discord.ui.Button): + if interaction.user.id != self.ticket.author_id: + await interaction.response.send_message("Only the ticket author can confirm close.", ephemeral=True) + return + + try: + # Respond to the button interaction first + await interaction.response.send_message("Closing ticket...", ephemeral=True) + + # Create a mock context object for close_ticket method + class MockContext: + def __init__(self, guild, channel, user): + self.guild = guild + self.channel = channel + self.user = user + self.author = user + + mock_ctx = MockContext(interaction.guild, interaction.channel, interaction.user) + + # Call close_ticket without respond_to_interaction parameter + await self.cog.close_ticket(mock_ctx, self.ticket, interaction.channel, self.author) + + # Clean up the confirmation message + if self.message: + with suppress(Exception): + await self.message.delete() + + except Exception as e: + logger.error(f"Error in confirm_close: {e}") + # Try to send an error message if the interaction hasn't been responded to + try: + if not interaction.response.is_done(): + await interaction.response.send_message( + "An error occurred while closing the ticket.", + ephemeral=True, + ) + else: + await interaction.followup.send("An error occurred while closing the ticket.", ephemeral=True) + except Exception: + pass + + async def on_timeout(self): + for item in self.children: + item.disabled = True + if self.message: + with suppress(Exception): + await self.message.edit(content="Request to close ticket has timed out.", view=self) + + +class TicketManagementView(discord.ui.View): + """View for managing tickets: only claim button is visible, all other management is via hidden slash commands.""" + + def __init__(self, bot: Tux, ticket_id: int | None = None) -> None: + super().__init__(timeout=None) + self.bot = bot + self.ticket_id = ticket_id + self.db = DatabaseController() + + async def update_view_for_ticket_state(self, interaction: discord.Interaction, ticket): + """Update the view buttons based on the current ticket state.""" + claim_button = discord.utils.get(self.children, custom_id="ticket_claim") + unclaim_button = discord.utils.get(self.children, custom_id="ticket_unclaim") + + if ticket.claimed_by: + claimed_user = interaction.guild.get_member(ticket.claimed_by) + claimed_name = claimed_user.display_name if claimed_user else "Unknown User" + + claim_button.label = f"Claimed by {claimed_name}" + claim_button.disabled = True + claim_button.style = discord.ButtonStyle.secondary + + if unclaim_button: + unclaim_button.disabled = False + unclaim_button.style = discord.ButtonStyle.danger + else: + claim_button.label = "Claim Ticket" + claim_button.disabled = False + claim_button.style = discord.ButtonStyle.primary + + if unclaim_button: + unclaim_button.disabled = True + unclaim_button.style = discord.ButtonStyle.secondary + + @discord.ui.button(label="Claim Ticket", style=discord.ButtonStyle.primary, emoji="šŸŽ«", custom_id="ticket_claim") + async def claim_ticket(self, interaction: discord.Interaction, button: discord.ui.Button) -> None: + """Claim a ticket for moderation.""" + if not interaction.guild or not isinstance(interaction.user, discord.Member): + await interaction.response.send_message("This command can only be used in a server.", ephemeral=True) + return + + try: + if not await checks.has_permission(interaction, 2, 9): + await interaction.response.send_message( + "You need to be a moderator or higher to claim tickets.", + ephemeral=True, + ) + return + except Exception: + await interaction.response.send_message( + "You need to be a moderator or higher to claim tickets.", + ephemeral=True, + ) + return + + ticket = await self.db.ticket.get_ticket_by_channel(interaction.channel.id) + if not ticket: + await interaction.response.send_message("This channel is not a valid ticket.", ephemeral=True) + return + + if ticket.claimed_by: + claimed_user = interaction.guild.get_member(ticket.claimed_by) + claimed_name = claimed_user.display_name if claimed_user else f"<@{ticket.claimed_by}>" + await interaction.response.send_message( + f"This ticket is already claimed by {claimed_name}.", + ephemeral=True, + ) + return + + updated_ticket = await self.db.ticket.claim_ticket(interaction.channel.id, interaction.user.id) + if not updated_ticket: + await interaction.response.send_message("Failed to claim the ticket.", ephemeral=True) + return + + channel = interaction.channel + if isinstance(channel, discord.TextChannel): + await channel.set_permissions( + interaction.guild.default_role, + view_channel=False, + reason="Ticket claimed - restricting access", + ) + + author = interaction.guild.get_member(ticket.author_id) + if author: + await channel.set_permissions( + author, + view_channel=True, + send_messages=True, + read_messages=True, + reason="Ticket author access", + ) + + await channel.set_permissions( + interaction.user, + view_channel=True, + send_messages=True, + read_messages=True, + manage_messages=True, + reason="Claiming moderator access", + ) + # Update the view to reflect the claimed state + await self.update_view_for_ticket_state(interaction, updated_ticket) + + embed = EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name=interaction.user.display_name, + user_display_avatar=interaction.user.display_avatar.url, + title="Ticket Claimed", + description=f"This ticket has been claimed by {interaction.user.mention}.\n" + f"Only the ticket author and claiming moderator can now access this channel.", + ) + + await interaction.response.edit_message(embed=embed, view=self) + + @discord.ui.button( + label="Unclaim Ticket", + style=discord.ButtonStyle.secondary, + emoji="šŸ”“", + custom_id="ticket_unclaim", + disabled=True, + ) + async def unclaim_ticket(self, interaction: discord.Interaction, button: discord.ui.Button) -> None: + """Unclaim a ticket, making it available for other moderators.""" + # Validate basic permissions and context + validation_error = await self._validate_unclaim_context(interaction) + if validation_error: + await interaction.response.send_message(validation_error, ephemeral=True) + return + + # Validate ticket state and unclaim permissions + ticket = await self.db.ticket.get_ticket_by_channel(interaction.channel.id) + permission_error = await self._validate_unclaim_permissions(interaction, ticket) + if permission_error: + await interaction.response.send_message(permission_error, ephemeral=True) + return + + # Perform unclaim operation + success = await self._perform_unclaim_operation(interaction, ticket) + if not success: + return + + async def _validate_unclaim_context(self, interaction: discord.Interaction) -> str | None: + """Validate basic context for unclaim operation.""" + if not interaction.guild or not isinstance(interaction.user, discord.Member): + return "This command can only be used in a server." + + try: + has_permission = await checks.has_permission(interaction, 2, 9) + except Exception: + has_permission = False + + if not has_permission: + return "You need to be a moderator or higher to unclaim tickets." + + return None + + async def _validate_unclaim_permissions(self, interaction: discord.Interaction, ticket) -> str | None: + """Validate ticket-specific unclaim permissions.""" + if not ticket: + return "This channel is not a valid ticket." + if not ticket.claimed_by: + return "This ticket is not currently claimed." + + # Check if user can unclaim this specific ticket + can_unclaim = ticket.claimed_by == interaction.user.id + if not can_unclaim: + try: + can_unclaim = await checks.has_permission(interaction, 3, 9) + except Exception: + can_unclaim = False + + if not can_unclaim: + claimed_user = interaction.guild.get_member(ticket.claimed_by) + claimed_name = claimed_user.display_name if claimed_user else f"<@{ticket.claimed_by}>" + return ( + "You can only unclaim tickets you have claimed yourself, or you need higher permissions. " + f"This ticket is claimed by {claimed_name}." + ) + + return None + + async def _perform_unclaim_operation(self, interaction: discord.Interaction, ticket) -> bool: + """Perform the actual unclaim operation.""" + updated_ticket = await self.db.ticket.unclaim_ticket(interaction.channel.id) + if not updated_ticket: + await interaction.response.send_message("Failed to unclaim the ticket.", ephemeral=True) + return False + + # Restore channel permissions + await self._restore_staff_permissions(interaction, ticket) + + # Update view and send response + await self.update_view_for_ticket_state(interaction, updated_ticket) + embed = EmbedCreator.create_embed( + bot=self.bot, + embed_type=EmbedType.INFO, + user_name=interaction.user.display_name, + user_display_avatar=interaction.user.display_avatar.url, + title="Ticket Unclaimed", + description=( + f"This ticket has been unclaimed by {interaction.user.mention}.\n" + "The ticket is now available for other staff members to claim." + ), + ) + await interaction.response.edit_message(embed=embed, view=self) + return True + + async def _restore_staff_permissions(self, interaction: discord.Interaction, ticket): + """Restore staff access to the unclaimed ticket.""" + channel = interaction.channel + if not isinstance(channel, discord.TextChannel): + return + + guild_config = await checks.fetch_guild_config(interaction.guild.id) + staff_role_ids = {guild_config.get(f"perm_level_{lvl}_role_id") for lvl in range(2, 8)} - {None} + + for role in (r for r in interaction.guild.roles if r.id in staff_role_ids): + await channel.set_permissions( + role, + view_channel=True, + send_messages=True, + read_messages=True, + manage_messages=True, + reason="Ticket unclaimed - restoring staff access", + ) + + author = interaction.guild.get_member(ticket.author_id) + if author: + await channel.set_permissions(author, view_channel=True, send_messages=True, read_messages=True) From 73efe95f293c1ee938afa5f1e2f6cf8faaa4fb12 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:04:49 +0100 Subject: [PATCH 02/13] Apply suggestion from @sourcery-ai[bot] Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- tux/cogs/moderation/tickets.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py index 452a3db6a..713a49a6c 100644 --- a/tux/cogs/moderation/tickets.py +++ b/tux/cogs/moderation/tickets.py @@ -50,8 +50,7 @@ async def show_transcript(self, interaction: discord.Interaction, ticket_id: int await interaction.followup.send("Transcript system not available.", ephemeral=True) return - channel = interaction.guild.get_channel(ticket.channel_id) - if channel: + if channel := interaction.guild.get_channel(ticket.channel_id): try: messages = [msg async for msg in channel.history(limit=100, oldest_first=True)] success = await ticket_log_cog.log_transcript(interaction.guild, channel, ticket, messages) From e63159de3afb2963f40ccb81d64163e1c7c6b108 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:05:00 +0100 Subject: [PATCH 03/13] Apply suggestion from @sourcery-ai[bot] Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- tux/cogs/moderation/tickets.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py index 713a49a6c..245ed44e3 100644 --- a/tux/cogs/moderation/tickets.py +++ b/tux/cogs/moderation/tickets.py @@ -105,8 +105,9 @@ def _log_ticket_closed_event(self, ctx, channel_id: int) -> None: ticket_log_cog = self.bot.get_cog("TicketLog") if not ticket_log_cog: return - closer_id = getattr(getattr(ctx, "author", None), "id", None) or getattr(getattr(ctx, "user", None), "id", None) - if closer_id: + if closer_id := getattr( + getattr(ctx, "author", None), "id", None + ) or getattr(getattr(ctx, "user", None), "id", None): ticket_log_cog.add_ticket_event( channel_id, "TICKET_CLOSED", From b78aebe2dd523536d8f5ce6a1d8835d513f76d24 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:05:18 +0100 Subject: [PATCH 04/13] Apply suggestion from @sourcery-ai[bot] Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- tux/cogs/moderation/tickets.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py index 245ed44e3..a46531820 100644 --- a/tux/cogs/moderation/tickets.py +++ b/tux/cogs/moderation/tickets.py @@ -139,8 +139,7 @@ async def _build_closure_overwrites(self, ctx, author, ticket) -> dict: claimed_by = getattr(ticket, "claimed_by", None) if not claimed_by: return overwrites - claimed_member = ctx.guild.get_member(claimed_by) - if claimed_member: + if claimed_member := ctx.guild.get_member(claimed_by): overwrites[claimed_member] = discord.PermissionOverwrite( view_channel=True, send_messages=False, From 1f3674539114b88c24f2db56ea12c112225f8b43 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:05:35 +0100 Subject: [PATCH 05/13] Apply suggestion from @sourcery-ai[bot] Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- tux/cogs/moderation/tickets.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py index a46531820..deeff243a 100644 --- a/tux/cogs/moderation/tickets.py +++ b/tux/cogs/moderation/tickets.py @@ -292,9 +292,7 @@ async def create(self, ctx: commands.Context, *, title: str | None = None): """Create a new support ticket (hybrid command).""" user, guild, send = self._extract_ctx(ctx) self._maybe_delete_invocation(ctx) - # Validate context & title - error = self._validate_creation_prereqs(guild, user, title) - if error: + if error := self._validate_creation_prereqs(guild, user, title): await send(error, ephemeral=True) return title = self._normalize_title(title) From 1d56e559dde5f203ab0f47f1f9caad1f862052a5 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:05:57 +0100 Subject: [PATCH 06/13] Apply suggestion from @sourcery-ai[bot] Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- tux/cogs/moderation/tickets.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py index deeff243a..a645429bc 100644 --- a/tux/cogs/moderation/tickets.py +++ b/tux/cogs/moderation/tickets.py @@ -825,9 +825,7 @@ async def add(self, ctx: commands.Context, target: str): reason="Added to ticket by staff", ) - # Track the event - ticket_log_cog = self.bot.get_cog("TicketLog") - if ticket_log_cog: + if ticket_log_cog := self.bot.get_cog("TicketLog"): actor_id = ctx.author.id if hasattr(ctx, "author") else ctx.user.id event_type = "ROLE_ADDED" if isinstance(target_obj, discord.Role) else "USER_ADDED" target_name = getattr(target_obj, "name", str(target_obj)) From 47ed272c7b39456f170d846d4891c20648b3a493 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:06:28 +0100 Subject: [PATCH 07/13] Apply suggestion from @sourcery-ai[bot] Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- tux/cogs/moderation/tickets.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py index a645429bc..b67652551 100644 --- a/tux/cogs/moderation/tickets.py +++ b/tux/cogs/moderation/tickets.py @@ -859,8 +859,7 @@ async def remove(self, ctx: commands.Context, target: str): async def _log_removal_event(self, ctx, target_obj): """Log the removal event to ticket log.""" - ticket_log_cog = self.bot.get_cog("TicketLog") - if ticket_log_cog: + if ticket_log_cog := self.bot.get_cog("TicketLog"): actor_id = ctx.author.id if hasattr(ctx, "author") else ctx.user.id event_type = "ROLE_REMOVED" if isinstance(target_obj, discord.Role) else "USER_REMOVED" target_name = getattr(target_obj, "name", str(target_obj)) From 043d697d9da2d66583aef3c79abc42ef69f74b1c Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:06:47 +0100 Subject: [PATCH 08/13] Apply suggestion from @sourcery-ai[bot] Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- tux/cogs/moderation/tickets.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py index b67652551..242aad578 100644 --- a/tux/cogs/moderation/tickets.py +++ b/tux/cogs/moderation/tickets.py @@ -934,7 +934,7 @@ def _base(title: str): inline=True, ) total_tickets = len(tickets) - total_claimed = sum(1 for t in tickets if t.claimed_by) + total_claimed = sum(bool(t.claimed_by) total_closed = sum(1 for t in tickets if t.status == TicketStatus.CLOSED) embed.add_field( name="šŸ“ˆ Server Summary", From a2adbb5e4ca51d1790ac7b573c987abf622cf514 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:07:42 +0100 Subject: [PATCH 09/13] Apply suggestion from @sourcery-ai[bot] Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- tux/ui/views/tickets.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tux/ui/views/tickets.py b/tux/ui/views/tickets.py index e2b4784b7..6b34300b5 100644 --- a/tux/ui/views/tickets.py +++ b/tux/ui/views/tickets.py @@ -295,6 +295,5 @@ async def _restore_staff_permissions(self, interaction: discord.Interaction, tic reason="Ticket unclaimed - restoring staff access", ) - author = interaction.guild.get_member(ticket.author_id) - if author: + if author := interaction.guild.get_member(ticket.author_id): await channel.set_permissions(author, view_channel=True, send_messages=True, read_messages=True) From 5be115e5084c4ee75472f30503daf983abc57951 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:07:52 +0100 Subject: [PATCH 10/13] Apply suggestion from @sourcery-ai[bot] Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- tux/ui/views/tickets.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tux/ui/views/tickets.py b/tux/ui/views/tickets.py index 6b34300b5..57460c41d 100644 --- a/tux/ui/views/tickets.py +++ b/tux/ui/views/tickets.py @@ -150,8 +150,7 @@ async def claim_ticket(self, interaction: discord.Interaction, button: discord.u reason="Ticket claimed - restricting access", ) - author = interaction.guild.get_member(ticket.author_id) - if author: + if author := interaction.guild.get_member(ticket.author_id): await channel.set_permissions( author, view_channel=True, From 89d3ca493e6a4a7ead7cbba03baa3677dc1d8230 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:08:20 +0100 Subject: [PATCH 11/13] Apply suggestion from @sourcery-ai[bot] Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- tux/cogs/moderation/tickets.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py index 242aad578..fe47fe5e0 100644 --- a/tux/cogs/moderation/tickets.py +++ b/tux/cogs/moderation/tickets.py @@ -935,7 +935,7 @@ def _base(title: str): ) total_tickets = len(tickets) total_claimed = sum(bool(t.claimed_by) - total_closed = sum(1 for t in tickets if t.status == TicketStatus.CLOSED) + total_closed = sum(bool(t.status == TicketStatus.CLOSED) embed.add_field( name="šŸ“ˆ Server Summary", value=f"Total: {total_tickets}\nClaimed: {total_claimed}\nClosed: {total_closed}", From 515a3c9caf73d89d81c99a213d98eb640f8d29e2 Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Fri, 22 Aug 2025 13:54:14 +0100 Subject: [PATCH 12/13] fix(command): fix sourcery issues --- prisma/schema/guild/config.prisma | 1 + tux/cogs/moderation/tickets.py | 19 ++- tux/cogs/utility/ticket_log.py | 206 +++++++++++++---------- tux/cogs/utility/ticket_log_config.py | 31 +--- tux/database/controllers/guild_config.py | 22 +++ tux/ui/views/tickets.py | 4 +- 6 files changed, 160 insertions(+), 123 deletions(-) diff --git a/prisma/schema/guild/config.prisma b/prisma/schema/guild/config.prisma index 8c08a0c27..9f8b10770 100644 --- a/prisma/schema/guild/config.prisma +++ b/prisma/schema/guild/config.prisma @@ -6,6 +6,7 @@ model GuildConfig { private_log_id BigInt? report_log_id BigInt? dev_log_id BigInt? + ticket_log_id BigInt? jail_channel_id BigInt? general_channel_id BigInt? starboard_channel_id BigInt? diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py index fe47fe5e0..7db6db465 100644 --- a/tux/cogs/moderation/tickets.py +++ b/tux/cogs/moderation/tickets.py @@ -106,7 +106,9 @@ def _log_ticket_closed_event(self, ctx, channel_id: int) -> None: if not ticket_log_cog: return if closer_id := getattr( - getattr(ctx, "author", None), "id", None + getattr(ctx, "author", None), + "id", + None, ) or getattr(getattr(ctx, "user", None), "id", None): ticket_log_cog.add_ticket_event( channel_id, @@ -181,7 +183,7 @@ async def _attempt_transcript_log(self, ctx, channel: discord.TextChannel, ticke try: messages = [msg async for msg in channel.history(oldest_first=True)] success = await ticket_log_cog.log_transcript(ctx.guild, channel, ticket, messages) - log_func = logger.info if success else logger.warning + log_func = logger.debug if success else logger.warning log_func( f"{'Successfully' if success else 'Failed to'} log transcript for ticket {ticket.ticket_id}", ) @@ -213,7 +215,7 @@ def _resolve_target(self, ctx, target): Returns discord.Member | discord.Role | None """ - if isinstance(target, discord.Member | discord.Role): # UP038 + if isinstance(target, discord.Member | discord.Role): return target s = str(target).strip() @@ -296,9 +298,10 @@ async def create(self, ctx: commands.Context, *, title: str | None = None): await send(error, ephemeral=True) return title = self._normalize_title(title) - - if hasattr(ctx, "response") and hasattr(ctx.response, "defer"): + is_interaction = hasattr(ctx, "response") and hasattr(ctx.response, "defer") + if is_interaction: await ctx.response.defer(ephemeral=True) + send = ctx.followup.send try: await self._ensure_guild_config(guild.id) @@ -805,7 +808,7 @@ async def add(self, ctx: commands.Context, target: str): # Support @role mention if target.startswith("<@&") and target.endswith(">"): role_id = target[3:-1] - target_obj = ctx.guild.get_role(int(role_id)) if role_id.isdigit() else None # SIM108 + target_obj = ctx.guild.get_role(int(role_id)) if role_id.isdigit() else None else: target_obj = self._resolve_target(ctx, target) @@ -934,8 +937,8 @@ def _base(title: str): inline=True, ) total_tickets = len(tickets) - total_claimed = sum(bool(t.claimed_by) - total_closed = sum(bool(t.status == TicketStatus.CLOSED) + total_claimed = sum(bool(t.claimed_by) for t in tickets) + total_closed = sum(bool(t.status == TicketStatus.CLOSED) for t in tickets) embed.add_field( name="šŸ“ˆ Server Summary", value=f"Total: {total_tickets}\nClaimed: {total_claimed}\nClosed: {total_closed}", diff --git a/tux/cogs/utility/ticket_log.py b/tux/cogs/utility/ticket_log.py index 534e493eb..8c695bd37 100644 --- a/tux/cogs/utility/ticket_log.py +++ b/tux/cogs/utility/ticket_log.py @@ -40,93 +40,125 @@ def clear_ticket_events(self, channel_id: int): async def log_transcript(self, guild, channel, ticket, messages): """Log the transcript as a plain text file with enhanced event detection.""" - log_cog = self.bot.get_cog("TicketLogConfig") - log_channel_id = None - if log_cog: - log_channel_id = log_cog.get_log_channel_id(guild.id) - - if log_channel_id: - log_channel = guild.get_channel(log_channel_id) - if log_channel: - # Get tracked events for this channel - tracked_events = self.get_ticket_events(channel.id) - - # Combine messages and tracked events, then sort by timestamp - all_entries = [{"timestamp": msg.created_at, "type": "message", "data": msg} for msg in messages] + [ - {"timestamp": event["timestamp"], "type": "event", "data": event} for event in tracked_events - ] - - # Sort by timestamp - all_entries.sort(key=lambda x: x["timestamp"]) - - lines = [] - for entry in all_entries: - timestamp_str = entry["timestamp"].strftime("%Y-%m-%d %H:%M") - - if entry["type"] == "message": - msg = entry["data"] - author = f"{msg.author.display_name} ({msg.author.id})" - content = msg.content - - # Enhanced message content analysis - tags = self._analyze_message_content(content, msg) - tag_str = "".join(f"[{tag}] " for tag in tags) - - lines.append(f"[{timestamp_str}] {author}: {tag_str}{content}") - - elif entry["type"] == "event": - event = entry["data"] - user = guild.get_member(event["user_id"]) - user_name = user.display_name if user else f"Unknown ({event['user_id']})" - - event_msg = self._format_event_message(event, user_name) - lines.append(f"[{timestamp_str}] SYSTEM: [EVENT] {event_msg}") - - # Add ticket metadata at the beginning - metadata_lines = [ - f"=== TICKET #{ticket.ticket_id} TRANSCRIPT ===", - f"Title: {ticket.title}", - f"Author: {guild.get_member(ticket.author_id).display_name if guild.get_member(ticket.author_id) else f'Unknown ({ticket.author_id})'}", - f"Created: {ticket.created_at.strftime('%Y-%m-%d %H:%M:%S')} UTC", - f"Channel: #{channel.name} ({channel.id})", - f"Status: {ticket.status.value if hasattr(ticket.status, 'value') else ticket.status}", - f"Claimed by: {guild.get_member(ticket.claimed_by).display_name if ticket.claimed_by and guild.get_member(ticket.claimed_by) else 'None'}", - "=" * 50, - "", - ] - - transcript_text = "\n".join(metadata_lines + lines) - file = discord.File(io.BytesIO(transcript_text.encode()), filename=f"transcript_{channel.id}.txt") - - # Create summary embed - embed = discord.Embed( - title=f"Ticket #{ticket.ticket_id} Transcript", - description=f"**Title:** {ticket.title}\n**Author:** {guild.get_member(ticket.author_id).mention if guild.get_member(ticket.author_id) else 'Unknown User'}", - color=0x2ECC71 if ticket.status == TicketStatus.CLOSED else 0xE74C3C, - timestamp=datetime.now(UTC), - ) - embed.add_field(name="Messages", value=str(len(messages)), inline=True) - embed.add_field(name="Events", value=str(len(tracked_events)), inline=True) - embed.add_field( - name="Status", - value=ticket.status.value if hasattr(ticket.status, "value") else str(ticket.status), - inline=True, - ) - - if ticket.claimed_by: - claimed_user = guild.get_member(ticket.claimed_by) - embed.add_field( - name="Claimed By", - value=claimed_user.mention if claimed_user else f"Unknown ({ticket.claimed_by})", - inline=True, - ) - - await log_channel.send(embed=embed, file=file) - - # Clear tracked events after logging - self.clear_ticket_events(channel.id) - return True - return False + # Early return if no log configuration + if not (log_cog := self.bot.get_cog("TicketLogConfig")): + return False + + if not (log_channel_id := await log_cog.get_log_channel_id(guild.id)): + return False + + if not (log_channel := guild.get_channel(log_channel_id)): + return False + + # Generate transcript content + transcript_content = self._generate_transcript_content(guild, channel, ticket, messages) + + # Create and send the transcript + file = discord.File(io.BytesIO(transcript_content.encode()), filename=f"transcript_{channel.id}.txt") + embed = self._create_transcript_embed(guild, ticket, messages) + + await log_channel.send(embed=embed, file=file) + self.clear_ticket_events(channel.id) + return True + + def _generate_transcript_content(self, guild, channel, ticket, messages): + """Generate the complete transcript content.""" + tracked_events = self.get_ticket_events(channel.id) + all_entries = self._combine_messages_and_events(messages, tracked_events) + lines = self._format_transcript_lines(guild, all_entries) + metadata_lines = self._create_metadata_lines(guild, ticket, channel) + return "\n".join(metadata_lines + lines) + + def _combine_messages_and_events(self, messages, tracked_events): + """Combine messages and events, sorted by timestamp.""" + all_entries = [{"timestamp": msg.created_at, "type": "message", "data": msg} for msg in messages] + [ + {"timestamp": event["timestamp"], "type": "event", "data": event} for event in tracked_events + ] + return sorted(all_entries, key=lambda x: x["timestamp"]) + + def _format_transcript_lines(self, guild, all_entries): + """Format all entries into transcript lines.""" + lines = [] + for entry in all_entries: + timestamp_str = entry["timestamp"].strftime("%Y-%m-%d %H:%M") + + if entry["type"] == "message": + line = self._format_message_line(timestamp_str, entry["data"]) + else: # event + line = self._format_event_line(timestamp_str, guild, entry["data"]) + + lines.append(line) + return lines + + def _format_message_line(self, timestamp_str, msg): + """Format a message into a transcript line.""" + author = f"{msg.author.display_name} ({msg.author.id})" + tags = self._analyze_message_content(msg.content, msg) + tag_str = "".join(f"[{tag}] " for tag in tags) + return f"[{timestamp_str}] {author}: {tag_str}{msg.content}" + + def _format_event_line(self, timestamp_str, guild, event): + """Format an event into a transcript line.""" + if user := guild.get_member(event["user_id"]): + user_name = user.display_name + else: + user_name = f"Unknown ({event['user_id']})" + + event_msg = self._format_event_message(event, user_name) + return f"[{timestamp_str}] SYSTEM: [EVENT] {event_msg}" + + def _create_metadata_lines(self, guild, ticket, channel): + """Create the metadata header for the transcript.""" + if author := guild.get_member(ticket.author_id): + author_name = author.display_name + else: + author_name = f"Unknown ({ticket.author_id})" + + if ticket.claimed_by and (claimed_user := guild.get_member(ticket.claimed_by)): + claimed_by = claimed_user.display_name + else: + claimed_by = "None" + + return [ + f"=== TICKET #{ticket.ticket_id} TRANSCRIPT ===", + f"Title: {ticket.title}", + f"Author: {author_name}", + f"Created: {ticket.created_at.strftime('%Y-%m-%d %H:%M:%S')} UTC", + f"Channel: #{channel.name} ({channel.id})", + f"Status: {ticket.status.value if hasattr(ticket.status, 'value') else ticket.status}", + f"Claimed by: {claimed_by}", + "=" * 50, + "", + ] + + def _create_transcript_embed(self, guild, ticket, messages): + """Create the summary embed for the transcript.""" + author_display = author.mention if (author := guild.get_member(ticket.author_id)) else "Unknown User" + + embed = discord.Embed( + title=f"Ticket #{ticket.ticket_id} Transcript", + description=f"**Title:** {ticket.title}\n**Author:** {author_display}", + color=0x2ECC71 if ticket.status == TicketStatus.CLOSED else 0xE74C3C, + timestamp=datetime.now(UTC), + ) + + tracked_events = self.get_ticket_events(ticket.channel_id) if hasattr(ticket, "channel_id") else [] + embed.add_field(name="Messages", value=str(len(messages)), inline=True) + embed.add_field(name="Events", value=str(len(tracked_events)), inline=True) + embed.add_field( + name="Status", + value=ticket.status.value if hasattr(ticket.status, "value") else str(ticket.status), + inline=True, + ) + + if ticket.claimed_by and (claimed_user := guild.get_member(ticket.claimed_by)): + embed.add_field( + name="Claimed By", + value=claimed_user.mention, + inline=True, + ) + + return embed def _analyze_message_content(self, content: str, msg: discord.Message) -> list: """Analyze message content for ticket-related events.""" diff --git a/tux/cogs/utility/ticket_log_config.py b/tux/cogs/utility/ticket_log_config.py index 4edbdc8dc..74d5b91e3 100644 --- a/tux/cogs/utility/ticket_log_config.py +++ b/tux/cogs/utility/ticket_log_config.py @@ -1,11 +1,8 @@ -from pathlib import Path - import discord from discord.ext import commands from tux.bot import Tux - -CONFIG_PATH = (Path(__file__).parent / "../../assets/embeds/ticket_log_channel.txt").resolve() +from tux.database.controllers import DatabaseController class TicketLogConfig(commands.Cog): @@ -13,23 +10,7 @@ class TicketLogConfig(commands.Cog): def __init__(self, bot: Tux): self.bot = bot - self._load_config() - - def _load_config(self): - self.config = {} - if CONFIG_PATH.exists(): - with CONFIG_PATH.open() as f: - for raw_line in f: - line = raw_line.strip() - if not line or ":" not in line: - continue - gid, cid = line.split(":", 1) - self.config[gid] = int(cid) - - def _save_config(self): - CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) - with CONFIG_PATH.open("w") as f: - f.writelines(f"{gid}:{cid}\n" for gid, cid in self.config.items()) + self.db = DatabaseController() @discord.app_commands.command( name="set_ticket_log_channel", @@ -39,12 +20,12 @@ def _save_config(self): @discord.app_commands.checks.has_permissions(administrator=True) async def set_ticket_log_channel(self, interaction: discord.Interaction, channel: discord.TextChannel): """Set the channel where ticket logs will be sent (slash command).""" - self.config[str(interaction.guild.id)] = channel.id - self._save_config() + await self.db.guild_config.update_ticket_log_id(interaction.guild.id, channel.id) await interaction.response.send_message(f"Ticket log channel set to {channel.mention}", ephemeral=True) - def get_log_channel_id(self, guild_id: int): - return self.config.get(str(guild_id)) + async def get_log_channel_id(self, guild_id: int): + """Get the ticket log channel ID for a guild.""" + return await self.db.guild_config.get_ticket_log_id(guild_id) async def setup(bot: Tux): diff --git a/tux/database/controllers/guild_config.py b/tux/database/controllers/guild_config.py index 5acda6552..2d4133f63 100644 --- a/tux/database/controllers/guild_config.py +++ b/tux/database/controllers/guild_config.py @@ -46,6 +46,7 @@ async def get_log_channel(self, guild_id: int, log_type: str) -> int | None: "private": "private_log_id", "report": "report_log_id", "dev": "dev_log_id", + "ticket": "ticket_log_id", } return await self.get_guild_config_field_value(guild_id, log_channel_ids[log_type]) @@ -129,6 +130,9 @@ async def get_report_log_id(self, guild_id: int) -> int | None: async def get_dev_log_id(self, guild_id: int) -> int | None: return await self.get_guild_config_field_value(guild_id, "dev_log_id") + async def get_ticket_log_id(self, guild_id: int) -> int | None: + return await self.get_guild_config_field_value(guild_id, "ticket_log_id") + async def get_jail_channel_id(self, guild_id: int) -> int | None: return await self.get_guild_config_field_value(guild_id, "jail_channel_id") @@ -300,6 +304,24 @@ async def update_dev_log_id( }, ) + async def update_ticket_log_id( + self, + guild_id: int, + ticket_log_id: int, + ) -> Any: + await self.ensure_guild_exists(guild_id) + + return await self.table.upsert( + where={"guild_id": guild_id}, + data={ + "create": { + "guild_id": guild_id, + "ticket_log_id": ticket_log_id, + }, + "update": {"ticket_log_id": ticket_log_id}, + }, + ) + async def update_jail_channel_id( self, guild_id: int, diff --git a/tux/ui/views/tickets.py b/tux/ui/views/tickets.py index 57460c41d..c5443e75f 100644 --- a/tux/ui/views/tickets.py +++ b/tux/ui/views/tickets.py @@ -49,7 +49,7 @@ def __init__(self, guild, channel, user): except Exception as e: logger.error(f"Error in confirm_close: {e}") # Try to send an error message if the interaction hasn't been responded to - try: + with suppress(Exception): if not interaction.response.is_done(): await interaction.response.send_message( "An error occurred while closing the ticket.", @@ -57,8 +57,6 @@ def __init__(self, guild, channel, user): ) else: await interaction.followup.send("An error occurred while closing the ticket.", ephemeral=True) - except Exception: - pass async def on_timeout(self): for item in self.children: From e47f79caea5b14880abc3035c9d1fd7c8ffcb74d Mon Sep 17 00:00:00 2001 From: nomnomshark41 Date: Sun, 24 Aug 2025 20:29:57 +0100 Subject: [PATCH 13/13] fix(command): fix sourcery issues again --- tux/cogs/moderation/tickets.py | 63 ++++++++++++++++++---------------- 1 file changed, 34 insertions(+), 29 deletions(-) diff --git a/tux/cogs/moderation/tickets.py b/tux/cogs/moderation/tickets.py index 7db6db465..bfa5773b6 100644 --- a/tux/cogs/moderation/tickets.py +++ b/tux/cogs/moderation/tickets.py @@ -45,8 +45,7 @@ async def show_transcript(self, interaction: discord.Interaction, ticket_id: int await interaction.followup.send("Ticket not found.", ephemeral=True) return - ticket_log_cog = self.bot.get_cog("TicketLog") - if not ticket_log_cog: + if not (ticket_log_cog := self.bot.get_cog("TicketLog")): await interaction.followup.send("Transcript system not available.", ephemeral=True) return @@ -102,8 +101,7 @@ async def close_ticket(self, ctx, ticket, channel, author, respond_to_interactio await self._delete_channel_and_category(channel) def _log_ticket_closed_event(self, ctx, channel_id: int) -> None: - ticket_log_cog = self.bot.get_cog("TicketLog") - if not ticket_log_cog: + if not (ticket_log_cog := self.bot.get_cog("TicketLog")): return if closer_id := getattr( getattr(ctx, "author", None), @@ -177,8 +175,7 @@ async def _send_closure_embed(self, channel: discord.TextChannel) -> None: logger.error(f"Failed sending closure embed in {channel.id}: {e}") async def _attempt_transcript_log(self, ctx, channel: discord.TextChannel, ticket) -> None: - ticket_log_cog = self.bot.get_cog("TicketLog") - if not ticket_log_cog: + if not (ticket_log_cog := self.bot.get_cog("TicketLog")): return try: messages = [msg async for msg in channel.history(oldest_first=True)] @@ -210,6 +207,21 @@ async def _ephemeral_msg(self, ctx, content): with suppress(Exception): await msg.delete() + def _parse_role_mention(self, guild: discord.Guild, target: str) -> discord.Role | None: + """Parse a role mention and return the role object if found. + + Args: + guild: The guild to search for the role + target: The target string (could be role mention, ID, or name) + + Returns: + discord.Role | None: The role if found, None otherwise + """ + if target.startswith("<@&") and target.endswith(">"): + role_id = target[3:-1] + return guild.get_role(int(role_id)) if role_id.isdigit() else None + return None + def _resolve_target(self, ctx, target): """Resolve a user or role from a string or object. @@ -218,30 +230,30 @@ def _resolve_target(self, ctx, target): if isinstance(target, discord.Member | discord.Role): return target + # First try role mention parsing + if role := self._parse_role_mention(ctx.guild, target): + return role + s = str(target).strip() guild = ctx.guild id_candidates: set[int] = set() if s.isdigit(): id_candidates.add(int(s)) for pattern in (r"<@!?([0-9]+)>", r"<@&([0-9]+)>"): - match = re.fullmatch(pattern, s) - if match: - id_candidates.add(int(match.group(1))) + if match := re.fullmatch(pattern, s): + id_candidates.add(int(match[1])) for cid in id_candidates: - member = guild.get_member(cid) - if member: + if member := guild.get_member(cid): return member - role = guild.get_role(cid) - if role: + if role := guild.get_role(cid): return role lowered = s.lower() - member = discord.utils.find( + if member := discord.utils.find( lambda m: m.name.lower() == lowered or m.display_name.lower() == lowered, guild.members, - ) - if member: + ): return member return discord.utils.find(lambda r: r.name.lower() == lowered, guild.roles) @@ -298,8 +310,7 @@ async def create(self, ctx: commands.Context, *, title: str | None = None): await send(error, ephemeral=True) return title = self._normalize_title(title) - is_interaction = hasattr(ctx, "response") and hasattr(ctx.response, "defer") - if is_interaction: + if hasattr(ctx, "response") and hasattr(ctx.response, "defer"): await ctx.response.defer(ephemeral=True) send = ctx.followup.send @@ -805,12 +816,7 @@ async def add(self, ctx: commands.Context, target: str): await ctx.message.delete() return - # Support @role mention - if target.startswith("<@&") and target.endswith(">"): - role_id = target[3:-1] - target_obj = ctx.guild.get_role(int(role_id)) if role_id.isdigit() else None - else: - target_obj = self._resolve_target(ctx, target) + target_obj = self._resolve_target(ctx, target) if not target_obj: await self._ephemeral_msg(ctx, f"Could not find user or role for '{target}'.") @@ -937,8 +943,8 @@ def _base(title: str): inline=True, ) total_tickets = len(tickets) - total_claimed = sum(bool(t.claimed_by) for t in tickets) - total_closed = sum(bool(t.status == TicketStatus.CLOSED) for t in tickets) + total_claimed = sum(t.claimed_by for t in tickets) + total_closed = sum(t.status == TicketStatus.CLOSED for t in tickets) embed.add_field( name="šŸ“ˆ Server Summary", value=f"Total: {total_tickets}\nClaimed: {total_claimed}\nClosed: {total_closed}", @@ -1015,9 +1021,8 @@ async def _validate_ticket_channel(self, ctx): return True, ticket def _parse_target(self, ctx, target: str): - if target.startswith("<@&") and target.endswith(">"): - role_id = target.strip("<@&>") - return ctx.guild.get_role(int(role_id)) if role_id.isdigit() else None + if role := self._parse_role_mention(ctx.guild, target): + return role return self._resolve_target(ctx, target) async def _maybe_delete_command_message(self, ctx):