Added warnings service #35

kdb-bot/src/modules/base/service/user_warnings_service.py

@@ -0,0 +1,98 @@
+from cpl_core.database.context import DatabaseContextABC
+from cpl_core.logging import LoggerABC
+from cpl_discord.service import DiscordBotServiceABC
+from cpl_translation import TranslatePipe
+
+from bot_core.abc.message_service_abc import MessageServiceABC
+from bot_data.abc.level_repository_abc import LevelRepositoryABC
+from bot_data.abc.server_repository_abc import ServerRepositoryABC
+from bot_data.abc.user_repository_abc import UserRepositoryABC
+from bot_data.abc.user_warning_repository_abc import UserWarningsRepositoryABC
+from bot_data.model.user_warnings import UserWarnings
+from modules.level.service.level_service import LevelService
+
+
+class UserWarningsService:
+    def __init__(
+        self,
+        logger: LoggerABC,
+        db: DatabaseContextABC,
+        bot: DiscordBotServiceABC,
+        warnings: UserWarningsRepositoryABC,
+        users: UserRepositoryABC,
+        servers: ServerRepositoryABC,
+        levels: LevelRepositoryABC,
+        level_service: LevelService,
+        message_service: MessageServiceABC,
+        t: TranslatePipe,
+    ):
+        self._logger = logger
+        self._db = db
+        self._bot = bot
+        self._warnings = warnings
+        self._users = users
+        self._servers = servers
+        self._levels = levels
+        self._level_service = level_service
+        self._message_service = message_service
+        self._t = t
+
+    async def add_warnings(self, guild_id: int, member_id: int, description: str, author_id: int = None):
+        server = self._servers.get_server_by_discord_id(guild_id)
+        user = self._users.get_user_by_discord_id_and_server_id(member_id, server.id)
+
+        guild = self._bot.get_guild(guild_id)
+        member = guild.get_member(member_id)
+
+        author = None
+        if author_id is not None:
+            author = self._users.get_user_by_discord_id_and_server_id(author_id, server.id)
+
+        warning = UserWarnings(description, user, author)
+        self._warnings.add_user_warnings(warning)
+        self._db.save_changes()
+
+        existing_warnings = self._warnings.get_user_warnings_by_user_id(user.id)
+        await self._message_service.send_dm_message(
+            self._t.transform("modules.base.warnings.message").format(warning.description), member
+        )
+
+        match existing_warnings.count():
+            case 1:
+                await self._message_service.send_dm_message(
+                    self._t.transform("modules.base.warnings.messages.first"), member
+                )
+            case 2:
+                server = self._servers.get_server_by_discord_id(guild_id)
+                user = self._users.get_user_by_discord_id_and_server_id(member_id, server.id)
+                level = self._level_service.get_level(user)
+                levels = self._levels.get_levels_by_server_id(server.id).order_by(lambda l: l.min_xp)
+
+                new_level = levels.where(lambda l: l.min_xp < level.min_xp).last()
+                user.xp = new_level.min_xp
+                self._users.update_user(user)
+                self._db.save_changes()
+                await self._message_service.send_dm_message(
+                    self._t.transform("modules.base.warnings.messages.second"), member
+                )
+            case 3:
+                server = self._servers.get_server_by_discord_id(guild_id)
+                user = self._users.get_user_by_discord_id_and_server_id(member_id, server.id)
+                levels = self._levels.get_levels_by_server_id(server.id)
+
+                new_level = levels.where(lambda l: l.min_xp > 0).order_by(lambda l: l.min_xp).last()
+                user.xp = new_level.min_xp
+                self._users.update_user(user)
+                self._db.save_changes()
+                await self._message_service.send_dm_message(
+                    self._t.transform("modules.base.warnings.messages.third"), member
+                )
+            case 4:
+                user.xp = 0
+                self._users.update_user(user)
+                self._db.save_changes()
+                await member.kick()
+
+    def remove_warnings(self, id: int):
+        warning = self._warnings.get_user_warnings_by_id(id)
+        self._warnings.delete_user_warnings(warning)