feat(moderation): added importer for moving db data between bots

moderation/importers/moderation.py

@@ -0,0 +1,95 @@
+from datetime import timedelta
+import json
+from redbot.core import commands
+from discord import Message, ButtonStyle, Interaction, ui
+from ..database import connect, create_guild_table, mysql_log
+
+class ImportModerationView(ui.View):
+    def __init__(self, timeout, ctx, message):
+        super().__init__()
+        self.ctx: commands.Context = ctx
+        self.message: Message = message
+
+    @ui.button(label="Yes", style=ButtonStyle.success)
+    async def import_button_y(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
+        await self.message.delete()
+        await interaction.response.send_message("Deleting original table...", ephemeral=True)
+
+        database = await connect()
+        cursor = database.cursor()
+
+        query = f"DROP TABLE IF EXISTS moderation_{self.ctx.guild.id};"
+        cursor.execute(query)
+
+        cursor.close()
+        database.commit()
+
+        await interaction.edit_original_response(content="Creating new table...")
+
+        await create_guild_table(self.ctx.guild)
+
+        await interaction.edit_original_response(content="Importing moderations...")
+
+        file = await self.ctx.message.attachments[0].read()
+        data = sorted(json.loads(file), key=lambda x: x['case'])
+
+        user_mod_types = [
+            'NOTE',
+            'WARN',
+            'MUTE',
+            'UNMUTE',
+            'KICK',
+            'BAN',
+            'UNBAN'
+        ]
+
+        channel_mod_types = [
+            'SLOWMODE',
+            'LOCKDOWN'
+        ]
+
+        failed_cases = []
+
+        for case in data:
+            if 'target_type' not in case or not case['target_type']:
+                if case['moderation_type'] in user_mod_types:
+                    case['target_type'] = 'USER'
+                elif case['moderation_type'] in channel_mod_types:
+                    case['target_type'] = 'CHANNEL'
+
+            if 'role_id' not in case or not case['role_id']:
+                case['role_id'] = 0
+
+            if 'changes' not in case or not case['changes']:
+                case['changes'] = []
+
+            if 'metadata' not in case or not case['metadata']:
+                case['metadata'] = {}
+
+            await mysql_log(
+                self.ctx.guild.id,
+                case['moderator_id'],
+                case['moderation_type'],
+                case['target_type'],
+                case['target_id'],
+                case['role_id'],
+                case['duration'],
+                case['reason'],
+                timestamp=case['timestamp'],
+                resolved=case['resolved'],
+                resolved_by=case['resolved_by'],
+                resolved_reason=case['resolved_reason'],
+                changes=case['changes'],
+                metadata=case['metadata'],
+                database=database
+            )
+
+        await interaction.edit_original_response(content="Import complete.")
+        if failed_cases:
+            await interaction.edit_original_response(content=f"Import complete.\n*Failed to import the following cases:*\n```{failed_cases}```")
+
+    @ui.button(label="No", style=ButtonStyle.danger)
+    async def import_button_n(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
+        await self.message.edit("Import cancelled.", view=None)
+        await self.message.delete(10)
+        await self.ctx.message.delete(10)

moderation/moderation.py

@@ -25,6 +25,7 @@ from .embed_factory import embed_factory
 from .logger import logger
 from .utils import check_conf, check_permissions, check_moddable, fetch_channel_dict, fetch_user_dict, generate_dict, log
 from .importers.galacticbot import ImportGalacticBotView
+from .importers.moderation import ImportModerationView
 
 class Moderation(commands.Cog):
     """Custom moderation cog.
@@ -1339,6 +1340,16 @@ class Moderation(commands.Cog):
     async def moderationset_import(self, ctx: commands.Context):
         """Import moderations from other bots."""
 
+    moderationset_import.command(name="moderation")
+    @checks.admin()
+    async def moderationset_import_moderation(self, ctx: commands.Context):
+        """Import moderations from another bot using this cog."""
+        if ctx.message.attachments and ctx.message.attachments[0].content_type == 'application/json; charset=utf-8':
+            message = await ctx.send("Are you sure you want to import moderations from another bot?\n**This will overwrite any moderations that already exist in this guild's moderation table.**\n*The import process will block the rest of your bot until it is complete.*")
+            await message.edit(view=ImportModerationView(60, ctx, message))
+        else:
+            await ctx.send("Please provide a valid moderation export file.")
+
     @moderationset_import.command(name="galacticbot")
     @checks.admin()
     async def moderationset_import_galacticbot(self, ctx: commands.Context):