Hikari rewrite (#25)

Co-authored-by: wizzdom <[email protected]>
Co-authored-by: XOREAX <[email protected]>

Dockerfile

@@ -8,4 +8,4 @@ RUN pip install --no-cache-dir -r requirements.txt
 
 COPY . .
 
-CMD ["python3", "src/main.py"]
+CMD ["python3", "-m", "src"]

requirements.txt

@@ -1,5 +1,3 @@
-discord-typings==0.5.1
-jurigged==0.5.6
-discord-py-interactions==5.0.0
-python-dotenv==0.19.1
-loguru==0.7.2
+hikari==2.0.0.dev122
+hikari-arc==1.1.0
+python-dotenv==1.0.1

src/__main__.py

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

src/bot.py

@@ -0,0 +1,23 @@
+import logging
+import sys
+
+import arc
+import hikari
+
+from src.config import DEBUG, TOKEN
+
+if TOKEN is None:
+    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",
+)
+
+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/")

src/config.py

@@ -1,7 +1,10 @@
-import os
-from dotenv import load_dotenv
-
-load_dotenv()
-
-TOKEN = os.environ.get("TOKEN") # required
-DEBUG = os.environ.get("DEBUG", False)
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+TOKEN = os.environ.get("TOKEN")  # required
+DEBUG = os.environ.get("DEBUG", False)
+
+CHANNEL_IDS = {"lobby": 627542044390457350}

src/extensions/boosts.py

@@ -0,0 +1,50 @@
+import arc
+import hikari
+
+from src.config import CHANNEL_IDS
+
+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
+) -> 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,
+    )
+
+    message = f"{author.display_name} just boosted the server{f' **{content}** times' if content else ''}!"
+
+    if (count := TIER_COUNT[message_type]) is not None:
+        message += f"{guild.name} has achieved **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)
+
+
+@arc.loader
+def loader(client: arc.GatewayClient) -> None:
+    client.add_plugin(plugin)

src/extensions/hello_world.py

@@ -1,102 +1,81 @@
-"""
-Example extension with simple commands
-"""
-import interactions as discord
-
-
-class HelloWorld(discord.Extension):
-    @discord.slash_command("hello", description="Say hello!")
-    async def hello(self, ctx: discord.SlashContext):
-        """A simple hello world command"""
-        await ctx.send("Hello, world!")
-
-    @discord.slash_command(
-        "base_command", description="A base command, to expand on"
-    )
-    async def base_command(self, ctx: discord.SlashContext):
-        ...
-
-    @base_command.subcommand(
-        "sub_command", sub_cmd_description="A sub command, to expand on"
-    )
-    async def sub_command(self, ctx: discord.SlashContext):
-        """A simple sub command"""
-        await ctx.send("Hello, world! This is a sub command")
-
-    @discord.slash_command("options", description="A command with options")
-    @discord.slash_option(
-        "option_str",
-        "A string option",
-        opt_type=discord.OptionType.STRING,
-        required=True,
-    )
-    @discord.slash_option(
-        "option_int",
-        "An integer option",
-        opt_type=discord.OptionType.INTEGER,
-        required=True,
-    )
-    @discord.slash_option(
-        "option_attachment",
-        "An attachment option",
-        opt_type=discord.OptionType.ATTACHMENT,
-        required=True,
-    )
-    async def options(
-        self,
-        ctx: discord.SlashContext,
-        option_str: str,
-        option_int: int,
-        option_attachment: discord.Attachment,
-    ):
-        """A command with lots of options"""
-        embed = discord.Embed(
-            "There are a lot of options here",
-            description="Maybe too many",
-            color=discord.BrandColors.BLURPLE,
-        )
-        embed.set_image(url=option_attachment.url)
-        embed.add_field(
-            "String option",
-            option_str,
-            inline=False,
-        )
-        embed.add_field(
-            "Integer option",
-            str(option_int),
-            inline=False,
-        )
-        await ctx.send(embed=embed)
-
-    @discord.slash_command("components", description="A command with components")
-    async def components(self, ctx: discord.SlashContext):
-        """A command with components"""
-        await ctx.send(
-            "Here are some components",
-            components=discord.spread_to_rows(
-                discord.Button(
-                    label="Click me!",
-                    custom_id="click_me",
-                    style=discord.ButtonStyle.PRIMARY,
-                ),
-                discord.StringSelectMenu(
-                    "Select me!",
-                    "No, select me!",
-                    "Select me too!",
-                    placeholder="I wonder what this does",
-                    min_values=1,
-                    max_values=2,
-                    custom_id="select_me",
-                ),
-            ),
-        )
-
-    @discord.component_callback("click_me")
-    async def click_me(self, ctx: discord.ComponentContext):
-        """A callback for the click me button"""
-        await ctx.send("You clicked me!")
-
-    @discord.component_callback("select_me")
-    async def select_me(self, ctx: discord.ComponentContext):
-        """A callback for the select me menu"""
-        await ctx.send(f"You selected {' '.join(ctx.values)}")
+"""
+Example extension with simple commands
+"""
+import arc
+import hikari
+
+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(
+    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")],
+) -> 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.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)
+    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!"
+    )
+
+    await ctx.respond("Here are some components", components=[builder, button])
+
+
+@plugin.listen()
+async def on_interaction(event: hikari.InteractionCreateEvent) -> None:
+    interaction = event.interaction
+
+    # Discussions are underway for allowing to listen for a "ComponentInteractionEvent" directly
+    # instead of doing this manual filtering: https://github.com/hikari-py/hikari/issues/1777
+    if not isinstance(interaction, hikari.ComponentInteraction):
+        return
+
+    if interaction.custom_id == "click_me":
+        await interaction.create_initial_response(
+            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/userroles.py

@@ -0,0 +1,77 @@
+import arc
+import hikari
+
+plugin = arc.GatewayPlugin("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
+
+    role_id = int(role)
+    if role_id not in ctx.member.role_ids:
+        await ctx.client.rest.add_role_to_member(
+            ctx.guild_id, ctx.author, int(role), reason=f"{ctx.author} added role."
+        )
+        await ctx.respond(f"Done! Added <@&{role}> to your roles.", flags=hikari.MessageFlag.EPHEMERAL)
+        return
+
+    await ctx.respond(f"You already have <@&{role}>!", 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
+
+    role_id = int(role)
+    if role_id in ctx.member.role_ids:
+        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}> from your roles.", flags=hikari.MessageFlag.EPHEMERAL)
+        return
+
+    await ctx.respond(f"You don't have <@&{role}>!", flags=hikari.MessageFlag.EPHEMERAL)
+
+
+@add_role.set_error_handler
+async def add_error_handler(ctx: arc.GatewayContext, exc: Exception) -> None:
+    await role_error_handler(ctx, exc, "obtain")
+
+
+@remove_role.set_error_handler
+async def remove_error_handler(ctx: arc.GatewayContext, exc: Exception) -> None:
+    await role_error_handler(ctx, exc, "remove")
+
+
+async def role_error_handler(ctx: arc.GatewayContext, exc: Exception, type: str) -> None:
+    role = ctx.get_option("role", arc.OptionType.STRING)
+    assert role is not None
+    role_id = int(role)
+
+    if isinstance(exc, hikari.ForbiddenError):
+        await ctx.respond(f"You don't have permission to {type} <@&{role_id}>.", flags=hikari.MessageFlag.EPHEMERAL)
+        return
+
+    raise exc
+
+
+@arc.loader
+def loader(client: arc.GatewayClient) -> None:
+    client.add_plugin(plugin)