SeaCogs/aurora/importers/galacticbot.py

161 lines
5.4 KiB
Python
Raw Normal View History

2024-02-28 10:58:57 -05:00
# pylint: disable=duplicate-code
import json
2023-12-18 17:24:40 -05:00
from datetime import timedelta
from time import time
2023-12-18 17:24:40 -05:00
from discord import ButtonStyle, Interaction, Message, ui
from redbot.core import commands
from redbot.core.utils.chat_formatting import box, warning
2023-12-18 17:24:40 -05:00
2024-05-06 21:39:43 -04:00
from ..models.moderation import Change, Moderation
2024-06-08 20:12:22 -04:00
from ..utilities.database import create_guild_table
2023-12-18 17:24:40 -05:00
class ImportGalacticBotView(ui.View):
def __init__(self, timeout, ctx, message):
super().__init__()
self.ctx: commands.Context = ctx
2024-06-08 20:12:22 -04:00
self.timeout = timeout
self.message: Message = message
@ui.button(label="Yes", style=ButtonStyle.success)
2024-02-03 13:41:57 -05:00
async def import_button_y(
self, interaction: Interaction, button: ui.Button
2024-02-28 10:56:13 -05:00
): # pylint: disable=unused-argument
await self.message.delete()
2024-02-03 13:41:57 -05:00
await interaction.response.send_message(
"Deleting original table...", ephemeral=True
)
2024-06-08 20:12:22 -04:00
database = await Moderation.connect()
2024-06-05 00:14:43 -04:00
cursor = await database.cursor()
query = f"DROP TABLE IF EXISTS moderation_{self.ctx.guild.id};"
2024-06-05 00:14:43 -04:00
await cursor.execute(query)
2024-06-05 00:14:43 -04:00
await cursor.close()
await 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 = [
2024-02-03 13:41:57 -05:00
"NOTE",
"WARN",
"MUTE",
"UNMUTE",
"KICK",
"SOFTBAN",
"BAN",
"UNBAN",
"SLOWMODE",
"LOCKDOWN",
]
file = await self.ctx.message.attachments[0].read()
2024-02-03 13:41:57 -05:00
data = sorted(json.loads(file), key=lambda x: x["case"])
failed_cases = []
for case in data:
2024-02-03 13:41:57 -05:00
if case["type"] not in accepted_types:
continue
2024-02-03 13:41:57 -05:00
timestamp = round(case["timestamp"] / 1000)
try:
2024-02-03 13:41:57 -05:00
if case["duration"] is not None and float(case["duration"]) != 0:
duration = timedelta(seconds=round(float(case["duration"]) / 1000))
else:
duration = None
except OverflowError:
2024-02-03 13:41:57 -05:00
failed_cases.append(case["case"])
continue
metadata = {"imported_from": "GalacticBot", "imported_timestamp": int(time())}
2024-02-03 13:41:57 -05:00
if case["type"] == "SLOWMODE":
metadata["seconds"] = case["data"]["seconds"]
2024-02-03 13:41:57 -05:00
if case["resolved"]:
resolved = 1
resolved_by = None
resolved_reason = None
resolved_timestamp = None
2024-02-03 13:41:57 -05:00
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:
2024-02-03 13:41:57 -05:00
resolved_by = "?"
if resolved_reason is None:
2024-02-03 13:41:57 -05:00
resolved_reason = (
"Could not get resolve reason during moderation import."
)
if resolved_timestamp is None:
resolved_timestamp = timestamp
changes = [
2024-05-06 16:47:21 -04:00
Change.from_dict(interaction.client, {
2024-02-03 13:41:57 -05:00
"type": "ORIGINAL",
"reason": case["reason"],
"user_id": case["executor"],
"timestamp": timestamp,
}),
2024-05-06 16:47:21 -04:00
Change.from_dict(interaction.client, {
2024-02-03 13:41:57 -05:00
"type": "RESOLVE",
"reason": resolved_reason,
"user_id": resolved_by,
"timestamp": resolved_timestamp,
}),
]
else:
resolved = None
resolved_by = None
resolved_reason = None
changes = None
2024-02-03 13:41:57 -05:00
if case["reason"] and case["reason"] != "N/A":
reason = case["reason"]
else:
reason = None
2024-06-05 00:14:43 -04:00
await Moderation.log(
self.ctx.guild.id,
2024-02-03 13:41:57 -05:00
case["executor"],
case["type"],
case["targetType"],
case["target"],
None,
duration,
reason,
timestamp=timestamp,
resolved=resolved,
resolved_by=resolved_by,
resolved_reason=resolved_reason,
changes=changes,
metadata=metadata,
2024-02-03 13:41:57 -05:00
database=database,
)
await interaction.edit_original_response(content="Import complete.")
if failed_cases:
2024-02-03 13:41:57 -05:00
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)
2024-02-03 13:41:57 -05:00
async def import_button_n(
self, interaction: Interaction, button: ui.Button
2024-02-28 10:56:13 -05:00
): # pylint: disable=unused-argument
await self.message.edit(content="Import cancelled.", view=None)
await self.message.delete(10)
await self.ctx.message.delete(10)