python-app-rocketbot/rocketbot/cogs/logcog.py

"""
Cog for detecting large numbers of guild joins in a short period of time.
"""
import difflib
import re
from collections.abc import Callable
from datetime import datetime, timedelta, timezone
from typing import Any

from discord import (
	AuditLogAction,
	AuditLogEntry,
	Guild,
	Invite,
	Member,
	Message,
	RawBulkMessageDeleteEvent,
	RawMessageDeleteEvent,
	RawMessageUpdateEvent,
	Role,
	User,
)
from discord.abc import GuildChannel
from discord.errors import DiscordException
from discord.ext import tasks
from discord.ext.commands import Cog
from discord.utils import escape_markdown

from config import CONFIG
from rocketbot.cogs.basecog import BaseCog, BotMessage, CogSetting
from rocketbot.storage import ConfigKey, Storage
from rocketbot.utils import dump_stacktrace, norm_datetime


class BufferedMessageEditEvent:
	def __init__(self, guild: Guild, channel: GuildChannel, before: Message | None, after: Message, data = None) -> None:
		self.guild = guild
		self.channel = channel
		self.before = before
		self.after = after
		self.data = data

class BufferedMessageDeleteEvent:
	def __init__(self, guild: Guild, channel: GuildChannel, message_id: int, message: Message | None = None) -> None:
		self.guild = guild
		self.channel = channel
		self.message_id = message_id
		self.message = message
		self.author = message.author if message is not None else None

class LoggingCog(BaseCog, name='Logging'):
	"""
	Cog for logging notable events to a designated logging channel.
	"""
	SETTING_ENABLED = CogSetting(
		'enabled',
		bool,
		default_value=False,
		brief='logging',
		description='Whether this module is enabled for a guild.',
	)
	SETTING_EDITS_ENABLED = CogSetting(
		'edits',
		bool,
		default_value=True,
		brief='whether to log message edit events',
		description='Whether to log member message edits.'
	)
	SETTING_DELETES_ENABLED = CogSetting(
		'deletes',
		bool,
		default_value=True,
		brief='whether to log message deletion events',
		description='Whether to log member message deletions.'
	)
	SETTING_JOIN_ENABLED = CogSetting(
		'joins',
		bool,
		default_value=True,
		brief='whether to log members joining the server',
		description='Whether to log members joining the server.'
	)
	SETTING_LEAVE_ENABLED = CogSetting(
		'leaves',
		bool,
		default_value=True,
		brief='whether to log members leaving the server',
		description='Whether to log members leaving the server.'
	)
	SETTING_MEMBER_UPDATES_ENABLED = CogSetting(
		'member_updates',
		bool,
		default_value=True,
		brief='whether to log changes to member details',
		description='Whether to log changes to member roles, flags, display names, etc.'
	)

	STATE_EVENT_BUFFER = 'LoggingCog.eventBuffer'

	def __init__(self, bot):
		super().__init__(
			bot,
			config_prefix='logging',
			short_description='Manages event logging.',
		)
		self.add_setting(Self.SETTING_ENABLED)
		self.add_setting(Self.SETTING_EDITS_ENABLED)
		self.add_setting(Self.SETTING_DELETES_ENABLED)
		self.add_setting(Self.SETTING_JOIN_ENABLED)
		self.add_setting(Self.SETTING_LEAVE_ENABLED)
		self.add_setting(Self.SETTING_MEMBER_UPDATES_ENABLED)
		self.flush_buffers.start()
		self.buffered_guilds: set[Guild] = set()

	# @overload
	def cog_unload(self) -> None:
		self.flush_buffers.cancel()

	def __is_cog_disabled(self, guild: Guild | None) -> bool:
		"""Whether this cog is disabled for a guild."""
		return guild is None or not self.get_guild_setting(guild, Self.SETTING_ENABLED)
	def __is_edits_disabled(self, guild: Guild | None) -> bool:
		"""Whether logging message edit events is disabled for a guild."""
		return guild is None or self.__is_cog_disabled(guild) or not self.get_guild_setting(guild, Self.SETTING_EDITS_ENABLED)
	def __is_deletes_disabled(self, guild: Guild | None) -> bool:
		"""Whether logging message delete events is disabled for a guild."""
		return guild is None or self.__is_cog_disabled(guild) or not self.get_guild_setting(guild, Self.SETTING_DELETES_ENABLED)
	def __is_joins_disabled(self, guild: Guild | None) -> bool:
		"""Whether logging member join events is disabled for a guild."""
		return guild is None or self.__is_cog_disabled(guild) or not self.get_guild_setting(guild, Self.SETTING_JOIN_ENABLED)
	def __is_leaves_disabled(self, guild: Guild | None) -> bool:
		"""Whether logging member departure events is disabled for a guild."""
		return guild is None or self.__is_cog_disabled(guild) or not self.get_guild_setting(guild, Self.SETTING_LEAVE_ENABLED)
	def __is_member_updates_disabled(self, guild: Guild | None) -> bool:
		"""Whether logging member profile updates is disabled for a guild."""
		return guild is None or self.__is_cog_disabled(guild) or not self.get_guild_setting(guild, Self.SETTING_MEMBER_UPDATES_ENABLED)
	def __is_excluded_user(self, user: User | None) -> bool:
		"""Whether member events should be ignored for the given user."""
		return user is None or \
			user.bot or \
			user.id == self.bot.user.id
	def __is_excluded_channel(self, channel: GuildChannel | None) -> bool:
		"""Whether message events should be ignored in the given channel."""
		if channel is None: return True
		warning_channel_id = Storage.get_config_value(channel.guild, ConfigKey.WARNING_CHANNEL_ID)
		return channel.id == warning_channel_id

	# Events - Channels

	@Cog.listener()
	async def on_guild_channel_delete(self, channel: GuildChannel) -> None:
		"""
		Called whenever a guild channel is deleted or created.

		Note that you can get the guild from guild.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_guild_channel_delete
		"""
		guild = channel.guild
		if self.__is_cog_disabled(guild): return

		text = f'Channel **{channel.name}** deleted.'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@Cog.listener()
	async def on_guild_channel_create(self, channel: GuildChannel) -> None:
		"""
		Called whenever a guild channel is deleted or created.

		Note that you can get the guild from guild.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_guild_channel_create
		"""
		guild = channel.guild
		if self.__is_cog_disabled(guild): return

		text = f'Channel **{channel.name}** created. {channel.mention}'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@Cog.listener()
	async def on_guild_channel_update(self, before: GuildChannel, after: GuildChannel) -> None:
		"""
		Called whenever a guild channel is updated. e.g. changed name, topic,
		permissions.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_guild_channel_update
		"""
		guild = after.guild
		if self.__is_cog_disabled(guild): return

		changes = []
		if after.name != before.name:
			changes.append(f'Name: `{before.name}` -> `{after.name}`')
		if after.category != before.category:
			changes.append(f'Category: {before.category.name if before.category else None}' + \
		  		f' -> {after.category.name if after.category else None}')
		if after.changed_roles != before.changed_roles:
			changes.append('Roles changed')
		if after.overwrites != before.overwrites:
			changes.append('Permission overwrites changed')
		if after.position != before.position:
			changes.append(f'Position: {before.position} -> {after.position}')

		if len(changes) == 0:
			return
		text = f'Channel **{before.name}** updated. Changes:\n'
		text += '* ' + '\n* '.join(changes)
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	# Events - Guilds

	# @Cog.listener()
	# async def on_guild_available(self, guild: Guild) -> None:
	# 	pass

	# @Cog.listener()
	# async def on_guild_unavailable(self, guild: Guild) -> None:
	# 	pass

	# @Cog.listener()
	# async def on_guild_update(self, before: Guild, after: Guild) -> None:
	# 	pass

	# @Cog.listener()
	# async def on_guild_emojis_update(self, guild: Guild, before: Sequence[Emoji], after: Sequence[Emoji]) -> None:
	# 	pass

	# @Cog.listener()
	# async def on_guild_stickers_update(self, guild: Guild, before: Sequence[GuildSticker], after: Sequence[GuildSticker]) -> None:
	# 	pass

	@Cog.listener()
	async def on_invite_create(self, invite: Invite) -> None:
		"""
		Called when an `Invite` is created. You must have manage_channels to receive this.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_invite_create
		"""
		guild = invite.guild
		if self.__is_cog_disabled(guild): return

		text = f'Invite code `{invite.code}` created by {self.__describe_user(invite.inviter)}. '
		if invite.max_age == 0:
			text += "Doesn't expire."
		else:
			text += f'Expires in {invite.max_age} seconds.'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@Cog.listener()
	async def on_invite_delete(self, invite: Invite) -> None:
		"""
		Called when an `Invite` is deleted. You must have manage_channels to receive this.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_invite_delete
		"""
		guild = invite.guild
		if self.__is_cog_disabled(guild): return

		if invite.inviter:
			text = f'Invite code `{invite.code}` deleted. Originally created by {self.__describe_user(invite.inviter)}.'
		else:
			text = f'Invite code `{invite.code}` deleted.'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	# Events - Members

	@Cog.listener()
	async def on_member_join(self, member: Member) -> None:
		"""
		Called when a Member joins a Guild.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_member_join

		Requires Members privileged intent.
		"""
		guild = member.guild
		if self.__is_joins_disabled(guild): return

		flags = []
		noteworthy = False
		badge_info_link = 'https://support.discord.com/hc/en-us/articles/360035962891-Profile-Badges-101'

		# Highest priority - should flag mods to be aware
		if member.public_flags.spammer:
			flags.append(f"{CONFIG['warning_emoji']} **Is flagged as a spammer by Discord**")
			noteworthy = True
		if member.flags.automod_quarantined_username:
			flags.append(f"{CONFIG['warning_emoji']} **Quarantined by AutoMod for their username**")
			noteworthy = True
		if member.flags.automod_quarantined_guild_tag:
			flags.append(f"{CONFIG['warning_emoji']} **Quarantined by AutoMod for their server tag**")
			noteworthy = True
		if member.public_flags.verified_bot:
			flags.append(f"{CONFIG['info_emoji']} **Is a verified bot**")
			noteworthy = True
		if member.public_flags.system:
			flags.append(f"{CONFIG['info_emoji']} **Is a Discord system user**")
			noteworthy = True
		if member.public_flags.staff:
			flags.append(f"{CONFIG['info_emoji']} **Is Discord staff**")
			noteworthy = True

		# Interesting but doesn't need flagging
		if member.public_flags.partner:
			flags.append(f"{CONFIG['info_emoji']} **Is a Discord partner**")

		# Guild-specific
		if member.flags.did_rejoin:
			flags.append('Rejoined this server')
		if member.flags.guest:
			flags.append('Guest (can only access voice channel they were invited to)')
		if member.flags.bypasses_verification:
			flags.append('Can bypass guild verification steps')
		if member.flags.completed_home_actions:
			flags.append('Completed guild Server Guide actions')
		elif member.flags.started_home_actions:
			flags.append('Started guild Server Guide actions')
		if member.flags.completed_onboarding:
			flags.append('Completed guild onboarding')
		elif member.flags.started_onboarding:
			flags.append('Started guild onboarding')
		# member.flags.dm_settings_upsell_acknowledged - no idea what the "DM settings upsell" is

		# Trivia
		if member.public_flags.verified_bot_developer:
			flags.append('In an Early Verified Bot Developer')
		elif member.public_flags.active_developer:
			flags.append('Is an active developer')
		if member.public_flags.hypesquad_bravery:
			flags.append(f'Was a HypeSquad Bravery member ([info](<{badge_info_link}>))')
		elif member.public_flags.hypesquad_brilliance:
			flags.append(f'Was a HypeSquad Brilliance member ([info](<{badge_info_link}>))')
		elif member.public_flags.hypesquad_balance:
			flags.append(f'Was a HypeSquad Balance member ([info](<{badge_info_link}>))')
		elif member.public_flags.hypesquad:
			flags.append(f'Was a HypeSquad Events member ([info](<{badge_info_link}>))')
		if member.public_flags.early_supporter:
			flags.append(f'Was a Nitro early supporter ([info](<{badge_info_link}>))')
		if member.public_flags.team_user:
			flags.append('Is a Team User')
		if member.public_flags.bug_hunter_level_2:
			flags.append('Is a Discord bug hunter level 2')
		elif member.public_flags.bug_hunter:
			flags.append('Is a Discord bug hunter')
		if member.public_flags.discord_certified_moderator:
			flags.append('Was a Discord Certified Moderator')
		if member.public_flags.early_verified_bot_developer:
			flags.append('Is a verified bot developer')
		# public_flags.bot_http_interactions isn't interesting

		text = f"Member joined server: {self.__describe_user(member, tag=noteworthy)}"
		if len(flags) > 0:
			text += '\n- ' + '\n- '.join(flags)
		bot_message = BotMessage(guild, text, BotMessage.TYPE_MOD_WARNING if noteworthy else BotMessage.TYPE_LOG)
		await bot_message.update()

	@Cog.listener()
	async def on_member_remove(self, member: Member) -> None:
		"""
		Called when a Member leaves a Guild.

		If the guild or member could not be found in the internal cache this event
		will not be called, you may use on_raw_member_remove() instead.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_member_remove

		Requires Members privileged intent.
		"""
		guild = member.guild
		if self.__is_cog_disabled(guild): return

		is_kick = False
		kicker = None
		kick_reason = None
		entry = await self.__find_audit_entry(member, AuditLogAction.kick)
		if entry:
			is_kick = True
			kicker = entry.user
			kick_reason = entry.reason

		if is_kick:
			if kicker and kicker != member:
				text = f'Member kicked from the server: {self.__describe_user(member)} by **{kicker.name}**'
			else:
				text = f'Member kicked from the server: {self.__describe_user(member)}'
		else:
			if self.__is_leaves_disabled(guild): return
			text = f'Member left server: {self.__describe_user(member)}'
		if kick_reason:
			text += f'\nReason: "{kick_reason}"'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@Cog.listener()
	async def on_member_update(self, before: Member, after: Member) -> None:
		"""
		Called when a Member updates their profile.

		This is called when one or more of the following things change:
		* nickname
		* roles
		* pending
		* timeout
		* guild avatar
		* flags

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_member_update

		Requires Members privileged intent.
		"""
		guild = after.guild
		if self.__is_member_updates_disabled(guild): return
		if self.__is_excluded_user(after): return

		changes = []
		noteworthy = False
		if after.nick != before.nick:
			changes.append(f'Nick: `{before.nick}` -> `{after.nick}`')
		if after.name != before.name:
			changes.append(f'Username: `@{before.name}` -> `@{after.name}`')
		if after.display_name != before.display_name:
			changes.append(f'Display name: `{before.display_name}` -> `{after.display_name}`')
		if after.roles != before.roles:
			added_role_names = []
			removed_role_names = []
			for role in before.roles:
				if role not in after.roles:
					removed_role_names.append(role.name)
			for role in after.roles:
				if role not in before.roles:
					added_role_names.append(role.name)
			if len(removed_role_names) > 0:
				changes.append(f'Removed roles: ~~**{"**~~, ~~**".join(removed_role_names)}**~~')
			if len(added_role_names) > 0:
				changes.append(f'Added roles: **{"**, **".join(added_role_names)}**')
		if after.pending != before.pending:
			pass  # not that interesting and probably noisy
		if after.timed_out_until != before.timed_out_until:
			if after.timed_out_until:
				delta = after.timed_out_until - datetime.now(timezone.utc)
				changes.append(f'Timed out for `{delta}`')
			elif before.timed_out_until:
				changes.append('Timeout cleared')
		before_guild_avatar = before.guild_avatar.url if before.guild_avatar else None
		after_guild_avatar = after.guild_avatar.url if after.guild_avatar else None
		if after_guild_avatar != before_guild_avatar:
			changes.append(f'Guild avatar: <{before_guild_avatar}> -> <{after_guild_avatar}>')
		if after.flags != before.flags:
			flag_changes = []
			noteworthy_flags = { 'automod_quarantined_username', 'automod_quarantined_guild_tag' }
			for (name, after_value) in iter(after.flags):
				before_value = getattr(before.flags, name)
				if after_value != before_value:
					flag_changes.append(f'`{name}` = `{before_value}` -> `{after_value}`')
					if after_value and name in noteworthy_flags:
						noteworthy = True
			if len(flag_changes) > 0:
				changes.append(f'Guild-specific flag changes: {", ".join(flag_changes)}')
		if after.public_flags != before.public_flags:
			flag_changes = []
			noteworthy_flags = { 'spammer' }
			for (name, after_value) in iter(after.public_flags):
				before_value = getattr(before.public_flags, name)
				if after_value != before_value:
					flag_changes.append(f'`{name}` = `{before_value}` -> `{after_value}`')
					if after_value and name in noteworthy_flags:
						noteworthy = True
			if len(flag_changes) > 0:
				changes.append(f'Public flag changes: {", ".join(flag_changes)}')

		if len(changes) == 0:
			return
		text = f'Details for member {self.__describe_user(before, tag=noteworthy)} changed:\n'
		text += '- ' + ('\n- '.join(changes))
		bot_message = BotMessage(guild, text,
			BotMessage.TYPE_MOD_WARNING if noteworthy else BotMessage.TYPE_LOG)
		await bot_message.update()

	@Cog.listener()
	async def on_user_update(self, before: User, after: User) -> None:
		"""
		Called when a User updates their profile.

		This is called when one or more of the following things change:
		* avatar
		* username
		* discriminator

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_user_update

		Requires Members privileged intent.
		"""
		if not hasattr(after, 'guild'): return
		guild = after.guild
		if self.__is_member_updates_disabled(guild): return
		if self.__is_excluded_user(after): return

		changes = []
		before_avatar_url = before.avatar.url if before.avatar else None
		after_avatar_url = after.avatar.url if after.avatar else None
		if after_avatar_url != before_avatar_url:
			changes.append(f'Avatar URL: <{before_avatar_url}> -> <{after_avatar_url}>')
		if after.name != before.name:
			changes.append(f'Username: `{before.name}` -> `{after.name}`')
		if len(changes) == 0:
			return
		text = f'Details for user {self.__describe_user(before)} changed:\n'
		text += '* ' + '\n* '.join(changes)
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@Cog.listener()
	async def on_member_ban(self, guild: Guild, user: User | Member) -> None:
		"""
		Called when user gets banned from a Guild.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_member_ban
		"""
		if self.__is_cog_disabled(guild): return

		banner = None
		ban_reason = None
		entry = await self.__find_audit_entry(user, AuditLogAction.ban)
		if entry:
			banner = entry.user
			ban_reason = entry.reason
		if banner:
			text = f'Member {self.__describe_user(user)} banned by **{banner.name}**.'
			if ban_reason:
				text += f'\nReason: "{ban_reason}"'
		else:
			text = f'Member {self.__describe_user(user)} banned.'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	async def __find_audit_entry(self, user: User | Member, action: AuditLogAction, max_age: int = 10) -> AuditLogEntry | None:
		"""
		Searches the audit log for the most recent entry of a given type for a
		given user. Intended for finding the relevant entry for a ban/kick that
		just occurred.
		"""
		if hasattr(user, 'guild') and user.guild:
			guild = user.guild
		else:
			return None
		now = datetime.now(timezone.utc)
		async for entry in guild.audit_logs():
			age_seconds = now.timestamp() - entry.created_at.timestamp()
			if entry.action == action and entry.target == user and age_seconds <= max_age:
				return entry
		return None

	@Cog.listener()
	async def on_member_unban(self, guild: Guild, user: User | Member) -> None:
		"""
		Called when a User gets unbanned from a Guild.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_member_unban
		"""
		if self.__is_cog_disabled(guild): return

		text = f'Member {self.__describe_user(user)} unbanned'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	# Events - Messages

	def __buffer_event(self, guild: Guild, event_type: str, event) -> None:
		buffers: dict[str, list] = Storage.get_state_value(guild, self.STATE_EVENT_BUFFER)
		if buffers is None:
			buffers = {}
			Storage.set_state_value(guild, self.STATE_EVENT_BUFFER, buffers)
		if buffers.get(event_type) is None:
			buffers[event_type] = [ event ]
		else:
			buffers[event_type].append(event)
		self.buffered_guilds.add(guild)

	@tasks.loop(seconds=3.0)
	async def flush_buffers(self) -> None:
		try:
			if len(self.buffered_guilds) == 0:
				return
			guilds = set(self.buffered_guilds)
			self.buffered_guilds.clear()
			for guild in guilds:
				await self.__flush_buffers_for_guild(guild)
		except DiscordException as e:
			dump_stacktrace(e)

	async def __flush_buffers_for_guild(self, guild: Guild) -> None:
		buffers: dict[str, list] = Storage.get_state_value(guild, self.STATE_EVENT_BUFFER)
		if buffers is None:
			return
		Storage.set_state_value(guild, self.STATE_EVENT_BUFFER, None)
		for event_type, buffer in buffers.items():
			if event_type == 'edit':
				await self.__flush_edit_buffers(guild, buffer)
			elif event_type == 'delete':
				await self.__flush_delete_buffers(guild, buffer)

	@flush_buffers.before_loop
	async def before_flush_buffers_start(self) -> None:
		await self.bot.wait_until_ready()

	@Cog.listener()
	async def on_message(self, message: Message) -> None:
		"""
		Called when a Message is created and sent.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_message

		Content fields blank without MessageContent privileged intent.
		"""
		# if self.__is_cog_disabled(message.guild): return
		# if self.__is_excluded_channel(message.channel): return
		# if self.__is_excluded_user(message.author): return

		# print(f"on_message:"
		# 	f"\n\tid: {message.id}"
		# 	f"\n\tauthor: {message.author.name}"
		# 	f"\n\tchannel: {message.channel.name}"
		# 	f"\n\tcontent: \"{message.content}\""
		# 	f"\n\tattachments: x{len(message.attachments)}"
		# 	f"\n\tembeds: x{len(message.embeds)}"
		# 	f"\n\tcomponents: x{len(message.components)}")

	@Cog.listener()
	async def on_message_edit(self, before: Message, after: Message) -> None:
		"""
		Called when a Message receives an update event. If the message is not
		found in the internal message cache, then these events will not be
		called. Messages might not be in cache if the message is too old or the
		client is participating in high traffic guilds.

		If this occurs increase the max_messages parameter or use the
		on_raw_message_edit() event instead.

		The following non-exhaustive cases trigger this event:
		* A message has been pinned or unpinned.
		* The message content has been changed.
		* The message has received an embed.
			* For performance reasons, the embed server does not do this in a
			“consistent” manner.
		* The message’s embeds were suppressed or unsuppressed.
		* A call message has received an update to its participants or ending time.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_message_edit

		Content fields blank without MessageContent privileged intent.
		"""
		guild = after.guild
		if self.__is_edits_disabled(guild): return
		if self.__is_excluded_user(after.author): return
		channel = after.channel
		if self.__is_excluded_channel(channel): return

		self.__buffer_event(guild, 'edit', BufferedMessageEditEvent(guild, channel, before, after))

	@Cog.listener()
	async def on_raw_message_edit(self, payload: RawMessageUpdateEvent) -> None:
		"""
		Called when a message is edited. Unlike on_message_edit(), this is called
		regardless of the state of the internal message cache.

		If the message is found in the message cache, it can be accessed via
		RawMessageUpdateEvent.cached_message. The cached message represents the
		message before it has been edited. For example, if the content of a
		message is modified and triggers the on_raw_message_edit() coroutine,
		the RawMessageUpdateEvent.cached_message will return a Message object
		that represents the message before the content was modified.

		Due to the inherently raw nature of this event, the data parameter
		coincides with the raw data given by the gateway.

		Since the data payload can be partial, care must be taken when accessing
		stuff in the dictionary. One example of a common case of partial data is
		when the 'content' key is inaccessible. This denotes an “embed” only
		edit, which is an edit in which only the embeds are updated by the
		Discord embed server.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_raw_message_edit

		Content fields blank without MessageContent privileged intent.
		"""
		if payload.cached_message: return  # already handled by on_message_edit
		guild = self.bot.get_guild(payload.guild_id) or await self.bot.fetch_guild(payload.guild_id)
		if not guild or self.__is_edits_disabled(guild): return
		channel = guild.get_channel(payload.channel_id) or await guild.fetch_channel(payload.channel_id)
		if not channel or self.__is_excluded_channel(channel): return

		self.__buffer_event(guild, 'edit', BufferedMessageEditEvent(
			guild, channel, None, payload.message, payload.data))

	async def __flush_edit_buffers(self, guild: Guild, events: list[BufferedMessageEditEvent]) -> None:
		simple_edits: list[BufferedMessageEditEvent] = []
		complex_edits: list[BufferedMessageEditEvent] = []
		old_cutoff = timedelta(days=1)
		now = datetime.now(timezone.utc)
		for event in events:
			if event.before is not None and (now - norm_datetime(event.after.created_at)) < old_cutoff:
				simple_edits.append(event)
			else:
				complex_edits.append(event)
		if len(simple_edits) <= 3:
			# A small number of edits with full details. Log them individually.
			for event in simple_edits:
				await self.__handle_complete_edit_event(event)
		else:
			complex_edits = events
		if len(complex_edits) > 0:
			# These messages are not cached, too old, or too numerous
			text = 'Multiple messages edited' if len(complex_edits) > 1 else 'Message edited'
			for event in complex_edits[:10]:
				text += f'\n- {event.after.jump_url} by {event.after.author.name} ' + \
						f'first posted <t:{int(event.after.created_at.timestamp())}:f>'
			if len(complex_edits) > 10:
				text += f'\n- ...{len(complex_edits) - 10} more...'
			bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG, suppress_embeds=True)
			await bot_message.update()

	async def __handle_complete_edit_event(self, event: BufferedMessageEditEvent) -> None:
		before = event.before
		after = event.after
		guild = after.guild

		content_changed = (after.content != before.content)
		attachments_changed = (after.attachments != before.attachments)
		embeds_changed = (after.embeds != before.embeds)
		embeds_add_only = len(before.embeds or []) == 0 and len(after.embeds or []) > 0

		if not content_changed and not attachments_changed and (not embeds_changed or embeds_add_only):
			# Most likely an embed being asynchronously populated by server
			return
		if content_changed:
			(before_markdown, after_markdown) = self.__diff(self.__quote_markdown(before.content),
													  self.__quote_markdown(after.content))
		else:
			before_markdown = self.__quote_markdown(before.content)
			after_markdown = before_markdown if len(before.content.strip()) == 0 else '> _<content unchanged>_'
		if attachments_changed:
			if len(before.attachments or []) > 0:
				for attachment in before.attachments:
					before_markdown += f'\n> * 📎 {attachment.url}'
					if attachment not in after.attachments or []:
						before_markdown += ' (removed)'
			else:
				before_markdown += '\n> * _<no attachments>_'
			if len(after.attachments or []) > 0:
				for attachment in after.attachments:
					after_markdown += f'\n> * 📎 {attachment.url}'
					if attachment not in before.attachments or []:
						after_markdown += ' (added)'
			else:
				after_markdown += '\n> * _<no attachments>_'
		if embeds_changed:
			if len(before.embeds or []) > 0:
				for embed in before.embeds:
					before_markdown += f'\n> * 🔗 {embed.url}'
					if embed not in after.embeds or []:
						before_markdown += ' (removed)'
			else:
				before_markdown += '\n> * _<no embeds>_'
			if len(after.embeds or []) > 0:
				for embed in after.embeds:
					after_markdown += f'\n> * 🔗 {embed.url}'
					if embed not in before.embeds or []:
						after_markdown += ' (added)'
			else:
				after_markdown += '\n> * _<no embeds>_'
		text = f'Message {after.jump_url} edited by {self.__describe_user(after.author)}.\n' + \
			f'Original:\n{before_markdown}\n' + \
			f'Updated:\n{after_markdown}'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG, suppress_embeds=True)
		await bot_message.update()

	@Cog.listener()
	async def on_raw_message_delete(self, payload: RawMessageDeleteEvent) -> None:
		"""
		Called when a message is deleted. Unlike on_message_delete(), this is
		called regardless of the message being in the internal message cache or not.

		If the message is found in the message cache, it can be accessed via
		RawMessageDeleteEvent.cached_message

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_raw_message_delete

		Content fields blank without MessageContent privileged intent.
		"""
		message = payload.cached_message
		if message and self.__is_excluded_user(message.author): return
		guild = (message.guild if message else None) or \
				self.bot.get_guild(payload.guild_id) or \
				await self.bot.fetch_guild(payload.guild_id)
		if guild is None or self.__is_deletes_disabled(guild): return
		channel = (message.channel if message else None) or \
				  self.bot.get_channel(payload.channel_id) or \
				  await guild.fetch_channel(payload.channel_id)
		if channel is None or self.__is_excluded_channel(channel): return

		self.__buffer_event(guild, 'delete', BufferedMessageDeleteEvent(guild, channel, payload.message_id, message))

	@Cog.listener()
	async def on_raw_bulk_message_delete(self, payload: RawBulkMessageDeleteEvent) -> None:
		"""
		Called when a bulk delete is triggered. Unlike on_bulk_message_delete(),
		this is called regardless of the messages being in the internal message
		cache or not.

		If the messages are found in the message cache, they can be accessed via
		RawBulkMessageDeleteEvent.cached_messages

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_raw_bulk_message_delete

		Content fields blank without MessageContent privileged intent.
		"""
		guild = self.bot.get_guild(payload.guild_id) or await self.bot.fetch_guild(payload.guild_id)
		if not guild: return
		if self.__is_deletes_disabled(guild): return
		channel = guild.get_channel(payload.channel_id) or await guild.fetch_channel(payload.channel_id)
		if self.__is_excluded_channel(channel): return

		for message_id in payload.message_ids:
			message = None
			for cached_message in payload.cached_messages:
				if cached_message.id == message_id:
					message = cached_message
			self.__buffer_event(guild, 'delete', BufferedMessageDeleteEvent(guild, channel, message_id, message))

	async def __flush_delete_buffers(self, guild: Guild, events: list[BufferedMessageDeleteEvent]) -> None:
		simple_deletes: list[BufferedMessageDeleteEvent] = []
		complex_deletes: list[BufferedMessageDeleteEvent] = []
		for event in events:
			if event.message is not None:
				simple_deletes.append(event)
			else:
				complex_deletes.append(event)
		if len(simple_deletes) <= 3:
			# Small number of deletes with complete info
			for event in simple_deletes:
				await self.__handle_complete_delete_event(event)
		else:
			complex_deletes = events
		if len(complex_deletes) > 0:
			messages_per_author: dict[User | None, list[BufferedMessageDeleteEvent]] = self.__groupby(complex_deletes, lambda e: e.author)
			text = 'Multiple messages deleted' if len(complex_deletes) > 1 else 'Message deleted'
			row_count = 0
			for author, messages in messages_per_author.items():
				row_count += 1
				if row_count > 10:
					break
				count = len(messages)
				text += f'\n- {count} {"message" if count == 1 else "messages"} by {author.mention if author else "unavailable user"}'
				if count == 1:
					text += f' in {messages[0].channel.mention}'
				else:
					messages_by_channel: dict[GuildChannel, list[BufferedMessageDeleteEvent]] = self.__groupby(messages, lambda e: e.channel)
					if len(messages_by_channel) == 1:
						text += f' in {messages[0].channel.mention}'
					else:
						for channel, ch_messages in messages_by_channel.items():
							row_count += 1
							if row_count > 10:
								break
							ch_count = len(ch_messages)
							text += f'\n   - {ch_count} in {channel.mention}'
			if row_count > 10:
				text += '- ...more omitted...'
			bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG, suppress_embeds=True)
			await bot_message.update()

	async def __handle_complete_delete_event(self, event: BufferedMessageDeleteEvent) -> None:
		message: Message = event.message
		text = f'Message by {self.__describe_user(message.author)} deleted from {message.channel.mention}. ' + \
			   f'Markdown:\n{self.__quote_markdown(message.content)}'
		for attachment in message.attachments or []:
			text += f'\n> * 📎 {attachment.url}'
		for embed in message.embeds or []:
			text += f'\n> * 🔗 {embed.url}'
		bot_message = BotMessage(message.guild, text, BotMessage.TYPE_LOG, suppress_embeds=True)
		await bot_message.update()

	# Events - Roles

	@Cog.listener()
	async def on_guild_role_create(self, role: Role) -> None:
		"""
		Called when a Guild creates or deletes a new Role.

		To get the guild it belongs to, use Role.guild.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_guild_role_create
		"""
		guild = role.guild
		if self.__is_cog_disabled(guild): return

		text = f'Role created: **{role.name}**'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@Cog.listener()
	async def on_guild_role_delete(self, role: Role) -> None:
		"""
		Called when a Guild creates or deletes a new Role.

		To get the guild it belongs to, use Role.guild.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_guild_role_delete
		"""
		guild = role.guild
		if self.__is_cog_disabled(guild): return

		text = f'Role removed: **{role.name}**'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@Cog.listener()
	async def on_guild_role_update(self, before: Role, after: Role) -> None:
		"""
		Called when a Role is changed guild-wide.

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_guild_role_update
		"""
		guild = after.guild
		if self.__is_cog_disabled(guild): return

		changes = []
		if after.name != before.name:
			changes.append(f'Name: `{before.name}` -> `{after.name}`')
		if after.hoist != before.hoist:
			changes.append(f'Hoisted: `{before.hoist}` -> `{after.hoist}`')
		if after.position != before.position:
			changes.append(f'Position: `{before.position}` -> `{after.position}`')
		if after.unicode_emoji != before.unicode_emoji:
			changes.append(f'Emoji: {before.unicode_emoji} -> {after.unicode_emoji}')
		if after.mentionable != before.mentionable:
			changes.append(f'Mentionable: `{before.mentionable}` -> `{after.mentionable}`')
		if after.permissions != before.permissions:
			changes.append('Permissions edited')
		if after.color != before.color:
			changes.append('Color edited')
		before_icon_url = before.icon.url if before.icon else None
		after_icon_url = after.icon.url if after.icon else None
		if after_icon_url != before_icon_url:
			changes.append(f'Icon: <{before_icon_url}> -> <{after_icon_url}>')
		before_icon_url = before.display_icon.url if before.display_icon else None
		after_icon_url = after.display_icon.url if after.display_icon else None
		if after_icon_url != before_icon_url:
			changes.append(f'Display icon: <{before_icon_url}> -> <{after_icon_url}>')

		if len(changes) == 0:
			return
		text = f'Role **{after.name}** updated. Changes:\n'
		text += '* ' + '\n* '.join(changes)
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	# Events - Threads

	# @Cog.listener()
	# async def on_thread_create(self, thread: Thread) -> None:
	# 	pass

	# @Cog.listener()
	# async def on_thread_update(self, before: Thread, after: Thread) -> None:
	# 	pass

	# @Cog.listener()
	# async def on_thread_delete(self, thread: Thread) -> None:
	# 	pass


	# ------------------------------------------------------------------------
	def __quote_markdown(self, s: str) -> str:
		if len(s.strip()) == 0:
			return '> _<no content>_'
		return '> ' + escape_markdown(s).replace('\n', '\n> ')

	def __describe_user(self, user: User | Member, tag: bool = False) -> str:
		"""
		Standardized Markdown describing a user or member.
		"""
		if tag:
			return f'<@{user.id}> (`@{user.name}`, display "{user.display_name}", id `{user.id}`)'
		return f'**`@{user.name}`** (display "{user.display_name}", id `{user.id}`)'

	def __diff(self, a: str, b: str) -> tuple[str, str]:
		# URLs don't work well in the diffs. Replace them with private use characters, one per unique URL.
		preserved_sequences = []
		def sub_token(match: re.Match) -> str:
			seq = match.group(0)
			sequence_index = len(preserved_sequences)
			if seq in preserved_sequences:
				sequence_index = preserved_sequences.index(seq)
			else:
				preserved_sequences.append(seq)
			return chr(0xe000 + sequence_index)
		url_regex = r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+'
		a = re.sub(url_regex, sub_token, a)
		b = re.sub(url_regex, sub_token, b)

		deletion_start = '~~'
		deletion_end = '~~'
		addition_start = '**'
		addition_end = '**'
		markdown_a = ''
		markdown_b = ''
		a_open = False
		b_open = False
		for i, s in enumerate(difflib.ndiff(a, b)):
			operation = s[0]
			content = s[2:]
			if operation != '-' and a_open:
				markdown_a += deletion_end
				a_open = False
			if operation != '+' and b_open:
				markdown_b += addition_end
				b_open = False
			if operation == ' ':
				markdown_a += content
				markdown_b += content
			elif operation == '-':
				if not a_open:
					markdown_a += deletion_start
					a_open = True
				markdown_a += content
			elif operation == '+':
				if not b_open:
					markdown_b += addition_start
					b_open = True
				markdown_b += content
		if a_open:
			markdown_a += deletion_end
		if b_open:
			markdown_b += addition_end

		# Sub URLs back in
		def unsub_token(match: re.Match) -> str:
			char = match.group(0)
			index = ord(char) - 0xe000
			if 0 <= index < len(preserved_sequences):
				return preserved_sequences[index]
			return char
		markdown_a = re.sub(r'[\ue000-\uefff]', unsub_token, markdown_a)
		markdown_b = re.sub(r'[\ue000-\uefff]', unsub_token, markdown_b)

		return markdown_a, markdown_b

	def __groupby(self, a_list: list[Any], grouper: Callable[[Any], Any]) -> dict:
		"""itertools.groupby just less annoying"""
		d = {}
		for elem in a_list:
			key = grouper(elem)
			if key in d:
				d[key].append(elem)
			else:
				d[key] = [elem]
		return d

Self = LoggingCog