Added sync commands #367

kdb-bot/src/modules/technician/command/sync_xp_command.py

@@ -0,0 +1,147 @@
+import discord
+from cpl_core.configuration import ConfigurationABC
+from cpl_core.database.context import DatabaseContextABC
+from cpl_discord.command import DiscordCommandABC
+from cpl_discord.service import DiscordBotServiceABC
+from cpl_translation import TranslatePipe
+from discord import app_commands
+from discord.ext import commands
+from discord.ext.commands import Context
+
+from bot_core.abc.client_utils_abc import ClientUtilsABC
+from bot_core.abc.message_service_abc import MessageServiceABC
+from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
+from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
+from bot_core.helper.command_checks import CommandChecks
+from bot_core.logging.command_logger import CommandLogger
+from bot_data.abc.server_repository_abc import ServerRepositoryABC
+from bot_data.abc.user_repository_abc import UserRepositoryABC
+from bot_data.model.server_config import ServerConfig
+from bot_data.model.technician_config import TechnicianConfig
+from bot_data.model.user import User
+from modules.permission.abc.permission_service_abc import PermissionServiceABC
+
+
+class SyncXpGroup(DiscordCommandABC):
+    def __init__(
+        self,
+        logger: CommandLogger,
+        config: ConfigurationABC,
+        message_service: MessageServiceABC,
+        bot: DiscordBotServiceABC,
+        client_utils: ClientUtilsABC,
+        translate: TranslatePipe,
+        servers: ServerRepositoryABC,
+        users: UserRepositoryABC,
+        permissions: PermissionServiceABC,
+        settings: TechnicianConfig,
+        db: DatabaseContextABC,
+    ):
+        DiscordCommandABC.__init__(self)
+
+        self._logger = logger
+        self._config = config
+        self._message_service = message_service
+        self._bot = bot
+        self._client_utils = client_utils
+        self._t = translate
+        self._servers = servers
+        self._users = users
+        self._permissions = permissions
+        self._settings = settings
+        self._db = db
+
+        self._logger.trace(__name__, f"Loaded command service: {type(self).__name__}")
+
+    @commands.hybrid_group(name="sync-xp")
+    @commands.guild_only()
+    async def sync_xp(self, ctx: Context):
+        pass
+
+    @sync_xp.command(name="all-members")
+    @commands.guild_only()
+    @CommandChecks.check_is_ready()
+    @CommandChecks.check_is_member_technician()
+    async def all_members(self, ctx: Context, server_id: int):
+        self._logger.debug(__name__, f"Received command sync xp {ctx}")
+
+        if ctx.guild is None:
+            return
+
+        settings: ServerConfig = self._config.get_configuration(f"ServerConfig_{ctx.guild.id}")
+        if not FeatureFlagsSettings.get_flag_from_dict(settings.feature_flags, FeatureFlagsEnum.sync_xp):
+            await self._message_service.send_ctx_msg(ctx, self._t.transform("common.feature_not_activated"))
+            return
+
+        other_server = self._servers.get_server_by_id(server_id)
+        users_on_other_server = self._users.get_users_by_server_id(other_server.id).where(lambda x: not x.left_server)
+        discord_ids_on_other_server = users_on_other_server.select(lambda x: x.discord_id)
+
+        for user in self._users.get_users_by_server_id(self._servers.get_server_by_discord_id(ctx.guild.id).id).where(
+            lambda x: not x.left_server
+        ):
+            try:
+                if user.discord_id not in discord_ids_on_other_server:
+                    continue
+
+                user_on_other_server: User = users_on_other_server.where(
+                    lambda x: x.discord_id == user.discord_id
+                ).first_or_default()
+                if user_on_other_server is None or user_on_other_server.xp <= user.xp:
+                    continue
+
+                user.xp = user_on_other_server.xp
+                self._users.update_user(user)
+                self._db.save_changes()
+            except Exception as e:
+                self._logger.error(__name__, f"Cannot sync user {user.name}", e)
+
+        await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.technician.synced_message"))
+        self._logger.trace(__name__, f"Finished sync xp command")
+
+    @sync_xp.command(name="by_member")
+    @commands.guild_only()
+    @CommandChecks.check_is_ready()
+    @CommandChecks.check_is_member_technician()
+    async def by_member(self, ctx: Context, server_id: int, member: discord.Member):
+        self._logger.debug(__name__, f"Received command sync xp {ctx}")
+
+        if ctx.guild is None:
+            return
+
+        settings: ServerConfig = self._config.get_configuration(f"ServerConfig_{ctx.guild.id}")
+        if not FeatureFlagsSettings.get_flag_from_dict(settings.feature_flags, FeatureFlagsEnum.sync_xp):
+            await self._message_service.send_ctx_msg(ctx, self._t.transform("common.feature_not_activated"))
+            return
+
+        other_server = self._servers.get_server_by_id(server_id)
+        user = self._users.get_user_by_discord_id_and_server_id(
+            self._servers.get_server_by_discord_id(ctx.guild.id).id, member.id
+        )
+
+        try:
+            user_on_other_server = (
+                self._users.get_users_by_server_id(other_server.id)
+                .where(lambda x: x.discord_id == member.id)
+                .first_or_default()
+            )
+            if user_on_other_server is None or user_on_other_server.xp <= user.xp:
+                return
+
+            user.xp = user_on_other_server.xp
+            self._users.update_user(user)
+            self._db.save_changes()
+        except Exception as e:
+            self._logger.error(__name__, f"Cannot sync user {user.name}", e)
+
+        await self._message_service.send_ctx_msg(ctx, self._t.transform("modules.technician.synced_message"))
+        self._logger.trace(__name__, f"Finished sync xp command")
+
+    @sync_xp.autocomplete("server_id")
+    async def list_autocomplete(self, interaction: discord.Interaction, current: str) -> list[app_commands.Choice]:
+        return [
+            app_commands.Choice(name=server.name, value=server.id)
+            for server in self._client_utils.get_auto_complete_list(
+                self._servers.get_servers(), current, lambda x: x.name
+            )
+        ]

kdb-bot/src/modules/technician/technician_module.py

@@ -11,6 +11,7 @@ from modules.technician.command.api_key_group import ApiKeyGroup
 from modules.technician.command.log_command import LogCommand
 from modules.technician.command.restart_command import RestartCommand
 from modules.technician.command.shutdown_command import ShutdownCommand
+from modules.technician.command.sync_xp_command import SyncXpGroup
 
 
 class TechnicianModule(ModuleABC):
@@ -27,4 +28,5 @@ class TechnicianModule(ModuleABC):
         self._dc.add_command(ShutdownCommand)
         self._dc.add_command(LogCommand)
         self._dc.add_command(ApiKeyGroup)
+        self._dc.add_command(SyncXpGroup)
         # events