SeaCogs/moderation/importers/moderation.py
SeaswimmerTheFsh faebd2629c
Some checks failed
Pylint / Pylint (3.11) (push) Failing after 44s
fix(moderation): pylint fixes
2023-12-18 19:00:16 -05:00

111 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 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:
metadata = {}
else:
metadata: Dict[str, any] = case["metadata"]
metadata.update({
'imported_from': 'SeaCogs/Moderation'
})
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)