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 Emoji, Guild, GuildSticker, Invite, Member, Message, RawMessageUpdateEvent, Role, Thread, User
from discord.abc import GuildChannel
from discord.ext import commands
from typing import List, Union

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

class LogCog(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.')
	SETTING_EDITS_ENABLED = CogSetting('edits_enabled', bool,
			brief='post edits',
			description='Whether to log when users edit their posts.')
	SETTING_JOINS_ENABLED = CogSetting('joins_enabled', bool,
			brief='joins',
			description='Whether to log when new users join the server.')
	SETTING_LEAVES_ENABLED = CogSetting('leaves_enabled', bool,
			brief='leaves',
			description='Whether to log when users leave the server.')

	def __init__(self, bot):
		super().__init__(bot)
		self.add_setting(LogCog.SETTING_ENABLED)
		self.add_setting(LogCog.SETTING_EDITS_ENABLED)
		self.add_setting(LogCog.SETTING_JOINS_ENABLED)
		self.add_setting(LogCog.SETTING_LEAVES_ENABLED)

	@commands.group(
		brief='Manages event logging',
	)
	@commands.has_permissions(ban_members=True)
	@commands.guild_only()
	async def log(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:
		pass

	@commands.Cog.listener()
	async def on_guild_channel_create(self, channel: GuildChannel) -> None:
		pass

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

	# 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:
		pass

	@commands.Cog.listener()
	async def on_invite_delete(self, invite: Invite) -> None:
		pass

	# Events - Members

	@commands.Cog.listener()
	async def on_member_join(self, member: Member) -> None:
		pass

	@commands.Cog.listener()
	async def on_member_remove(self, member: Member) -> None:
		pass

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

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

	@commands.Cog.listener()
	async def on_member_ban(self, user: Union[User, Member]) -> None:
		pass

	@commands.Cog.listener()
	async def on_member_unban(self, guild: Guild, user: Union[User, Member]) -> None:
		pass

	# Events - Messages

	@commands.Cog.listener()
	async def on_message(self, message: Message):
		print(f"Saw message \"{message.content}\"")

	@commands.Cog.listener()
	async def on_message_edit(self, before: Message, after: Message) -> None:
		print(f"Message \"{before.content}\" changed to \"{after.content}\"")

	@commands.Cog.listener()
	async def on_raw_message_edit(self, payload: RawMessageUpdateEvent) -> None:
		if payload.cached_message:
			print("Edited message is cached")
			return  # already handled by on_message_edit
		print(f"Fetching guild {payload.guild_id}")
		guild = await self.bot.fetch_guild(payload.guild_id)
		if not guild:
			print(f"Couldn't fetch edited message guild {payload.guild_id}")
			return
		print(f"Fetching channel {payload.channel_id}")
		channel = await guild.fetch_channel(payload.channel_id)
		if not channel:
			print(f"Couldn't fetch edited message channel {payload.channel_id}")
			return
		print(f"Fetching message {payload.message_id}")
		message = await channel.fetch_message(payload.message_id)
		if not message:
			print(f"Couldn't fetch edited message {payload.message_id}")
			return
		print(f"Message was edited, but I don't have the old version. Current version is \"{message.content}\"")

	@commands.Cog.listener()
	async def on_message_delete(self, message: Message) -> None:
		print(f"Message deleted \"{message.content}\"")

	@commands.Cog.listener()
	async def on_bulk_message_delete(self, messages: List[Message]) -> None:
		pass

	# Events - Roles

	@commands.Cog.listener()
	async def on_guild_role_create(self, role: Role) -> None:
		pass

	@commands.Cog.listener()
	async def on_guild_role_delete(self, role: Role) -> None:
		pass

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

	# 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 remove_me():
		pass


	# @commands.Cog.listener()
	# async def on_member_join(self, member: Member) -> None:
	# 	'Event handler'
	# 	guild: Guild = member.guild
	# 	if not self.get_guild_setting(guild, self.SETTING_ENABLED):
	# 		return
	# 	min_count = self.get_guild_setting(guild, self.SETTING_JOIN_COUNT)
	# 	seconds = self.get_guild_setting(guild, self.SETTING_JOIN_TIME)
	# 	timespan: timedelta = timedelta(seconds=seconds)

	# 	last_raid: JoinRaidContext = Storage.get_state_value(guild, self.STATE_KEY_LAST_RAID)
	# 	recent_joins: AgeBoundList = Storage.get_state_value(guild, self.STATE_KEY_RECENT_JOINS)
	# 	if recent_joins is None:
	# 		recent_joins = AgeBoundList(timespan, lambda i, member : member.joined_at)
	# 		Storage.set_state_value(guild, self.STATE_KEY_RECENT_JOINS, recent_joins)
	# 	if last_raid:
	# 		if member.joined_at - last_raid.last_join_time() > timespan:
	# 			# Last raid is over
	# 			Storage.set_state_value(guild, self.STATE_KEY_LAST_RAID, None)
	# 			recent_joins.append(member)
	# 			return
	# 		# Add join to existing raid
	# 		last_raid.join_members.append(member)
	# 		self.record_warning(member)
	# 		if len(last_raid.banned_members) > 0:
	# 			self.log(guild, f'Banning as part of last join raid: {member.name}')
	# 			await member.ban(
	# 				reason='Rocketbot: Part of join raid.',
	# 				delete_message_days=0)
	# 			last_raid.banned_members.add(member)
	# 		elif len(last_raid.kicked_members) > 0:
	# 			self.log(guild, f'Kicking as part of last join raid: {member.name}')
	# 			await member.kick(
	# 				reason='Rocketbot: Part of join raid.')
	# 			last_raid.kicked_members.add(member)
	# 		await self.__update_warning_message(last_raid)
	# 	else:
	# 		# Add join to the general, non-raid recent join list
	# 		recent_joins.append(member)
	# 		if len(recent_joins) >= min_count:
	# 			self.log(guild, '\u0007Join raid detected')
	# 			last_raid = JoinRaidContext(recent_joins)
	# 			Storage.set_state_value(guild, self.STATE_KEY_LAST_RAID, last_raid)
	# 			recent_joins.clear()
	# 			msg = BotMessage(guild,
	# 					text='',
	# 					type=BotMessage.TYPE_MOD_WARNING,
	# 					context=last_raid)
	# 			self.record_warnings(recent_joins)
	# 			last_raid.warning_message_ref = weakref.ref(msg)
	# 			await self.__update_warning_message(last_raid)
	# 			await self.post_message(msg)