Merge pull request 'v0.1.0' (#20) from dev into main

Reviewed-on: #20

README.md

@@ -10,7 +10,7 @@
 ## Installation from release
 
 1. Install MongoDB using the [official installation manual](https://www.mongodb.com/docs/manual/installation)
-2. Install Python 3.9+ (3.11+ is recommended)
+2. Install Python 3.11+
 3. Download the [latest release](https://git.end-play.xyz/HoloUA/Discord/releases/latest)'s archive
 4. Extract the archive
 5. Navigate to the extracted folder and subfolder `Discord` in it
@@ -19,14 +19,14 @@
 7. Activate the virtual environment:  
    Windows: `.venv\Scripts\activate.bat`  
    Linux/macOS: `.venv/bin/activate`
-8. Install dependencies:  
+8. Install the dependencies:  
    `python -m pip install -r requirements.txt`
 9. Run the bot with `python main.py` after completing the [configuration](#Configuration)
 
 ## Installation with Git
 
 1. Install MongoDB using the [official installation manual](https://www.mongodb.com/docs/manual/installation)
-2. Install Python 3.9+ (3.11+ is recommended)
+2. Install Python 3.11+
 3. Clone the repository:  
    `git clone https://git.end-play.xyz/HoloUA/Discord.git`
 4. `cd Discord`

classes/holo_user.py

@@ -1,5 +1,6 @@
 import logging
-from typing import Any, Union, Dict
+from logging import Logger
+from typing import Any, Dict
 
 from bson import ObjectId
 from discord import User, Member
@@ -8,15 +9,15 @@ from libbot.utils import config_get
 from errors import UserNotFoundError
 from modules.database import col_warnings, sync_col_users, sync_col_warnings, col_users
 
-logger = logging.getLogger(__name__)
+logger: Logger = logging.getLogger(__name__)
 
 
 class HoloUser:
-    def __init__(self, user: Union[User, Member, int]) -> None:
+    def __init__(self, user: User | Member | int) -> None:
         """Get an object that has a proper binding between Discord ID and database
 
         ### Args:
-            * `user` (Union[User, Member, int]): Object from which ID can be extracted
+            * `user` (User | Member | int): Object from which ID can be extracted
 
         ### Raises:
             * `UserNotFoundError`: User with such ID does not seem to exist in database
@@ -24,17 +25,15 @@ class HoloUser:
 
         self.id: int = user if not hasattr(user, "id") else user.id
 
-        jav_user: Union[Dict[str, Any], None] = sync_col_users.find_one(
+        jav_user: Dict[str, Any] | None = sync_col_users.find_one({"user": self.id})
-            {"user": self.id}
-        )
 
         if jav_user is None:
             raise UserNotFoundError(user=user, user_id=self.id)
 
         self.db_id: ObjectId = jav_user["_id"]
 
-        self.customrole: Union[int, None] = jav_user["customrole"]
+        self.customrole: int | None = jav_user["customrole"]
-        self.customchannel: Union[int, None] = jav_user["customchannel"]
+        self.customchannel: int | None = jav_user["customchannel"]
         self.warnings: int = self.warns()
 
     def warns(self) -> int:
@@ -43,9 +42,7 @@ class HoloUser:
         ### Returns:
             * `int`: Number of warnings
         """
-        warns: Union[Dict[str, Any], None] = sync_col_warnings.find_one(
+        warns: Dict[str, Any] | None = sync_col_warnings.find_one({"user": self.id})
-            {"user": self.id}
-        )
 
         return 0 if warns is None else warns["warns"]
 
@@ -55,9 +52,7 @@ class HoloUser:
         ### Args:
             * `count` (int, optional): Count of warnings to be added. Defaults to 1.
         """
-        warns: Union[Dict[str, Any], None] = await col_warnings.find_one(
+        warns: Dict[str, Any] | None = await col_warnings.find_one({"user": self.id})
-            {"user": self.id}
-        )
 
         if warns is not None:
             await col_warnings.update_one(
@@ -88,11 +83,11 @@ class HoloUser:
         logger.info("Set attribute %s of user %s to %s", key, self.id, value)
 
     @staticmethod
-    async def is_moderator(member: Union[User, Member]) -> bool:
+    async def is_moderator(member: User | Member) -> bool:
         """Check if user is moderator or council member
 
         ### Args:
-            * `member` (Union[User, Member]): Member object
+            * `member` (User | Member): Member object
 
         ### Returns:
             `bool`: `True` if member is a moderator or member of council and `False` if not
@@ -100,8 +95,8 @@ class HoloUser:
         if isinstance(member, User):
             return False
 
-        moderator_role: Union[int, None] = await config_get("moderators", "roles")
+        moderator_role: int | None = await config_get("moderators", "roles")
-        council_role: Union[int, None] = await config_get("council", "roles")
+        council_role: int | None = await config_get("council", "roles")
 
         for role in member.roles:
             if role.id in (moderator_role, council_role):
@@ -110,11 +105,11 @@ class HoloUser:
         return False
 
     @staticmethod
-    async def is_council(member: Union[User, Member]) -> bool:
+    async def is_council(member: User | Member) -> bool:
         """Check if user is a member of council
 
         ### Args:
-            * `member` (Union[User, Member]): Member object
+            * `member` (User | Member): Member object
 
         ### Returns:
             `bool`: `True` if member is a member of council and `False` if not
@@ -129,8 +124,3 @@ class HoloUser:
                 return True
 
         return False
-
-    # def purge(self) -> None:
-    #     """Completely remove data from database. Will not remove transactions logs and warnings."""
-    #     col_users.delete_one(filter={"_id": self.db_id})
-    #     self.unauthorize()

cogs/admin.py

@@ -1,6 +1,6 @@
 import logging
 import sys
-from typing import Union
+from logging import Logger
 
 from discord import (
     ApplicationContext,
@@ -21,7 +21,7 @@ from modules.scheduler import scheduler
 from modules.utils_sync import guild_name
 from modules.waifu_pics import waifu_pics
 
-logger = logging.getLogger(__name__)
+logger: Logger = logging.getLogger(__name__)
 
 
 class Admin(commands.Cog):
@@ -30,89 +30,6 @@ class Admin(commands.Cog):
     def __init__(self, client: HoloBot):
         self.client: HoloBot = client
 
-    # Disabled because warning functionality is temporarily not needed
-    # @slash_command(
-    #     name="warning",
-    #     description="Попередити юзера про порушення правил",
-    #     guild_ids=[config_get_sync("guild")],
-    # )
-    # @option("user", description="Користувач")
-    # @option("reason", description="Причина")
-    # async def warn_cmd(
-    #     self,
-    #     ctx: ApplicationContext,
-    #     user: User,
-    #     reason: str = "Не вказана",
-    # ):
-    #     logging.info(f"User {ctx.user.id} warned {user.id} for {reason}")
-    #     await ctx.defer()
-    #     jav_user = HoloUser(user)
-    #     if ctx.user.id in await config_get("admins"):
-    #         logging.info(
-    #             f"Moderator {guild_name(ctx.user)} warned {guild_name(user)} for {reason} (has {jav_user.warnings} warns)"
-    #         )
-    #         if jav_user.warnings >= 5:
-    #             logging.info(
-    #                 f"User {guild_name(user)} was banned due to a big amount of warns ({jav_user.warnings})"
-    #             )
-    #             await user.send(
-    #                 embed=Embed(
-    #                     title="Перманентне блокування",
-    #                     description=f"Вас було заблоковано за неодноразове порушення правил сервера.",
-    #                     color=Color.fail,
-    #                 )
-    #             )
-    #             await user.ban(reason=reason)
-    #         elif jav_user.warnings >= 2:
-    #             logging.info(
-    #                 f"User {guild_name(user)} was muted due to a big amount of warns ({jav_user.warnings})"
-    #             )
-    #             jav_user.warn(reason=reason)
-    #             await user.send(
-    #                 embed=Embed(
-    #                     title="Тимчасове блокування",
-    #                     description=f"Причина: `{reason}`\n\nНа вашому рахунку вже {jav_user.warnings} попереджень. Вас було тимчасово заблоковано на **1 годину**.\n\nЯкщо Ви продовжите порушувати правила сервера – згодом Вас заблокують.",
-    #                     color=0xDED56B,
-    #                 )
-    #             )
-    #             await user.timeout_for(timedelta(hours=1), reason=reason)
-    #         else:
-    #             jav_user.warn()
-
-    #         await ctx.respond(
-    #             embed=Embed(
-    #                 title="Попередження",
-    #                 description=f"{user.mention} Будь ласка, не порушуйте правила. Ви отримали попередження з причини `{reason}`.\n\nЯкщо Ви продовжите порушувати правила – це може призвести до блокування в спільноті.",
-    #                 color=0xDED56B,
-    #             )
-    #         )
-    #     else:
-    #         logging.warning(
-    #             f"User {guild_name(ctx.user)} tried to use /warn but permission denied"
-    #         )
-    #         await ctx.respond(
-    #             embed=Embed(
-    #                 title="Відмовлено в доступі",
-    #                 description="Здається, це команда лише для модераторів",
-    #                 color=Color.fail,
-    #             )
-    #         )
-    #         mod_role = ds_utils.get(
-    #             ctx.user.guild.roles, id=await config_get("moderators", "roles")
-    #         )
-    #         admin_chan = ds_utils.get(
-    #             ctx.user.guild.channels,
-    #             id=await config_get("adminchat", "channels", "text"),
-    #         )
-    #         await admin_chan.send(
-    #             content=f"{mod_role.mention}",
-    #             embed=Embed(
-    #                 title="Неавторизований запит",
-    #                 description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
-    #                 color=Color.fail,
-    #             ),
-    #         )
-
     @slash_command(
         name="clear",
         description="Видалити деяку кількість повідомлень в каналі",
@@ -126,6 +43,10 @@ class Admin(commands.Cog):
         amount: int,
         user: User,
     ) -> None:
+        """Command /clear <amount> [<user>]
+
+        Removes last <amount> messages in the current channel. Optionally from a specific user.
+        """
         if ctx.user.id in self.client.owner_ids:
             logging.info(
                 "User %s removed %s message(s) in %s",
@@ -160,10 +81,10 @@ class Admin(commands.Cog):
             )
         )
 
-        mod_role: Union[Role, None] = ds_utils.get(
+        mod_role: Role | None = ds_utils.get(
             ctx.user.guild.roles, id=await config_get("moderators", "roles")
         )
-        admin_chan: Union[TextChannel, None] = ds_utils.get(
+        admin_chan: TextChannel | None = ds_utils.get(
             ctx.user.guild.channels,
             id=await config_get("adminchat", "channels", "text"),
         )
@@ -184,6 +105,10 @@ class Admin(commands.Cog):
         guild_ids=[config_get("guild")],
     )
     async def reboot_cmd(self, ctx: ApplicationContext) -> None:
+        """Command /reboot
+
+        Stops the bot. Is called "reboot" because it's assumed that the bot has automatic restart.
+        """
         await ctx.defer(ephemeral=True)
 
         if ctx.user.id in self.client.owner_ids:
@@ -216,10 +141,10 @@ class Admin(commands.Cog):
             )
         )
 
-        mod_role: Union[Role, None] = ds_utils.get(
+        mod_role: Role | None = ds_utils.get(
             ctx.user.guild.roles, id=await config_get("moderators", "roles")
         )
-        admin_chan: Union[TextChannel, None] = ds_utils.get(
+        admin_chan: TextChannel | None = ds_utils.get(
             ctx.user.guild.channels,
             id=await config_get("adminchat", "channels", "text"),
         )

cogs/analytics.py

@@ -1,4 +1,5 @@
 import logging
+from logging import Logger
 from typing import Dict, List, Any
 
 from discord import Cog, Message
@@ -7,7 +8,7 @@ from discord.ext import commands
 from classes.holo_bot import HoloBot
 from modules.database import col_analytics
 
-logger = logging.getLogger(__name__)
+logger: Logger = logging.getLogger(__name__)
 
 
 class Analytics(commands.Cog):
@@ -16,11 +17,13 @@ class Analytics(commands.Cog):
 
     @Cog.listener()
     async def on_message(self, message: Message) -> None:
+        """Listener that collects analytical data (stickers, attachments, messages)."""
         if (
             (message.author != self.client.user)
             and (message.author.bot is False)
             and (message.author.system is False)
         ):
+            # Handle stickers
             stickers: List[Dict[str, Any]] = []
 
             for sticker in message.stickers:
@@ -33,6 +36,7 @@ class Analytics(commands.Cog):
                     }
                 )
 
+            # Handle attachments
             attachments: List[Dict[str, Any]] = []
 
             for attachment in message.attachments:
@@ -49,6 +53,7 @@ class Analytics(commands.Cog):
                     }
                 )
 
+            # Insert entry into the database
             await col_analytics.insert_one(
                 {
                     "user": message.author.id,

cogs/custom_channels.py

@@ -1,5 +1,6 @@
 import logging
-from typing import Any, Dict, Union
+from logging import Logger
+from typing import Any, Dict
 
 from discord import ApplicationContext, Embed, option, TextChannel, Role
 from discord import utils as ds_utils
@@ -14,7 +15,7 @@ from enums import Color
 from modules.database import col_users
 from modules.utils_sync import guild_name
 
-logger = logging.getLogger(__name__)
+logger: Logger = logging.getLogger(__name__)
 
 
 class CustomChannels(commands.Cog):
@@ -42,6 +43,10 @@ class CustomChannels(commands.Cog):
     async def custom_channel_get_cmd(
         self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool
     ) -> None:
+        """Command /customchannel get <name> <reactions> <threads>
+
+        Command to create a custom channel for a user.
+        """
         holo_user_ctx: HoloUser = HoloUser(ctx.user)
 
         # Return if the user is using the command outside of a guild
@@ -108,9 +113,7 @@ class CustomChannels(commands.Cog):
         bots: Dict[str, Any] = await config_get("bots")
 
         for bot in bots:
-            role: Union[Role, None] = ds_utils.get(
+            role: Role | None = ds_utils.get(ctx.user.guild.roles, id=bots[bot]["role"])
-                ctx.user.guild.roles, id=bots[bot]["role"]
-            )
 
             if role is not None:
                 await created_channel.set_permissions(
@@ -129,9 +132,13 @@ class CustomChannels(commands.Cog):
     async def custom_channel_edit_cmd(
         self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool
     ) -> None:
+        """Command /customchannel edit <name> <reactions> <threads>
+
+        Command to change properties of a custom channel.
+        """
         holo_user_ctx: HoloUser = HoloUser(ctx.user)
 
-        custom_channel: Union[TextChannel, None] = ds_utils.get(
+        custom_channel: TextChannel | None = ds_utils.get(
             ctx.guild.channels, id=holo_user_ctx.customchannel
         )
 
@@ -172,6 +179,9 @@ class CustomChannels(commands.Cog):
     async def custom_channel_remove_cmd(
         self, ctx: ApplicationContext, confirm: bool = False
     ) -> None:
+        """Command /customchannel remove [<confirm>]
+
+        Command to remove a custom channel. Requires additional confirmation."""
         holo_user_ctx: HoloUser = HoloUser(ctx.user)
 
         # Return if the user does not have a custom channel
@@ -188,7 +198,7 @@ class CustomChannels(commands.Cog):
 
         await ctx.defer()
 
-        custom_channel: Union[TextChannel, None] = ds_utils.get(
+        custom_channel: TextChannel | None = ds_utils.get(
             ctx.guild.channels, id=holo_user_ctx.customchannel
         )
 

cogs/data.py

@@ -2,7 +2,7 @@ import logging
 from logging import Logger
 from os import makedirs
 from pathlib import Path
-from typing import Union, List, Dict, Any
+from typing import List, Dict, Any
 from uuid import uuid4
 
 from discord import ApplicationContext, Embed, File, option, Role, TextChannel
@@ -35,6 +35,9 @@ class Data(commands.Cog):
         "kind", description="Тип даних, які треба експортувати", choices=["Користувачі"]
     )
     async def data_export_cmd(self, ctx: ApplicationContext, kind: str) -> None:
+        """Command /data export <kind>
+
+        Command to export specific kind of data."""
         await ctx.defer()
 
         # Return if the user is not an owner and not in the council
@@ -54,10 +57,10 @@ class Data(commands.Cog):
                 )
             )
 
-            mod_role: Union[Role, None] = ds_utils.get(
+            mod_role: Role | None = ds_utils.get(
                 ctx.user.guild.roles, id=await config_get("moderators", "roles")
             )
-            admin_chan: Union[TextChannel, None] = ds_utils.get(
+            admin_chan: TextChannel | None = ds_utils.get(
                 ctx.user.guild.channels,
                 id=await config_get("adminchat", "channels", "text"),
             )
@@ -107,6 +110,11 @@ class Data(commands.Cog):
         "kind", description="Тип даних, які треба експортувати", choices=["Користувачі"]
     )
     async def data_migrate_cmd(self, ctx: ApplicationContext, kind: str) -> None:
+        """Command /migrate <kind>
+
+        Command to migrate specific kind of data.
+
+        Migration of users in this case means creation of their DB entries."""
         await ctx.defer()
 
         # Return if the user is not an owner and not in the council
@@ -126,10 +134,10 @@ class Data(commands.Cog):
                 )
             )
 
-            mod_role: Union[Role, None] = ds_utils.get(
+            mod_role: Role | None = ds_utils.get(
                 ctx.user.guild.roles, id=await config_get("moderators", "roles")
             )
-            admin_chan: Union[TextChannel, None] = ds_utils.get(
+            admin_chan: TextChannel | None = ds_utils.get(
                 ctx.user.guild.channels,
                 id=await config_get("adminchat", "channels", "text"),
             )

cogs/fun.py

@@ -1,4 +1,5 @@
 import logging
+from logging import Logger
 
 from discord import ApplicationContext, Embed, User, option, slash_command
 from discord.ext import commands
@@ -8,7 +9,7 @@ from classes.holo_bot import HoloBot
 from modules.utils_sync import guild_name
 from modules.waifu_pics import waifu_pics
 
-logger = logging.getLogger(__name__)
+logger: Logger = logging.getLogger(__name__)
 
 
 class Fun(commands.Cog):
@@ -27,6 +28,9 @@ class Fun(commands.Cog):
     )
     @option("user", description="Користувач")
     async def action_cmd(self, ctx: ApplicationContext, type: str, user: User) -> None:
+        """Command /action <type> <user>
+
+        Command to perform some RP action on a user and send them a GIF."""
         await ctx.defer()
 
         action: str = await config_get("category", "actions", type)

cogs/logger.py

@@ -1,6 +1,8 @@
-from typing import Dict, Any, Union
+import logging
+from logging import Logger
+from typing import Dict, Any
 
-from discord import Member, Message, TextChannel
+from discord import Member, Message, TextChannel, MessageType
 from discord import utils as ds_utils
 from discord.ext import commands
 from libbot.utils import config_get
@@ -8,6 +10,8 @@ from libbot.utils import config_get
 from classes.holo_bot import HoloBot
 from modules.database import col_users
 
+logger: Logger = logging.getLogger(__name__)
+
 
 class Logger(commands.Cog):
     def __init__(self, client: HoloBot):
@@ -15,6 +19,7 @@ class Logger(commands.Cog):
 
     @commands.Cog.listener()
     async def on_message(self, message: Message):
+        """Message listener. All actions on messages remain here for now."""
         if (
             (message.author != self.client.user)
             and (message.author.bot is False)
@@ -31,17 +36,42 @@ class Logger(commands.Cog):
 
                 await col_users.insert_one(document=user)
 
+        if (
+            (message.type == MessageType.thread_created)
+            and (message.channel is not None)
+            and (
+                await col_users.count_documents({"customchannel": message.channel.id})
+                > 0
+            )
+        ):
+            try:
+                logger.info(
+                    "Deleting the thread creation message in a custom channel %s",
+                    message.channel.id,
+                )
+                await message.delete()
+            except Exception as exc:
+                logger.warning(
+                    "Could not delete the thread creation message in a custom channel %s due to %s",
+                    message.channel.id,
+                    exc,
+                )
+
     @commands.Cog.listener()
     async def on_member_join(self, member: Member) -> None:
-        welcome_chan: Union[TextChannel, None] = ds_utils.get(
+        """Member join handler. All actions on member join remain here for now."""
+        welcome_chan: TextChannel | None = ds_utils.get(
             self.client.get_guild(await config_get("guild")).channels,
             id=await config_get("welcome", "channels", "text"),
         )
-        rules_chan: Union[TextChannel, None] = ds_utils.get(
+        rules_chan: TextChannel | None = ds_utils.get(
             self.client.get_guild(await config_get("guild")).channels,
             id=await config_get("rules", "channels", "text"),
         )
 
+        if welcome_chan is None:
+            logger.warning("Could not find a welcome channel by its id")
+
         if (
             (member != self.client.user)
             and (member.bot is False)

cogs/utility.py

@@ -16,6 +16,7 @@ class Utility(commands.Cog):
 
     @commands.Cog.listener()
     async def on_ready(self) -> None:
+        """Listener for the event when bot connects to Discord and becomes "ready"."""
         logger.info("Logged in as %s", self.client.user)
 
         activity_type: str = await config_get("type", "status")

modules/utils_sync.py

@@ -1,9 +1,7 @@
-from typing import Union
-
 from discord import Member, User
 
 
-def guild_name(member: Union[Member, User]) -> str:
+def guild_name(member: Member | User) -> str:
     if isinstance(member, User):
         return member.name