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

"""
Base cog class and helper classes.
"""
from datetime import datetime, timedelta, timezone
from typing import Optional

import discord
from discord import (
	Guild,
	Interaction,
	Member,
	Message,
	RawReactionActionEvent,
	TextChannel,
)
from discord.abc import GuildChannel
from discord.app_commands import AppCommandError
from discord.app_commands.errors import CommandInvokeError
from discord.ext.commands import Cog

from config import CONFIG
from rocketbot.bot import Rocketbot
from rocketbot.botmessage import BotMessage, BotMessageReaction
from rocketbot.cogsetting import CogSetting
from rocketbot.collections import AgeBoundDict
from rocketbot.storage import Storage
from rocketbot.utils import bot_log, dump_stacktrace


class WarningContext:
	def __init__(self, member: Member, warn_time: datetime):
		self.member = member
		self.last_warned = warn_time

class BaseCog(Cog):
	STATE_KEY_RECENT_WARNINGS = "BaseCog.recent_warnings"

	"""
	Superclass for all Rocketbot cogs. Provides lots of conveniences for
	common tasks.
	"""
	def __init__(
			self,
			bot: Rocketbot,
			config_prefix: Optional[str],
			short_description: str,
			long_description: Optional[str] = None,
	):
		"""
		Parameters
		----------
		bot: Rocketbot
		config_prefix: str
		    Prefix to show on variables in /set and /get commands to namespace
		    configuration variables. E.g. if config_prefix is "foo", a config
		    variable named "bar" in that cog will show as "foo.bar". If None,
		    config variable acts as a top-level variable with no prefix.
		"""
		self.bot: Rocketbot = bot
		self.are_settings_setup: bool = False
		self.settings: list[CogSetting] = []
		self.config_prefix: Optional[str] = config_prefix
		self.short_description: str = short_description
		self.long_description: str = long_description

	async def cog_app_command_error(self, interaction: Interaction, error: AppCommandError) -> None:
		if isinstance(error, CommandInvokeError):
			error = error.original
		dump_stacktrace(error)
		message = f"\nException: {error.__class__.__name__}, "\
				  f"Command: {interaction.command.qualified_name if interaction.command else None}, "\
				  f"User: {interaction.user}, "\
				  f"Time: {discord.utils.format_dt(interaction.created_at, style='F')}"
		try:
			await interaction.response.send_message(f"An error occurred: {message}", ephemeral=True)
		except discord.InteractionResponded:
			await interaction.followup.send(f"An error occurred: {message}", ephemeral=True)

	@property
	def basecogs(self) -> list['BaseCog']:
		"""
		List of BaseCog instances. Cogs that do not inherit from BaseCog are omitted.
		"""
		return [
			bcog
			for bcog in sorted(self.bot.cogs.values(), key=lambda c: c.qualified_name)
			if isinstance(bcog, BaseCog)
		]

	@property
	def basecog_map(self) -> dict[str, 'BaseCog']:
		"""
		Map of qualified names to BaseCog instances. Cogs that do not inherit
		from BaseCog are omitted.
		"""
		return {
			qname: bcog
			for qname, bcog in self.bot.cogs.items()
			if isinstance(bcog, BaseCog)
		}

	# Config

	def add_setting(self, setting: CogSetting) -> None:
		"""
		Called by a subclass in __init__ to register a mod-configurable
		guild setting. A "get" and "set" command will be generated. If the
		setting is named "enabled" (exactly) then "enable" and "disable"
		commands will be created instead which set the setting to True/False.

		If the cog has a command group it will be detected automatically and
		the commands added to that. Otherwise, the commands will be added at
		the top level.

		Changes to settings can be detected by overriding `on_setting_updated`.
		"""
		self.settings.append(setting)

	@classmethod
	def get_guild_setting(cls,
			guild: Optional[Guild],
			setting: CogSetting,
			use_cog_default_if_not_set: bool = True):
		"""
		Returns the configured value for a setting for the given guild. If no
		setting is configured the default for the cog will be returned,
		unless the optional `use_cog_default_if_not_set` is `False`, then
		`None` will be returned.
		"""
		if guild:
			key = f'{cls.__name__}.{setting.name}'
			value = Storage.get_config_value(guild, key)
			if value is not None:
				return value
		if use_cog_default_if_not_set:
			return setting.default_value
		return None

	@classmethod
	def set_guild_setting(cls,
			guild: Guild,
			setting: CogSetting,
			new_value) -> None:
		"""
		Manually sets a setting for the given guild. BaseCog creates "get" and
		"set" commands for guild administrators to configure values themselves,
		but this method can be used for hidden settings from code. A ValueError
		will be raised if the new value does not pass validation specified in
		the CogSetting.
		"""
		setting.validate_value(new_value)
		key = f'{cls.__name__}.{setting.name}'
		Storage.set_config_value(guild, key, new_value)

	# @commands.Cog.listener()
	async def __on_ready(self):
		"""Event listener"""
		if not self.are_settings_setup:
			self.are_settings_setup = True
			CogSetting.set_up_all(self, self.bot, self.settings)

	async def on_setting_updated(self, guild: Guild, setting: CogSetting) -> None:
		"""
		Subclass override point for being notified when a CogSetting is edited.
		"""

	# Warning squelch

	def was_warned_recently(self, member: Member) -> bool:
		"""
		Tests if a given member was included in a mod warning message recently.
		Used to suppress redundant messages. Should be checked before pinging
		mods for relatively minor warnings about single users, but warnings
		about larger threats involving several members (e.g. join raids) should
		issue warnings regardless. Call record_warning or record_warnings after
		triggering a mod warning.
		"""
		recent_warns: AgeBoundDict[int, WarningContext, datetime, timedelta] = Storage.get_state_value(member.guild,
			BaseCog.STATE_KEY_RECENT_WARNINGS)
		if recent_warns is None:
			return False
		context: WarningContext = recent_warns.get(member.id)
		if context is None:
			return False
		squelch_warning_seconds: int = CONFIG['squelch_warning_seconds']
		return datetime.now() - context.last_warned < timedelta(seconds=squelch_warning_seconds)

	def record_warning(self, member: Member):
		"""
		Records that mods have been warned about a member and do not need to be
		warned about them again for a short while.
		"""
		recent_warns: AgeBoundDict[int, WarningContext, datetime, timedelta] = Storage.get_state_value(member.guild,
			BaseCog.STATE_KEY_RECENT_WARNINGS)
		if recent_warns is None:
			recent_warns = AgeBoundDict(timedelta(seconds=CONFIG['squelch_warning_seconds']),
				lambda i, context0 : context0.last_warned)
			Storage.set_state_value(member.guild, BaseCog.STATE_KEY_RECENT_WARNINGS, recent_warns)
		context: WarningContext = recent_warns.get(member.id)
		if context is None:
			context = WarningContext(member, datetime.now())
			recent_warns[member.id] = context
		else:
			context.last_warned = datetime.now()

	def record_warnings(self, members: list[Member]):
		"""
		Records that mods have been warned about some members and do not need to
		be warned about them again for a short while.
		"""
		for member in members:
			self.record_warning(member)

	# Bot message handling

	@classmethod
	def __bot_messages(cls, guild: Guild) -> AgeBoundDict[int, BotMessage, datetime, timedelta]:
		bm: AgeBoundDict[int, BotMessage, datetime, timedelta] = Storage.get_state_value(guild, 'bot_messages')
		if bm is None:
			far_future = datetime.now(timezone.utc) + timedelta(days=1000)
			bm = AgeBoundDict(timedelta(seconds=600),
				lambda k, v : v.message_sent_at() or far_future)
			Storage.set_state_value(guild, 'bot_messages', bm)
		return bm

	async def post_message(self, message: BotMessage) -> bool:
		"""
		Posts a BotMessage to a guild. Returns whether it was successful. If
		the caller wants to listen to reactions they should be added before
		calling this method. Listen to reactions by overriding `on_mod_react`.
		"""
		message.source_cog = self
		await message.update()
		return message.is_sent()

	@Cog.listener()
	async def on_raw_reaction_add(self, payload: RawReactionActionEvent):
		"""Event handler"""
		# Avoid any unnecessary requests. Gets called for every reaction
		# multiplied by every active cog.
		if payload.user_id == self.bot.user.id:
			# Ignore bot's own reactions
			return

		guild: Guild = self.bot.get_guild(payload.guild_id) or await self.bot.fetch_guild(payload.guild_id)
		if guild is None:
			# Possibly a DM
			return

		guild_messages: dict[int, BotMessage] = Storage.get_bot_messages(guild)
		bot_message = guild_messages.get(payload.message_id)
		if bot_message is None:
			# Unknown message (expired or was never tracked)
			return
		if self is not bot_message.source_cog:
			# Belongs to a different cog
			return
		reaction = bot_message.reaction_for_emoji(payload.emoji)
		if reaction is None or not reaction.is_enabled:
			# Can't use this reaction with this message
			return

		g_channel: GuildChannel = guild.get_channel(payload.channel_id) or await guild.fetch_channel(payload.channel_id)
		if g_channel is None:
			# Possibly a DM
			return
		if not isinstance(g_channel, TextChannel):
			return
		channel: TextChannel = g_channel
		member: Member = payload.member
		if member is None:
			return
		if not channel.permissions_for(member).ban_members:
			# Not a mod (could make permissions configurable per BotMessageReaction some day)
			return

		message: Message = await channel.fetch_message(payload.message_id)
		if message is None:
			# Message deleted?
			return
		if message.author.id != self.bot.user.id:
			# Bot didn't author this
			return

		await self.on_mod_react(bot_message, reaction, member)

	async def on_mod_react(self,
			bot_message: BotMessage,
			reaction: BotMessageReaction,
			reacted_by: Member) -> None:
		"""
		Subclass override point for receiving mod reactions to bot messages sent
		via `post_message()`.
		"""

	# Helpers

	@classmethod
	def log(cls, guild: Optional[Guild], message) -> None:
		"""
		Writes a message to the console. Intended for significant events only.
		"""
		bot_log(guild, cls, message)