import json from datetime import timedelta from discord import ButtonStyle, Interaction, Message, ui from redbot.core import commands from redbot.core.utils.chat_formatting import box, warning from ..utilities.database import connect, create_guild_table, mysql_log class ImportGalacticBotView(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 = 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...") accepted_types = [ 'NOTE', 'WARN', 'MUTE', 'UNMUTE', 'KICK', 'SOFTBAN', 'BAN', 'UNBAN', 'SLOWMODE', 'LOCKDOWN' ] file = await self.ctx.message.attachments[0].read() data = sorted(json.loads(file), key=lambda x: x['case']) failed_cases = [] for case in data: if case['type'] not in accepted_types: continue timestamp = round(case['timestamp'] / 1000) try: if case['duration'] is not None and float(case['duration']) != 0: duration = timedelta(seconds=round(float(case['duration']) / 1000)) else: duration = 'NULL' except OverflowError: failed_cases.append(case['case']) continue metadata = { 'imported_from': 'GalacticBot' } if case['type'] == 'SLOWMODE': metadata['seconds'] = case['data']['seconds'] if case['resolved']: resolved = 1 resolved_by = None resolved_reason = None resolved_timestamp = None if case['changes']: for change in case['changes']: if change['type'] == 'RESOLVE': resolved_by = change['staff'] resolved_reason = change['reason'] resolved_timestamp = round(change['timestamp'] / 1000) break if resolved_by is None: resolved_by = '?' if resolved_reason is None: resolved_reason = 'Could not get resolve reason during moderation import.' if resolved_timestamp is None: resolved_timestamp = timestamp changes = [ { 'type': "ORIGINAL", 'reason': case['reason'], 'user_id': case['executor'], 'timestamp': timestamp }, { 'type': "RESOLVE", 'reason': resolved_reason, 'user_id': resolved_by, 'timestamp': resolved_timestamp } ] else: resolved = 0 resolved_by = 'NULL' resolved_reason = 'NULL' changes = [] if case['reason'] and case['reason'] != "N/A": reason = case['reason'] else: reason = "NULL" await mysql_log( self.ctx.guild.id, case['executor'], case['type'], case['targetType'], case['target'], 0, duration, reason, timestamp=timestamp, resolved=resolved, resolved_by=resolved_by, resolved_reason=resolved_reason, changes=changes, metadata=metadata, database=database ) await interaction.edit_original_response(content="Import complete.") if failed_cases: await interaction.edit_original_response(content="Import complete.\n" + warning("Failed to import the following cases:\n") + box(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(content="Import cancelled.", view=None) await self.message.delete(10) await self.ctx.message.delete(10)