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

"""
Cog for detecting large numbers of guild joins in a short period of time.
"""
import weakref
from collections.abc import Sequence
from datetime import datetime, timedelta
from discord import AuditLogAction, AuditLogEntry, Emoji, Guild, GuildSticker, Invite, Member, Message, RawBulkMessageDeleteEvent, RawMessageDeleteEvent, RawMessageUpdateEvent, Role, Thread, User
from discord.abc import GuildChannel
from discord.ext import commands
from discord.utils import escape_markdown
from typing import List, Optional, Union
import traceback

from config import CONFIG
from rocketbot.cogs.basecog import BaseCog, BotMessage, BotMessageReaction, CogSetting
from rocketbot.collections import AgeBoundList
from rocketbot.storage import Storage

class LoggingCog(BaseCog, name='Logging'):
	"""
	Cog for logging notable events to a designated logging channel.
	"""
	SETTING_ENABLED = CogSetting('enabled', bool,
			brief='logging',
			description='Whether this cog is enabled for a guild.')

	def __init__(self, bot):
		super().__init__(bot)
		self.add_setting(LoggingCog.SETTING_ENABLED)

	@commands.group(
		brief='Manages event logging',
	)
	@commands.has_permissions(ban_members=True)
	@commands.guild_only()
	async def logging(self, context: commands.Context):
		'Logging command group'
		if context.invoked_subcommand is None:
			await context.send_help()

	# Events - Channels

	@commands.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 not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		text = f'Channel **{channel.name}** deleted.'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@commands.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 not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		text = f'Channel **{channel.name}** created. {channel.mention}'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@commands.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 not self.get_guild_setting(guild, self.SETTING_ENABLED):
			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

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

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

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

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

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

	@commands.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 not self.get_guild_setting(guild, self.SETTING_ENABLED):
			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()

	@commands.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 not self.get_guild_setting(guild, self.SETTING_ENABLED):
			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

	@commands.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
		"""
		guild = member.guild
		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		text = f'Member joined server: {self.__describe_user(member)}.'
		flags = []
		noteworthy = False

		if member.flags.did_rejoin:
			flags.append('Rejoined this server')

		if member.public_flags.active_developer:
			flags.append('Is an active developer')
		if member.public_flags.hypesquad:
			flags.append('Is a HypeSquad Events member')
		if member.public_flags.hypesquad_bravery:
			flags.append('Is a HypeSquad Bravery member')
		if member.public_flags.hypesquad_brilliance:
			flags.append('Is a HypeSquad Brilliance member')
		if member.public_flags.hypesquad_balance:
			flags.append('Is a HypeSquad Balance member')
		if member.public_flags.early_supporter:
			flags.append('Is an early supporter')

		if member.public_flags.spammer:
			flags.append('**Is flagged as a spammer**')
			noteworthy = True
		if member.public_flags.discord_certified_moderator:
			flags.append('**Is a Discord Certified Moderator**')
			noteworthy = True
		if member.public_flags.early_verified_bot_developer:
			flags.append('**Is a verified bot developer**')
			noteworthy = True
		if member.public_flags.verified_bot:
			flags.append('**Is a verified bot**')
			noteworthy = True
		if member.public_flags.bug_hunter or member.public_flags.bug_hunter_level_2:
			flags.append('**Is a bug hunter**')
			noteworthy = True
		if member.public_flags.system:
			flags.append('**Is a Discord system user**')
			noteworthy = True
		if member.public_flags.staff:
			flags.append('**Is Discord staff**')
			noteworthy = True
		if member.public_flags.partner:
			flags.append('**Is a Discord partner**')
			noteworthy = True

		if len(flags) > 0:
			text += '\n* ' + '\n* '.join(flags)
		if noteworthy:
			text += f'\n\nLink: {member.mention}'
			bot_message = BotMessage(guild, text, BotMessage.TYPE_MOD_WARNING)
		else:
			bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@commands.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
		"""
		guild = member.guild
		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
			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:
			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()

	@commands.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
		"""
		guild = after.guild
		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		changes = []
		if after.nick != before.nick:
			changes.append(f'Nick: `{before.nick}` -> `{after.nick}`')
		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: ~~**{"**~~, ~~**".joined(removed_role_names)}**~~')
			if len(added_role_names) > 0:
				changes.append(f'Added roles: **{"**, **".joined(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()
				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 = []
			for (name, after_value) in after.iter():
				before_value = getattr(before, name)
				if after_value != before_value:
					flag_changes.append(f'`{name}` = `{before_value}` -> `{after_value}`')
			if len(flag_changes) > 0:
				changes.append(f'Flag changes: {", ".join(flag_changes)}')

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

	@commands.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
		"""
		if hasattr(after, 'guild'):
			guild = after.guild
		else:
			return
		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
			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()

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

		https://discordpy.readthedocs.io/en/stable/api.html#discord.on_member_ban
		"""
		if hasattr(user, 'guild'):
			guild = user
		else:
			return
		guild = user.guild
		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
			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: Union[User, Member], action: AuditLogAction, max_age: int = 10) -> Optional[AuditLogEntry]:
		"""
		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()
		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

	@commands.Cog.listener()
	async def on_member_unban(self, guild: Guild, user: Union[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 not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		text = f'Member {self.__describe_user(user)} unbanned'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	# Events - Messages

	@commands.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
		"""
		# print(f"Saw message {message.id} \"{message.content}\"")

	@commands.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
		"""
		guild = after.guild
		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		if after.author.id == self.bot.user.id:
			return
		if after.content == before.content:
			# Most likely an embed being updated
			return
		text = f'Message {after.jump_url} edited by {self.__describe_user(after.author)}.\n' + \
			f'Original markdown:\n{self.__quote_markdown(before.content)}\n' + \
			f'Updated markdown:\n{self.__quote_markdown(after.content)}'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG, suppress_embeds=True)
		await bot_message.update()

	@commands.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
		"""
		if payload.cached_message:
			return  # already handled by on_message_edit
		guild = await self.bot.fetch_guild(payload.guild_id)
		if not guild:
			return
		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		channel = await guild.fetch_channel(payload.channel_id)
		if not channel:
			return
		message = await channel.fetch_message(payload.message_id)
		if not message:
			return
		text = f'Message {message.jump_url} edited by {self.__describe_user(message.author)}.\n' + \
			'Original markdown unavailable in cache.\n' + \
			f'Updated markdown:\n{self.__quote_markdown(message.content)}'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG, suppress_embeds=True)
		await bot_message.update()

	@commands.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
		"""
		if payload.cached_message:
			message = payload.cached_message
			guild = message.guild
			if not self.get_guild_setting(guild, self.SETTING_ENABLED):
				return
			if message.author.id == self.bot.user.id:
				return
			text = f'Message by {self.__describe_user(message.author)} deleted from {message.channel.mention}. ' + \
				f'Markdown:\n{self.__quote_markdown(message.content)}'
			bot_message = BotMessage(message.guild, text, BotMessage.TYPE_LOG, suppress_embeds=True)
			await bot_message.update()
		else:
			guild = await self.bot.fetch_guild(payload.guild_id)
			if not guild:
				return
			if not self.get_guild_setting(guild, self.SETTING_ENABLED):
				return
			channel = await guild.fetch_channel(payload.channel_id)
			if not channel:
				return
			text = f'Message {payload.message_id} deleted in ' + channel.mention + ' but content and author not available in cache.'
			bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG, suppress_embeds=True)
			await bot_message.update()

	@commands.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
		"""
		guild = await self.bot.fetch_guild(payload.guild_id)
		if not guild:
			return
		if not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		channel = await guild.fetch_channel(payload.channel_id)
		count = len(payload.message_ids)
		cached_count = len(payload.cached_messages)
		uncached_count = count - cached_count
		text = f'Bulk deletion of {count} message(s) from {channel.mention}.'
		if uncached_count == count:
			text += f' No cached content available for any of them.'
		elif uncached_count > 0:
			text += f' No cached content available for {uncached_count} of them.'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

		for message in payload.cached_messages:
			text = f'Message by {self.__describe_user(message.author)} bulk deleted from {message.channel.mention}. ' + \
				f'Markdown:\n{self.__quote_markdown(message.content)}'
			bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
			await bot_message.update()

	# Events - Roles

	@commands.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 not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		text = f'Role created: **{role.name}**'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@commands.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 not self.get_guild_setting(guild, self.SETTING_ENABLED):
			return
		text = f'Role removed: **{role.name}**'
		bot_message = BotMessage(guild, text, BotMessage.TYPE_LOG)
		await bot_message.update()

	@commands.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 not self.get_guild_setting(guild, self.SETTING_ENABLED):
			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

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

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

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


	# ------------------------------------------------------------------------
	def __quote_markdown(self, s: str) -> str:
		return '> ' + escape_markdown(s).replace('\n', '\n> ')

	def __describe_user(self, user: Union[User, Member]) -> str:
		"""
		Standardized markdown describing a user or member.
		"""
		return f'**{user.name}** ({user.display_name} {user.id})'