Merge pull request 'v0.1.0-rc.3' (#15) from dev into main

Reviewed-on: #15

README.md

@@ -1,2 +1,38 @@
-# HoloBotDiscord
+<h1 align="center">HoloBot Discord</h1>
 
+<p align="center">Small Discord bot made on Py-Cord</p>
+
+<p align="center">
+<a href="https://git.end-play.xyz/HoloUA/Discord/src/branch/master/LICENSE"><img alt="License: GPL" src="https://img.shields.io/badge/License-GPL-blue"></a>
+<a href="https://git.end-play.xyz/HoloUA/Discord"><img alt="Code style: black" src="https://img.shields.io/badge/code%20style-black-000000.svg"></a>
+</p>  
+
+## Installation
+
+1. Install MongoDB using the [official installation manual](https://www.mongodb.com/docs/manual/installation/).
+2. `git clone https://git.end-play.xyz/HoloUA/Discord.git`
+3. `cd Discord`
+4. Install Python 3.9+ (at least 3.11 is recommended) for your OS
+5. `python3 -m pip install -r requirements.txt`
+6. Run it with `python3 main.py` after configuring
+
+## Configuration
+
+There's a file `config_example.json` which contains default configuration
+and should be used as a base config.  
+
+Copy this file to `config.json` and open it with any text editor of your liking.  
+
+Modify the newly created configuration file to fit your needs.  
+
+Mandatory keys to modify:
+
+- guild
+- bot.owner
+- bot.bot_token
+- database.*
+- categories.*
+- channels.*
+- roles.*
+
+After all of that you're good to go! Happy using :)

classes/holo_user.py

@@ -1,157 +1,41 @@
 import logging
-from typing import Any, Union
+from typing import Any, Union, Dict
 
-import discord
+from bson import ObjectId
-import discord.member
+from discord import User, Member
+from libbot import config_get
 
-from modules.database import col_users, col_warnings
+from errors import UserNotFoundError
+from modules.database import col_warnings, sync_col_users, sync_col_warnings, col_users
 
 logger = logging.getLogger(__name__)
 
 
-class NotEnoughMoneyError(Exception):
-    """User does not have enough money to do that"""
-
-    pass
-
-
-class UserNotFoundError(Exception):
-    """HoloUser could not find user with such an ID in database"""
-
-    def __init__(self, user, user_id):
-        self.user = user
-        self.user_id = user_id
-        super().__init__(
-            f"User of type {type(self.user)} with id {self.user_id} was not found"
-        )
-
-
 class HoloUser:
-    def __init__(
+    def __init__(self, user: Union[User, Member, int]) -> None:
-        self, user: Union[discord.User, discord.Member, discord.member.Member, int]
-    ) -> None:
         """Get an object that has a proper binding between Discord ID and database
 
         ### Args:
-            * `user` (Union[discord.User, discord.Member, discord.member.Member, int]): Object from which ID can be extracted
+            * `user` (Union[User, Member, int]): Object from which ID can be extracted
 
         ### Raises:
             * `UserNotFoundError`: User with such ID does not seem to exist in database
         """
 
-        if hasattr(user, "id"):
+        self.id: int = user if not hasattr(user, "id") else user.id
-            self.id = user.id  # type: ignore
-        else:
-            self.id = user
 
-        jav_user = col_users.find_one({"user": self.id})
+        jav_user: Union[Dict[str, Any], None] = sync_col_users.find_one(
+            {"user": self.id}
+        )
 
         if jav_user is None:
             raise UserNotFoundError(user=user, user_id=self.id)
 
-        self.db_id = jav_user["_id"]
+        self.db_id: ObjectId = jav_user["_id"]
 
-        # self.xp = jav_user["xp"]
+        self.customrole: Union[int, None] = jav_user["customrole"]
-        # self.xp_next = jav_user["xp_next"]
+        self.customchannel: Union[int, None] = jav_user["customchannel"]
-        # self.level = jav_user["level"]
+        self.warnings: int = self.warns()
-        # self.work_xp = jav_user["work_xp"]
-        # self.balance = jav_user["balance"]
-        self.customrole = jav_user["customrole"]
-        self.customchannel = jav_user["customchannel"]
-        # self.married = jav_user["married"]
-        # self.marriage_request = jav_user["marriage_request"]
-        # self.marriage_request_sent = jav_user["marriage_request_sent"]
-        # self.cooldown = jav_user["cooldown"]
-        self.warnings = self.warns()
-
-    # def xp_add(self, amount: int = 1) -> None:
-    #     """Add some amount of XP points
-
-    #     ### Args:
-    #         * `amount` (int, optional): Amount of XP points to give. Defaults to 1.
-    #     """
-    #     self.xp += amount
-    #     col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "xp": self.xp } })
-
-    # def xp_level_up(self) -> None:
-    #     """Add 1 to the current XP level"""
-    #     xp_diff = int(self.xp - self.xp_next)
-    #     xp_next = int(self.xp_next*configGet("multiplier", "leveling")+configGet("addition", "leveling"))
-    #     self.xp = xp_diff
-    #     self.xp_next = xp_next
-    #     self.level += 1
-    #     col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "xp": xp_diff } })
-    #     col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "xp_next": xp_next } })
-    #     col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "level": self.level } })
-
-    # def balance_set(self, amount: int) -> None:
-    #     """Set the balance to amount
-
-    #     ### Args:
-    #         * `amount` (int): Amount of currency to be set
-    #     """
-    #     self.balance = amount
-    #     col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "balance": self.balance } })
-
-    # def balance_add(self, amount: int) -> None:
-    #     """Add amount to the balance
-
-    #     ### Args:
-    #         * `amount` (int): Amount to be added
-    #     """
-    #     self.balance_set(self.balance+amount)
-
-    # def balance_take(self, amount: int) -> bool:
-    #     """Take amount from the balance
-
-    #     ### Args:
-    #         * `amount` (int): Amount to be taken
-
-    #     ### Returns:
-    #         * `bool`: True if successful and False if not
-    #     """
-    #     if self.balance >= amount:
-    #         self.balance_set(self.balance-amount)
-    #         return True
-    #     else:
-    #         return False
-    #         #raise NotEnoughMoneyError()
-
-    # def balance_transfer(self, amount: int, destination: Union[Any, int]) -> None:
-    #     """Transfer money to another user
-
-    #     ### Args:
-    #         * `amount` (int): Amount to be transferred
-    #         * `destination` (Union[Any, int]): Destination user of the transfer (should have attribute "id" or be id itself if int)
-
-    #     ### Raises:
-    #         * `NotEnoughMoneyError`: Not enough money to perform this transaction
-    #     """
-    #     if self.balance >= amount:
-    #         if isinstance(destination, int):
-    #             destination = HoloUser(destination)
-    #         self.balance_take(amount)
-    #         destination.balance_add(amount) # type: ignore
-    #     else:
-    #         raise NotEnoughMoneyError()
-
-    # def salary_get(self) -> int:
-    #     """Get the salary level depending on work_xp
-
-    #     ### Returns:
-    #         * `int`: Amount of money to be earned
-    #     """
-    #     if self.work_xp >= 100:
-    #         return randint(configGet("min", "work", "level", "4"), configGet("max", "work", "level", "4"))
-    #     elif self.work_xp >= 50:
-    #         return randint(configGet("min", "work", "level", "3"), configGet("max", "work", "level", "3"))
-    #     elif self.work_xp > 10:
-    #         return randint(configGet("min", "work", "level", "2"), configGet("max", "work", "level", "2"))
-    #     else:
-    #         return randint(configGet("min", "work", "level", "1"), configGet("max", "work", "level", "1"))
-
-    # def work_xp_add(self) -> None:
-    #     self.set("work_xp", self.work_xp+1)
 
     def warns(self) -> int:
         """Get number of warnings user has
@@ -159,38 +43,33 @@ class HoloUser:
         ### Returns:
             * `int`: Number of warnings
         """
-        warns = col_warnings.find_one({"user": self.id})
+        warns: Union[Dict[str, Any], None] = sync_col_warnings.find_one(
-        if warns == None:
+            {"user": self.id}
-            return 0
+        )
-        else:
-            return warns["warns"]
 
-    def warn(self, count=1, reason: str = "Not provided") -> None:
+        return 0 if warns is None else warns["warns"]
+
+    async def warn(self, count=1, reason: str = "Not provided") -> None:
         """Warn and add count to warns number
 
         ### Args:
             * `count` (int, optional): Count of warnings to be added. Defaults to 1.
         """
-        warns = col_warnings.find_one({"user": self.id})
+        warns: Union[Dict[str, Any], None] = await col_warnings.find_one(
-        if warns != None:
+            {"user": self.id}
-            col_warnings.update_one(
+        )
-                filter={"_id": self.db_id},
+
-                update={"$set": {"warns": warns["warns"] + count}},
+        if warns is not None:
+            await col_warnings.update_one(
+                {"_id": self.db_id},
+                {"$set": {"warns": warns["warns"] + count}},
             )
         else:
-            col_warnings.insert_one(document={"user": self.id, "warns": count})
+            await col_warnings.insert_one(document={"user": self.id, "warns": count})
-        logger.info(f"User {self.id} was warned {count} times due to: {reason}")
 
-    # def cooldown_go(self, kind: Literal["work", "daily", "weekly", "monthly", "steal"]) -> None:
+        logger.info("User %s was warned %s times due to: %s", self.id, count, reason)
-    #     """Set cooldown start of kind now
 
-    #     ### Args:
+    async def set(self, key: str, value: Any) -> None:
-    #         * `kind` (Literal["work", "daily", "weekly", "monthly", "steal"]): Kind of a cooldown
-    #     """
-    #     self.cooldown[kind] = datetime.now(tz=timezone.utc)
-    #     col_users.update_one(filter={"_id": self.db_id}, update={ "$set": { "cooldown": self.cooldown } })
-
-    def set(self, key: str, value: Any) -> None:
         """Set attribute data and save it into database
 
         ### Args:
@@ -199,32 +78,59 @@ class HoloUser:
         """
         if not hasattr(self, key):
             raise AttributeError()
+
         setattr(self, key, value)
-        col_users.update_one(
+
-            filter={"_id": self.db_id}, update={"$set": {key: value}}, upsert=True
+        await col_users.update_one(
+            {"_id": self.db_id}, {"$set": {key: value}}, upsert=True
         )
-        logger.info(f"Set attribute {key} of user {self.id} to {value}")
+
+        logger.info("Set attribute %s of user %s to %s", key, self.id, value)
+
+    @staticmethod
+    async def is_moderator(member: Union[User, Member]) -> bool:
+        """Check if user is moderator or council member
+
+        ### Args:
+            * `member` (Union[User, Member]): Member object
+
+        ### Returns:
+            `bool`: `True` if member is a moderator or member of council and `False` if not
+        """
+        if isinstance(member, User):
+            return False
+
+        moderator_role: Union[int, None] = await config_get("moderators", "roles")
+        council_role: Union[int, None] = await config_get("council", "roles")
+
+        for role in member.roles:
+            if role.id in (moderator_role, council_role):
+                return True
+
+        return False
+
+    @staticmethod
+    async def is_council(member: Union[User, Member]) -> bool:
+        """Check if user is a member of council
+
+        ### Args:
+            * `member` (Union[User, Member]): Member object
+
+        ### Returns:
+            `bool`: `True` if member is a member of council and `False` if not
+        """
+        if isinstance(member, User):
+            return False
+
+        council_role = await config_get("council", "roles")
+
+        for role in member.roles:
+            if role.id == council_role:
+                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()
-
-    # def unauthorize(self) -> None:
-    #     """Cancel Oauth2 authorization"""
-    #     col_authorized.find_one_and_delete({"user": self.id})
-
-    # def is_authorized(self) -> bool:
-    #     """Check if user provided Oauth2 authorization
-
-    #     ### Returns:
-    #         * `bool`: True if yes and False if no
-    #     """
-    #     if configGet("mode") == "secure":
-    #         authorized = col_authorized.find_one({"user": self.id})
-    #         if authorized is not None:
-    #             return True
-    #         else:
-    #             return False
-    #     else:
-    #         return True

cogs/admin.py

@@ -1,21 +1,37 @@
 import logging
+import sys
+from typing import Union
 
-from discord import ApplicationContext, Embed, User, option, slash_command
+from discord import (
+    ApplicationContext,
+    Embed,
+    User,
+    option,
+    slash_command,
+    Role,
+    TextChannel,
+)
 from discord import utils as ds_utils
 from discord.ext import commands
+from libbot import config_get
+from libbot.pycord.classes import PycordBot
+from libbot.sync import config_get as sync_config_get
 
-from enums.colors import Color
+from enums import Color
-from modules.scheduled import scheduler
+from modules.scheduler import scheduler
-from modules.utils import config_get
+from modules.utils_sync import guild_name
-from modules.utils_sync import config_get_sync, guild_name
+from modules.waifu_pics import waifu_pics
 
 logger = logging.getLogger(__name__)
 
 
 class Admin(commands.Cog):
-    def __init__(self, client):
+    """Cog with utility commands for admins."""
-        self.client = client
 
+    def __init__(self, client: PycordBot):
+        self.client: PycordBot = client
+
+    # Disabled because warning functionality is temporarily not needed
     # @slash_command(
     #     name="warning",
     #     description="Попередити юзера про порушення правил",
@@ -83,7 +99,7 @@ class Admin(commands.Cog):
     #             )
     #         )
     #         mod_role = ds_utils.get(
-    #             ctx.user.guild.roles, id=await config_get("moderator", "roles")
+    #             ctx.user.guild.roles, id=await config_get("moderators", "roles")
     #         )
     #         admin_chan = ds_utils.get(
     #             ctx.user.guild.channels,
@@ -101,7 +117,7 @@ class Admin(commands.Cog):
     @slash_command(
         name="clear",
         description="Видалити деяку кількість повідомлень в каналі",
-        guild_ids=[config_get_sync("guild")],
+        guild_ids=[sync_config_get("guild")],
     )
     @option("amount", description="Кількість")
     @option("user", description="Користувач", default=None)
@@ -110,88 +126,115 @@ class Admin(commands.Cog):
         ctx: ApplicationContext,
         amount: int,
         user: User,
-    ):
+    ) -> None:
-        if ctx.user.id in await config_get("admins"):
+        if ctx.user.id in self.client.owner_ids:
             logging.info(
-                f"User {ctx.user.id} removed {amount} message(s) in {ctx.channel.id}"
+                "User %s removed %s message(s) in %s",
+                ctx.user.id,
+                amount,
+                ctx.channel.id,
             )
+
             await ctx.respond(
                 embed=Embed(description="Видаляю..."), ephemeral=True, delete_after=2.0
             )
-            if user == None:
+
+            if user is None:
                 await ctx.channel.purge(limit=amount)
             else:
                 await ctx.channel.purge(
                     limit=amount, check=lambda msg: msg.author == user
                 )
-        else:
+
-            logging.warning(
+            return
-                f"User {guild_name(ctx.user)} tried to use /clear but permission denied"
+
-            )
+        logging.warning(
-            await ctx.respond(
+            "User %s tried to use /clear but permission denied",
-                embed=Embed(
+            guild_name(ctx.user),
-                    title="Відмовлено в доступі",
+        )
-                    description="Здається, це команда лише для модераторів",
+
-                    color=Color.fail,
+        await ctx.respond(
-                )
+            embed=Embed(
-            )
+                title="Відмовлено в доступі",
-            mod_role = ds_utils.get(
+                description="Здається, це команда лише для модераторів",
-                ctx.user.guild.roles, id=await config_get("moderator", "roles")
+                color=Color.FAIL,
-            )
-            admin_chan = ds_utils.get(
-                ctx.user.guild.channels,
-                id=await config_get("adminchat", "channels", "text"),
             )
+        )
+
+        mod_role: Union[Role, None] = ds_utils.get(
+            ctx.user.guild.roles, id=await config_get("moderators", "roles")
+        )
+        admin_chan: Union[TextChannel, None] = ds_utils.get(
+            ctx.user.guild.channels,
+            id=await config_get("adminchat", "channels", "text"),
+        )
+
+        if admin_chan is not None:
             await admin_chan.send(
-                content=f"{mod_role.mention}",
+                content="" if mod_role is None else mod_role.mention,
                 embed=Embed(
                     title="Неавторизований запит",
                     description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
-                    color=Color.fail,
+                    color=Color.FAIL,
                 ),
             )
 
     @slash_command(
         name="reboot",
         description="Перезапустити бота",
-        guild_ids=[config_get_sync("guild")],
+        guild_ids=[sync_config_get("guild")],
     )
-    async def reboot_cmd(self, ctx: ApplicationContext):
+    async def reboot_cmd(self, ctx: ApplicationContext) -> None:
         await ctx.defer(ephemeral=True)
-        if ctx.user.id in await config_get("admins"):
+
-            logging.info(f"Calling shutdown initiated by {guild_name(ctx.user)}")
+        if ctx.user.id in self.client.owner_ids:
+            logging.info("Calling shutdown initiated by %s", guild_name(ctx.user))
+
             await ctx.respond(
                 embed=Embed(
                     title="Вимикаюсь...",
                     description="Спробую перезавантажитись за 5 секунд",
                 )
             )
+
             scheduler.shutdown()
+
             await self.client.close()
-            exit()
+            await waifu_pics._client_session.close()
-        else:
+
-            logging.warning(
+            sys.exit()
-                f"User {guild_name(ctx.user)} tried to use /reboot but permission denied"
+
-            )
+        logging.warning(
-            await ctx.respond(
+            "User %s tried to use /reboot but permission denied",
-                embed=Embed(
+            guild_name(ctx.user),
-                    title="Відмовлено в доступі",
+        )
-                    description="Здається, це команда лише для модераторів",
+
-                    color=Color.fail,
+        await ctx.respond(
-                )
+            embed=Embed(
-            )
+                title="Відмовлено в доступі",
-            mod_role = ds_utils.get(
+                description="Здається, це команда лише для модераторів",
-                ctx.user.guild.roles, id=await config_get("moderator", "roles")
+                color=Color.FAIL,
-            )
-            admin_chan = ds_utils.get(
-                ctx.user.guild.channels,
-                id=await config_get("adminchat", "channels", "text"),
             )
+        )
+
+        mod_role: Union[Role, None] = ds_utils.get(
+            ctx.user.guild.roles, id=await config_get("moderators", "roles")
+        )
+        admin_chan: Union[TextChannel, None] = ds_utils.get(
+            ctx.user.guild.channels,
+            id=await config_get("adminchat", "channels", "text"),
+        )
+
+        if admin_chan is not None:
             await admin_chan.send(
-                content=f"{mod_role.mention}",
+                content="" if mod_role is None else mod_role.mention,
                 embed=Embed(
                     title="Неавторизований запит",
                     description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
-                    color=Color.fail,
+                    color=Color.FAIL,
                 ),
             )
+
+
+def setup(client: PycordBot) -> None:
+    client.add_cog(Admin(client))

cogs/analytics.py

@@ -1,7 +1,9 @@
 import logging
+from typing import Dict, List, Any
 
 from discord import Cog, Message
 from discord.ext import commands
+from libbot.pycord.classes import PycordBot
 
 from modules.database import col_analytics
 
@@ -9,20 +11,54 @@ logger = logging.getLogger(__name__)
 
 
 class Analytics(commands.Cog):
-    def __init__(self, client):
+    def __init__(self, client: PycordBot):
-        self.client = client
+        self.client: PycordBot = client
 
     @Cog.listener()
-    async def on_message(self, message: Message):
+    async def on_message(self, message: Message) -> None:
         if (
             (message.author != self.client.user)
-            and (message.author.bot == False)
+            and (message.author.bot is False)
-            and (message.author.system == False)
+            and (message.author.system is False)
         ):
-            col_analytics.insert_one(
+            stickers: List[Dict[str, Any]] = []
+
+            for sticker in message.stickers:
+                stickers.append(
+                    {
+                        "id": sticker.id,
+                        "name": sticker.name,
+                        "format": sticker.format,
+                        "url": sticker.url,
+                    }
+                )
+
+            attachments: List[Dict[str, Any]] = []
+
+            for attachment in message.attachments:
+                attachments.append(
+                    {
+                        "content_type": attachment.content_type,
+                        "description": attachment.description,
+                        "filename": attachment.filename,
+                        "is_spoiler": attachment.is_spoiler(),
+                        "size": attachment.size,
+                        "url": attachment.url,
+                        "width": attachment.width,
+                        "height": attachment.height,
+                    }
+                )
+
+            await col_analytics.insert_one(
                 {
-                    "message": message.content,
                     "user": message.author.id,
                     "channel": message.channel.id,
+                    "content": message.content,
+                    "stickers": stickers,
+                    "attachments": attachments,
                 }
             )
+
+
+def setup(client: PycordBot) -> None:
+    client.add_cog(Analytics(client))

cogs/custom_channels.py

@@ -1,115 +1,152 @@
-from discord import ApplicationContext, Embed, option
+import logging
+from typing import Any, Dict, Union
+
+from discord import ApplicationContext, Embed, option, TextChannel, Role
 from discord import utils as ds_utils
 from discord.abc import GuildChannel
 from discord.commands import SlashCommandGroup
 from discord.ext import commands
+from libbot import config_get
+from libbot.pycord.classes import PycordBot
+from libbot.sync import config_get as sync_config_get
 
 from classes.holo_user import HoloUser
-from enums.colors import Color
+from enums import Color
 from modules.database import col_users
-from modules.utils import config_get
+from modules.utils_sync import guild_name
-from modules.utils_sync import config_get_sync, guild_name
+
+logger = logging.getLogger(__name__)
 
 
 class CustomChannels(commands.Cog):
-    def __init__(self, client):
+    def __init__(self, client: PycordBot):
-        self.client = client
+        self.client: PycordBot = client
 
     @commands.Cog.listener()
-    async def on_guild_channel_delete(self, channel: GuildChannel):
+    async def on_guild_channel_delete(self, channel: GuildChannel) -> None:
-        col_users.find_one_and_update(
+        await col_users.find_one_and_update(
             {"customchannel": channel.id}, {"$set": {"customchannel": None}}
         )
 
-    customchannel = SlashCommandGroup("customchannel", "Керування особистим каналом")
+    custom_channel_group: SlashCommandGroup = SlashCommandGroup(
+        "customchannel", "Керування особистим каналом"
+    )
 
-    @customchannel.command(
+    @custom_channel_group.command(
         name="get",
         description="Отримати персональний текстовий канал",
-        guild_ids=[config_get_sync("guild")],
+        guild_ids=[sync_config_get("guild")],
     )
     @option("name", description="Назва каналу")
     @option("reactions", description="Дозволити реакції")
     @option("threads", description="Дозволити гілки")
-    async def customchannel_get_cmd(
+    async def custom_channel_get_cmd(
         self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool
-    ):
+    ) -> None:
-        holo_user_ctx = HoloUser(ctx.user)
+        holo_user_ctx: HoloUser = HoloUser(ctx.user)
 
-        if holo_user_ctx.customchannel == None:
+        # Return if the user is using the command outside of a guild
-            await ctx.defer()
+        if not hasattr(ctx.author, "guild"):
-            created_channel = await ctx.user.guild.create_text_channel(
-                name=name,
-                reason=f"Користувач {guild_name(ctx.user)} отримав власний приватний канал",
-                category=ds_utils.get(
-                    ctx.author.guild.categories,
-                    id=await config_get("customchannels", "categories"),
-                ),
-            )
-            await created_channel.set_permissions(
-                ctx.user.guild.default_role,
-                send_messages=False,
-                add_reactions=reactions,
-                create_public_threads=threads,
-                create_private_threads=threads,
-            )
-            await created_channel.set_permissions(
-                ctx.user,
-                attach_files=True,
-                manage_messages=True,
-                send_messages=True,
-                embed_links=True,
-                manage_channels=True,
-            )
-            holo_user_ctx.set("customchannel", created_channel.id)
-            await ctx.respond(
-                embed=Embed(
-                    title="Створено канал",
-                    description=f"Вітаємо! Ви створили канал {created_channel.mention}. Для керування ним користуйтесь меню налаштувань каналу а також командою `/customchannel edit`",
-                    color=Color.success,
-                )
-            )
-            bots = await config_get("bots")
-            for bot in bots:
-                await created_channel.set_permissions(
-                    ds_utils.get(ctx.user.guild.roles, id=bots[bot]["role"]),
-                    view_channel=False,
-                )
-        else:
             await ctx.defer(ephemeral=True)
             await ctx.respond(
                 embed=Embed(
                     title="Помилка виконання",
-                    description=f"У вас вже є особистий канал.\nДля редагування каналу є `/customchannel edit` або просто відкрийте меню керування вашим каналом.",
+                    description="Виконання за межами сервера не є можливим.",
-                    color=Color.fail,
+                    color=Color.FAIL,
                 )
             )
+            return
 
-    @customchannel.command(
+        # Return if the user already has a custom channel
+        if holo_user_ctx.customchannel is not None:
+            await ctx.defer(ephemeral=True)
+            await ctx.respond(
+                embed=Embed(
+                    title="Помилка виконання",
+                    description="У вас вже є особистий канал.\nДля редагування каналу є `/customchannel edit` або просто відкрийте меню керування вашим каналом.",
+                    color=Color.FAIL,
+                )
+            )
+            return
+
+        await ctx.defer()
+
+        created_channel: TextChannel = await ctx.user.guild.create_text_channel(
+            name=name,
+            reason=f"Користувач {guild_name(ctx.user)} отримав власний приватний канал",
+            category=ds_utils.get(
+                ctx.author.guild.categories,
+                id=await config_get("customchannels", "categories"),
+            ),
+        )
+
+        await created_channel.set_permissions(
+            ctx.user.guild.default_role,
+            send_messages=False,
+            add_reactions=reactions,
+            create_public_threads=threads,
+            create_private_threads=threads,
+        )
+        await created_channel.set_permissions(
+            ctx.user,
+            attach_files=True,
+            manage_messages=True,
+            send_messages=True,
+            embed_links=True,
+            manage_channels=True,
+        )
+
+        await holo_user_ctx.set("customchannel", created_channel.id)
+
+        await ctx.respond(
+            embed=Embed(
+                title="Створено канал",
+                description=f"Вітаємо! Ви створили канал {created_channel.mention}. Для керування ним користуйтесь меню налаштувань каналу а також командою `/customchannel edit`",
+                color=Color.SUCCESS,
+            )
+        )
+
+        bots: Dict[str, Any] = await config_get("bots")
+
+        for bot in bots:
+            role: Union[Role, None] = ds_utils.get(
+                ctx.user.guild.roles, id=bots[bot]["role"]
+            )
+
+            if role is not None:
+                await created_channel.set_permissions(
+                    role,
+                    view_channel=False,
+                )
+
+    @custom_channel_group.command(
         name="edit",
         description="Змінити параметри особистого каналу",
-        guild_ids=[config_get_sync("guild")],
+        guild_ids=[sync_config_get("guild")],
     )
     @option("name", description="Назва каналу")
     @option("reactions", description="Дозволити реакції")
     @option("threads", description="Дозволити гілки")
-    async def customchannel_edit_cmd(
+    async def custom_channel_edit_cmd(
         self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool
-    ):
+    ) -> None:
-        holo_user_ctx = HoloUser(ctx.user)
+        holo_user_ctx: HoloUser = HoloUser(ctx.user)
 
-        custom_channel = ds_utils.get(
+        custom_channel: Union[TextChannel, None] = ds_utils.get(
             ctx.guild.channels, id=holo_user_ctx.customchannel
         )
+
+        # Return if the channel was not found
         if custom_channel is None:
             await ctx.respond(
                 embed=Embed(
                     title="Канал не знайдено",
-                    description=f"Канал, вказаний як ваш, не існує. Можливо, його було вручну видалено раніше.",
+                    description="Канал, вказаний як ваш, не існує. Можливо, його було вручну видалено раніше.",
-                    color=Color.fail,
+                    color=Color.FAIL,
                 )
             )
             return
+
         await custom_channel.edit(name=name)
         await custom_channel.set_permissions(
             ctx.user.guild.default_role,
@@ -118,64 +155,84 @@ class CustomChannels(commands.Cog):
             create_public_threads=threads,
             create_private_threads=threads,
         )
+
         await ctx.respond(
             embed=Embed(
                 title="Канал змінено",
                 description=f"Назва каналу тепер `{name}`, реакції `{reactions}` та дозволено треди `{threads}`",
-                color=Color.fail,
+                color=Color.FAIL,
             )
         )
 
-    @customchannel.command(
+    @custom_channel_group.command(
         name="remove",
         description="Відібрати канал, знищуючи його, та частково повернути кошти",
-        guild_ids=[config_get_sync("guild")],
+        guild_ids=[sync_config_get("guild")],
     )
     @option("confirm", description="Підтвердження операції")
-    async def customchannel_remove_cmd(
+    async def custom_channel_remove_cmd(
         self, ctx: ApplicationContext, confirm: bool = False
-    ):
+    ) -> None:
-        holo_user_ctx = HoloUser(ctx.user)
+        holo_user_ctx: HoloUser = HoloUser(ctx.user)
 
-        if holo_user_ctx.customchannel is not None:
+        # Return if the user does not have a custom channel
-            await ctx.defer()
+        if holo_user_ctx.customchannel is None:
-            custom_channel = ds_utils.get(
-                ctx.guild.channels, id=holo_user_ctx.customchannel
-            )
-            if custom_channel is None:
-                await ctx.respond(
-                    embed=Embed(
-                        title="Канал не знайдено",
-                        description=f"Канал, вказаний як ваш, не існує. Можливо, його було вручну видалено раніше.",
-                        color=Color.fail,
-                    )
-                )
-                holo_user_ctx.set("customchannel", None)
-                return
-            if not confirm:
-                await ctx.respond(
-                    embed=Embed(
-                        title="Підтвердження не надано",
-                        description=f"Для підтвердження операції додайте до команди параметр `confirm` зі значенням `True`.",
-                        color=Color.fail,
-                    )
-                )
-                return
-            await custom_channel.delete(reason="Власник запросив видалення")
-            holo_user_ctx.set("customchannel", None)
-            await ctx.respond(
-                embed=Embed(
-                    title="Канал знищено",
-                    description=f"Ви відмовились від каналу та видалили його.",
-                    color=Color.default,
-                )
-            )
-        else:
             await ctx.defer(ephemeral=True)
             await ctx.respond(
                 embed=Embed(
                     title="Помилка виконання",
-                    description=f"У вас немає особистого каналу.",
+                    description="У вас немає особистого каналу.",
-                    color=Color.fail,
+                    color=Color.FAIL,
                 )
             )
+            return
+
+        await ctx.defer()
+
+        custom_channel: Union[TextChannel, None] = ds_utils.get(
+            ctx.guild.channels, id=holo_user_ctx.customchannel
+        )
+
+        # Return if the channel was not found
+        if custom_channel is None:
+            await ctx.respond(
+                embed=Embed(
+                    title="Канал не знайдено",
+                    description="Канал, вказаний як ваш, не існує. Можливо, його було вручну видалено раніше.",
+                    color=Color.FAIL,
+                )
+            )
+            await holo_user_ctx.set("customchannel", None)
+            return
+
+        # Return if the confirmation is missing
+        if not confirm:
+            await ctx.respond(
+                embed=Embed(
+                    title="Підтвердження не надано",
+                    description="Для підтвердження операції додайте до команди параметр `confirm` зі значенням `True`.",
+                    color=Color.FAIL,
+                )
+            )
+            return
+
+        await custom_channel.delete(reason="Власник запросив видалення")
+
+        await holo_user_ctx.set("customchannel", None)
+
+        try:
+            await ctx.respond(
+                embed=Embed(
+                    title="Канал знищено",
+                    description="Ви відмовились від каналу та видалили його.",
+                    color=Color.DEFAULT,
+                )
+            )
+        except Exception as exc:
+            logger.warning(
+                "Could not send a custom channel removal confirmation due to: %s", exc
+            )
+
+
+def setup(client: PycordBot) -> None:
+    client.add_cog(CustomChannels(client))

cogs/data.py

@@ -1,152 +1,183 @@
 import logging
 from os import makedirs
 from pathlib import Path
+from typing import Union, List, Dict, Any
 from uuid import uuid4
 
-from discord import ApplicationContext, Embed, File, option
+from discord import ApplicationContext, Embed, File, option, Role, TextChannel
 from discord import utils as ds_utils
 from discord.commands import SlashCommandGroup
 from discord.ext import commands
+from libbot import config_get
+from libbot.pycord.classes import PycordBot
+from libbot.sync import config_get as sync_config_get
+from libbot.sync import json_write as sync_json_write
 
-from enums.colors import Color
+from classes.holo_user import HoloUser
+from enums import Color
 from modules.database import col_users
-from modules.utils import config_get
+from modules.utils_sync import guild_name
-from modules.utils_sync import config_get_sync, guild_name, json_write_sync
 
 logger = logging.getLogger(__name__)
 
 
 class Data(commands.Cog):
-    def __init__(self, client):
+    def __init__(self, client: PycordBot):
-        self.client = client
+        self.client: PycordBot = client
 
-    data = SlashCommandGroup("data", "Керування даними користувачів")
+    data: SlashCommandGroup = SlashCommandGroup("data", "Керування даними користувачів")
 
     @data.command(
         name="export",
         description="Експортувати дані",
-        guild_ids=[config_get_sync("guild")],
+        guild_ids=[sync_config_get("guild")],
     )
     @option(
         "kind", description="Тип даних, які треба експортувати", choices=["Користувачі"]
     )
-    async def data_export_cmd(self, ctx: ApplicationContext, kind: str):
+    async def data_export_cmd(self, ctx: ApplicationContext, kind: str) -> None:
         await ctx.defer()
-        if ctx.user.id in await config_get("admins"):
+
+        # Return if the user is not an owner and not in the council
+        if (ctx.user.id not in self.client.owner_ids) and not (
+            await HoloUser.is_council(ctx.author)
+        ):
             logging.info(
-                f"Moderator {guild_name(ctx.user)} exported current users list"
+                "User %s tried to use /export but permission denied",
+                guild_name(ctx.user),
             )
-            makedirs("tmp", exist_ok=True)
-            uuid = str(uuid4())
 
-            if kind == "Користувачі":
-                users = []
-
-                for member in ctx.guild.members:
-                    users.append(
-                        {
-                            "id": member.id,
-                            "nick": member.nick,
-                            "username": f"{member.name}#{member.discriminator}",
-                            "bot": member.bot,
-                        }
-                    )
-
-                json_write_sync(users, str(Path(f"tmp/{uuid}")))
-
-                await ctx.respond(
-                    file=File(str(Path(f"tmp/{uuid}")), filename="users.json")
-                )
-        else:
-            logging.info(
-                f"User {guild_name(ctx.user)} tried to use /export but permission denied"
-            )
             await ctx.respond(
                 embed=Embed(
                     title="Відмовлено в доступі",
                     description="Здається, це команда лише для модераторів",
-                    color=Color.fail,
+                    color=Color.FAIL,
                 )
             )
-            mod_role = ds_utils.get(
+
-                ctx.user.guild.roles, id=await config_get("moderator", "roles")
+            mod_role: Union[Role, None] = ds_utils.get(
+                ctx.user.guild.roles, id=await config_get("moderators", "roles")
             )
-            admin_chan = ds_utils.get(
+            admin_chan: Union[TextChannel, None] = ds_utils.get(
                 ctx.user.guild.channels,
                 id=await config_get("adminchat", "channels", "text"),
             )
+
             await admin_chan.send(
-                content=f"{mod_role.mention}",
+                content="" if mod_role is None else mod_role.mention,
                 embed=Embed(
                     title="Неавторизований запит",
                     description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
-                    color=Color.fail,
+                    color=Color.FAIL,
                 ),
             )
 
+            return
+
+        logging.info("Moderator %s exported current users list", guild_name(ctx.user))
+
+        makedirs("tmp", exist_ok=True)
+
+        uuid: str = str(uuid4())
+
+        if kind == "Користувачі":
+            users: List[Dict[str, Any]] = []
+
+            for member in ctx.guild.members:
+                users.append(
+                    {
+                        "id": member.id,
+                        "nick": member.nick,
+                        "username": f"{member.name}#{member.discriminator}",
+                        "bot": member.bot,
+                    }
+                )
+
+            sync_json_write(users, Path(f"tmp/{uuid}"))
+
+            await ctx.respond(file=File(Path(f"tmp/{uuid}"), filename="users.json"))
+
     @data.command(
         name="migrate",
         description="Мігрувати всіх користувачів до бази",
-        guild_ids=[config_get_sync("guild")],
+        guild_ids=[sync_config_get("guild")],
     )
     @option(
         "kind", description="Тип даних, які треба експортувати", choices=["Користувачі"]
     )
-    async def data_migrate_cmd(self, ctx: ApplicationContext, kind: str):
+    async def data_migrate_cmd(self, ctx: ApplicationContext, kind: str) -> None:
         await ctx.defer()
-        if ctx.user.id in await config_get("admins"):
+
+        # Return if the user is not an owner and not in the council
+        if (ctx.user.id not in self.client.owner_ids) and not (
+            await HoloUser.is_council(ctx.author)
+        ):
             logging.info(
-                f"Moderator {guild_name(ctx.user)} started migration of all members to the database"
+                "User %s tried to use /migrate but permission denied",
+                guild_name(ctx.user),
             )
 
-            if kind == "Користувачі":
-                for member in ctx.guild.members:
-                    if member.bot:
-                        continue
-                    if col_users.find_one({"user": member.id}) is None:
-                        user = {}
-                        defaults = await config_get("user", "defaults")
-
-                        user["user"] = member.id
-
-                        for key in defaults:
-                            user[key] = defaults[key]
-
-                        col_users.insert_one(document=user)
-                        logging.info(
-                            f"Added DB record for user {member.id} during migration"
-                        )
-
-            await ctx.respond(
-                embed=Embed(
-                    title="Міграцію завершено",
-                    description="Всім користувачам сервера було створено записи в базі даних.",
-                    color=Color.success,
-                )
-            )
-        else:
-            logging.info(
-                f"User {guild_name(ctx.user)} tried to use /migrate but permission denied"
-            )
             await ctx.respond(
                 embed=Embed(
                     title="Відмовлено в доступі",
                     description="Здається, це команда лише для модераторів",
-                    color=Color.fail,
+                    color=Color.FAIL,
                 )
             )
-            mod_role = ds_utils.get(
+
-                ctx.user.guild.roles, id=await config_get("moderator", "roles")
+            mod_role: Union[Role, None] = ds_utils.get(
+                ctx.user.guild.roles, id=await config_get("moderators", "roles")
             )
-            admin_chan = ds_utils.get(
+            admin_chan: Union[TextChannel, None] = ds_utils.get(
                 ctx.user.guild.channels,
                 id=await config_get("adminchat", "channels", "text"),
             )
-            await admin_chan.send(
+
-                content=f"{mod_role.mention}",
+            if admin_chan is not None:
-                embed=Embed(
+                await admin_chan.send(
-                    title="Неавторизований запит",
+                    content="" if mod_role is None else mod_role.mention,
-                    description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
+                    embed=Embed(
-                    color=Color.fail,
+                        title="Неавторизований запит",
-                ),
+                        description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
+                        color=Color.FAIL,
+                    ),
+                )
+
+            return
+
+        logging.info(
+            "Moderator %s started migration of all members to the database",
+            guild_name(ctx.user),
+        )
+
+        if kind == "Користувачі":
+            for member in ctx.guild.members:
+                if member.bot:
+                    continue
+
+                if (await col_users.find_one({"user": member.id})) is None:
+                    user: Dict[str, Any] = {}
+                    defaults: Dict[str, Any] = await config_get("user", "defaults")
+
+                    user["user"] = member.id
+
+                    for key in defaults:
+                        user[key] = defaults[key]
+
+                    await col_users.insert_one(document=user)
+
+                    logging.info(
+                        "Added DB record for user %s during migration", member.id
+                    )
+
+        await ctx.respond(
+            embed=Embed(
+                title="Міграцію завершено",
+                description="Всім користувачам сервера було створено записи в базі даних.",
+                color=Color.SUCCESS,
             )
+        )
+
+
+def setup(client: PycordBot) -> None:
+    client.add_cog(Data(client))

cogs/fun.py

@@ -0,0 +1,58 @@
+import logging
+
+from discord import ApplicationContext, Embed, User, option, slash_command
+from discord.ext import commands
+from libbot import config_get
+from libbot.pycord.classes import PycordBot
+from libbot.sync import config_get as sync_config_get
+
+from modules.utils_sync import guild_name
+from modules.waifu_pics import waifu_pics
+
+logger = logging.getLogger(__name__)
+
+
+class Fun(commands.Cog):
+    def __init__(self, client: PycordBot):
+        self.client: PycordBot = client
+
+    @slash_command(
+        name="action",
+        description="Провести над користувачем РП дію",
+        guild_ids=[sync_config_get("guild")],
+    )
+    @option(
+        "type",
+        description="Тип дії, яку хочете провести з користувачем",
+        choices=sync_config_get("actions").keys(),
+    )
+    @option("user", description="Користувач")
+    async def action_cmd(self, ctx: ApplicationContext, type: str, user: User) -> None:
+        await ctx.defer()
+
+        action: str = await config_get("category", "actions", type)
+        action_verb: str = await config_get("action", "actions", type)
+
+        image_url: str = await waifu_pics.sfw(action)
+
+        logger.info(
+            "User %s (%s) %s %s (%s) with image %s",
+            guild_name(ctx.user),
+            ctx.user.id,
+            action_verb,
+            guild_name(user),
+            user.id,
+            image_url,
+        )
+
+        embed: Embed = Embed(
+            description=f"**{guild_name(ctx.user)}** {action_verb} **{guild_name(user)}**",
+            color=0x2F3136,
+        )
+        embed.set_image(url=image_url)
+
+        await ctx.respond(embed=embed)
+
+
+def setup(client: PycordBot) -> None:
+    client.add_cog(Fun(client))

cogs/logger.py

@@ -1,48 +1,51 @@
-from discord import Member, Message
+from typing import Dict, Any, Union
+
+from discord import Member, Message, TextChannel
 from discord import utils as ds_utils
 from discord.ext import commands
+from libbot import config_get
+from libbot.pycord.classes import PycordBot
 
 from modules.database import col_users
-from modules.utils import config_get
 
 
 class Logger(commands.Cog):
-    def __init__(self, client):
+    def __init__(self, client: PycordBot):
-        self.client = client
+        self.client: PycordBot = client
 
     @commands.Cog.listener()
     async def on_message(self, message: Message):
         if (
             (message.author != self.client.user)
-            and (message.author.bot == False)
+            and (message.author.bot is False)
-            and (message.author.system == False)
+            and (message.author.system is False)
         ):
-            if col_users.find_one({"user": message.author.id}) is None:
+            if (await col_users.find_one({"user": message.author.id})) is None:
-                user = {}
+                user: Dict[str, Any] = {}
-                defaults = await config_get("user", "defaults")
+                defaults: Dict[str, Any] = await config_get("user", "defaults")
 
                 user["user"] = message.author.id
 
                 for key in defaults:
                     user[key] = defaults[key]
 
-                col_users.insert_one(document=user)
+                await col_users.insert_one(document=user)
 
     @commands.Cog.listener()
-    async def on_member_join(self, member: Member):
+    async def on_member_join(self, member: Member) -> None:
-        welcome_chan = ds_utils.get(
+        welcome_chan: Union[TextChannel, None] = ds_utils.get(
             self.client.get_guild(await config_get("guild")).channels,
             id=await config_get("welcome", "channels", "text"),
         )
-        rules_chan = ds_utils.get(
+        rules_chan: Union[TextChannel, None] = ds_utils.get(
             self.client.get_guild(await config_get("guild")).channels,
             id=await config_get("rules", "channels", "text"),
         )
 
         if (
             (member != self.client.user)
-            and (member.bot == False)
+            and (member.bot is False)
-            and (member.system == False)
+            and (member.system is False)
         ):
             await welcome_chan.send(
                 content=(await config_get("welcome", "messages")).format(
@@ -50,13 +53,17 @@ class Logger(commands.Cog):
                 )
             )
 
-            if col_users.find_one({"user": member.id}) is None:
+            if (await col_users.find_one({"user": member.id})) is None:
-                user = {}
+                user: Dict[str, Any] = {}
-                defaults = await config_get("user", "defaults")
+                defaults: Dict[str, Any] = await config_get("user", "defaults")
 
                 user["user"] = member.id
 
                 for key in defaults:
                     user[key] = defaults[key]
 
-                col_users.insert_one(document=user)
+                await col_users.insert_one(document=user)
+
+
+def setup(client: PycordBot) -> None:
+    client.add_cog(Logger(client))

config_example.json

@@ -1,8 +1,16 @@
 {
-    "token": "",
+    "locale": "en",
-    "owner": 0,
+    "debug": false,
     "guild": 0,
-    "admins": [],
+    "bot": {
+        "owners": [
+            0
+        ],
+        "debug_guilds": [
+            0
+        ],
+        "bot_token": ""
+    },
     "status": {
         "type": "playing",
         "message": "on your nerves"
@@ -36,10 +44,45 @@
         "voice": {}
     },
     "roles": {
-        "moderator": 0
+        "council": 0,
+        "moderators": 0
     },
     "bots": {},
     "messages": {
         "welcome": "Вітаємо {mention} на сервері HoloUA! Будь ласка, ознайомся з правилами серверу на каналі {rules}. Сподіваємося, тобі тут сподобається!"
+    },
+    "actions": {
+        "Вкусити": {
+            "category": "bite",
+            "action": "робить кусь"
+        },
+        "Обійняти": {
+            "category": "hug",
+            "action": "обіймає"
+        },
+        "Поцілувати": {
+            "category": "kiss",
+            "action": "цілує"
+        },
+        "Лизнути": {
+            "category": "lick",
+            "action": "лиже"
+        },
+        "Погладити": {
+            "category": "pat",
+            "action": "гладить"
+        },
+        "Тикнути": {
+            "category": "poke",
+            "action": "тикає в"
+        },
+        "Помахати": {
+            "category": "wave",
+            "action": "махає"
+        },
+        "Підморгнути": {
+            "category": "wink",
+            "action": "підморгує"
+        }
     }
 }

enums/__init__.py

@@ -0,0 +1 @@
+from .colors import Color

enums/colors.py

@@ -1,6 +1,7 @@
 from enum import IntEnum
 
+
 class Color(IntEnum):
-    fail = 0xd6345b
+    FAIL = 0xD6345B
-    success = 0x84d961
+    SUCCESS = 0x84D961
-    default = 0xa7a6ab
+    DEFAULT = 0xA7A6AB

errors/__init__.py

@@ -0,0 +1 @@
+from .user import UserNotFoundError

errors/user.py

@@ -0,0 +1,10 @@
+class UserNotFoundError(Exception):
+    """HoloUser could not find user with such an ID in database"""
+
+    def __init__(self, user, user_id):
+        self.user = user
+        self.user_id = user_id
+
+        super().__init__(
+            f"User of type {type(self.user)} with id {self.user_id} was not found"
+        )

locale/en.json

@@ -0,0 +1,9 @@
+{
+    "metadata": {
+        "flag": "🇬🇧",
+        "name": "English",
+        "codes": [
+            "en"
+        ]
+    }
+}

locale/uk.json

@@ -0,0 +1,9 @@
+{
+    "metadata": {
+        "flag": "🇺🇦",
+        "name": "Українська",
+        "codes": [
+            "uk"
+        ]
+    }
+}

main.py

@@ -1,16 +1,13 @@
 import logging
+import sys
+from logging import Logger
 
 from discord import Activity, ActivityType
+from libbot import config_get
+from libbot.sync import config_get as sync_config_get
 
-from cogs.admin import Admin
-from cogs.analytics import Analytics
-from cogs.custom_channels import CustomChannels
-from cogs.data import Data
-from cogs.logger import Logger
 from modules.client import client
-from modules.scheduled import scheduler
+from modules.scheduler import scheduler
-from modules.utils import config_get
-from modules.utils_sync import config_get_sync
 
 logging.basicConfig(
     level=logging.INFO,
@@ -18,7 +15,7 @@ logging.basicConfig(
     datefmt="[%X]",
 )
 
-logger = logging.getLogger(__name__)
+logger: Logger = logging.getLogger(__name__)
 
 try:
     import uvloop  # type: ignore
@@ -29,11 +26,11 @@ except ImportError:
 
 
 @client.event
-async def on_ready():
+async def on_ready() -> None:
-    logger.info(f"Logged in as {client.user}")
+    logger.info("Logged in as %s", client.user)
 
-    activity_type = await config_get("type", "status")
+    activity_type: str = await config_get("type", "status")
-    activity_message = await config_get("message", "status")
+    activity_message: str = await config_get("message", "status")
 
     if activity_type == "playing":
         await client.change_presence(
@@ -62,22 +59,20 @@ async def on_ready():
     else:
         return
 
-    logger.info(f"Set activity type to {activity_type} with message {activity_message}")
+    logger.info(
+        "Set activity type to %s with message %s", activity_type, activity_message
+    )
 
 
-def main():
+def main() -> None:
-    client.add_cog(Admin(client))
+    client.load_extension("cogs")
-    client.add_cog(Analytics(client))
-    client.add_cog(CustomChannels(client))
-    client.add_cog(Data(client))
-    client.add_cog(Logger(client))
 
     try:
         scheduler.start()
-        client.run(config_get_sync("token"))
+        client.run(sync_config_get("bot_token", "bot"))
     except KeyboardInterrupt:
         scheduler.shutdown()
-        exit()
+        sys.exit()
 
 
 if __name__ == "__main__":

modules/client.py

@@ -1,5 +1,10 @@
-from discord import Intents, Bot
+from discord import Intents
+from libbot.pycord.classes import PycordBot
+
+from modules.scheduler import scheduler
+
+intents: Intents = Intents().all()
 
-intents = Intents().all()
 intents.members = True
-client = Bot(intents=intents)
+
+client: PycordBot = PycordBot(intents=intents, scheduler=scheduler)

modules/database.py

@@ -1,12 +1,19 @@
+from typing import Dict, Any
+
+from async_pymongo import AsyncClient, AsyncCollection, AsyncDatabase
+from libbot.sync import config_get as sync_config_get
 from pymongo import MongoClient
-from ujson import loads
+from pymongo.synchronous.collection import Collection
+from pymongo.synchronous.database import Database
 
-with open("config.json", "r", encoding="utf-8") as f:
+db_config: Dict[str, Any] = sync_config_get("database")
-    db_config = loads(f.read())["database"]
-    f.close()
 
-db_client = MongoClient(
+con_string: str = (
-    "mongodb://{0}:{1}@{2}:{3}/{4}".format(
+    "mongodb://{0}:{1}/{2}".format(
+        db_config["host"], db_config["port"], db_config["name"]
+    )
+    if db_config["user"] is None or db_config["password"] is None
+    else "mongodb://{0}:{1}@{2}:{3}/{4}".format(
         db_config["user"],
         db_config["password"],
         db_config["host"],
@@ -14,18 +21,20 @@ db_client = MongoClient(
         db_config["name"],
     )
 )
-db = db_client.get_database(name=db_config["name"])
 
-collections = db.list_collection_names()
+db_client: AsyncClient = AsyncClient(con_string)
+db_client_sync: MongoClient = MongoClient(con_string)
 
-for collection in ["users", "warnings", "scheduler", "analytics"]:
+# Async declarations per default
-    if not collection in collections:
+db: AsyncDatabase = db_client.get_database(name=db_config["name"])
-        db.create_collection(collection)
 
-col_users = db.get_collection("users")
+col_users: AsyncCollection = db.get_collection("users")
-col_warnings = db.get_collection("warnings")
+col_warnings: AsyncCollection = db.get_collection("warnings")
-col_analytics = db.get_collection("analytics")
+col_analytics: AsyncCollection = db.get_collection("analytics")
-# col_checkouts = db.get_collection("checkouts")
+
-# col_trackings = db.get_collection("trackings")
+# Sync declarations as a fallback
-# col_authorized = db.get_collection("authorized")
+sync_db: Database = db_client_sync.get_database(name=db_config["name"])
-# col_transactions = db.get_collection("transactions")
+
+sync_col_users: Collection = sync_db.get_collection("users")
+sync_col_warnings: Collection = sync_db.get_collection("warnings")
+sync_col_analytics: Collection = sync_db.get_collection("analytics")

modules/scheduler.py

@@ -1,3 +1,3 @@
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 
-scheduler = AsyncIOScheduler()
+scheduler: AsyncIOScheduler = AsyncIOScheduler()

modules/utils.py

@@ -1,38 +0,0 @@
-from typing import Any
-
-import aiofiles
-from ujson import dumps, loads
-
-
-async def json_read(path: str) -> Any:
-    async with aiofiles.open(path, mode="r", encoding="utf-8") as f:
-        data = await f.read()
-    return loads(data)
-
-
-async def json_write(data: Any, path: str) -> None:
-    async with aiofiles.open(path, mode="w", encoding="utf-8") as f:
-        await f.write(
-            dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4)
-        )
-
-
-async def config_get(key: str, *path: str) -> Any:
-    this_key = await json_read("config.json")
-    for dict_key in path:
-        this_key = this_key[dict_key]
-    return this_key[key]
-
-
-async def config_set(key: str, value: Any, *path: str) -> None:
-    this_dict = await json_read("config.json")
-    string = "this_dict"
-    for arg in path:
-        string += f'["{arg}"]'
-    if type(value) in [str]:
-        string += f'["{key}"] = "{value}"'
-    else:
-        string += f'["{key}"] = {value}'
-    exec(string)
-    await json_write(this_dict, "config.json")
-    return

modules/utils_sync.py

@@ -1,43 +1,10 @@
-from typing import Any
+from typing import Union
 
-from discord import Member
+from discord import Member, User
-from ujson import dumps, loads
 
 
-def json_read_sync(path: str) -> Any:
+def guild_name(member: Union[Member, User]) -> str:
-    with open(path, mode="r", encoding="utf-8") as f:
+    if isinstance(member, User):
-        data = f.read()
-    return loads(data)
-
-
-def json_write_sync(data: Any, path: str) -> None:
-    with open(path, mode="w", encoding="utf-8") as f:
-        f.write(dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4))
-
-
-def config_get_sync(key: str, *path: str) -> Any:
-    this_key = json_read_sync("config.json")
-    for dict_key in path:
-        this_key = this_key[dict_key]
-    return this_key[key]
-
-
-def config_set_sync(key: str, value: Any, *path: str) -> None:
-    this_dict = json_read_sync("config.json")
-    string = "this_dict"
-    for arg in path:
-        string += f'["{arg}"]'
-    if type(value) in [str]:
-        string += f'["{key}"] = "{value}"'
-    else:
-        string += f'["{key}"] = {value}'
-    exec(string)
-    json_write_sync(this_dict, "config.json")
-    return
-
-
-def guild_name(member: Member):
-    if member.nick == None:
         return member.name
-    else:
+
-        return member.nick
+    return member.name if member.nick is None else member.nick

modules/waifu_pics.py

@@ -0,0 +1,3 @@
+from WaifuPicsPython import WaifuAsync
+
+waifu_pics: WaifuAsync = WaifuAsync()

requirements.txt

@@ -1,5 +1,10 @@
-aiofiles==23.1.0
+# Waifu pics related dependencies (not listed directly by waifupics)
-apscheduler==3.10.1
+aiohttp>=3.10.0
-py-cord[speed]==2.4.1
+requests>=2.32.2
-pymongo==4.3.3
+
-ujson==5.7.0
+aiofiles~=24.1.0
+apscheduler>=3.10.0
+async_pymongo==0.1.11
+libbot[speed,pycord]==3.2.3
+ujson~=5.10.0
+WaifuPicsPython==0.2.0

validation/analytics.json

@@ -0,0 +1,54 @@
+{
+    "$jsonSchema": {
+        "required": [
+            "user",
+            "channel",
+            "content",
+            "stickers",
+            "attachments"
+        ],
+        "properties": {
+            "user": {
+                "bsonType": "long",
+                "description": "Discord ID of user"
+            },
+            "channel": {
+                "bsonType": "int",
+                "description": "Discord ID of a channel"
+            },
+            "content": {
+                "bsonType": ["null", "string"],
+                "description": "Text of the message"
+            },
+            "stickers": {
+                "bsonType": "array",
+                "items": {
+                    "bsonType": "object",
+                    "required": [
+                        "id",
+                        "name",
+                        "format",
+                        "url"
+                    ],
+                    "properties": {
+                        "id": {
+                            "bsonType": "int"
+                        },
+                        "name": {
+                            "bsonType": "string"
+                        },
+                        "format": {
+                            "bsonType": "array"
+                        },
+                        "user": {
+                            "bsonType": "string"
+                        }
+                    }
+                }
+            },
+            "attachments": {
+                "bsonType": "array"
+            }
+        }
+    }
+}

validation/users.json

@@ -0,0 +1,24 @@
+{
+    "$jsonSchema": {
+        "required": [
+            "user",
+            "customrole",
+            "customchannel"
+
+        ],
+        "properties": {
+            "user": {
+                "bsonType": "long",
+                "description": "Discord ID of user"
+            },
+            "customrole": {
+                "bsonType": ["null", "long"],
+                "description": "Discord ID of custom role or 'null' if not set"
+            },
+            "customchannel": {
+                "bsonType": ["null", "long"],
+                "description": "Discord ID of custom channel or 'null' if not set"
+            }
+        }
+    }
+}

validation/warnings.json

@@ -0,0 +1,18 @@
+{
+    "$jsonSchema": {
+        "required": [
+            "user",
+            "warns"
+        ],
+        "properties": {
+            "user": {
+                "bsonType": "long",
+                "description": "Discord ID of user"
+            },
+            "warns": {
+                "bsonType": "int",
+                "description": "Number of warnings on count"
+            }
+        }
+    }
+}