SeaCogs/moderation/importers/moderation.py

from datetime import timedelta, strptime
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: [dict] = sorted(json.loads(file), key=lambda x: x['moderation_id'])

        user_mod_types = [
            'NOTE',
            'WARN',
            'MUTE',
            'UNMUTE',
            'KICK',
            'BAN',
            'UNBAN'
        ]

        channel_mod_types = [
            'SLOWMODE',
            'LOCKDOWN'
        ]

        failed_cases = []

        for case in data:
            if case['moderation_id'] == 0:
                continue

            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'] = {}

            duration_t = strptime(case['duration'], "%H:%M:%S")
            duration = timedelta(
                hours=duration_t.hours,
                minutes=duration_t.minutes,
                seconds=duration_t.seconds
            )

            await mysql_log(
                self.ctx.guild.id,
                case['moderator_id'],
                case['moderation_type'],
                case['target_type'],
                case['target_id'],
                case['role_id'],
                duration,
                case['reason'],
                timestamp=case['timestamp'],
                resolved=case['resolved'],
                resolved_by=case['resolved_by'],
                resolved_reason=case['resolve_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)