Imported all cogs

.gitignore

@@ -167,3 +167,5 @@ venv_linux
 venv_windows
 
 config.json
+
+tmp

cogs/admin.py

@@ -0,0 +1,201 @@
+import logging
+from datetime import timedelta
+from os import getpid, system
+from typing import Union
+
+from discord import ApplicationContext, Embed, User, slash_command, option
+from discord import utils as ds_utils
+from discord.ext import commands
+
+from classes.holo_user import HoloUser
+from enums.colors import Color
+from modules.utils import config_get
+from modules.utils_sync import config_get_sync, guild_name
+from modules.scheduled import scheduler
+
+logger = logging.getLogger(__name__)
+
+
+class Admin(commands.Cog):
+    def __init__(self, client):
+        self.client = client
+
+    # @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("moderator", "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="Видалити деяку кількість повідомлень в каналі",
+        guild_ids=[config_get_sync("guild")],
+    )
+    @option("amount", description="Кількість")
+    @option("user", description="Користувач", default=None)
+    async def clear_cmd(
+        self,
+        ctx: ApplicationContext,
+        amount: int,
+        user: User,
+    ):
+        if ctx.user.id in await config_get("admins"):
+            logging.info(
+                f"User {ctx.user.id} removed {amount} message(s) in {ctx.channel.id}"
+            )
+            await ctx.respond(
+                embed=Embed(description="Видаляю..."), ephemeral=True, delete_after=2.0
+            )
+            if user == None:
+                await ctx.channel.purge(limit=amount)
+            else:
+                await ctx.channel.purge(
+                    limit=amount, check=lambda msg: msg.author == user
+                )
+        else:
+            logging.warning(
+                f"User {guild_name(ctx.user)} tried to use /clear 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("moderator", "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="reboot",
+        description="Перезапустити бота",
+        guild_ids=[config_get_sync("guild")],
+    )
+    async def reboot_cmd(self, ctx: ApplicationContext):
+        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)}")
+            await ctx.respond(
+                embed=Embed(
+                    title="Вимикаюсь...",
+                    description="Спробую перезавантажитись за 5 секунд",
+                )
+            )
+            scheduler.shutdown()
+            await self.client.close()
+            exit()
+        else:
+            logging.warning(
+                f"User {guild_name(ctx.user)} tried to use /reboot 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("moderator", "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,
+                ),
+            )

cogs/analytics.py

@@ -0,0 +1,30 @@
+import logging
+from discord import (
+    Cog,
+    Message,
+)
+from discord.ext import commands
+from modules.database import col_analytics
+
+
+logger = logging.getLogger(__name__)
+
+
+class Analytics(commands.Cog):
+    def __init__(self, client):
+        self.client = client
+
+    @Cog.listener()
+    async def on_message(self, message: Message):
+        if (
+            (message.author != self.client.user)
+            and (message.author.bot == False)
+            and (message.author.system == False)
+        ):
+            col_analytics.insert_one(
+                {
+                    "message": message.content,
+                    "user": message.author.id,
+                    "channel": message.channel.id,
+                }
+            )

cogs/data.py

@@ -0,0 +1,79 @@
+import logging
+from datetime import timedelta
+from os import getpid, makedirs, system
+from pathlib import Path
+from uuid import uuid4
+
+from discord import ApplicationContext, Embed, Option, User, slash_command, File
+from discord import utils as ds_utils
+from discord.ext import commands
+
+from classes.holo_user import HoloUser
+from enums.colors import Color
+from modules.utils import config_get
+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):
+        self.client = client
+
+    @slash_command(
+        name="export",
+        description="Експортувати дані",
+        guild_ids=[config_get_sync("guild")],
+    )
+    async def warn_cmd(
+        self,
+        ctx: ApplicationContext,
+    ):
+        await ctx.defer()
+        if ctx.user.id in await config_get("admins"):
+            logging.info(
+                f"Moderator {guild_name(ctx.user)} exported current users list"
+            )
+            makedirs("tmp", exist_ok=True)
+            uuid = str(uuid4())
+            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,
+                )
+            )
+            mod_role = ds_utils.get(
+                ctx.user.guild.roles, id=await config_get("moderator", "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,
+                ),
+            )

config_example.json

@@ -27,10 +27,14 @@
     "channels": {
         "text": {
             "rules": 0,
-            "welcome": 0
+            "welcome": 0,
+            "adminchat": 0
         },
         "voice": {}
     },
+    "roles": {
+        "moderator": 0
+    },
     "bots": {},
     "messages": {
         "welcome": "Вітаємо {mention} на сервері HoloUA! Будь ласка, ознайомся з правилами серверу на каналі {rules}. Сподіваємося, тобі тут сподобається!"

main.py

@@ -8,7 +8,10 @@ from modules.client import client
 from modules.utils import config_get
 from modules.utils_sync import config_get_sync
 
+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
 
 logging.basicConfig(
@@ -36,7 +39,10 @@ async def on_ready():
 
 
 def main():
+    client.add_cog(Admin(client))
+    client.add_cog(Analytics(client))
     client.add_cog(CustomChannels(client))
+    client.add_cog(Data(client))
     client.add_cog(Logger(client))
 
     try:

modules/database.py

@@ -18,12 +18,13 @@ db = db_client.get_database(name=db_config["name"])
 
 collections = db.list_collection_names()
 
-for collection in ["users", "warnings", "scheduler"]:
+for collection in ["users", "warnings", "scheduler", "analytics"]:
     if not collection in collections:
         db.create_collection(collection)
 
 col_users = db.get_collection("users")
 col_warnings = db.get_collection("warnings")
+col_analytics = db.get_collection("analytics")
 # col_checkouts = db.get_collection("checkouts")
 # col_trackings = db.get_collection("trackings")
 # col_authorized = db.get_collection("authorized")

modules/utils.py

@@ -12,7 +12,7 @@ async def json_read(path: str) -> Any:
 
 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))
+        await f.write(dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4))
 
 
 async def config_get(key: str, *path: str) -> Any:

modules/utils_sync.py

@@ -12,7 +12,7 @@ def json_read_sync(path: str) -> Any:
 
 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))
+        f.write(dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4))
 
 
 def config_get_sync(key: str, *path: str) -> Any: