from datetime import datetime from logging import Logger from zoneinfo import ZoneInfo from discord import Cog, Message from classes.pycord_bot import PycordBot from enums import AnalyticsEventType from modules.database import col_analytics from modules.utils import get_logger logger: Logger = get_logger(__name__) class CogAnalytics(Cog): def __init__(self, bot: PycordBot): self.bot: PycordBot = bot @Cog.listener() async def on_message(self, message: Message) -> None: if ( message.guild is None or message.channel is None or message.author is None or message.author.bot ): return await col_analytics.insert_one( { "event_type": AnalyticsEventType.GUILD_MESSAGE_SENT.value, "event_date": message.created_at, "guild_id": message.guild.id, "channel_id": message.channel.id, "message_id": message.id, "user_id": message.author.id, "is_deleted": False, } ) @Cog.listener() async def on_message_edit(self, before: Message, after: Message) -> None: if ( after.guild is None or after.channel is None or after.author is None or after.author.bot ): return await col_analytics.insert_one( { "event_type": AnalyticsEventType.GUILD_MESSAGE_EDITED.value, "event_date": after.edited_at, "guild_id": after.guild.id, "channel_id": after.channel.id, "message_id": after.id, "user_id": after.author.id, } ) @Cog.listener() async def on_message_delete(self, message: Message) -> None: if ( message.guild is None or message.channel is None or message.author is None or message.author.bot ): return await col_analytics.insert_one( { "event_type": AnalyticsEventType.GUILD_MESSAGE_DELETED.value, "event_date": datetime.now(tz=ZoneInfo("UTC")), "guild_id": message.guild.id, "channel_id": message.channel.id, "message_id": message.id, "user_id": message.author.id, } ) await col_analytics.update_one( { "event_type": AnalyticsEventType.GUILD_MESSAGE_SENT.value, "guild_id": message.guild.id, "channel_id": message.channel.id, "message_id": message.id, }, { "$set": { "is_deleted": True, } }, ) def setup(bot: PycordBot) -> None: bot.add_cog(CogAnalytics(bot))