feat(admin): Add Admin-only Commands

src/commands/admin_commands.py

@@ -1,6 +1,7 @@
 import os
 import logging
-from discord import app_commands, Interaction
+from discord import app_commands, Interaction, Embed
+from models.databases.admin_settings_db import AdminSettingsDB
 
 # Retrieve the list of admin usernames from the .env file
 ADMIN_USERS = os.getenv("ADMIN_USERS", "").split(",")
@@ -9,9 +10,12 @@
 class AdminCommands(app_commands.Group):
     def __init__(self, gemini_bot):
         super().__init__(name="admin", description="Admin commands for DuckBot setup.")
+
+        # Initialise the database
+        self.settings_db = AdminSettingsDB()
 
         # Add subgroups to the main admin group
-        self.set = SetSubGroup(self.check_admin)
+        self.set = SetSubGroup(self.check_admin, self.settings_db)
         self.reset = ResetSubGroup(self.check_admin, gemini_bot)
 
         # Register subgroups
@@ -23,60 +27,50 @@ async def check_admin(self, interaction: Interaction) -> bool:
         logging.info(f"Checking admin status for user: {user_name}")
 
         if user_name in ADMIN_USERS:
-            logging.info(f"User {user_name} is authorized.")
+            logging.info(f"User {user_name} is authorised.")
             return True
         else:
-            logging.warning(f"User {user_name} is not authorized.")
+            await interaction.response.send_message("You don't have permission to execute that command.", ephemeral=True)
+            logging.warning(f"User {user_name} is not authorised.")
             return False
 
-    async def interaction_check(self, interaction: Interaction) -> bool:
-        """Restrict all admin commands visibility to authorized users only."""
-        is_admin = await self.check_admin(interaction)
-        if is_admin:
-            return True
-        await interaction.response.send_message("Unauthorized", ephemeral=True)
-        return False
-
     @app_commands.command(
-        name="log-info", description="Display all current environment variables."
+        name="log-variables", description="Display all current environment variables."
     )
     async def log_info(self, interaction: Interaction):
         """Command to log and display all relevant environment variables."""
         if not await self.check_admin(interaction):
-            await interaction.response.send_message("Unauthorized", ephemeral=True)
             return
 
-        # Collect environment variable values
-        guild_id = os.getenv("GUILD_ID", "Not Set")
-        skullboard_channel_id = os.getenv("SKULLBOARD_CHANNEL_ID", "Not Set")
-        required_reactions = os.getenv("REQUIRED_REACTIONS", "Not Set")
-        tenor_api_key = os.getenv("TENOR_API_KEY", "Not Set")
-        gemini_api_key = os.getenv("GEMINI_API_KEY", "Not Set")
-
-        # Construct a formatted message for environment variables
-        config_message = (
-            "**Current Environment Variables:**\n"
-            f"Guild ID: `{guild_id}`\n"
-            f"Skullboard Channel ID: `{skullboard_channel_id}`\n"
-            f"Required Reactions: `{required_reactions}`\n"
+        # Get values from database instead of env
+        guild_id = self.settings_db.get_setting("GUILD_ID") or "Not Set"
+        skullboard_channel_id = self.settings_db.get_setting("SKULLBOARD_CHANNEL_ID") or "Not Set"
+        required_reactions = self.settings_db.get_setting("REQUIRED_REACTIONS") or "Not Set"
+
+        embed = Embed(
+            title="Current Settings",
+            color=0x00ff00
         )
+        embed.add_field(name="Guild ID", value=f"`{guild_id}`", inline=False)
+        embed.add_field(name="Skullboard Channel ID", value=f"`{skullboard_channel_id}`", inline=False)
+        embed.add_field(name="Required Reactions", value=f"`{required_reactions}`", inline=False)
 
-        await interaction.response.send_message(config_message, ephemeral=True)
+        await interaction.response.send_message(embed=embed, ephemeral=True)
 
 
 class SetSubGroup(app_commands.Group):
-    def __init__(self, check_admin):
+    def __init__(self, check_admin, settings_db):
         super().__init__(
             name="set", description="Set configuration values for DuckBot."
         )
         self.check_admin = check_admin
+        self.settings_db = settings_db
 
     @app_commands.command(name="guild-id", description="Set the guild ID for DuckBot.")
     async def set_guild_id(self, interaction: Interaction, guild_id: str):
         if not await self.check_admin(interaction):
-            await interaction.response.send_message("Unauthorized", ephemeral=True)
             return
-        os.environ["GUILD_ID"] = guild_id
+        self.settings_db.set_setting("GUILD_ID", guild_id)
         await interaction.response.send_message(
             f"Guild ID set to {guild_id}.", ephemeral=True
         )
@@ -88,9 +82,8 @@ async def set_skullboard_channel_id(
         self, interaction: Interaction, channel_id: str
     ):
         if not await self.check_admin(interaction):
-            await interaction.response.send_message("Unauthorized", ephemeral=True)
             return
-        os.environ["SKULLBOARD_CHANNEL_ID"] = channel_id
+        self.settings_db.set_setting("SKULLBOARD_CHANNEL_ID", channel_id)
         await interaction.response.send_message(
             f"Skullboard channel ID set to {channel_id}.", ephemeral=True
         )
@@ -100,9 +93,8 @@ async def set_skullboard_channel_id(
     )
     async def set_required_reactions(self, interaction: Interaction, reactions: int):
         if not await self.check_admin(interaction):
-            await interaction.response.send_message("Unauthorized", ephemeral=True)
             return
-        os.environ["REQUIRED_REACTIONS"] = str(reactions)
+        self.settings_db.set_setting("REQUIRED_REACTIONS", str(reactions))
         await interaction.response.send_message(
             f"Required reactions set to {reactions}.", ephemeral=True
         )
@@ -117,7 +109,6 @@ def __init__(self, check_admin, gemini_bot):
     @app_commands.command(name="chat-history", description="Reset Gemini chat history.")
     async def reset_chat_history(self, interaction: Interaction):
         if not await self.check_admin(interaction):
-            await interaction.response.send_message("Unauthorized", ephemeral=True)
             return
 
         # Call the method to reset Gemini chat history

src/models/databases/admin_settings_db.py

@@ -0,0 +1,55 @@
+import os
+import sqlite3
+from pathlib import Path
+from models.schema.admin_settings_sql import AdminSettingsSQL
+
+class AdminSettingsDB:
+    def __init__(self, db_path: str = "db/admin_settings.db"):
+        # Ensure the data directory exists
+        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
+
+        self.db_path = db_path
+        self.init_db()
+
+    def get_db_connection(self):
+        return sqlite3.connect(self.db_path)
+
+    def init_db(self):
+        """Initialise the database with tables and default values from .env"""
+        with self.get_db_connection() as conn:
+            cursor = conn.cursor()
+
+            # Create tables
+            for statement in AdminSettingsSQL.initialisation_tables:
+                cursor.execute(statement)
+
+            # Initialise with default values from .env
+            default_settings = {
+                "GUILD_ID": os.getenv("GUILD_ID", ""),
+                "SKULLBOARD_CHANNEL_ID": os.getenv("SKULLBOARD_CHANNEL_ID", ""),
+                "REQUIRED_REACTIONS": os.getenv("REQUIRED_REACTIONS", "3")
+            }
+
+            # Only set defaults if the settings don't exist
+            for key, value in default_settings.items():
+                cursor.execute(
+                    "INSERT OR IGNORE INTO settings (key, value) VALUES (?, ?)",
+                    (key, value)
+                )
+
+            conn.commit()
+
+    def get_setting(self, key: str) -> str:
+        """Get a setting value from the database"""
+        with self.get_db_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(AdminSettingsSQL.get_setting, (key,))
+            result = cursor.fetchone()
+            return result[0] if result else None
+
+    def set_setting(self, key: str, value: str):
+        """Set a setting value in the database"""
+        with self.get_db_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(AdminSettingsSQL.set_setting, (key, value))
+            conn.commit() 

src/models/schema/admin_settings_sql.py

@@ -0,0 +1,21 @@
+class AdminSettingsSQL:
+    """Store SQL statements for the admin settings functionalities."""
+
+    initialisation_tables = [
+        """
+        CREATE TABLE IF NOT EXISTS settings (
+            key TEXT PRIMARY KEY,
+            value TEXT NOT NULL
+        );
+        """
+    ]
+
+    get_setting = """
+    SELECT value FROM settings WHERE key = ?;
+    """
+
+    set_setting = """
+    INSERT INTO settings (key, value) 
+    VALUES (?, ?)
+    ON CONFLICT(key) DO UPDATE SET value = excluded.value;
+    """ 

src/utils/settings.py

@@ -0,0 +1,10 @@
+from models.admin_settings_db import AdminSettingsDB
+import os
+
+def get_setting_with_fallback(key: str, default: str = None) -> str:
+    """Get a setting from the database, falling back to env vars if not found"""
+    db = AdminSettingsDB()
+    value = db.get_setting(key)
+    if value is None:
+        value = os.getenv(key, default)
+    return value