SeaCogs/moderation/importers/moderation.py

104 lines
3.6 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":
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=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)