""" 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, 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, 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.id} \"{message.content}\"") pass @commands.Cog.listener() async def on_message_edit(self, before: Message, after: Message) -> None: text = f'Message {after.jump_url} edited by **{after.author.name}** ({after.author.display_name} {after.author.id}).\n\n' + \ f'Original markdown:\n> {escape_markdown(before.content)}\n\n' + \ f'Updated markdown:\n> {escape_markdown(after.content)}' bot_message = BotMessage(after.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: if payload.cached_message: return # already handled by on_message_edit guild = await self.bot.fetch_guild(payload.guild_id) if not guild: 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 **{message.author.name}** ({message.author.display_name} {message.author.id}).\n\n' + \ 'Original markdown unavailable in cache.\n\n' + \ f'Updated markdown:\n> {escape_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: print('Raw message deleted') if payload.cached_message: message = payload.cached_message text = f'Message by **{message.author.name}** ({message.author.display_name} {message.author.id}) deleted from {message.channel.mention}\n\n' + \ f'Markdown:\n> {escape_markdown(message.content)}' bot_message = BotMessage(message.guild, text, BotMessage.TYPE_LOG, suppress_embeds=True) await bot_message.update() else: print(f'Looking up guild {payload.guild_id}') guild = await self.bot.fetch_guild(payload.guild_id) if not guild: return print(f'Looking up channel {payload.channel_id}') 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: guild = await self.bot.fetch_guild(payload.guild_id) if not guild: 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 **{message.author.name}** ({message.author.display_name} {message.author.id}) bulk deleted from {message.channel.mention}\n\n' + \ f'Markdown:\n> {escape_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: 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)