First draft of /bangcommand

rocketbot/bot.py

@@ -91,6 +91,7 @@ def __create_bot():
 __create_bot()
 
 from rocketbot.cogs.autokickcog import AutoKickCog
+from rocketbot.cogs.bangcommandcog import BangCommandCog
 from rocketbot.cogs.configcog import ConfigCog
 from rocketbot.cogs.crosspostcog import CrossPostCog
 from rocketbot.cogs.gamescog import GamesCog
@@ -111,6 +112,7 @@ async def start_bot():
 
 	# Optional
 	await rocketbot.add_cog(AutoKickCog(rocketbot))
+	await rocketbot.add_cog(BangCommandCog(rocketbot))
 	await rocketbot.add_cog(CrossPostCog(rocketbot))
 	await rocketbot.add_cog(GamesCog(rocketbot))
 	await rocketbot.add_cog(HelpCog(rocketbot))

rocketbot/cogs/bangcommandcog.py

@@ -0,0 +1,170 @@
+import re
+from typing import Optional, TypedDict
+
+from discord import Interaction, Guild, Message, TextChannel
+from discord.app_commands import Group
+from discord.ext.commands import Cog
+
+from config import CONFIG
+from rocketbot.bot import Rocketbot
+from rocketbot.cogs.basecog import BaseCog
+from rocketbot.cogsetting import CogSetting
+from rocketbot.ui.pagedcontent import PAGE_BREAK, update_paged_content, paginate
+from rocketbot.utils import MOD_PERMISSIONS, blockquote_markdown, indent_markdown
+
+_CURRENT_DATA_VERSION = 1
+
+class BangCommand(TypedDict):
+    content: str
+    mod_only: bool
+    version: int
+
+class BangCommandCog(BaseCog, name='Bang Commands'):
+    SETTING_COMMANDS = CogSetting(
+        name='commands',
+        datatype=dict[str, BangCommand],
+        default_value={},
+    )
+
+    def __init__(self, bot: Rocketbot):
+        super().__init__(
+            bot,
+            config_prefix='bangcommand',
+            short_description='Provides custom informational chat !commands.',
+            long_description='Bang commands are simple one-word messages starting with an exclamation '
+                             '(bang) that will make the bot respond with simple informational replies. '
+                             'Useful for posting answers to frequently asked questions, reminding users '
+                             'of rules, and similar.'
+        )
+
+    def _get_commands(self, guild: Guild) -> dict[str, BangCommand]:
+        return self.get_guild_setting(guild, BangCommandCog.SETTING_COMMANDS)
+
+    def _set_commands(self, guild: Guild, commands: dict[str, BangCommand]) -> None:
+        self.set_guild_setting(guild, BangCommandCog.SETTING_COMMANDS, commands)
+
+    bang = Group(
+        name='bangcommand',
+        description='Provides custom informational chat !commands.',
+        guild_only=True,
+        default_permissions=MOD_PERMISSIONS,
+    )
+
+    @bang.command()
+    async def define(self, interaction: Interaction, name: str, definition: str, mod_only: bool = False) -> None:
+        """
+        Defines or redefines a bang command.
+
+        Parameters
+        ----------
+        interaction: Interaction
+        name: string
+            name of the command (lowercase a-z, underscores, and hyphens)
+        definition: string
+            content of the command
+        mod_only: bool
+            whether the command will only be recognized when a mod uses it
+        """
+        if not BangCommandCog._is_valid_name(name):
+            self.log(interaction.guild, f'{interaction.user.name} used command /bangcommand define {name} {definition}')
+            await interaction.response.send_message(
+                f'{CONFIG["failure_emoji"]} Invalid command name. Names must consist of lowercase letters, underscores, and hyphens (no spaces).',
+                ephemeral=True,
+            )
+            return
+        name = BangCommandCog._normalize_name(name)
+        cmds = self._get_commands(interaction.guild)
+        cmds[name] = {
+            'content': definition,
+            'mod_only': mod_only,
+            'version': _CURRENT_DATA_VERSION,
+        }
+        self._set_commands(interaction.guild, cmds)
+        await interaction.response.send_message(
+            f'{CONFIG["success_emoji"]} Command `!{name}` has been defined.\n\n{blockquote_markdown(definition)}',
+            ephemeral=True,
+        )
+
+    @bang.command()
+    async def undefine(self, interaction: Interaction, name: str) -> None:
+        """
+        Removes a bang command.
+
+        Parameters
+        ----------
+        interaction: Interaction
+        name: string
+            name of the previously defined command
+        """
+        name = BangCommandCog._normalize_name(name)
+        cmds = self._get_commands(interaction.guild)
+        if name not in cmds:
+            await interaction.response.send_message(
+                f'{CONFIG["failure_emoji"]} Command `!{name}` does not exist.',
+                ephemeral=True,
+            )
+            return
+        del cmds[name]
+        self._set_commands(interaction.guild, cmds)
+        await interaction.response.send_message(
+            f'{CONFIG["success_emoji"]} Command `!{name}` removed.',
+            ephemeral=True,
+        )
+
+    @bang.command()
+    async def list(self, interaction: Interaction) -> None:
+        """
+        Lists all defined bang commands.
+
+        Parameters
+        ----------
+        interaction: Interaction
+        """
+        cmds = self._get_commands(interaction.guild)
+        if cmds is None or len(cmds) == 0:
+            await interaction.response.send_message(
+                f'{CONFIG["info_emoji"]} No commands defined.',
+                ephemeral=True,
+            )
+            return
+        text = '## Commands'
+        for name, cmd in sorted(cmds.items()):
+            text += PAGE_BREAK + f'\n- `!{name}`'
+            if cmd['mod_only']:
+                text += ' - **mod only**'
+            text += f'\n{indent_markdown(cmd["content"])}'
+        pages = paginate(text)
+        await update_paged_content(interaction, None, 0, pages)
+
+    @Cog.listener()
+    async def on_message(self, message: Message) -> None:
+        if message.guild is None or message.channel is None or not isinstance(message.channel, TextChannel):
+            return
+        content = message.content
+        if content is None or not content.startswith('!') or not BangCommandCog._is_valid_name(content):
+            return
+        name = BangCommandCog._normalize_name(content)
+        cmds = self._get_commands(message.guild)
+        cmd = cmds.get(name, None)
+        if cmd is None:
+            return
+        if cmd['mod_only'] and not message.author.guild_permissions.ban_members:
+            return
+        text = cmd["content"]
+        # text = f'{text}\n\n-# {message.author.name} used `!{name}`'
+        await message.channel.send(
+            text,
+        )
+
+    @staticmethod
+    def _normalize_name(name: str) -> str:
+        name = name.lower().strip()
+        if name.startswith('!'):
+            name = name[1:]
+        return name
+
+    @staticmethod
+    def _is_valid_name(name: Optional[str]) -> bool:
+        if name is None:
+            return False
+        return re.match(r'^!?([a-z]+)([_-][a-z]+)*$', name) is not None

rocketbot/cogs/helpcog.py

@@ -3,13 +3,13 @@ import re
 import time
 from typing import Union, Optional, TypedDict
 
-from discord import Interaction, Permissions, AppCommandType, ButtonStyle, InteractionResponse
+from discord import Interaction, Permissions, AppCommandType
 from discord.app_commands import Group, Command, autocomplete, guild_only, command, Choice
-from discord.ui import ActionRow, Button, LayoutView, TextDisplay
 
 from config import CONFIG
 from rocketbot.bot import Rocketbot
 from rocketbot.cogs.basecog import BaseCog
+from rocketbot.ui.pagedcontent import update_paged_content, paginate
 from rocketbot.utils import MOD_PERMISSIONS, dump_stacktrace
 
 HelpTopic = Union[Command, Group, BaseCog]
@@ -420,88 +420,8 @@ class HelpCog(BaseCog, name='Help'):
 		await self.__send_paged_help(interaction, text)
 
 	async def __send_paged_help(self, interaction: Interaction, text: str) -> None:
-		pages = _paginate(text)
-		if len(pages) == 1:
-			await interaction.response.send_message(pages[0], ephemeral=True)
-		else:
-			await _update_paged_help(interaction, None, 0, pages)
-
-def _paginate(text: str) -> list[str]:
-	max_page_size = 2000
-	chunks = text.split(PAGE_BREAK)
-	pages = [ '' ]
-	for chunk in chunks:
-		if len(chunk) > max_page_size:
-			raise ValueError('Help content needs more page breaks! One chunk is too big for message.')
-		if len(pages[-1] + chunk) < max_page_size:
-			pages[-1] += chunk
-		else:
-			pages.append(chunk)
-	page_count = len(pages)
-	if page_count == 1:
-		return pages
-
-	# Do another pass and try to even out the page lengths
-	indices = [ i * len(chunks) // page_count for i in range(page_count + 1) ]
-	even_pages = [
-		''.join(chunks[indices[i]:indices[i + 1]])
-		for i in range(page_count)
-	]
-	for page in even_pages:
-		if len(page) > max_page_size:
-			# We made a page too big. Give up.
-			return pages
-	return even_pages
-
-async def _update_paged_help(interaction: Interaction, original_interaction: Optional[Interaction], current_page: int, pages: list[str]) -> None:
-	try:
-		view = _PagingLayoutView(current_page, pages, original_interaction or interaction)
-		resolved = interaction
-		if original_interaction is not None:
-			# We have an original interaction from the initial command and a
-			# new one from the button press. Use the original to swap in the
-			# new page in place, then acknowledge the new one to satisfy the
-			# API that we didn't fail.
-			await original_interaction.edit_original_response(
-				view=view,
-			)
-			if interaction is not original_interaction:
-				await interaction.response.defer(ephemeral=True, thinking=False)
-		else:
-			# Initial send
-			await resolved.response.send_message(
-				view=view,
-				ephemeral=True,
-				delete_after=60,
-			)
-	except BaseException as e:
-		dump_stacktrace(e)
-
-class _PagingLayoutView(LayoutView):
-	def __init__(self, current_page: int, pages: list[str], original_interaction: Optional[Interaction]):
-		super().__init__()
-		self.current_page: int = current_page
-		self.pages: list[str] = pages
-		self.text.content = self.pages[self.current_page]
-		self.original_interaction = original_interaction
-		if current_page <= 0:
-			self.handle_prev_button.disabled = True
-		if current_page >= len(self.pages) - 1:
-			self.handle_next_button.disabled = True
-
-	text = TextDisplay('')
-
-	row = ActionRow()
-
-	@row.button(label='< Prev')
-	async def handle_prev_button(self, interaction: Interaction, button: Button) -> None:
-		new_page = max(0, self.current_page - 1)
-		await _update_paged_help(interaction, self.original_interaction, new_page, self.pages)
-
-	@row.button(label='Next >')
-	async def handle_next_button(self, interaction: Interaction, button: Button) -> None:
-		new_page = min(len(self.pages) - 1, self.current_page + 1)
-		await _update_paged_help(interaction, self.original_interaction, new_page, self.pages)
+		pages = paginate(text)
+		await update_paged_content(interaction, None, 0, pages, delete_after=60)
 
 # Exclusions from keyword indexing
 trivial_words = {

rocketbot/ui/pagedcontent.py

@@ -0,0 +1,133 @@
+"""
+Provides a means of presenting long messages by paging them. Paging requires the
+source message to insert `PAGE_BREAK` characters at meaningful breaks, preferably
+at fairly uniform intervals.
+"""
+
+from typing import Optional
+
+from discord import Interaction
+from discord.ui import LayoutView, TextDisplay, ActionRow, Button
+
+from rocketbot.utils import dump_stacktrace
+
+PAGE_BREAK = '\f'
+
+def paginate(text: str) -> list[str]:
+	"""
+	Breaks long message text into one or more pages, using page break markers as
+	potential clean break points.
+	"""
+	max_page_size = 2000
+	chunks = text.split(PAGE_BREAK)
+	pages = [ '' ]
+	for chunk in chunks:
+		if len(chunk) > max_page_size:
+			raise ValueError('Help content needs more page breaks! One chunk is too big for message.')
+		if len(pages[-1] + chunk) < max_page_size:
+			pages[-1] += chunk
+		else:
+			pages.append(chunk)
+	page_count = len(pages)
+	if page_count == 1:
+		return pages
+
+	# Do another pass and try to even out the page lengths
+	indices = [ i * len(chunks) // page_count for i in range(page_count + 1) ]
+	even_pages = [
+		''.join(chunks[indices[i]:indices[i + 1]])
+		for i in range(page_count)
+	]
+	for page in even_pages:
+		if len(page) > max_page_size:
+			# We made a page too big. Give up.
+			return pages
+	return even_pages
+
+async def update_paged_content(
+		interaction: Interaction,
+		original_interaction: Optional[Interaction],
+		current_page: int,
+		pages: list[str],
+		**send_args,
+) -> None:
+	"""
+	Posts and/or updates the content of a message from paged content.
+
+	Parameters
+	----------
+	interaction : Interaction
+		the current interaction, either the initial one or from a button press
+	original_interaction : Interaction
+		the first interaction that triggered presentation of the content, or None
+		if this already is the first interaction
+	current_page : int
+		page index to show (assumed to be in bounds)
+	pages : list[str]
+		array of page content
+	**send_args
+		additional arguments to pass to the send_message method
+	"""
+	if len(pages) == 1:
+		# No paging needed
+		await interaction.response.send_message(
+			pages[0],
+			ephemeral=True,
+		)
+		return
+
+	try:
+		view = _PagingLayoutView(current_page, pages, original_interaction or interaction)
+		resolved = interaction
+		if original_interaction is not None:
+			# We have an original interaction from the initial command and a
+			# new one from the button press. Use the original to swap in the
+			# new page in place, then acknowledge the new one to satisfy the
+			# API that we didn't fail.
+			await original_interaction.edit_original_response(
+				view=view,
+			)
+			if interaction is not original_interaction:
+				await interaction.response.defer(ephemeral=True, thinking=False)
+		else:
+			# Initial send
+			await resolved.response.send_message(
+				view=view,
+				ephemeral=True,
+				**send_args,
+			)
+	except BaseException as e:
+		dump_stacktrace(e)
+
+class _PagingLayoutView(LayoutView):
+	def __init__(
+			self,
+			current_page: int,
+			pages: list[str],
+			original_interaction: Optional[Interaction],
+			**send_args,
+	) -> None:
+		super().__init__()
+		self.current_page: int = current_page
+		self.pages: list[str] = pages
+		self.text.content = self.pages[self.current_page] + f'\n\n_Page {self.current_page + 1} of {len(self.pages)}_'
+		self.original_interaction = original_interaction
+		if current_page <= 0:
+			self.handle_prev_button.disabled = True
+		if current_page >= len(self.pages) - 1:
+			self.handle_next_button.disabled = True
+		self.send_args = send_args
+
+	text = TextDisplay('')
+
+	row = ActionRow()
+
+	@row.button(label='< Prev')
+	async def handle_prev_button(self, interaction: Interaction, button: Button) -> None:
+		new_page = max(0, self.current_page - 1)
+		await update_paged_content(interaction, self.original_interaction, new_page, self.pages, **self.send_args)
+
+	@row.button(label='Next >')
+	async def handle_next_button(self, interaction: Interaction, button: Button) -> None:
+		new_page = min(len(self.pages) - 1, self.current_page + 1)
+		await update_paged_content(interaction, self.original_interaction, new_page, self.pages, **self.send_args)

rocketbot/utils.py

@@ -171,6 +171,14 @@ def str_from_quoted_str(val: str) -> str:
 		raise ValueError(f'Not a quoted string: {val}')
 	return val[1:-1]
 
+def blockquote_markdown(markdown: str) -> str:
+	"""Encloses some Markdown in a blockquote."""
+	return '> ' + (markdown.replace('\n', '\n> '))
+
+def indent_markdown(markdown: str) -> str:
+	"""Indents a block of Markdown by one level."""
+	return '    ' + (markdown.replace('\n', '\n    '))
+
 MOD_PERMISSIONS: Permissions = Permissions(Permissions.manage_messages.flag)
 
 from discord import Interaction