Refactor strings, add utils + root error handler

src/__main__.py

@@ -3,5 +3,5 @@
 
 from src.bot import bot
 
-if __name__ == "__main__":
-    bot.run(activity=hikari.Activity(name="Webgroup issues", type=hikari.ActivityType.WATCHING))
+if __name__ == '__main__':
+    bot.run(activity=hikari.Activity(name='Webgroup issues', type=hikari.ActivityType.WATCHING))

src/bot.py

@@ -7,17 +7,27 @@
 from src.config import DEBUG, TOKEN
 
 if TOKEN is None:
-    print("TOKEN environment variable not set. Exiting.")
+    print('TOKEN environment variable not set. Exiting.')
     sys.exit(1)
 
 bot = hikari.GatewayBot(
     token=TOKEN,
     banner=None,
     intents=hikari.Intents.ALL_UNPRIVILEGED | hikari.Intents.MESSAGE_CONTENT,
-    logs="DEBUG" if DEBUG else "INFO",
+    logs='DEBUG' if DEBUG else 'INFO',
 )
 
-logging.info(f"Debug mode is {DEBUG}; You can safely ignore this.")
+logging.info(f'Debug mode is {DEBUG}; You can safely ignore this.')
 
-arc_client = arc.GatewayClient(bot, is_dm_enabled=False)
-arc_client.load_extensions_from("./src/extensions/")
+client = arc.GatewayClient(bot, is_dm_enabled=False)
+client.load_extensions_from('./src/extensions/')
+
+@client.set_error_handler
+async def error_handler(ctx: arc.GatewayContext, exc: Exception) -> None:
+    if DEBUG:
+        message = f'```{exc}```'
+    else:
+        message = 'If this persists, create an issue at <https://webgroup-issues.redbrick.dcu.ie/>.'
+
+    await ctx.respond(f':x: Blockbot encountered an unhandled exception. {message}')
+    logging.error(exc)

src/config.py

@@ -4,7 +4,9 @@
 
 load_dotenv()
 
-TOKEN = os.environ.get("TOKEN")  # required
-DEBUG = os.environ.get("DEBUG", False)
+TOKEN = os.environ.get('TOKEN') # required
+DEBUG = os.environ.get('DEBUG', False)
 
-CHANNEL_IDS = {"lobby": 627542044390457350}
+CHANNEL_IDS = {
+    'lobby': '627542044390457350'
+}

src/extensions/boosts.py

@@ -2,47 +2,53 @@
 import hikari
 
 from src.config import CHANNEL_IDS
+from src.utils import get_guild
 
-plugin = arc.GatewayPlugin(name="Boosts")
+plugin = arc.GatewayPlugin(name='Boosts')
 
-TIER_COUNT: dict[hikari.MessageType, None | int] = {
-    hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION: None,
-    hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION_TIER_1: 1,
-    hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION_TIER_2: 2,
-    hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION_TIER_3: 3,
-}
+BOOST_TIERS: list[hikari.MessageType] = [
+    hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION_TIER_1,
+    hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION_TIER_2,
+    hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION_TIER_3,
+]
 
+BOOST_MESSAGE_TYPES: list[hikari.MessageType] = BOOST_TIERS + [
+    hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION,
+]
 
-# NOTE: this is baked into discord-interactions-py, so I extracted and cleaned up the logic
-def get_boost_message(
-    message_type: hikari.MessageType | int, content: str | None, author: hikari.Member, guild: hikari.Guild
+def build_boost_message(
+    message_type: hikari.MessageType | int,
+    number_of_boosts: str | None,
+    booster_user: hikari.Member,
+    guild: hikari.Guild
 ) -> str:
-    assert message_type in (
-        hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION,
-        hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION_TIER_1,
-        hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION_TIER_2,
-        hikari.MessageType.USER_PREMIUM_GUILD_SUBSCRIPTION_TIER_3,
-    )
+    assert message_type in BOOST_MESSAGE_TYPES
 
-    message = f"{author.display_name} just boosted the server{f' **{content}** times' if content else ''}!"
+    base_message = f'{booster_user.display_name} just boosted the server'
+    multiple_boosts_message = f' **{number_of_boosts}** times' if number_of_boosts else ''
 
-    if (count := TIER_COUNT[message_type]) is not None:
-        message += f"{guild.name} has achieved **Level {count}!**"
+    message = base_message + multiple_boosts_message + '!'
 
-    return message
+    if (message_type in BOOST_TIERS):
+        count = BOOST_TIERS.index(message_type)
+        message += f'\n{guild.name} has reached **Level {count}!**'
 
+    return message
 
 @plugin.listen()
 async def on_message(event: hikari.GuildMessageCreateEvent):
-    if event.message.type in TIER_COUNT:
-        assert event.member is not None
-        message = get_boost_message(
-            event.message.type,
-            event.content,
-            event.member,
-            event.get_guild() or await plugin.client.rest.fetch_guild(event.guild_id),
-        )
-        await plugin.client.rest.create_message(CHANNEL_IDS["lobby"], content=message)
+    if not event.message.type in BOOST_MESSAGE_TYPES:
+        return
+
+    assert event.member is not None
+    message = build_boost_message(
+        event.message.type,
+        number_of_boosts=event.content,
+        booster_user=event.member,
+        guild=await get_guild(plugin.client, event)
+    )
+
+    await plugin.client.rest.create_message(CHANNEL_IDS['lobby'], content=message)
 
 
 @arc.loader

src/extensions/hello_world.py

@@ -4,57 +4,60 @@
 import arc
 import hikari
 
-plugin = arc.GatewayPlugin(name="hello_world")
-
+plugin = arc.GatewayPlugin(name='Hello World')
 
 @plugin.include
-@arc.slash_command("hello", "Say hello!")
+@arc.slash_command('hello', 'Say hello!')
 async def hello(ctx: arc.GatewayContext) -> None:
     """A simple hello world command"""
-    await ctx.respond("Hello from hikari!")
-
-
-group = plugin.include_slash_group("base_command", "A base command, to expand on")
+    await ctx.respond('Hello from hikari!')
 
+group = plugin.include_slash_group('base_command', 'A base command, to expand on')
 
 @group.include
-@arc.slash_subcommand("sub_command", "A sub command, to expand on")
+@arc.slash_subcommand('sub_command', 'A sub command, to expand on')
 async def sub_command(ctx: arc.GatewayContext) -> None:
     """A simple sub command"""
-    await ctx.respond("Hello, world! This is a sub command")
-
+    await ctx.respond('Hello, world! This is a sub command')
 
 @plugin.include
-@arc.slash_command("options", "A command with options")
+@arc.slash_command('options', 'A command with options')
 async def options(
     ctx: arc.GatewayContext,
-    option_str: arc.Option[str, arc.StrParams("A string option")],
-    option_int: arc.Option[int, arc.IntParams("An integer option")],
-    option_attachment: arc.Option[hikari.Attachment, arc.AttachmentParams("An attachment option")],
+    option_str: arc.Option[str, arc.StrParams('A string option')],
+    option_int: arc.Option[int, arc.IntParams('An integer option')],
+    option_attachment: arc.Option[hikari.Attachment, arc.AttachmentParams('An attachment option')],
 ) -> None:
     """A command with lots of options"""
-    embed = hikari.Embed(title="There are a lot of options here", description="Maybe too many", colour=0x5865F2)
+    embed = hikari.Embed(
+      title='There are a lot of options here',
+      description='Maybe too many',
+      colour=0x5865F2
+    )
     embed.set_image(option_attachment)
-    embed.add_field("String option", option_str, inline=False)
-    embed.add_field("Integer option", str(option_int), inline=False)
+    embed.add_field('String option', option_str, inline=False)
+    embed.add_field('Integer option', str(option_int), inline=False)
     await ctx.respond(embed=embed)
 
-
 @plugin.include
-@arc.slash_command("components", "A command with components")
+@arc.slash_command('components', 'A command with components')
 async def components(ctx: arc.GatewayContext) -> None:
     """A command with components"""
     builder = ctx.client.rest.build_message_action_row()
-    select_menu = builder.add_text_menu("select_me", placeholder="I wonder what this does", min_values=1, max_values=2)
-    for opt in ("Select me!", "No, select me!", "Select me too!"):
+    select_menu = builder.add_text_menu(
+      'select_me',
+      placeholder='I wonder what this does',
+      min_values=1,
+      max_values=2
+    )
+    for opt in ('Select me!', 'No, select me!', 'Select me too!'):
         select_menu.add_option(opt, opt)
 
     button = ctx.client.rest.build_message_action_row().add_interactive_button(
-        hikari.ButtonStyle.PRIMARY, "click_me", label="Click me!"
+        hikari.ButtonStyle.PRIMARY, 'click_me', label='Click me!'
     )
 
-    await ctx.respond("Here are some components", components=[builder, button])
-
+    await ctx.respond('Here are some components', components=[builder, button])
 
 @plugin.listen()
 async def on_interaction(event: hikari.InteractionCreateEvent) -> None:
@@ -65,17 +68,17 @@ async def on_interaction(event: hikari.InteractionCreateEvent) -> None:
     if not isinstance(interaction, hikari.ComponentInteraction):
         return
 
-    if interaction.custom_id == "click_me":
+    if interaction.custom_id == 'click_me':
         await interaction.create_initial_response(
-            hikari.ResponseType.MESSAGE_CREATE, f"{interaction.user.mention}, you clicked me!"
+            hikari.ResponseType.MESSAGE_CREATE,
+            f'{interaction.user.mention}, you clicked me!'
         )
-    elif interaction.custom_id == "select_me":
+    elif interaction.custom_id == 'select_me':
         await interaction.create_initial_response(
             hikari.ResponseType.MESSAGE_CREATE,
             f"{interaction.user.mention}, you selected {' '.join(interaction.values)}",
         )
 
-
 @arc.loader
 def loader(client: arc.GatewayClient) -> None:
     client.add_plugin(plugin)

src/extensions/user_roles.py

@@ -0,0 +1,83 @@
+import arc
+import hikari
+
+from src.utils import role_mention
+
+plugin = arc.GatewayPlugin(name='User Roles')
+
+role = plugin.include_slash_group('role', 'Get/remove assignable roles.')
+
+role_choices = [
+    hikari.CommandChoice(name='Webgroup', value='1166751688598761583'),
+    hikari.CommandChoice(name='Gamez', value='1089204642241581139'),
+    hikari.CommandChoice(name='Croomer', value='1172696659097047050'),
+]
+
+@role.include
+@arc.slash_subcommand('add', 'Add an assignable role.')
+async def add_role(
+    ctx: arc.GatewayContext,
+    role: arc.Option[str, arc.StrParams('The role to add.', choices=role_choices)]
+) -> None:
+    assert ctx.guild_id
+    assert ctx.member
+
+    if int(role) in ctx.member.role_ids:
+        return await ctx.respond(
+          f'You already have the {role_mention(role)} role.',
+          flags=hikari.MessageFlag.EPHEMERAL
+        )
+
+    await ctx.client.rest.add_role_to_member(
+        ctx.guild_id, ctx.author, int(role), reason='Self-service role.'
+    )
+    await ctx.respond(
+      f'Done! Added {role_mention(role)} to your roles.',
+      flags=hikari.MessageFlag.EPHEMERAL
+    )
+
+@role.include
+@arc.slash_subcommand('remove', 'Remove an assignable role.')
+async def remove_role(
+    ctx: arc.GatewayContext,
+    role: arc.Option[str, arc.StrParams('The role to remove.', choices=role_choices)]
+) -> None:
+    assert ctx.guild_id
+    assert ctx.member
+
+    if int(role) not in ctx.member.role_ids:
+        return await ctx.respond(
+          f"You don't have the {role_mention(role)} role.",
+          flags=hikari.MessageFlag.EPHEMERAL
+        )
+
+    await ctx.client.rest.remove_role_from_member(
+        ctx.guild_id, ctx.author, int(role), reason=f'{ctx.author} removed role.'
+    )
+    await ctx.respond(
+      f'Done! Removed {role_mention(role)} from your roles.',
+      flags=hikari.MessageFlag.EPHEMERAL
+    )
+
+@role.set_error_handler
+async def role_error_handler(ctx: arc.GatewayContext, exc: Exception) -> None:
+    role = ctx.get_option('role', arc.OptionType.STRING)
+    assert role is not None
+
+    if isinstance(exc, hikari.ForbiddenError):
+        return await ctx.respond(
+          f':x: Blockbot is not permitted to self-service the {role_mention(role)} role.',
+          flags=hikari.MessageFlag.EPHEMERAL
+        )
+
+    if isinstance(exc, hikari.NotFoundError):
+        return await ctx.respond(
+          f":x: Blockbot can't find that role.",
+          flags=hikari.MessageFlag.EPHEMERAL
+        )
+
+    raise exc
+
+@arc.loader
+def loader(client: arc.GatewayClient) -> None:
+    client.add_plugin(plugin)