Database loading checks

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ repos:
       hooks:
       - id: black
         language: python
-        types: [python]
+        types: [python3]
         args: ["--line-length=120"]
 
   -   repo: https://github.com/PyCQA/autoflake

Dockerfile

@@ -1,5 +1,5 @@
 #FROM python:3
-FROM python:3.8.6-alpine
+FROM python:3.11.6-alpine
 # use python:3.11.0rc2-slim for less vulnerabilities ? (from `docker scan`)
 # use python:3.8.6 for no pip dependencies build errors ?
 # use python:alpine for reduced size

README.md

@@ -4,7 +4,7 @@
 
 This is a small discord bot written in python using the [disnake library](https://github.com/DisnakeDev/disnake) to make a registration system for ULB discord servers.
 
-The bot checks that a user is a ULB student by verifying their ULB email adress using a one-time generated token sent to their email adress. It then gives them the role and adds their Discord user ID and ULB email adress to a database. The user will is then automatically verified on every server that the bot is running. The bot also has a rename functionality (optional, per server), names are extracted from the email adress.
+**_[Version 1]_**: The bot checks that a user is a ULB student by verifying their ULB email adress using a one-time generated token sent to their email adress. It then gives them the role and adds their Discord user ID and ULB email adress to a database. The user will is then automatically verified on every server that the bot is running. The bot also has a rename functionality (optional, per server), names are extracted from the email adress.
 
 # ➕ Add the bot to your server
 
@@ -196,16 +196,20 @@ To see the bot logs when running with docker in detached mode (`-d`), use the [d
 
 * `/setup`
 
-(Admin permission needed) When adding the bot to a new server, you need to set the @ULB role with the command `/setup`. This command also allows you to choose if you want to force the registered member to get renamed with their real name or not (yes by default).
+**[Admin permission needed]** When adding the bot to a new server, you need to set the @ULB role with the command `/setup`. This command also allows you to choose if you want to force the registered member to get renamed with their real name or not (yes by default).
 
 * `/info`
 
-(Admin permission needed) Get current server information (@ULB role, if rename is enabled, and checks for permission conflicts).
+**[Admin permission needed]** Get current server information (@ULB role, if rename is enabled, and checks for permission conflicts).
 
 * `/ulb`
 
 Once the ULB role is set, when a new user joins the server, either they are already registered (from another of your servers) in which case they will get the `@ULB` role and get renamed, or they are not registered yet and will receive a DM message with the instructions to register themselves using the `/ulb` command.
 
+* `/feedback`
+
+Send feedback directly from discord.
+
 ### Admin server
 
 * `/user add`
@@ -224,9 +228,17 @@ Edit info of a user.
 
 Delete a user.
 
+* `/server info`
+
+Get information about a guild (ULB role, number of registered members, ...)
+
+* `/stats`
+
+Get statistics about the bot usage (number of configured servers, number of registered users, ...)
+
 * `/update`
 
-This forces a total update of the database and of all the servers. Since the bot already does this automatically at startup and after each reconnection, the only normal usecase for this would be if you manually add an entry (server or user) to the google sheet instead of using the `/user add` command above, we don't recommend manually editing the google sheet.
+This forces a total update of the database and of all the servers. Since the bot already does this automatically at startup and after each reconnection, the only normal usecase for this would be if you manually add an entry (server or user) to the google sheet instead of using the `/user add` command above, we don't recommend manually editing the google sheet. It also includes an option (`/update Non`) to not rename users (on servers where it is enabled) when running this command (default: do **not** force rename on force update).
 
 ## 👤 Author
 

bot/bot.py

@@ -96,11 +96,24 @@ def load_commands(self) -> None:
                     )
                     self.cog_not_loaded.append(extension)
 
-    async def send_error_log(self, interaction: ApplicationCommandInteraction, error: Exception):
+    async def send_error_log(self, tb: str):
+
+        n = len(tb) // 4050
+
+        #Logs need to be diveded into multiple embed due to size limitation
+        # TODO Check if we can use a list of embeds and one message
+        # TODO Make it dynamic base on the message size from the lib (check library version, maybe need to upgrade)
+        for i in range(n):
+            await self.log_channel.send(embed=disnake.Embed(description=f"```python\n{tb[4050*i:4050*(i+1)]}```"))
+        await self.log_channel.send(embed=disnake.Embed(description=f"```python\n{tb[4050*n:]}```"))
+
+    async def send_cmd_error_log(self, interaction: ApplicationCommandInteraction, error: Exception):
         tb = self.tracebackEx(error)
         logging.error(
             f"{error} raised on command /{interaction.application_command.name} from {interaction.guild.name+'#'+interaction.channel.name if interaction.guild else 'DM'} by {interaction.author.name}.\n{tb}"
         )
+
+        #Send error msg to the user
         await interaction.send(
             content=self.owner.mention,
             embed=disnake.Embed(
@@ -110,17 +123,16 @@ async def send_error_log(self, interaction: ApplicationCommandInteraction, error
             ),
             delete_after=10,
         )
+
+        #Send logs to admins
         await self.log_channel.send(
             embed=disnake.Embed(title=f":x: __** ERROR**__ :x:", description=f"```{error}```").add_field(
                 name=f"Raised on command :",
                 value=f"**/{interaction.application_command.name}:{interaction.id}** from {interaction.guild.name+'#'+interaction.channel.name if interaction.guild else 'DM'} by {interaction.author.mention} at {interaction.created_at} with options\n```{interaction.filled_options}```"
                 + (f" and target\n``'{interaction.target}``'." if interaction.target else "."),
             )
         )
-        n = len(tb) // 4050
-        for i in range(n):
-            await self.log_channel.send(embed=disnake.Embed(description=f"```python\n{tb[4050*i:4050*(i+1)]}```"))
-        await self.log_channel.send(embed=disnake.Embed(description=f"```python\n{tb[4050*n:]}```"))
+        await self.send_error_log(tb)
 
     async def on_slash_command(self, interaction: disnake.ApplicationCommandInteraction) -> None:
         logging.trace(
@@ -138,13 +150,13 @@ async def on_message_command(self, interaction: disnake.MessageCommandInteractio
         )
 
     async def on_slash_command_error(self, interaction: ApplicationCommandInteraction, error: Exception) -> None:
-        await self.send_error_log(interaction, error)
+        await self.send_cmd_error_log(interaction, error)
 
     async def on_user_command_error(self, interaction: disnake.UserCommandInteraction, error: Exception) -> None:
-        await self.send_error_log(interaction, error)
+        await self.send_cmd_error_log(interaction, error)
 
     async def on_message_command_error(self, interaction: disnake.MessageCommandInteraction, error: Exception) -> None:
-        await self.send_error_log(interaction, error)
+        await self.send_cmd_error_log(interaction, error)
 
     async def on_slash_command_completion(self, interaction: disnake.ApplicationCommandInteraction) -> None:
         logging.trace(

classes/database.py

@@ -94,7 +94,7 @@ def loaded(cls) -> bool:
         return cls._loaded
 
     @classmethod
-    def load(cls, bot: Bot) -> None:
+    async def load(cls, bot: Bot) -> bool:
         """Load the data from the google sheet.
 
         Returns
@@ -104,33 +104,37 @@ def load(cls, bot: Bot) -> None:
             - Guild: `Dict[disnake.Guild, disnake.Role]`
             - Users: `Dict[disnake.User, UlbUser]]`
         """
-        # First time this is call, we need to load the credentials and the sheet
-        if not cls._sheet:
-            cred_dict = {}
-            cred_dict["type"] = os.getenv("GS_TYPE")
-            cred_dict["project_id"] = os.getenv("GS_PROJECT_ID")
-            cred_dict["auth_uri"] = os.getenv("GS_AUTHOR_URI")
-            cred_dict["token_uri"] = os.getenv("GS_TOKEN_URI")
-            cred_dict["auth_provider_x509_cert_url"] = os.getenv("GS_AUTH_PROV")
-            cred_dict["client_x509_cert_url"] = os.getenv("GS_CLIENT_CERT_URL")
-            cred_dict["private_key"] = os.getenv("GS_PRIVATE_KEY").replace(
-                "\\n", "\n"
-            )  # Python add a '\' before any '\n' when loading a str
-            cred_dict["private_key_id"] = os.getenv("GS_PRIVATE_KEY_ID")
-            cred_dict["client_email"] = os.getenv("GS_CLIENT_EMAIL")
-            cred_dict["client_id"] = int(os.getenv("GS_CLIENT_ID"))
-            creds = ServiceAccountCredentials.from_json_keyfile_dict(cred_dict, cls._scope)
-            cls._client = gspread.authorize(creds)
-            logging.info("[Database] Google sheet credentials loaded.")
-
-            # Open google sheet
-            cls._sheet = cls._client.open_by_url(os.getenv("GOOGLE_SHEET_URL"))
-            cls._users_ws = cls._sheet.worksheet("users")
-            cls._guilds_ws = cls._sheet.worksheet("guilds")
-
-            logging.info("[Database] Spreadsheed loaded")
-
-        logging.info("[Database] Loading data...")
+        try:
+            # First time this is call, we need to load the credentials and the sheet
+            if not cls._sheet:
+                cred_dict = {}
+                cred_dict["type"] = os.getenv("GS_TYPE")
+                cred_dict["project_id"] = os.getenv("GS_PROJECT_ID")
+                cred_dict["auth_uri"] = os.getenv("GS_AUTHOR_URI")
+                cred_dict["token_uri"] = os.getenv("GS_TOKEN_URI")
+                cred_dict["auth_provider_x509_cert_url"] = os.getenv("GS_AUTH_PROV")
+                cred_dict["client_x509_cert_url"] = os.getenv("GS_CLIENT_CERT_URL")
+                cred_dict["private_key"] = os.getenv("GS_PRIVATE_KEY").replace(
+                    "\\n", "\n"
+                )  # Python add a '\' before any '\n' when loading a str
+                cred_dict["private_key_id"] = os.getenv("GS_PRIVATE_KEY_ID")
+                cred_dict["client_email"] = os.getenv("GS_CLIENT_EMAIL")
+                cred_dict["client_id"] = int(os.getenv("GS_CLIENT_ID"))
+                creds = ServiceAccountCredentials.from_json_keyfile_dict(cred_dict, cls._scope)
+                cls._client = gspread.authorize(creds)
+                logging.info("[Database] Google sheet credentials loaded.")
+
+                # Open google sheet
+                cls._sheet = cls._client.open_by_url(os.getenv("GOOGLE_SHEET_URL"))
+                cls._users_ws = cls._sheet.worksheet("users")
+                cls._guilds_ws = cls._sheet.worksheet("guilds")
+
+                logging.info("[Database:load] Spreadsheed loaded")
+        except (ValueError, gspread.exceptions.SpreadsheetNotFound, gspread.exceptions.WorksheetNotFound) as err:
+            await bot.send_error_log(bot.tracebackEx(err))
+            return
+
+        logging.info("[Database:load] Loading data...")
 
         # Load guilds
         cls.ulb_guilds = {}
@@ -142,28 +146,30 @@ def load(cls, bot: Bot) -> None:
                 if role:
                     cls.ulb_guilds.setdefault(guild, UlbGuild(role, rename))
                     logging.trace(
-                        f"[Database] Role {role.name}:{role.id} loaded from guild {guild.name}:{guild.id} with {rename=}"
+                        f"[Database:load] Role {role.name}:{role.id} loaded from guild {guild.name}:{guild.id} with {rename=}"
                     )
                 else:
                     logging.warning(
-                        f"[Database] Not able to find role from id={guild_data.get('role_id', int)} in guild {guild.name}:{guild.id}."
+                        f"[Database:load] Not able to find role from id={guild_data.get('role_id', int)} in guild {guild.name}:{guild.id}."
                     )
             else:
-                logging.warning(f"[GoogleSheet] Not able to find guild from id={guild_data.get('guild_id', int)}.")
-        logging.info(f"[Database] Found {len(cls.ulb_guilds)} guilds.")
+                logging.warning(f"[Database:load] Not able to find guild from id={guild_data.get('guild_id', int)}.")
+        logging.info(f"[Database:load] Found {len(cls.ulb_guilds)} guilds.")
 
         # Load users
         cls.ulb_users = {}
+        not_found_counter = 0
         for user_data in cls._users_ws.get_all_records():
             user = bot.get_user(user_data.get("user_id", int))
             if user:
                 cls.ulb_users.setdefault(user, UlbUser(user_data.get("name", str), user_data.get("email", str)))
                 logging.trace(
-                    f"[Database] User {user.name}:{user.id} loaded with name={user_data.get('name')} and email={user_data.get('email')}"
+                    f"[Database:load] User {user.name}:{user.id} loaded with name={user_data.get('name')} and email={user_data.get('email')}"
                 )
             else:
+                not_found_counter += 1
                 logging.warning(f"[Database] Not able to find user from id={user_data.get('user_id',int)}.")
-        logging.info(f"[Database] Found {len(cls.ulb_users)} users.")
+        logging.info(f"[Database:load] {len(cls.ulb_users)} users found, {not_found_counter} not found.")
 
         cls._loaded = True
 
@@ -183,15 +189,15 @@ async def _set_user_task(cls, user_id: int, name: str, email: str):
         user_cell: gspread.cell.Cell = cls._users_ws.find(str(user_id), in_column=1)
         await asyncio.sleep(0.1)
         if user_cell:
-            logging.debug(f"[Database] {user_id=} found")
+            logging.trace(f"[Database:_set_user_task] {user_id=} found")
             cls._users_ws.update_cell(user_cell.row, 2, name)
             await asyncio.sleep(0.1)
             cls._users_ws.update_cell(user_cell.row, 3, email)
-            logging.info(f"[Database] {user_id=} updated with {name=} and {email=}")
+            logging.info(f"[Database:_set_user_task] {user_id=} updated with {name=} and {email=}")
         else:
-            logging.debug(f"[Database] {user_id=} not found")
+            logging.trace(f"[Database:_set_user_task] {user_id=} not found")
             cls._users_ws.append_row(values=[str(user_id), name, email])
-            logging.info(f"[Database] {user_id=} added with {name=} and {email=}")
+            logging.info(f"[Database:_set_user_task] {user_id=} added with {name=} and {email=}")
 
     @classmethod
     def set_user(cls, user: disnake.User, name: str, email: str):
@@ -224,10 +230,10 @@ async def _delete_user_task(cls, user_id: int):
         """
         user_cell: gspread.cell.Cell = cls._users_ws.find(str(user_id), in_column=1)
         await asyncio.sleep(0.1)
-        logging.trace(f"[Database] {user_id=} found")
+        logging.trace(f"[Database:_delete_user_task] {user_id=} found")
         cls._users_ws.delete_row(user_cell.row)
         await asyncio.sleep(0.1)
-        logging.info(f"[Database] {user_id=} deleted.")
+        logging.info(f"[Database:_delete_user_task] {user_id=} deleted.")
 
     @classmethod
     def delete_user(cls, user: disnake.User):
@@ -261,14 +267,14 @@ async def _set_guild_task(cls, guild_id: int, role_id: int, rename: bool):
         guild_cell: gspread.cell.Cell = cls._guilds_ws.find(str(guild_id), in_column=1)
         await asyncio.sleep(0.1)
         if guild_cell:
-            logging.debug(f"[Database] {guild_id=} found.")
+            logging.trace(f"[Database:_set_guild_task] {guild_id=} found.")
             cls._guilds_ws.update_cell(guild_cell.row, 2, str(role_id))
             cls._guilds_ws.update_cell(guild_cell.row, 3, rename)
-            logging.info(f"[Database] {guild_id=} update with {role_id=} and {rename=}.")
+            logging.info(f"[Database:_set_guild_task] {guild_id=} update with {role_id=} and {rename=}.")
         else:
-            logging.debug(f"[Database] {guild_id=} not found.")
+            logging.trace(f"[Database:_set_guild_task] {guild_id=} not found.")
             cls._guilds_ws.append_row(values=[str(guild_id), str(role_id), rename])
-            logging.info(f"[Database] {guild_id=} added with {role_id=} and {rename=}.")
+            logging.info(f"[Database:_set_guild_task] {guild_id=} added with {role_id=} and {rename=}.")
 
     @classmethod
     def set_guild(cls, guild: disnake.Guild, role: disnake.Role, rename: bool):
@@ -299,10 +305,10 @@ async def _delete_guild_task(cls, guild_id: int):
         """
         guild_cell: gspread.cell.Cell = cls._guilds_ws.find(str(guild_id), in_column=1)
         await asyncio.sleep(0.1)
-        logging.trace(f"[Database] {guild_id=} found")
+        logging.trace(f"[Database:_delete_guild_task] {guild_id=} found")
         cls._guilds_ws.delete_row(guild_cell.row)
         await asyncio.sleep(0.1)
-        logging.info(f"[Database] {guild_id=} deleted.")
+        logging.info(f"[Database:_delete_guild_task] {guild_id=} deleted.")
 
     @classmethod
     def delete_guild(cls, guild: disnake.Guild):

classes/registration.py

@@ -9,6 +9,7 @@
 from typing import Dict
 from typing import List
 from typing import Tuple
+from string import ascii_letters
 
 import disnake
 from disnake.ext import commands
@@ -519,7 +520,10 @@ async def _register_user_step(self, inter: disnake.ModalInteraction) -> None:
             The modal interaction that triggered the step
         """
         # Extract name and store the user
-        name = " ".join([name.title() for name in self.email.split("@")[0].split(".")])
+        allowed_chars = set(ascii_letters + " ") # or set('abcdefghijklmnopqrstuvwxyz ABCDEFGHIJKLMNOPQRSTUVWXYZ')
+        unfiltered_name = " ".join([name.title() for name in self.email.split("@")[0].split(".")])
+        name = "".join(filter(allowed_chars.__contains__, unfiltered_name)) # remove all characters that aren't ASCII letters or a space
+
         logging.trace(f"[RegistrationForm] [User:{self.target.id}] Extracted name from email= {name}")
         Database.set_user(self.target, name, self.email)
         await self._stop()