SeaCogs/moderation/importers/moderation.py

107 lines
3.7 KiB
Python
Raw Normal View History

2023-12-18 16:45:34 -05:00
from datetime import timedelta, datetime
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'] = {}
if case['duration'] != 'NULL':
2023-12-18 16:45:34 -05:00
duration_t = datetime.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)