Merge pull request #22 from albertsgarde/21-change-role-capabilities

21 change role capabilities

.gitignore

@@ -162,6 +162,7 @@ cython_debug/
 #.idea/
 
 db*.json
+.bot_config.toml
 
 .vscode/
 .ruff_cache/

Makefile

@@ -8,4 +8,4 @@ cov:
 	uv run pytest --cov --cov-report=term-missing
 
 launch:
-	. ./.env && uv run eadk_discord
+	export EADK_DISCORD_CONFIG_PATH=.bot_config.toml && uv run eadk_discord

eadk_discord/__main__.py

@@ -1,34 +1,33 @@
 # pragma: coverage exclude file
 import logging
 import os
+import sys
+import tomllib
 from pathlib import Path
 
-import discord
-from discord.abc import Snowflake
-
 from eadk_discord import bot_setup
 
-if __name__ == "__main__":
-    logging.basicConfig(level=logging.DEBUG)
 
-    bot_token_option: str | None = os.getenv("DISCORD_BOT_TOKEN")
-    if bot_token_option is None:
-        raise ValueError("DISCORD_BOT_TOKEN is not set in environment variables")
-    else:
-        bot_token: str = bot_token_option
+def get_config_path(env_var_name: str) -> str:
+    match len(sys.argv):
+        case 1:
+            env_var_option: str | None = os.getenv(env_var_name)
+            if env_var_option is None:
+                raise ValueError(f"environment variable '{env_var_name}' is not set")
+            else:
+                return env_var_option
+        case 2:
+            return sys.argv[1]
+        case num_args:
+            raise ValueError(f"expected 0 or 1 argument, got {num_args-1}")
 
-    database_path_option: str | None = os.getenv("DATABASE_PATH")
-    if database_path_option is None:
-        raise ValueError("DATABASE_PATH is not set in environment variables")
-    else:
-        database_path: Path = Path(database_path_option)
 
-    guild_ids_option: str | None = os.getenv("GUILD_IDS")
-    if guild_ids_option is None:
-        raise ValueError("GUILD_IDS is not set in environment variables")
-    else:
-        guilds: list[Snowflake] = [discord.Object(id=int(guild_id)) for guild_id in guild_ids_option.split(",")]
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
+
+    config_path = Path(get_config_path("EADK_DISCORD_CONFIG_PATH"))
 
-    bot = bot_setup.setup_bot(database_path, guilds)
+    with open(config_path, "rb") as config_file:
+        config = bot_setup.BotConfig.model_validate(tomllib.load(config_file))
 
-    bot.run(bot_token)
+    config.run_bot()

eadk_discord/bot.py

@@ -20,14 +20,23 @@ class CommandInfo(BaseModel):
     now: datetime = Field()
     format_user: Callable[[int], str] = Field()
     author_id: int = Field()
+    author_role_ids: set[int] = Field()
 
     @beartype
     @staticmethod
     def from_interaction(interaction: discord.Interaction) -> "CommandInfo":
+        match interaction.user:
+            case discord.Member() as member:
+                role_ids = set(role.id for role in member.roles)
+            case discord.User():
+                role_ids = set()
+            case _:
+                raise ValueError("Invalid interaction user type")
         return CommandInfo(
             now=datetime.now(TIME_ZONE),
             format_user=lambda user: fmt.user(interaction, user),
             author_id=interaction.user.id,
+            author_role_ids=role_ids,
         )
 
 
@@ -53,10 +62,20 @@ async def send(self, interaction: discord.Interaction) -> None:  # pragma: no co
 
 class EADKBot:
     _database: Database
+    _regular_role_ids: set[int]
+    _admin_role_ids: set[int]
 
     @beartype
-    def __init__(self, database: Database) -> None:
+    def __init__(self, database: Database, regular_role_ids: set[int], admin_role_ids: set[int]) -> None:
         self._database = database
+        self._regular_role_ids = regular_role_ids
+        self._admin_role_ids = admin_role_ids
+
+    def _is_author_regular(self, info: CommandInfo) -> bool:
+        return bool(info.author_role_ids.intersection(self._regular_role_ids.union(self._admin_role_ids)))
+
+    def _is_author_admin(self, info: CommandInfo) -> bool:
+        return bool(info.author_role_ids.intersection(self._admin_role_ids))
 
     @property
     def database(self) -> Database:
@@ -123,12 +142,15 @@ def book(
         if end_date is not None:
             days = self._database.state.day_range(booking_date, end_date)
             for day in days:
-                if day.desk(desk_index).owner is not info.author_id:
+                if day.desk(desk_index).owner is not info.author_id and not self._is_author_admin(info):
                     return Response(
                         message="Range bookings are only allowed for desks you own for the entire range.",
                         ephemeral=True,
                     )
 
+        if user_id != info.author_id and not self._is_author_regular(info):
+            return Response(message="You do not have permission to book desks for other users.", ephemeral=True)
+
         self._database.handle_event(
             Event(
                 author=info.author_id,
@@ -173,7 +195,7 @@ def unbook(
                 return Response(message="A desk must be specified for range unbookings.", ephemeral=True)
             desk_index = desk_num - 1
             for booking_day in booking_days:
-                if booking_day.desk(desk_index).owner != info.author_id:
+                if booking_day.desk(desk_index).owner != info.author_id and not self._is_author_admin(info):
                     return Response(
                         message="Range unbookings are only allowed for desks you own for the entire range.",
                         ephemeral=True,
@@ -213,6 +235,8 @@ def unbook(
             return Response(message=(f"Desk {desk_num} has been unbooked from {date_str} to {fmt.date(end_date)}."))
         else:
             [booking_day] = booking_days
+            if booking_day.desk(desk_index).booker != info.author_id and not self._is_author_regular(info):
+                return Response(message="You do not have permission to unbook desks for other users.", ephemeral=True)
             desk_booker = booking_day.desk(desk_index).booker
             if desk_booker is not None:
                 self._database.handle_event(