Add files via upload

src/__main__.py

@@ -1,7 +1,6 @@
-"""Entrypoint script to load extensions and start the client."""
-import hikari
+from hikari import Activity, ActivityType
 
 from src.bot import bot
 
 if __name__ == "__main__":
-    bot.run(activity=hikari.Activity(name="Webgroup issues", type=hikari.ActivityType.WATCHING))
+    bot.run(activity=Activity(name="Webgroup issues", type=ActivityType.WATCHING))

src/bot.py

@@ -19,5 +19,17 @@
 
 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"❌ Blockbot encountered an unhandled exception. {message}")
+    logging.error(exc)
+
+    raise exc

src/config.py

@@ -13,4 +13,6 @@
     ("Croomer", "1172696659097047050"),
 )
 
-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")
 
-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,
-}
-
-
-# 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
+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,
+]
+
+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) + 1
+        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,26 +4,22 @@
 import arc
 import hikari
 
-plugin = arc.GatewayPlugin(name="hello_world")
-
+plugin = arc.GatewayPlugin(name="Hello World")
 
 @plugin.include
 @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")
 
-
 @group.include
 @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")
 
-
 @plugin.include
 @arc.slash_command("options", "A command with options")
 async def options(
@@ -33,19 +29,27 @@ async def options(
     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)
     await ctx.respond(embed=embed)
 
-
 @plugin.include
 @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)
+    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)
 
@@ -55,7 +59,6 @@ async def components(ctx: arc.GatewayContext) -> None:
 
     await ctx.respond("Here are some components", components=[builder, button])
 
-
 @plugin.listen()
 async def on_interaction(event: hikari.InteractionCreateEvent) -> None:
     interaction = event.interaction
@@ -67,15 +70,15 @@ async def on_interaction(event: hikari.InteractionCreateEvent) -> None:
 
     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":
         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/uptime.py

@@ -0,0 +1,25 @@
+from datetime import datetime
+
+import arc
+
+start_time = datetime.now()
+
+plugin = arc.GatewayPlugin("Blockbot Uptime")
+
+@plugin.include
+@arc.slash_command("uptime", "Show formatted uptime of Blockbot")
+async def uptime(ctx):
+    up_time = datetime.now() - start_time
+    d = up_time.days
+    h, ms = divmod(up_time.seconds, 3600)
+    m, s = divmod(ms, 60)
+
+    format = lambda val, str: f"{val} {str}{'s' if val != 1 else ''}"
+    message_parts = [(d, "day"), (h, "hour"), (m, "minute"), (s, "second")]
+    formatted_parts = [format(val, str) for val, str in message_parts if val]
+
+    await ctx.respond(f"Uptime: **{', '.join(formatted_parts)}**")
+
+@arc.loader
+def loader(client):
+    client.add_plugin(plugin)

src/extensions/user_roles.py

@@ -0,0 +1,80 @@
+import arc
+import hikari
+
+from src.utils import role_mention
+from src.config import ROLE_IDS
+
+plugin = arc.GatewayPlugin("User Roles")
+
+role = plugin.include_slash_group("role", "Get/remove assignable roles.")
+
+ROLE_CHOICES = [hikari.CommandChoice(name=name, value=value) for name,value in ROLE_IDS] # ROLE_IDS <- ROLE_CHOICES into CommandChoice-es; thus ROLE_CHOICES can just store easy-to-update tuples
+
+@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"❌ 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"❌ Blockbot can't find that role.",
+          flags=hikari.MessageFlag.EPHEMERAL
+        )
+
+    raise exc
+
+@arc.loader
+def loader(client: arc.GatewayClient) -> None:
+    client.add_plugin(plugin)

src/utils.py

@@ -0,0 +1,8 @@
+import hikari
+from arc import GatewayClient
+
+async def get_guild(client: GatewayClient, event: hikari.GuildMessageCreateEvent):
+    return event.get_guild() or await client.rest.fetch_guild(event.guild_id)
+
+def role_mention(role_id: hikari.Snowflake | int | str):
+    return f"<@&{role_id}>"