cswimr/SeaCogs
cswimr/SeaCogs
1
1
Fork 1
Code Issues 10 Pull requests 4 Projects 1 Packages 1 Wiki Activity Actions

feat(moderation): allow importing lockdown and slowmode moderation types
Some checks failed
Pylint / Pylint (3.10) (push) Failing after 49s
Details

Browse source
This commit is contained in:
Seaswimmer 2023-12-17 12:54:41 -05:00
parent 53290e7062
commit 141c0459d4
Signed by: cswimr
GPG key ID: 1EBC234EEDA901AE
4 changed files with 94 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
moderation/database.py
View file

@ -50,6 +50,7 @@ async def create_guild_table(guild: Guild):
moderation_id INT UNIQUE PRIMARY KEY NOT NULL,
timestamp INT NOT NULL,
moderation_type LONGTEXT NOT NULL,
target_type LONGTEXT NOT NULL,
target_id LONGTEXT NOT NULL,
moderator_id LONGTEXT NOT NULL,
role_id LONGTEXT,
@ -79,13 +80,14 @@ async def create_guild_table(guild: Guild):
insert_query = f"""
INSERT INTO `moderation_{guild.id}`
(moderation_id, timestamp, moderation_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
(moderation_id, timestamp, moderation_type, target_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
insert_values = (
0,
0,
"NULL",
"NULL",
0,
0,
0,
@ -117,6 +119,7 @@ async def mysql_log(
guild_id: str,
author_id: str,
moderation_type: str,
target_type: str,
target_id: int,
role_id: int,
duration,
@ -160,11 +163,12 @@ async def mysql_log(
moderation_id = await get_next_case_number(guild_id=guild_id, cursor=cursor)
sql = f"INSERT INTO `moderation_{guild_id}` (moderation_id, timestamp, moderation_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
sql = f"INSERT INTO `moderation_{guild_id}` (moderation_id, timestamp, moderation_type, target_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
val = (
moderation_id,
timestamp,
moderation_type,
target_type,
target_id,
author_id,
role_id,
@ -186,11 +190,12 @@ async def mysql_log(
database.close()
logger.debug(
"MySQL row inserted into moderation_%s!\n%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s",
"MySQL row inserted into moderation_%s!\n%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s",
guild_id,
moderation_id,
timestamp,
moderation_type,
target_type,
target_id,
author_id,
role_id,

41
moderation/embed_factory.py
View file

@ -3,7 +3,7 @@
from datetime import datetime, timedelta
import humanize
from discord import Color, Embed, Guild, Interaction, InteractionMessage
from .utils import get_next_case_number, fetch_user_dict
from .utils import get_next_case_number, fetch_user_dict, fetch_channel_dict
async def embed_factory(embed_type: str, color: Color, /, interaction: Interaction = None, case_dict: dict = None, guild: Guild = None, reason: str = None, moderation_type: str = None, response: InteractionMessage = None, duration: timedelta = None, resolved: bool = False):
"""This method creates an embed from set parameters, meant for either moderation logging or contacting the moderated user.
@ -54,10 +54,17 @@ async def embed_factory(embed_type: str, color: Color, /, interaction: Interact
return embed
if embed_type == 'case':
target_user = await fetch_user_dict(interaction, case_dict['target_id'])
moderator_user = await fetch_user_dict(interaction, case_dict['moderator_id'])
if case_dict['target_type'] == 'USER':
target_user = await fetch_user_dict(interaction, case_dict['target_id'])
target_name = f"`{target_user['name']}`" if target_user['discriminator'] == "0" else f"`{target_user['name']}#{target_user['discriminator']}`"
elif case_dict['target_type'] == 'CHANNEL':
target_user = await fetch_channel_dict(interaction, case_dict['target_id'])
if target_user['mention']:
target_name = f"{target_user['mention']} ({target_user['id']})"
else:
target_name = f"`{target_user['name']}` ({target_user['id']})"
target_name = f"`{target_user['name']}`" if target_user['discriminator'] == "0" else f"`{target_user['name']}#{target_user['discriminator']}`"
moderator_user = await fetch_user_dict(interaction, case_dict['moderator_id'])
moderator_name = f"`{moderator_user['name']}`" if moderator_user['discriminator'] == "0" else f"`{moderator_user['name']}#{moderator_user['discriminator']}`"
embed = Embed(title=f"📕 Case #{case_dict['moderation_id']:,}", color=color)
@ -114,10 +121,17 @@ async def embed_factory(embed_type: str, color: Color, /, interaction: Interact
if embed_type == 'log':
if resolved:
target_user = await fetch_user_dict(interaction, case_dict['target_id'])
moderator_user = await fetch_user_dict(interaction, case_dict['moderator_id'])
if case_dict['target_type'] == 'USER':
target_user = await fetch_user_dict(interaction, case_dict['target_id'])
target_name = f"`{target_user['name']}`" if target_user['discriminator'] == "0" else f"`{target_user['name']}#{target_user['discriminator']}`"
elif case_dict['target_type'] == 'CHANNEL':
target_user = await fetch_channel_dict(interaction, case_dict['target_id'])
if target_user['mention']:
target_name = f"{target_user['mention']} ({target_user['id']})"
else:
target_name = f"`{target_user['name']}` ({target_user['id']})"
target_name = f"`{target_user['name']}`" if target_user['discriminator'] == "0" else f"`{target_user['name']}#{target_user['discriminator']}`"
moderator_user = await fetch_user_dict(interaction, case_dict['moderator_id'])
moderator_name = moderator_user['name'] if moderator_user['discriminator'] == "0" else f"{moderator_user['name']}#{moderator_user['discriminator']}"
embed = Embed(title=f"📕 Case #{case_dict['moderation_id']:,} Resolved", color=color)
@ -135,10 +149,17 @@ async def embed_factory(embed_type: str, color: Color, /, interaction: Interact
resolved_name = resolved_user['name'] if resolved_user['discriminator'] == "0" else f"{resolved_user['name']}#{resolved_user['discriminator']}"
embed.add_field(name='Resolve Reason', value=f"Resolved by {resolved_name} ({resolved_user['id']}) for:\n```{case_dict['resolve_reason']}```", inline=False)
else:
target_user = await fetch_user_dict(interaction, case_dict['target_id'])
moderator_user = await fetch_user_dict(interaction, case_dict['moderator_id'])
if case_dict['target_type'] == 'USER':
target_user = await fetch_user_dict(interaction, case_dict['target_id'])
target_name = f"`{target_user['name']}`" if target_user['discriminator'] == "0" else f"`{target_user['name']}#{target_user['discriminator']}`"
elif case_dict['target_type'] == 'CHANNEL':
target_user = await fetch_channel_dict(interaction, case_dict['target_id'])
if target_user['mention']:
target_name = target_user['mention']
else:
target_name = target_user['name']
target_name = f"`{target_user['name']}`" if target_user['discriminator'] == "0" else f"`{target_user['name']}#{target_user['discriminator']}`"
moderator_user = await fetch_user_dict(interaction, case_dict['moderator_id'])
moderator_name = moderator_user['name'] if moderator_user['discriminator'] == "0" else f"{moderator_user['name']}#{moderator_user['discriminator']}"
embed = Embed(title=f"📕 Case #{case_dict['moderation_id']:,}", color=color)

34
moderation/moderation.py
View file

@ -18,7 +18,7 @@ from redbot.core import app_commands, checks, Config, commands, data_manager
from redbot.core.app_commands import Choice
from .database import connect, create_guild_table, fetch_case, mysql_log
from .embed_factory import embed_factory
from .utils import check_conf, check_permissions, check_moddable, fetch_user_dict, generate_dict, log
from .utils import check_conf, check_permissions, check_moddable, fetch_channel_dict, fetch_user_dict, generate_dict, log
from .logger import logger
class Moderation(commands.Cog):
@ -640,18 +640,24 @@ class Moderation(commands.Cog):
for case in result_dict_list[start_index:end_index]:
if case['target_id'] not in memory_dict:
memory_dict[str(case['target_id'])] = await fetch_user_dict(interaction, case['target_id'])
if case['target_type'] == 'USER':
memory_dict[str(case['target_id'])] = await fetch_user_dict(interaction, case['target_id'])
elif case['target_type'] == 'CHANNEL':
memory_dict[str(case['target_id'])] = await fetch_channel_dict(interaction, case['target_id'])
target_user = memory_dict[str(case['target_id'])]
if case['target_type'] == 'USER':
target_name = f"`{target_user['name']}`" if target_user['discriminator'] == "0" else f"`{target_user['name']}#{target_user['discriminator']}`"
elif case['target_type'] == 'CHANNEL':
target_name = f"`{target_user['mention']}`"
if case['moderator_id'] not in memory_dict:
memory_dict[str(case['moderator_id'])] = await fetch_user_dict(interaction, case['moderator_id'])
moderator_user = memory_dict[str(case['moderator_id'])]
target_name = target_user['name'] if target_user['discriminator'] == "0" else f"{target_user['name']}#{target_user['discriminator']}"
moderator_name = moderator_user['name'] if moderator_user['discriminator'] == "0" else f"{moderator_user['name']}#{moderator_user['discriminator']}"
moderator_name = f"`{moderator_user['name']}`" if moderator_user['discriminator'] == "0" else f"`{moderator_user['name']}#{moderator_user['discriminator']}`"
field_name = f"Case #{case['moderation_id']:,} ({str.title(case['moderation_type'])})"
field_value = f"**Target:** `{target_name}` ({target_user['id']})\n**Moderator:** `{moderator_name}` ({moderator_user['id']})"
field_value = f"**Target:** {target_name} ({target_user['id']})\n**Moderator:** {moderator_name} ({moderator_user['id']})"
if len(case['reason']) > 125:
field_value += f"\n**Reason:** `{str(case['reason'])[:125]}...`"
@ -1399,7 +1405,9 @@ class Moderation(commands.Cog):
'KICK',
'SOFTBAN',
'BAN',
'UNBAN'
'UNBAN',
'SLOWMODE',
'LOCKDOWN'
]
file = await self.ctx.message.attachments[0].read()
@ -1422,6 +1430,13 @@ class Moderation(commands.Cog):
failed_cases.append(case['case'])
continue
metadata = {
'imported_from': 'GalacticBot'
}
if case['type'] == 'SLOWMODE':
metadata['seconds'] = case['data']['seconds']
if case['resolved']:
resolved = 1
resolved_by = None
@ -1464,6 +1479,7 @@ class Moderation(commands.Cog):
self.ctx.guild.id,
case['executor'],
case['type'],
case['targetType'],
case['target'],
0,
duration,
@ -1473,9 +1489,7 @@ class Moderation(commands.Cog):
resolved_by=resolved_by,
resolved_reason=resolved_reason,
changes=changes,
metadata={
'imported_from': 'GalacticBot'
},
metadata=metadata,
database=database
)

46
moderation/utils.py
View file

@ -127,18 +127,19 @@ def generate_dict(result):
"moderation_id": result[0],
"timestamp": result[1],
"moderation_type": result[2],
"target_id": result[3],
"moderator_id": result[4],
"role_id": result[5],
"duration": result[6],
"end_timestamp": result[7],
"reason": result[8],
"resolved": result[9],
"resolved_by": result[10],
"resolve_reason": result[11],
"expired": result[12],
"changes": json.loads(result[13]),
"metadata": json.loads(result[14]),
"target_type": result[3],
"target_id": result[4],
"moderator_id": result[5],
"role_id": result[6],
"duration": result[7],
"end_timestamp": result[8],
"reason": result[9],
"resolved": result[10],
"resolved_by": result[11],
"resolve_reason": result[12],
"expired": result[13],
"changes": json.loads(result[14]),
"metadata": json.loads(result[15])
}
return case
@ -170,16 +171,29 @@ async def fetch_user_dict(interaction: Interaction, user_id: str):
return user_dict
async def fetch_role_dict(interaction: Interaction, role_id: str):
"""This method returns a dictionary containing either role information or a standard deleted role template."""
async def fetch_channel_dict(interaction: Interaction, channel_id: str):
"""This method returns a dictionary containing either channel information or a standard deleted channel template."""
try:
role = interaction.guild.get_role(role_id)
channel = interaction.guild.get_channel(channel_id)
if not channel:
channel = await interaction.guild.fetch_channel(channel_id)
role_dict = {"id": role.id, "name": role.name}
channel_dict = {"id": channel.id, "name": channel.name, "mention": channel.mention}
except NotFound:
channel_dict = {"id": channel_id, "name": "Deleted Channel", "mention": None}
return channel_dict
async def fetch_role_dict(interaction: Interaction, role_id: str):
"""This method returns a dictionary containing either role information or a standard deleted role template."""
role = interaction.guild.get_role(role_id)
if not role:
role_dict = {"id": role_id, "name": "Deleted Role"}
role_dict = {"id": role.id, "name": role.name}
return role_dict