Migrated reaction role cog to new extension format

bot/config.py

@@ -1,6 +1,5 @@
 import json
 import logging
-from typing import Dict, List
 
 from pydantic import BaseModel, BaseSettings, PostgresDsn, ValidationError, validator
 
@@ -14,7 +13,7 @@ class AoC(BaseModel):
 
 
 class Bot(BaseModel):
-    commands_channels_ids: List[int]
+    commands_channels_ids: list[int]
     games_channel_id: int  # #bot-games
     token: str
 
@@ -48,7 +47,7 @@ class Guild(BaseModel):
 
 
 class Moderation(BaseModel):
-    admin_roles_ids: List[int]
+    admin_roles_ids: list[int]
     staff_role_id: int
 
     @validator("admin_roles_ids", pre=True)
@@ -62,16 +61,6 @@ class Postgres(BaseModel):
     uri: PostgresDsn
 
 
-class ReactionRoles(BaseModel):
-    required_role_id: int  # [lvl 20] Developer
-    roles: Dict[int, int]  # Dict[emoji_id, role_id]
-    message_id: int
-
-    @validator("roles", pre=True)
-    def val_func(cls, val):
-        return {int(k): v for k, v in json.loads(val).items()}
-
-
 class Tags(BaseModel):
     log_channel_id: int
     required_role_id: int  # [lvl 30] Engineer
@@ -110,7 +99,6 @@ class Settings(BaseSettings):
     postgres: Postgres
     guild: Guild
     moderation: Moderation
-    reaction_roles: ReactionRoles
     tags: Tags
     timathon: Timathon
     hastebin: Hastebin

bot/extensions/selectable_roles/__init__.py

@@ -0,0 +1,7 @@
+from bot.core import DiscordBot
+
+from .commands import SelectableRoleCommands
+
+
+async def setup(bot: DiscordBot) -> None:
+    await bot.add_cog(SelectableRoleCommands(bot=bot))

bot/extensions/selectable_roles/commands.py

@@ -0,0 +1,137 @@
+from typing import List
+
+import discord
+from discord import app_commands
+from discord.ext import commands
+from pydantic import BaseModel
+
+from bot import core
+from bot.models import SelectableRole
+
+
+class Role(BaseModel):
+    name: str
+    id: int
+
+
+class SelectableRoleCommands(commands.Cog):
+    admin_commands = app_commands.Group(
+        name="selectable-roles",
+        description="Commands for managing selectable roles",
+        default_permissions=discord.Permissions(administrator=True),
+        guild_only=True,
+    )
+
+    def __init__(self, bot: core.DiscordBot):
+        self.bot = bot
+        self.roles: dict[int, list[Role]] = {}
+
+    def update_roles(self, guild_id: int, data: tuple[str, int]) -> None:
+        if self.roles.get(guild_id):
+            for role in self.roles[guild_id]:
+                if role.id == data[1]:
+                    return
+            self.roles[guild_id].append(Role(name=data[0], id=data[1]))
+        else:
+            self.roles[guild_id] = [Role(name=data[0], id=data[1])]
+
+    async def cog_load(self) -> None:
+        query = "SELECT * FROM selectable_roles"
+        records = await SelectableRole.fetch(query)
+
+        for record in records:
+            self.update_roles(record.guild_id, (record.role_name, record.role_id))
+
+    async def role_autocomplete(self, interaction: discord.Interaction, current: str) -> List[app_commands.Choice[str]]:
+        if not self.roles.get(interaction.guild.id):
+            return []
+
+        return [
+            app_commands.Choice(name=role.name, value=str(role.id))
+            for role in self.roles[interaction.guild.id]
+            if current.lower() in role.name.lower()
+        ][:25]
+
+    @app_commands.command(name="get-role")
+    @app_commands.guild_only()
+    @app_commands.autocomplete(role=role_autocomplete)
+    async def get(
+        self,
+        interaction: core.InteractionType,
+        role: str,
+    ):
+        """Get the selected role"""
+
+        if not self.roles.get(interaction.guild.id) or not role.isdigit():
+            return await interaction.response.send_message("That role isn't selectable!", ephemeral=True)
+
+        role = interaction.guild.get_role(int(role))
+        if role is None or not any(role.id == role_.id for role_ in self.roles[interaction.guild.id]):
+            return await interaction.response.send_message("That role isn't selectable!", ephemeral=True)
+
+        await interaction.user.add_roles(role, reason="Selectable role")
+
+        to_remove = []
+        for role_ in self.roles[interaction.guild.id]:
+            if role_.id != role.id:
+                to_remove.append(interaction.guild.get_role(role_.id))
+        await interaction.user.remove_roles(*to_remove, reason="Selectable role")
+
+        await interaction.response.send_message(f"Successfully added {role.mention} to you!", ephemeral=True)
+
+    @admin_commands.command()
+    async def add(
+        self,
+        interaction: core.InteractionType,
+        role: discord.Role,
+    ):
+        """Add a selectable role to the database"""
+
+        if not role.is_assignable():
+            return await interaction.response.send_message(
+                "That role is non-assignable by the bot. Please ensure the bot has the necessary permissions.",
+                ephemeral=True,
+            )
+
+        await SelectableRole.ensure_exists(interaction.guild.id, role.id, role.name)
+        self.update_roles(interaction.guild.id, (role.name, role.id))
+        await interaction.response.send_message(f"Successfully added {role.mention} to the database!", ephemeral=True)
+
+    @admin_commands.command()
+    @app_commands.autocomplete(role=role_autocomplete)
+    async def remove(
+        self,
+        interaction: core.InteractionType,
+        role: str,
+    ):
+        """Remove a selectable role from the database"""
+
+        if not self.roles.get(interaction.guild.id):
+            return await interaction.response.send_message("There are no selectable roles!", ephemeral=True)
+
+        role = interaction.guild.get_role(int(role))
+        query = "DELETE FROM selectable_roles WHERE guild_id = $1 AND role_id = $2"
+        await SelectableRole.execute(query, interaction.guild.id, role.id)
+
+        for i, role_ in enumerate(self.roles[interaction.guild.id]):
+            if role_.id == role.id:
+                del self.roles[interaction.guild.id][i]
+                break
+
+        await interaction.response.send_message(
+            f"Successfully removed {role.mention} from the database!", ephemeral=True
+        )
+
+    @admin_commands.command()
+    async def list(
+        self,
+        interaction: core.InteractionType,
+    ):
+        """List all selectable roles"""
+
+        if not self.roles.get(interaction.guild.id):
+            return await interaction.response.send_message("There are no selectable roles!", ephemeral=True)
+
+        roles = [f"<@&{role.id}>" for role in self.roles[interaction.guild.id]]
+        embed = discord.Embed(title="Selectable roles", description="\n".join(roles), color=discord.Color.gold())
+        await interaction.response.send_message(embed=embed)

bot/models/__init__.py

@@ -7,6 +7,7 @@
 from .model import Model
 from .persisted_role import PersistedRole
 from .rep import Rep
+from .selectable_roles import SelectableRole
 from .tag import Tag
 from .user import User
 
@@ -22,4 +23,5 @@
     IgnoredChannel,
     LevellingRole,
     CustomRole,
+    SelectableRole,
 )  # Fixes F401

bot/models/migrations/006_down__selectable_roles.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS selectable_roles;

bot/models/migrations/006_up__selectable_roles.sql

@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS selectable_roles (
+    guild_id           BIGINT NOT NULL,
+    role_id            BIGINT NOT NULL,
+    role_name          VARCHAR NOT NULL,
+    PRIMARY KEY (guild_id, role_id)
+);

bot/models/selectable_roles.py

@@ -0,0 +1,20 @@
+from .model import Model
+
+
+class SelectableRole(Model):
+    guild_id: int
+    role_id: int
+    role_name: str
+
+    @classmethod
+    async def ensure_exists(cls, guild_id: int, role_id: int, role_name: str):
+        """Inserts or updates the selectable role."""
+        query = """
+            INSERT INTO selectable_roles (guild_id, role_id, role_name)
+                 VALUES ($1, $2, $3)
+            ON CONFLICT (guild_id, role_id)
+                DO UPDATE SET
+                     role_name = $3
+        """
+
+        return await cls.fetchrow(query, guild_id, role_id, role_name)

cli.py

@@ -140,9 +140,9 @@ async def main(ctx):
         "bot.extensions.custom_roles",
         "bot.extensions.polls",
         "bot.extensions.youtube",
+        "bot.extensions.selectable_roles",
         "bot.cogs._help",
         "bot.cogs.clashofcode",
-        "bot.cogs.roles",
     )
 
     intents = discord.Intents.all()

example.env

@@ -47,14 +47,6 @@ MODERATION__STAFF_ROLE_ID=
 NOTIFICATION__CHANNEL_ID=0
 NOTIFICATION__ROLE_ID=0
 
-# --- Reaction Roles
-# Access to reaction roles
-REACTION_ROLES__REQUIRED_ROLE_ID=0
-# Dict[emoji_id: int, role_id: int]
-# Leave no space or use double quotes `"` e.g: "{\"0\": 0}"
-REACTION_ROLES__ROLES={"0":0}
-REACTION_ROLES__MESSAGE_ID=0
-
 # --- Tags
 TAGS__LOG_CHANNEL_ID=0
 # Access to tag commands

utils/transformers.py

@@ -9,12 +9,18 @@
 
 
 class MessageTransformer(app_commands.Transformer):
+    """
+    Transform any of the given formats to a Message instance:
+
+    - 1176092816762486784  # message_id
+    - 1024375283857506436/1176092816762486784  # channel_id/message_id
+    - https://discord.com/channels/1024375277679284315/1024375283857506436/1176092816762486784  # message_url
+    """
+
     async def transform(self, interaction: core.InteractionType, value: str, /):
+        parts: list[str] = value.split("/")
         try:
-            parts: list[str] = value.split("/")
-
-            # check that there are 2 parts
-            if len(parts) != 2:
+            if len(parts) == 1:
                 return await interaction.channel.fetch_message(int(value))
 
             message_id = int(parts[-1])
@@ -23,7 +29,11 @@ async def transform(self, interaction: core.InteractionType, value: str, /):
             channel = interaction.guild.get_channel(channel_id)
             return await channel.fetch_message(message_id)
         except (ValueError, TypeError, IndexError, AttributeError):
-            await interaction.response.send_message("Please provide a valid message URL.", ephemeral=True)
+            if len(parts) == 1:
+                message = "Please provide a valid message ID."
+            else:
+                message = "Please provide a valid message URL."
+            await interaction.response.send_message(message, ephemeral=True)
         except discord.HTTPException:
             await interaction.response.send_message("Sorry, I couldn't find that message...", ephemeral=True)