slipstream/discordbot
slipstream/discordbot
1
1
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Refactor logging cog to use layout views

Browse Source
This commit is contained in:
Codex 2025-06-06 05:43:23 +00:00 committed by Slipstream
parent fd3fb3fa90
commit 6549537d7d
Signed by: slipstream
GPG Key ID: 13E498CE010AC6FD
1 changed files with 124 additions and 232 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

356
cogs/logging_cog.py
View File

@ -2,6 +2,7 @@ import discord
from discord.ext import commands, tasks from discord.ext import commands, tasks
from discord import AllowedMentions, ui from discord import AllowedMentions, ui
import datetime import datetime
import difflib
import asyncio import asyncio
import aiohttp # Added for webhook sending import aiohttp # Added for webhook sending
import logging # Use logging instead of print import logging # Use logging instead of print
@ -16,6 +17,12 @@ except ImportError:
log = logging.getLogger(__name__) # Setup logger for this cog log = logging.getLogger(__name__) # Setup logger for this cog
# Mapping for consistent event styling
EVENT_STYLES = {
"message_edit": ("✏️", discord.Color.light_grey()),
"message_delete": ("🗑️", discord.Color.dark_grey()),
}
# Define all possible event keys for toggling # Define all possible event keys for toggling
# Keep this list updated if new loggable events are added # Keep this list updated if new loggable events are added
ALL_EVENT_KEYS = sorted( ALL_EVENT_KEYS = sorted(
@ -82,9 +89,7 @@ class LoggingCog(commands.Cog):
def __init__(self, bot: commands.Bot): def __init__(self, bot: commands.Bot):
self.bot = bot self.bot = bot
self.session: Optional[aiohttp.ClientSession] = None # Session for webhooks self.session: Optional[aiohttp.ClientSession] = None # Session for webhooks
self.last_audit_log_ids: dict[int, Optional[int]] = ( self.last_audit_log_ids: dict[int, Optional[int]] = {} # Store last ID per guild
{}
) # Store last ID per guild
# Start the audit log poller task if the bot is ready, otherwise wait # Start the audit log poller task if the bot is ready, otherwise wait
if bot.is_ready(): if bot.is_ready():
asyncio.create_task(self.initialize_cog()) # Use async init helper asyncio.create_task(self.initialize_cog()) # Use async init helper
@ -110,9 +115,7 @@ class LoggingCog(commands.Cog):
self.add_item(self.container) self.add_item(self.container)
if author is not None: if author is not None:
header = ui.Section( header = ui.Section(accessory=ui.Thumbnail(media=author.display_avatar.url))
accessory=ui.Thumbnail(media=author.display_avatar.url)
)
else: else:
header = ui.Section() header = ui.Section()
@ -120,13 +123,13 @@ class LoggingCog(commands.Cog):
if description: if description:
header.add_item(ui.TextDisplay(description)) header.add_item(ui.TextDisplay(description))
self.container.add_item(header) self.container.add_item(header)
self.container.add_item( self.container.add_item(ui.Separator(spacing=discord.SeparatorSpacing.small))
ui.Separator(spacing=discord.SeparatorSpacing.small)
)
footer_text = footer or f"Bot ID: {bot.user.id}" + ( timestamp = discord.utils.format_dt(datetime.datetime.utcnow(), style="f")
f" | User ID: {author.id}" if author else "" parts = [timestamp, footer or f"Bot ID: {bot.user.id}"]
) if author:
parts.append(f"User ID: {author.id}")
footer_text = " | ".join(parts)
self.footer_display = ui.TextDisplay(footer_text) self.footer_display = ui.TextDisplay(footer_text)
self.container.add_item(self.footer_display) self.container.add_item(self.footer_display)
@ -153,9 +156,7 @@ class LoggingCog(commands.Cog):
"""Fetch the latest audit log ID for each guild the bot is in.""" """Fetch the latest audit log ID for each guild the bot is in."""
log.info("Initializing last audit log IDs for guilds...") log.info("Initializing last audit log IDs for guilds...")
for guild in self.bot.guilds: for guild in self.bot.guilds:
if ( if guild.id not in self.last_audit_log_ids: # Only initialize if not already set
guild.id not in self.last_audit_log_ids
): # Only initialize if not already set
try: try:
if guild.me.guild_permissions.view_audit_log: if guild.me.guild_permissions.view_audit_log:
async for entry in guild.audit_logs(limit=1): async for entry in guild.audit_logs(limit=1):
@ -168,26 +169,20 @@ class LoggingCog(commands.Cog):
log.warning( log.warning(
f"Missing 'View Audit Log' permission in guild {guild.id}. Cannot initialize audit log ID." f"Missing 'View Audit Log' permission in guild {guild.id}. Cannot initialize audit log ID."
) )
self.last_audit_log_ids[guild.id] = ( self.last_audit_log_ids[guild.id] = None # Mark as unable to fetch
None # Mark as unable to fetch
)
except discord.Forbidden: except discord.Forbidden:
log.warning( log.warning(
f"Forbidden error fetching initial audit log ID for guild {guild.id}." f"Forbidden error fetching initial audit log ID for guild {guild.id}."
) )
self.last_audit_log_ids[guild.id] = None self.last_audit_log_ids[guild.id] = None
except discord.HTTPException as e: except discord.HTTPException as e:
log.error( log.error(f"HTTP error fetching initial audit log ID for guild {guild.id}: {e}")
f"HTTP error fetching initial audit log ID for guild {guild.id}: {e}"
)
self.last_audit_log_ids[guild.id] = None self.last_audit_log_ids[guild.id] = None
except Exception as e: except Exception as e:
log.exception( log.exception(
f"Unexpected error fetching initial audit log ID for guild {guild.id}: {e}" f"Unexpected error fetching initial audit log ID for guild {guild.id}: {e}"
) )
self.last_audit_log_ids[guild.id] = ( self.last_audit_log_ids[guild.id] = None # Mark as unable on other errors
None # Mark as unable on other errors
)
log.info("Finished initializing audit log IDs.") log.info("Finished initializing audit log IDs.")
async def start_audit_log_poller_when_ready(self): async def start_audit_log_poller_when_ready(self):
@ -231,9 +226,7 @@ class LoggingCog(commands.Cog):
) )
# log.debug(f"Sent log embed via webhook for guild {guild.id}") # Can be noisy # log.debug(f"Sent log embed via webhook for guild {guild.id}") # Can be noisy
except ValueError as e: except ValueError as e:
log.exception( log.exception(f"ValueError sending log via webhook for guild {guild.id}. Error: {e}")
f"ValueError sending log via webhook for guild {guild.id}. Error: {e}"
)
# Consider notifying an admin or disabling logging for this guild temporarily # Consider notifying an admin or disabling logging for this guild temporarily
# await settings_manager.set_logging_webhook(guild.id, None) # Example: Auto-disable on invalid URL # await settings_manager.set_logging_webhook(guild.id, None) # Example: Auto-disable on invalid URL
except (discord.Forbidden, discord.NotFound): except (discord.Forbidden, discord.NotFound):
@ -245,13 +238,9 @@ class LoggingCog(commands.Cog):
except discord.HTTPException as e: except discord.HTTPException as e:
log.error(f"HTTP error sending log via webhook for guild {guild.id}: {e}") log.error(f"HTTP error sending log via webhook for guild {guild.id}: {e}")
except aiohttp.ClientError as e: except aiohttp.ClientError as e:
log.error( log.error(f"aiohttp client error sending log via webhook for guild {guild.id}: {e}")
f"aiohttp client error sending log via webhook for guild {guild.id}: {e}"
)
except Exception as e: except Exception as e:
log.exception( log.exception(f"Unexpected error sending log via webhook for guild {guild.id}: {e}")
f"Unexpected error sending log via webhook for guild {guild.id}: {e}"
)
def _create_log_embed( def _create_log_embed(
self, self,
@ -284,8 +273,16 @@ class LoggingCog(commands.Cog):
if target_id: if target_id:
existing_footer = getattr(embed, "footer_display", None) existing_footer = getattr(embed, "footer_display", None)
if existing_footer: if existing_footer:
parts = [f"{id_name}: {target_id}"]
link = None
if hasattr(obj, "jump_url"):
link = f"[Jump]({obj.jump_url})"
elif isinstance(obj, discord.abc.GuildChannel):
link = obj.mention
if link:
parts.append(link)
sep = " | " if existing_footer.content else "" sep = " | " if existing_footer.content else ""
existing_footer.content += f"{sep}{id_name}: {target_id}" existing_footer.content += sep + " | ".join(parts)
async def _check_log_enabled(self, guild_id: int, event_key: str) -> bool: async def _check_log_enabled(self, guild_id: int, event_key: str) -> bool:
"""Checks if logging is enabled for a specific event key in a guild.""" """Checks if logging is enabled for a specific event key in a guild."""
@ -363,9 +360,7 @@ class LoggingCog(commands.Cog):
allowed_mentions=AllowedMentions.none(), allowed_mentions=AllowedMentions.none(),
) )
except Exception as e: except Exception as e:
log.exception( log.exception(f"Error fetching existing webhook during setup for guild {guild.id}")
f"Error fetching existing webhook during setup for guild {guild.id}"
)
await ctx.send( await ctx.send(
f"⚠️ An error occurred while checking the existing webhook. Proceeding to create a new one for {channel.mention}.", f"⚠️ An error occurred while checking the existing webhook. Proceeding to create a new one for {channel.mention}.",
allowed_mentions=AllowedMentions.none(), allowed_mentions=AllowedMentions.none(),
@ -379,9 +374,7 @@ class LoggingCog(commands.Cog):
try: try:
avatar_bytes = await self.bot.user.display_avatar.read() avatar_bytes = await self.bot.user.display_avatar.read()
except Exception: except Exception:
log.warning( log.warning(f"Could not read bot avatar for webhook creation in guild {guild.id}.")
f"Could not read bot avatar for webhook creation in guild {guild.id}."
)
new_webhook = await channel.create_webhook( new_webhook = await channel.create_webhook(
name=webhook_name, name=webhook_name,
@ -392,9 +385,7 @@ class LoggingCog(commands.Cog):
f"Created logging webhook '{webhook_name}' in channel {channel.id} for guild {guild.id}" f"Created logging webhook '{webhook_name}' in channel {channel.id} for guild {guild.id}"
) )
except discord.HTTPException as e: except discord.HTTPException as e:
log.error( log.error(f"Failed to create webhook in {channel.mention} for guild {guild.id}: {e}")
f"Failed to create webhook in {channel.mention} for guild {guild.id}: {e}"
)
await ctx.send( await ctx.send(
f"❌ Failed to create webhook. Error: {e}. This could be due to hitting the channel webhook limit (15).", f"❌ Failed to create webhook. Error: {e}. This could be due to hitting the channel webhook limit (15).",
allowed_mentions=AllowedMentions.none(), allowed_mentions=AllowedMentions.none(),
@ -431,9 +422,7 @@ class LoggingCog(commands.Cog):
allowed_mentions=AllowedMentions.none(), allowed_mentions=AllowedMentions.none(),
) )
except Exception as e: except Exception as e:
log.error( log.error(f"Failed to send test message via new webhook for guild {guild.id}: {e}")
f"Failed to send test message via new webhook for guild {guild.id}: {e}"
)
await ctx.send( await ctx.send(
"⚠️ Could not send a test message via the new webhook, but the URL has been saved.", "⚠️ Could not send a test message via the new webhook, but the URL has been saved.",
allowed_mentions=AllowedMentions.none(), allowed_mentions=AllowedMentions.none(),
@ -449,9 +438,7 @@ class LoggingCog(commands.Cog):
# Attempt to delete the created webhook to avoid orphans # Attempt to delete the created webhook to avoid orphans
try: try:
await new_webhook.delete(reason="Failed to save URL to settings") await new_webhook.delete(reason="Failed to save URL to settings")
log.info( log.info(f"Deleted orphaned webhook '{new_webhook.name}' for guild {guild.id}")
f"Deleted orphaned webhook '{new_webhook.name}' for guild {guild.id}"
)
except Exception as del_e: except Exception as del_e:
log.error( log.error(
f"Failed to delete orphaned webhook '{new_webhook.name}' for guild {guild.id}: {del_e}" f"Failed to delete orphaned webhook '{new_webhook.name}' for guild {guild.id}: {del_e}"
@ -493,9 +480,7 @@ class LoggingCog(commands.Cog):
new_status = enabled_status new_status = enabled_status
# Save the new status # Save the new status
success = await settings_manager.set_log_event_enabled( success = await settings_manager.set_log_event_enabled(guild_id, event_key, new_status)
guild_id, event_key, new_status
)
if success: if success:
status_str = "ENABLED" if new_status else "DISABLED" status_str = "ENABLED" if new_status else "DISABLED"
@ -529,15 +514,11 @@ class LoggingCog(commands.Cog):
# Paginate if too long for one embed description # Paginate if too long for one embed description
description = "" description = ""
for line in lines: for line in lines:
if ( if len(description) + len(line) + 1 > 4000: # Embed description limit (approx)
len(description) + len(line) + 1 > 4000
): # Embed description limit (approx)
embed.description = description embed.description = description
await ctx.send(embed=embed, allowed_mentions=AllowedMentions.none()) await ctx.send(embed=embed, allowed_mentions=AllowedMentions.none())
description = line + "\n" # Start new description description = line + "\n" # Start new description
embed = discord.Embed( embed = discord.Embed(color=discord.Color.blue()) # New embed for continuation
color=discord.Color.blue()
) # New embed for continuation
else: else:
description += line + "\n" description += line + "\n"
@ -548,9 +529,7 @@ class LoggingCog(commands.Cog):
@log_group.command(name="list_keys") @log_group.command(name="list_keys")
async def log_list_keys(self, ctx: commands.Context): async def log_list_keys(self, ctx: commands.Context):
"""Lists all valid event keys for use with the 'log toggle' command.""" """Lists all valid event keys for use with the 'log toggle' command."""
embed = discord.Embed( embed = discord.Embed(title="Available Logging Event Keys", color=discord.Color.purple())
title="Available Logging Event Keys", color=discord.Color.purple()
)
keys_text = "\n".join(f"`{key}`" for key in ALL_EVENT_KEYS) keys_text = "\n".join(f"`{key}`" for key in ALL_EVENT_KEYS)
# Paginate if needed # Paginate if needed
@ -594,9 +573,7 @@ class LoggingCog(commands.Cog):
footer=f"Thread ID: {thread.id} | Parent ID: {thread.parent_id}", footer=f"Thread ID: {thread.id} | Parent ID: {thread.parent_id}",
) )
if thread.owner: # Sometimes owner isn't cached immediately if thread.owner: # Sometimes owner isn't cached immediately
embed.set_author( embed.set_author(name=str(thread.owner), icon_url=thread.owner.display_avatar.url)
name=str(thread.owner), icon_url=thread.owner.display_avatar.url
)
await self._send_log_embed(guild, embed) await self._send_log_embed(guild, embed)
@commands.Cog.listener() @commands.Cog.listener()
@ -630,9 +607,7 @@ class LoggingCog(commands.Cog):
if before.locked != after.locked: if before.locked != after.locked:
changes.append(f"**Locked:** `{before.locked}` → `{after.locked}`") changes.append(f"**Locked:** `{before.locked}` → `{after.locked}`")
if before.slowmode_delay != after.slowmode_delay: if before.slowmode_delay != after.slowmode_delay:
changes.append( changes.append(f"**Slowmode:** `{before.slowmode_delay}s` → `{after.slowmode_delay}s`")
f"**Slowmode:** `{before.slowmode_delay}s` → `{after.slowmode_delay}s`"
)
if before.auto_archive_duration != after.auto_archive_duration: if before.auto_archive_duration != after.auto_archive_duration:
changes.append( changes.append(
f"**Auto-Archive:** `{before.auto_archive_duration} mins` → `{after.auto_archive_duration} mins`" f"**Auto-Archive:** `{before.auto_archive_duration} mins` → `{after.auto_archive_duration} mins`"
@ -711,9 +686,7 @@ class LoggingCog(commands.Cog):
# Initialization is now handled by initialize_cog called from __init__ or start_audit_log_poller_when_ready # Initialization is now handled by initialize_cog called from __init__ or start_audit_log_poller_when_ready
# Ensure the poller is running if it wasn't started earlier # Ensure the poller is running if it wasn't started earlier
if self.bot.is_ready() and not self.poll_audit_log.is_running(): if self.bot.is_ready() and not self.poll_audit_log.is_running():
log.warning( log.warning("Poll audit log task was not running after on_ready, attempting to start.")
"Poll audit log task was not running after on_ready, attempting to start."
)
await self.initialize_cog() # Re-initialize just in case await self.initialize_cog() # Re-initialize just in case
@commands.Cog.listener() @commands.Cog.listener()
@ -730,14 +703,10 @@ class LoggingCog(commands.Cog):
) )
break break
else: else:
log.warning( log.warning(f"Missing 'View Audit Log' permission in new guild {guild.id}.")
f"Missing 'View Audit Log' permission in new guild {guild.id}."
)
self.last_audit_log_ids[guild.id] = None self.last_audit_log_ids[guild.id] = None
except Exception as e: except Exception as e:
log.exception( log.exception(f"Error fetching initial audit log ID for new guild {guild.id}: {e}")
f"Error fetching initial audit log ID for new guild {guild.id}: {e}"
)
self.last_audit_log_ids[guild.id] = None self.last_audit_log_ids[guild.id] = None
@commands.Cog.listener() @commands.Cog.listener()
@ -790,9 +759,7 @@ class LoggingCog(commands.Cog):
await self._send_log_embed(member.guild, embed) await self._send_log_embed(member.guild, embed)
@commands.Cog.listener() @commands.Cog.listener()
async def on_member_ban( async def on_member_ban(self, guild: discord.Guild, user: Union[discord.User, discord.Member]):
self, guild: discord.Guild, user: Union[discord.User, discord.Member]
):
event_key = "member_ban_event" event_key = "member_ban_event"
if not await self._check_log_enabled(guild.id, event_key): if not await self._check_log_enabled(guild.id, event_key):
return return
@ -832,9 +799,7 @@ class LoggingCog(commands.Cog):
changes = [] changes = []
# Nickname change # Nickname change
if before.nick != after.nick: if before.nick != after.nick:
changes.append( changes.append(f"**Nickname:** `{before.nick or 'None'}` → `{after.nick or 'None'}`")
f"**Nickname:** `{before.nick or 'None'}` → `{after.nick or 'None'}`"
)
# Role changes (handled more reliably by audit log for who did it) # Role changes (handled more reliably by audit log for who did it)
if before.roles != after.roles: if before.roles != after.roles:
added_roles = [r.mention for r in after.roles if r not in before.roles] added_roles = [r.mention for r in after.roles if r not in before.roles]
@ -846,9 +811,7 @@ class LoggingCog(commands.Cog):
# Timeout change # Timeout change
if before.timed_out_until != after.timed_out_until: if before.timed_out_until != after.timed_out_until:
if after.timed_out_until: if after.timed_out_until:
timeout_duration = discord.utils.format_dt( timeout_duration = discord.utils.format_dt(after.timed_out_until, style="R")
after.timed_out_until, style="R"
)
changes.append(f"**Timed Out Until:** {timeout_duration}") changes.append(f"**Timed Out Until:** {timeout_duration}")
else: else:
changes.append("**Timeout Removed**") changes.append("**Timeout Removed**")
@ -857,9 +820,7 @@ class LoggingCog(commands.Cog):
# Add avatar change detection # Add avatar change detection
if before.display_avatar != after.display_avatar: if before.display_avatar != after.display_avatar:
changes.append( changes.append(f"**Avatar Changed**") # URL is enough, no need to show old/new
f"**Avatar Changed**"
) # URL is enough, no need to show old/new
if changes: if changes:
embed = self._create_log_embed( embed = self._create_log_embed(
@ -917,9 +878,7 @@ class LoggingCog(commands.Cog):
if before.hoist != after.hoist: if before.hoist != after.hoist:
changes.append(f"**Hoisted:** `{before.hoist}` → `{after.hoist}`") changes.append(f"**Hoisted:** `{before.hoist}` → `{after.hoist}`")
if before.mentionable != after.mentionable: if before.mentionable != after.mentionable:
changes.append( changes.append(f"**Mentionable:** `{before.mentionable}` → `{after.mentionable}`")
f"**Mentionable:** `{before.mentionable}` → `{after.mentionable}`"
)
if before.permissions != after.permissions: if before.permissions != after.permissions:
# Comparing permissions can be complex, just note that they changed. # Comparing permissions can be complex, just note that they changed.
# Audit log provides specifics on permission changes. # Audit log provides specifics on permission changes.
@ -988,28 +947,20 @@ class LoggingCog(commands.Cog):
if before.name != after.name: if before.name != after.name:
changes.append(f"**Name:** `{before.name}` → `{after.name}`") changes.append(f"**Name:** `{before.name}` → `{after.name}`")
if isinstance(before, discord.TextChannel) and isinstance( if isinstance(before, discord.TextChannel) and isinstance(after, discord.TextChannel):
after, discord.TextChannel
):
if before.topic != after.topic: if before.topic != after.topic:
changes.append( changes.append(f"**Topic:** `{before.topic or 'None'}` → `{after.topic or 'None'}`")
f"**Topic:** `{before.topic or 'None'}` → `{after.topic or 'None'}`"
)
if before.slowmode_delay != after.slowmode_delay: if before.slowmode_delay != after.slowmode_delay:
changes.append( changes.append(
f"**Slowmode:** `{before.slowmode_delay}s` → `{after.slowmode_delay}s`" f"**Slowmode:** `{before.slowmode_delay}s` → `{after.slowmode_delay}s`"
) )
if before.nsfw != after.nsfw: if before.nsfw != after.nsfw:
changes.append(f"**NSFW:** `{before.nsfw}` → `{after.nsfw}`") changes.append(f"**NSFW:** `{before.nsfw}` → `{after.nsfw}`")
if isinstance(before, discord.VoiceChannel) and isinstance( if isinstance(before, discord.VoiceChannel) and isinstance(after, discord.VoiceChannel):
after, discord.VoiceChannel
):
if before.bitrate != after.bitrate: if before.bitrate != after.bitrate:
changes.append(f"**Bitrate:** `{before.bitrate}` → `{after.bitrate}`") changes.append(f"**Bitrate:** `{before.bitrate}` → `{after.bitrate}`")
if before.user_limit != after.user_limit: if before.user_limit != after.user_limit:
changes.append( changes.append(f"**User Limit:** `{before.user_limit}` → `{after.user_limit}`")
f"**User Limit:** `{before.user_limit}` → `{after.user_limit}`"
)
# Permission overwrites change # Permission overwrites change
if before.overwrites != after.overwrites: if before.overwrites != after.overwrites:
# Identify changes without detailing every permission bit # Identify changes without detailing every permission bit
@ -1031,17 +982,13 @@ class LoggingCog(commands.Cog):
f"Removed overwrites for: {', '.join([f'<@{t.id}>' if isinstance(t, discord.Member) else f'<@&{t.id}>' for t in removed_targets])}" f"Removed overwrites for: {', '.join([f'<@{t.id}>' if isinstance(t, discord.Member) else f'<@&{t.id}>' for t in removed_targets])}"
) )
# Check if any *values* changed for targets present both before and after # Check if any *values* changed for targets present both before and after
if any( if any(before.overwrites[t] != after.overwrites[t] for t in updated_targets):
before.overwrites[t] != after.overwrites[t] for t in updated_targets
):
overwrite_changes.append( overwrite_changes.append(
f"Modified overwrites for: {', '.join([f'<@{t.id}>' if isinstance(t, discord.Member) else f'<@&{t.id}>' for t in updated_targets if before.overwrites[t] != after.overwrites[t]])}" f"Modified overwrites for: {', '.join([f'<@{t.id}>' if isinstance(t, discord.Member) else f'<@&{t.id}>' for t in updated_targets if before.overwrites[t] != after.overwrites[t]])}"
) )
if overwrite_changes: if overwrite_changes:
changes.append( changes.append(f"**Permission Overwrites:**\n - " + "\n - ".join(overwrite_changes))
f"**Permission Overwrites:**\n - " + "\n - ".join(overwrite_changes)
)
else: else:
changes.append( changes.append(
"**Permission Overwrites Updated** (No specific target changes detected by event)" "**Permission Overwrites Updated** (No specific target changes detected by event)"
@ -1081,23 +1028,22 @@ class LoggingCog(commands.Cog):
if not await self._check_log_enabled(guild.id, event_key): if not await self._check_log_enabled(guild.id, event_key):
return return
emoji, color = EVENT_STYLES.get("message_edit", ("", discord.Color.light_grey()))
embed = self._create_log_embed( embed = self._create_log_embed(
title="✏️ Message Edited", title=f"{emoji} Message Edited",
description=f"Message edited in {after.channel.mention} [Jump to Message]({after.jump_url})", description=f"Message edited in {after.channel.mention}",
color=discord.Color.light_grey(), color=color,
author=after.author, author=after.author,
) )
# Add fields for before and after, handling potential length limits diff = "\n".join(difflib.ndiff(before.content.splitlines(), after.content.splitlines()))
if len(diff) > 1000:
diff = diff[:997] + "..."
embed.add_field(name="Jump", value=f"[Link]({after.jump_url})")
embed.add_field( embed.add_field(
name="Before", name="Changes",
value=before.content[:1020] + ("..." if len(before.content) > 1020 else "") value=(
or "`Empty Message`", f"```diff\n{diff}\n```" if diff.strip() else "`(only embeds/attachments changed)`"
inline=False, ),
)
embed.add_field(
name="After",
value=after.content[:1020] + ("..." if len(after.content) > 1020 else "")
or "`Empty Message`",
inline=False, inline=False,
) )
self._add_id_footer(embed, after, id_name="Message ID") # Add message ID self._add_id_footer(embed, after, id_name="Message ID") # Add message ID
@ -1122,21 +1068,18 @@ class LoggingCog(commands.Cog):
if not await self._check_log_enabled(guild.id, event_key): if not await self._check_log_enabled(guild.id, event_key):
return return
desc = f"Message deleted in {message.channel.mention}" emoji, color = EVENT_STYLES.get("message_delete", ("", discord.Color.dark_grey()))
# Audit log needed for *who* deleted it, if not the author themselves
# We can add a placeholder here and update it if the audit log confirms a moderator deletion later
embed = self._create_log_embed( embed = self._create_log_embed(
title="🗑️ Message Deleted", title=f"{emoji} Message Deleted",
description=f"{desc}\n*Audit log may contain deleter if not the author.*", description=f"Message deleted in {message.channel.mention}",
color=discord.Color.dark_grey(), color=color,
author=message.author, author=message.author,
) )
embed.add_field(name="Jump", value=f"[Link]({message.jump_url})")
if message.content: if message.content:
embed.add_field( embed.add_field(
name="Content", name="Content",
value=message.content[:1020] value=message.content[:1020] + ("..." if len(message.content) > 1020 else "")
+ ("..." if len(message.content) > 1020 else "")
or "`Empty Message`", or "`Empty Message`",
inline=False, inline=False,
) )
@ -1196,9 +1139,7 @@ class LoggingCog(commands.Cog):
await self._send_log_embed(guild, embed) await self._send_log_embed(guild, embed)
@commands.Cog.listener() @commands.Cog.listener()
async def on_reaction_clear( async def on_reaction_clear(self, message: discord.Message, _: list[discord.Reaction]):
self, message: discord.Message, _: list[discord.Reaction]
):
guild = message.guild guild = message.guild
if not guild: if not guild:
return # Should not happen in guilds but safety check return # Should not happen in guilds but safety check
@ -1418,9 +1359,7 @@ class LoggingCog(commands.Cog):
if invite.max_age: if invite.max_age:
# Use invite.created_at if available, otherwise fall back to current time # Use invite.created_at if available, otherwise fall back to current time
created_time = ( created_time = (
invite.created_at invite.created_at if invite.created_at is not None else discord.utils.utcnow()
if invite.created_at is not None
else discord.utils.utcnow()
) )
expires_at = created_time + datetime.timedelta(seconds=invite.max_age) expires_at = created_time + datetime.timedelta(seconds=invite.max_age)
desc += f"\nExpires: {discord.utils.format_dt(expires_at, style='R')}" desc += f"\nExpires: {discord.utils.format_dt(expires_at, style='R')}"
@ -1478,9 +1417,7 @@ class LoggingCog(commands.Cog):
# await self._send_log_embed(ctx.guild, embed) # await self._send_log_embed(ctx.guild, embed)
@commands.Cog.listener() @commands.Cog.listener()
async def on_command_error( async def on_command_error(self, ctx: commands.Context, error: commands.CommandError):
self, ctx: commands.Context, error: commands.CommandError
):
# Log only significant errors, ignore things like CommandNotFound or CheckFailure if desired # Log only significant errors, ignore things like CommandNotFound or CheckFailure if desired
ignored = ( ignored = (
commands.CommandNotFound, commands.CommandNotFound,
@ -1508,14 +1445,10 @@ class LoggingCog(commands.Cog):
# Get traceback if available (might need error handling specific to your bot's setup) # Get traceback if available (might need error handling specific to your bot's setup)
import traceback import traceback
tb = "".join( tb = "".join(traceback.format_exception(type(error), error, error.__traceback__))
traceback.format_exception(type(error), error, error.__traceback__)
)
embed.add_field( embed.add_field(
name="Error Details", name="Error Details",
value=( value=(f"```py\n{tb[:1000]}\n...```" if len(tb) > 1000 else f"```py\n{tb}```"),
f"```py\n{tb[:1000]}\n...```" if len(tb) > 1000 else f"```py\n{tb}```"
),
inline=False, inline=False,
) )
@ -1579,9 +1512,7 @@ class LoggingCog(commands.Cog):
) )
break break
except Exception as e: except Exception as e:
log.exception( log.exception(f"Error re-initializing audit log ID for guild {guild.id}: {e}")
f"Error re-initializing audit log ID for guild {guild.id}: {e}"
)
continue # Skip this cycle if re-init fails continue # Skip this cycle if re-init fails
last_id = self.last_audit_log_ids.get(guild_id) last_id = self.last_audit_log_ids.get(guild_id)
@ -1648,14 +1579,10 @@ class LoggingCog(commands.Cog):
) )
# Consider adding backoff logic here if errors persist # Consider adding backoff logic here if errors persist
except Exception as e: except Exception as e:
log.exception( log.exception(f"Unexpected error in poll_audit_log for guild {guild.id}: {e}")
f"Unexpected error in poll_audit_log for guild {guild.id}: {e}"
)
# Don't update last_audit_log_id on unexpected error, retry next time # Don't update last_audit_log_id on unexpected error, retry next time
async def _process_audit_log_entry( async def _process_audit_log_entry(self, guild: discord.Guild, entry: discord.AuditLogEntry):
self, guild: discord.Guild, entry: discord.AuditLogEntry
):
"""Processes a single relevant audit log entry and sends an embed.""" """Processes a single relevant audit log entry and sends an embed."""
user = entry.user # Moderator/Actor user = entry.user # Moderator/Actor
target = entry.target # User/Channel/Role/Message affected target = entry.target # User/Channel/Role/Message affected
@ -1673,9 +1600,7 @@ class LoggingCog(commands.Cog):
if not await self._check_log_enabled(guild.id, audit_event_key): if not await self._check_log_enabled(guild.id, audit_event_key):
return return
title = "🛡️ Audit Log: Member Banned" title = "🛡️ Audit Log: Member Banned"
action_desc = ( action_desc = f"{self._user_display(user)} banned {self._user_display(target)}"
f"{self._user_display(user)} banned {self._user_display(target)}"
)
color = discord.Color.red() color = discord.Color.red()
# self._add_id_footer(embed, target, id_name="Target ID") # Footer set later # self._add_id_footer(embed, target, id_name="Target ID") # Footer set later
elif entry.action == discord.AuditLogAction.unban: elif entry.action == discord.AuditLogAction.unban:
@ -1683,9 +1608,7 @@ class LoggingCog(commands.Cog):
if not await self._check_log_enabled(guild.id, audit_event_key): if not await self._check_log_enabled(guild.id, audit_event_key):
return return
title = "🛡️ Audit Log: Member Unbanned" title = "🛡️ Audit Log: Member Unbanned"
action_desc = ( action_desc = f"{self._user_display(user)} unbanned {self._user_display(target)}"
f"{self._user_display(user)} unbanned {self._user_display(target)}"
)
color = discord.Color.blurple() color = discord.Color.blurple()
# self._add_id_footer(embed, target, id_name="Target ID") # Footer set later # self._add_id_footer(embed, target, id_name="Target ID") # Footer set later
elif entry.action == discord.AuditLogAction.kick: elif entry.action == discord.AuditLogAction.kick:
@ -1693,9 +1616,7 @@ class LoggingCog(commands.Cog):
if not await self._check_log_enabled(guild.id, audit_event_key): if not await self._check_log_enabled(guild.id, audit_event_key):
return return
title = "🛡️ Audit Log: Member Kicked" title = "🛡️ Audit Log: Member Kicked"
action_desc = ( action_desc = f"{self._user_display(user)} kicked {self._user_display(target)}"
f"{self._user_display(user)} kicked {self._user_display(target)}"
)
color = discord.Color.brand_red() color = discord.Color.brand_red()
# self._add_id_footer(embed, target, id_name="Target ID") # Footer set later # self._add_id_footer(embed, target, id_name="Target ID") # Footer set later
elif entry.action == discord.AuditLogAction.member_prune: elif entry.action == discord.AuditLogAction.member_prune:
@ -1705,7 +1626,9 @@ class LoggingCog(commands.Cog):
title = "🛡️ Audit Log: Member Prune" title = "🛡️ Audit Log: Member Prune"
days = entry.extra.get("delete_member_days") days = entry.extra.get("delete_member_days")
count = entry.extra.get("members_removed") count = entry.extra.get("members_removed")
action_desc = f"{self._user_display(user)} pruned {count} members inactive for {days} days." action_desc = (
f"{self._user_display(user)} pruned {count} members inactive for {days} days."
)
color = discord.Color.dark_red() color = discord.Color.dark_red()
# No specific target ID here # No specific target ID here
@ -1739,9 +1662,7 @@ class LoggingCog(commands.Cog):
return return
title = "🛡️ Audit Log: Member Timeout Update" title = "🛡️ Audit Log: Member Timeout Update"
if after_timed_out: if after_timed_out:
timeout_duration = discord.utils.format_dt( timeout_duration = discord.utils.format_dt(after_timed_out, style="R")
after_timed_out, style="R"
)
action_desc = f"{self._user_display(user)} timed out {self._user_display(target)} ({target.id}) until {timeout_duration}" action_desc = f"{self._user_display(user)} timed out {self._user_display(target)} ({target.id}) until {timeout_duration}"
color = discord.Color.orange() color = discord.Color.orange()
else: else:
@ -1799,9 +1720,7 @@ class LoggingCog(commands.Cog):
and hasattr(entry.after, "hoist") and hasattr(entry.after, "hoist")
and entry.before.hoist != entry.after.hoist and entry.before.hoist != entry.after.hoist
): ):
changes.append( changes.append(f"Hoisted: `{entry.before.hoist}` → `{entry.after.hoist}`")
f"Hoisted: `{entry.before.hoist}` → `{entry.after.hoist}`"
)
if ( if (
hasattr(entry.before, "mentionable") hasattr(entry.before, "mentionable")
and hasattr(entry.after, "mentionable") and hasattr(entry.after, "mentionable")
@ -1837,7 +1756,9 @@ class LoggingCog(commands.Cog):
title = "🛡️ Audit Log: Channel Created" title = "🛡️ Audit Log: Channel Created"
channel = target channel = target
ch_type = str(channel.type).capitalize() ch_type = str(channel.type).capitalize()
action_desc = f"{user.mention} created {ch_type} channel {channel.mention} (`{channel.name}`)" action_desc = (
f"{user.mention} created {ch_type} channel {channel.mention} (`{channel.name}`)"
)
color = discord.Color.green() color = discord.Color.green()
# self._add_id_footer(embed, channel, id_name="Channel ID") # Footer set later # self._add_id_footer(embed, channel, id_name="Channel ID") # Footer set later
elif entry.action == discord.AuditLogAction.channel_delete: elif entry.action == discord.AuditLogAction.channel_delete:
@ -1849,7 +1770,9 @@ class LoggingCog(commands.Cog):
channel_name = entry.before.name channel_name = entry.before.name
channel_id = entry.target.id channel_id = entry.target.id
ch_type = str(entry.before.type).capitalize() ch_type = str(entry.before.type).capitalize()
action_desc = f"{user.mention} deleted {ch_type} channel `{channel_name}` ({channel_id})" action_desc = (
f"{user.mention} deleted {ch_type} channel `{channel_name}` ({channel_id})"
)
color = discord.Color.red() color = discord.Color.red()
# self._add_id_footer(embed, obj_id=channel_id, id_name="Channel ID") # Footer set later # self._add_id_footer(embed, obj_id=channel_id, id_name="Channel ID") # Footer set later
elif entry.action == discord.AuditLogAction.channel_update: elif entry.action == discord.AuditLogAction.channel_update:
@ -1892,9 +1815,7 @@ class LoggingCog(commands.Cog):
and hasattr(entry.after, "bitrate") and hasattr(entry.after, "bitrate")
and entry.before.bitrate != entry.after.bitrate and entry.before.bitrate != entry.after.bitrate
): ):
changes.append( changes.append(f"Bitrate: `{entry.before.bitrate}` → `{entry.after.bitrate}`")
f"Bitrate: `{entry.before.bitrate}` → `{entry.after.bitrate}`"
)
# Process detailed changes from entry.changes # Process detailed changes from entry.changes
detailed_changes = [] detailed_changes = []
@ -1907,33 +1828,21 @@ class LoggingCog(commands.Cog):
before_val = change.before before_val = change.before
after_val = change.after after_val = change.after
if attr == "name": if attr == "name":
detailed_changes.append( detailed_changes.append(f"Name: `{before_val}` → `{after_val}`")
f"Name: `{before_val}` → `{after_val}`"
)
elif attr == "topic": elif attr == "topic":
detailed_changes.append( detailed_changes.append(
f"Topic: `{before_val or 'None'}` → `{after_val or 'None'}`" f"Topic: `{before_val or 'None'}` → `{after_val or 'None'}`"
) )
elif attr == "nsfw": elif attr == "nsfw":
detailed_changes.append( detailed_changes.append(f"NSFW: `{before_val}` → `{after_val}`")
f"NSFW: `{before_val}` → `{after_val}`"
)
elif attr == "slowmode_delay": elif attr == "slowmode_delay":
detailed_changes.append( detailed_changes.append(f"Slowmode: `{before_val}s` → `{after_val}s`")
f"Slowmode: `{before_val}s` → `{after_val}s`"
)
elif attr == "bitrate": elif attr == "bitrate":
detailed_changes.append( detailed_changes.append(f"Bitrate: `{before_val}` → `{after_val}`")
f"Bitrate: `{before_val}` → `{after_val}`"
)
elif attr == "user_limit": elif attr == "user_limit":
detailed_changes.append( detailed_changes.append(f"User Limit: `{before_val}` → `{after_val}`")
f"User Limit: `{before_val}` → `{after_val}`"
)
elif attr == "position": elif attr == "position":
detailed_changes.append( detailed_changes.append(f"Position: `{before_val}` → `{after_val}`")
f"Position: `{before_val}` → `{after_val}`"
)
elif attr == "category": elif attr == "category":
detailed_changes.append( detailed_changes.append(
f"Category: {getattr(before_val, 'mention', 'None')} → {getattr(after_val, 'mention', 'None')}" f"Category: {getattr(before_val, 'mention', 'None')} → {getattr(after_val, 'mention', 'None')}"
@ -1954,9 +1863,7 @@ class LoggingCog(commands.Cog):
) )
# Determine if added, removed, or updated (before/after values are PermissionOverwrite objects) # Determine if added, removed, or updated (before/after values are PermissionOverwrite objects)
if before_val is None and after_val is not None: if before_val is None and after_val is not None:
detailed_changes.append( detailed_changes.append(f"Added overwrite for {target_mention}")
f"Added overwrite for {target_mention}"
)
elif before_val is not None and after_val is None: elif before_val is not None and after_val is None:
detailed_changes.append( detailed_changes.append(
f"Removed overwrite for {target_mention}" f"Removed overwrite for {target_mention}"
@ -1977,9 +1884,7 @@ class LoggingCog(commands.Cog):
else: else:
# Handle AuditLogChanges as a non-iterable object # Handle AuditLogChanges as a non-iterable object
# We can access the before and after attributes directly # We can access the before and after attributes directly
if hasattr(entry.changes, "before") and hasattr( if hasattr(entry.changes, "before") and hasattr(entry.changes, "after"):
entry.changes, "after"
):
before = entry.changes.before before = entry.changes.before
after = entry.changes.after after = entry.changes.after
@ -1989,9 +1894,7 @@ class LoggingCog(commands.Cog):
and hasattr(after, "name") and hasattr(after, "name")
and before.name != after.name and before.name != after.name
): ):
detailed_changes.append( detailed_changes.append(f"Name: `{before.name}` → `{after.name}`")
f"Name: `{before.name}` → `{after.name}`"
)
if ( if (
hasattr(before, "topic") hasattr(before, "topic")
and hasattr(after, "topic") and hasattr(after, "topic")
@ -2005,9 +1908,7 @@ class LoggingCog(commands.Cog):
and hasattr(after, "nsfw") and hasattr(after, "nsfw")
and before.nsfw != after.nsfw and before.nsfw != after.nsfw
): ):
detailed_changes.append( detailed_changes.append(f"NSFW: `{before.nsfw}` → `{after.nsfw}`")
f"NSFW: `{before.nsfw}` → `{after.nsfw}`"
)
if ( if (
hasattr(before, "slowmode_delay") hasattr(before, "slowmode_delay")
and hasattr(after, "slowmode_delay") and hasattr(after, "slowmode_delay")
@ -2078,21 +1979,15 @@ class LoggingCog(commands.Cog):
channel_display = "" channel_display = ""
if hasattr(channel_target, "mention"): if hasattr(channel_target, "mention"):
channel_display = channel_target.mention channel_display = channel_target.mention
elif isinstance(channel_target, discord.Object) and hasattr( elif isinstance(channel_target, discord.Object) and hasattr(channel_target, "id"):
channel_target, "id"
):
# If it's an Object, it might be a deleted channel or not fully loaded. # If it's an Object, it might be a deleted channel or not fully loaded.
# Using <#id> is a safe way to reference it. # Using <#id> is a safe way to reference it.
channel_display = f"<#{channel_target.id}>" channel_display = f"<#{channel_target.id}>"
else: else:
# Fallback if it's not an object with 'mention' or an 'Object' with 'id' # Fallback if it's not an object with 'mention' or an 'Object' with 'id'
channel_display = ( channel_display = f"an unknown channel (ID: {getattr(channel_target, 'id', 'N/A')})"
f"an unknown channel (ID: {getattr(channel_target, 'id', 'N/A')})"
)
action_desc = ( action_desc = f"{user.mention} bulk deleted {count} messages in {channel_display}"
f"{user.mention} bulk deleted {count} messages in {channel_display}"
)
color = discord.Color.dark_grey() color = discord.Color.dark_grey()
# self._add_id_footer(embed, channel_target, id_name="Channel ID") # Footer set later # self._add_id_footer(embed, channel_target, id_name="Channel ID") # Footer set later
@ -2146,9 +2041,7 @@ class LoggingCog(commands.Cog):
if invite.max_age: if invite.max_age:
# Use invite.created_at if available, otherwise fall back to current time # Use invite.created_at if available, otherwise fall back to current time
created_time = ( created_time = (
invite.created_at invite.created_at if invite.created_at is not None else discord.utils.utcnow()
if invite.created_at is not None
else discord.utils.utcnow()
) )
expires_at = created_time + datetime.timedelta(seconds=invite.max_age) expires_at = created_time + datetime.timedelta(seconds=invite.max_age)
desc += f"\nExpires: {discord.utils.format_dt(expires_at, style='R')}" desc += f"\nExpires: {discord.utils.format_dt(expires_at, style='R')}"
@ -2166,7 +2059,9 @@ class LoggingCog(commands.Cog):
invite_code = entry.before.code invite_code = entry.before.code
channel_id = entry.before.channel_id channel_id = entry.before.channel_id
channel_mention = f"<#{channel_id}>" if channel_id else "Unknown Channel" channel_mention = f"<#{channel_id}>" if channel_id else "Unknown Channel"
action_desc = f"{user.mention} deleted invite `{invite_code}` for channel {channel_mention}" action_desc = (
f"{user.mention} deleted invite `{invite_code}` for channel {channel_mention}"
)
color = discord.Color.dark_red() color = discord.Color.dark_red()
# Cannot get invite ID after deletion easily, use code in footer later # Cannot get invite ID after deletion easily, use code in footer later
@ -2221,8 +2116,7 @@ class LoggingCog(commands.Cog):
if ( if (
hasattr(entry.before, "explicit_content_filter") hasattr(entry.before, "explicit_content_filter")
and hasattr(entry.after, "explicit_content_filter") and hasattr(entry.after, "explicit_content_filter")
and entry.before.explicit_content_filter and entry.before.explicit_content_filter != entry.after.explicit_content_filter
!= entry.after.explicit_content_filter
): ):
changes.append( changes.append(
f"Explicit Content Filter: `{entry.before.explicit_content_filter}` → `{entry.after.explicit_content_filter}`" f"Explicit Content Filter: `{entry.before.explicit_content_filter}` → `{entry.after.explicit_content_filter}`"
@ -2253,7 +2147,9 @@ class LoggingCog(commands.Cog):
# Generic fallback log # Generic fallback log
title = f"🛡️ Audit Log: {str(entry.action).replace('_', ' ').title()}" title = f"🛡️ Audit Log: {str(entry.action).replace('_', ' ').title()}"
# Determine the generic audit key based on the action category if possible # Determine the generic audit key based on the action category if possible
generic_audit_key = f"audit_{str(entry.action).split('.')[0]}" # e.g., audit_member, audit_channel generic_audit_key = (
f"audit_{str(entry.action).split('.')[0]}" # e.g., audit_member, audit_channel
)
if generic_audit_key in ALL_EVENT_KEYS: if generic_audit_key in ALL_EVENT_KEYS:
if not await self._check_log_enabled(guild.id, generic_audit_key): if not await self._check_log_enabled(guild.id, generic_audit_key):
return return
@ -2271,9 +2167,7 @@ class LoggingCog(commands.Cog):
# self._add_id_footer(embed, target, id_name="Target ID") # Footer set later # self._add_id_footer(embed, target, id_name="Target ID") # Footer set later
color = discord.Color.light_grey() color = discord.Color.light_grey()
if ( if not action_desc: # If no description was generated (e.g., skipped update), skip logging
not action_desc
): # If no description was generated (e.g., skipped update), skip logging
# log.debug(f"Skipping audit log entry {entry.id} (action: {entry.action}) as no action description was generated.") # log.debug(f"Skipping audit log entry {entry.id} (action: {entry.action}) as no action description was generated.")
return return
@ -2285,24 +2179,22 @@ class LoggingCog(commands.Cog):
author=user, # The moderator/actor is the author of the log entry author=user, # The moderator/actor is the author of the log entry
) )
if reason: if reason:
embed.add_field( embed.add_field(name="Reason", value=reason[:1024], inline=False) # Limit reason length
name="Reason", value=reason[:1024], inline=False
) # Limit reason length
# Add relevant IDs to footer (target ID if available, otherwise just mod/entry ID) # Add relevant IDs to footer (target ID if available, otherwise just mod/entry ID)
target_id_str = "" target_id_str = ""
if target: if target:
target_id_str = f" | Target ID: {target.id}" target_id_str = f" | Target ID: {target.id}"
elif entry.action == discord.AuditLogAction.role_delete: elif entry.action == discord.AuditLogAction.role_delete:
target_id_str = f" | Role ID: {entry.target.id}" # Get ID from target even if object deleted target_id_str = (
f" | Role ID: {entry.target.id}" # Get ID from target even if object deleted
)
elif entry.action == discord.AuditLogAction.channel_delete: elif entry.action == discord.AuditLogAction.channel_delete:
target_id_str = f" | Channel ID: {entry.target.id}" target_id_str = f" | Channel ID: {entry.target.id}"
elif entry.action == discord.AuditLogAction.emoji_delete: elif entry.action == discord.AuditLogAction.emoji_delete:
target_id_str = f" | Emoji ID: {entry.target.id}" target_id_str = f" | Emoji ID: {entry.target.id}"
elif entry.action == discord.AuditLogAction.invite_delete: elif entry.action == discord.AuditLogAction.invite_delete:
target_id_str = ( target_id_str = f" | Invite Code: {entry.before.code}" # Use code for deleted invites
f" | Invite Code: {entry.before.code}" # Use code for deleted invites
)
embed.set_footer( embed.set_footer(
text=f"Audit Log Entry ID: {entry.id} | Moderator ID: {user.id}{target_id_str}" text=f"Audit Log Entry ID: {entry.id} | Moderator ID: {user.id}{target_id_str}"