SeaCogs/aurora/importers/aurora.py
SeaswimmerTheFsh ca50deedd5
Some checks failed
Pylint / Pylint (3.11) (push) Failing after 1m12s
feat(aurora): renamed moderation cog to aurora, converted to sqlite3
2023-12-28 04:23:55 -05:00

112 lines
3.8 KiB
Python

import json
from datetime import timedelta
from typing import Dict
from discord import ButtonStyle, Interaction, Message, ui
from redbot.core import commands
from ..utilities.database import connect, create_guild_table, mysql_log
class ImportAuroraView(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...")
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:
metadata = {}
else:
metadata: Dict[str, any] = case["metadata"]
if not metadata['imported_from']:
metadata.update({
'imported_from': 'Aurora'
})
if case["duration"] != "NULL":
hours, minutes, seconds = map(int, case["duration"].split(":"))
duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
else:
duration = "NULL"
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"],
expired=case["expired"],
changes=case["changes"],
metadata=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)