feat(aurora): moved get_next_case_number() into the Moderation class, along with removing from_sql and get_all_cases. also added some other classmethods to replace those.

also modified message_factory and its calls to add a new kwarg
This commit is contained in:
Seaswimmer 2024-06-04 00:04:11 -04:00
parent 460d5a31fc
commit 9b0f977016
Signed by untrusted user: cswimr
GPG key ID: 5D671B5D03D65A7F
4 changed files with 60 additions and 54 deletions

View file

@ -221,7 +221,8 @@ class Aurora(commands.Cog):
if silent is False:
try:
embed = await message_factory(
await self.bot.get_embed_color(interaction.channel),
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
@ -279,7 +280,8 @@ class Aurora(commands.Cog):
if silent is False:
try:
embed = await message_factory(
await self.bot.get_embed_color(interaction.channel),
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
@ -374,7 +376,8 @@ class Aurora(commands.Cog):
if silent is False:
try:
embed = await message_factory(
await self.bot.get_embed_color(interaction.channel),
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
@ -479,7 +482,8 @@ class Aurora(commands.Cog):
if silent is False:
try:
embed = await message_factory(
await self.bot.get_embed_color(interaction.channel),
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
@ -575,7 +579,8 @@ class Aurora(commands.Cog):
if silent is False:
try:
embed = await message_factory(
await self.bot.get_embed_color(interaction.channel),
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
@ -650,7 +655,8 @@ class Aurora(commands.Cog):
if silent is False:
try:
embed = await message_factory(
await self.bot.get_embed_color(interaction.channel),
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
@ -708,7 +714,8 @@ class Aurora(commands.Cog):
if silent is False:
try:
embed = await message_factory(
await self.bot.get_embed_color(interaction.channel),
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
@ -810,7 +817,8 @@ class Aurora(commands.Cog):
try:
embed = await message_factory(
await self.bot.get_embed_color(interaction.channel),
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
@ -853,8 +861,9 @@ class Aurora(commands.Cog):
silent = not await config.guild(interaction.guild).dm_users()
if silent is False:
try:
embed = embed = await message_factory(
await self.bot.get_embed_color(interaction.channel),
embed = await message_factory(
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
@ -936,7 +945,8 @@ class Aurora(commands.Cog):
if silent is False:
try:
embed = await message_factory(
await self.bot.get_embed_color(interaction.channel),
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
@ -1089,7 +1099,7 @@ class Aurora(commands.Cog):
+ f"moderation_{interaction.guild.id}.json"
)
cases = Moderation.get_all_cases(bot=interaction.client, guild_id=interaction.guild.id)
cases = Moderation.get_latest(bot=interaction.client, guild_id=interaction.guild.id)
with open(filename, "w", encoding="utf-8") as f:
dump(obj=cases, fp=f, indent=2)
@ -1118,7 +1128,7 @@ class Aurora(commands.Cog):
elif moderator:
moderations = Moderation.find_by_moderator(interaction.client, interaction.guild.id, moderator.id)
else:
moderations = Moderation.get_all_cases(interaction.client, interaction.guild.id)
moderations = Moderation.get_latest(interaction.client, interaction.guild.id)
case_quantity = len(moderations)
page_quantity = ceil(case_quantity / pagesize)
@ -1204,7 +1214,7 @@ class Aurora(commands.Cog):
return
try:
moderation = Moderation.from_sql(interaction.client, case, interaction.guild.id)
moderation = Moderation.find_by_id(interaction.client, case, interaction.guild.id)
except ValueError:
await interaction.response.send_message(
content=error(f"Case #{case:,} does not exist!"), ephemeral=True
@ -1288,7 +1298,7 @@ class Aurora(commands.Cog):
)
try:
mod = Moderation.from_sql(interaction.client, case, interaction.guild.id)
mod = Moderation.find_by_id(interaction.client, case, interaction.guild.id)
except ValueError:
await interaction.response.send_message(
content=error(f"Case #{case:,} does not exist!"), ephemeral=True
@ -1381,7 +1391,7 @@ class Aurora(commands.Cog):
return
try:
moderation = Moderation.from_sql(interaction.client, case, interaction.guild.id)
moderation = Moderation.find_by_id(interaction.client, case, interaction.guild.id)
old_moderation = moderation
except ValueError:
await interaction.response.send_message(
@ -1511,7 +1521,8 @@ class Aurora(commands.Cog):
)
embed = await message_factory(
await self.bot.get_embed_color(guild.channels[0]),
bot=self.bot,
color=await self.bot.get_embed_color(guild.channels[0]),
guild=guild,
reason=f"Automatic unban from case #{moderation_id}",
moderation_type="unbanned",

View file

@ -1,6 +1,7 @@
import json
import sqlite3
from datetime import datetime, timedelta
from sqlite3 import Cursor
from time import time
from typing import Dict, Iterable, List, Optional, Tuple, Union
@ -9,7 +10,6 @@ from discord import NotFound
from redbot.core.bot import Red
from ..utilities.logger import logger
from ..utilities.utils import get_next_case_number
from .base import AuroraGuildModel
from .change import Change
from .partials import PartialChannel, PartialRole, PartialUser
@ -229,42 +229,48 @@ class Moderation(AuroraGuildModel):
return ()
@classmethod
def from_sql(cls, bot: Red, moderation_id: int, guild_id: int) -> "Moderation":
return cls.find_by_id(bot=bot, moderation_id=moderation_id, guild_id=guild_id)
def get_latest(cls, bot: Red, guild_id: int, limit: int | None = None, types: Iterable | None = None, cursor: Cursor | None = None) -> Tuple["Moderation"]:
params = []
query = f"SELECT * FROM moderation_{guild_id} ORDER BY moderation_id DESC"
if limit:
query += " LIMIT ?"
params.append(limit)
if types:
query += f" WHERE moderation_type IN ({', '.join(['?' for _ in types])})"
params.extend(types)
query += ";"
return cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=tuple(params) if limit else (), cursor=cursor)
@classmethod
def find_by_id(cls, bot: Red, moderation_id: int, guild_id: int) -> "Moderation":
def get_next_case_number(cls, bot: Red, guild_id: int, cursor: Cursor | None = None) -> int:
result = cls.get_latest(bot=bot, guild_id=guild_id, cursor=cursor)
return (result[0].moderation_id + 1) if result else 1
@classmethod
def find_by_id(cls, bot: Red, moderation_id: int, guild_id: int, cursor: Cursor | None = None) -> "Moderation":
query = f"SELECT * FROM moderation_{guild_id} WHERE moderation_id = ?;"
case = cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=(moderation_id,))
case = cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=(moderation_id,), cursor=cursor)
if case:
return case[0]
raise ValueError(f"Case {moderation_id} not found in moderation_{guild_id}!")
@classmethod
def find_by_target(cls, bot: Red, guild_id: int, target: int, types: list | None = None) -> Tuple["Moderation"]:
def find_by_target(cls, bot: Red, guild_id: int, target: int, types: Iterable | None = None, cursor: Cursor | None = None) -> Tuple["Moderation"]:
query = f"SELECT * FROM moderation_{guild_id} WHERE target_id = ?"
if types:
query += f" AND moderation_type IN ({', '.join(['?' for _ in types])})"
query += " ORDER BY moderation_id DESC;"
return cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=(target, *types) if types else (target,))
return cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=(target, *types) if types else (target,), cursor=cursor)
@classmethod
def find_by_moderator(cls, bot: Red, guild_id: int, moderator: int, types: list | None = None) -> Tuple["Moderation"]:
def find_by_moderator(cls, bot: Red, guild_id: int, moderator: int, types: Iterable | None = None, cursor: Cursor | None = None) -> Tuple["Moderation"]:
query = f"SELECT * FROM moderation_{guild_id} WHERE moderator_id = ?"
if types:
query += f" AND moderation_type IN ({', '.join(['?' for _ in types])})"
query += " ORDER BY moderation_id DESC;"
return cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=(moderator, *types) if types else (moderator,))
@classmethod
def get_all_cases(cls, bot: Red, guild_id: int, types: list | None = None) -> Tuple["Moderation"]:
query = f"SELECT * FROM moderation_{guild_id}"
if types:
query += f" WHERE moderation_type IN ({', '.join(['?' for _ in types])})"
query += " ORDER BY moderation_id DESC;"
return cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=tuple(iterable=types) if types else None)
return cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=(moderator, *types) if types else (moderator,), cursor=cursor)
@classmethod
def log(
@ -328,7 +334,7 @@ class Moderation(AuroraGuildModel):
close_db = False
cursor = database.cursor()
moderation_id = get_next_case_number(guild_id=guild_id, cursor=cursor)
moderation_id = cls.get_next_case_number(bot=bot, guild_id=guild_id, cursor=cursor)
case = {
"moderation_id": moderation_id,
@ -378,4 +384,4 @@ class Moderation(AuroraGuildModel):
case["metadata"],
)
return cls.from_sql(bot=bot, moderation_id=moderation_id, guild_id=guild_id)
return cls.find_by_id(bot=bot, moderation_id=moderation_id, guild_id=guild_id)

View file

@ -4,15 +4,17 @@ from typing import Union
from discord import Color, Embed, Guild, Interaction, InteractionMessage, Member, Role, User
from redbot.core import commands
from redbot.core.bot import Red
from redbot.core.utils.chat_formatting import bold, box, error, humanize_timedelta, warning
from ..models.moderation import Moderation
from ..models.partials import PartialUser
from .config import config
from .utils import get_bool_emoji, get_next_case_number, get_pagesize_str
from .utils import get_bool_emoji, get_pagesize_str
async def message_factory(
bot: Red,
color: Color,
guild: Guild,
reason: str,
@ -25,6 +27,7 @@ async def message_factory(
"""This function creates a message from set parameters, meant for contacting the moderated user.
Args:
bot (Red): The bot instance.
color (Color): The color of the embed.
guild (Guild): The guild the moderation occurred in.
reason (str): The reason for the moderation.
@ -88,7 +91,7 @@ async def message_factory(
embed.set_author(name=guild.name)
embed.set_footer(
text=f"Case #{get_next_case_number(guild.id):,}",
text=f"Case #{Moderation.get_next_case_number(bot=bot, guild_id=guild.id):,}",
icon_url="attachment://arrow.png",
)

View file

@ -110,20 +110,6 @@ async def check_moddable(
return True
def get_next_case_number(guild_id: str, cursor=None) -> int:
"""This function returns the next case number from the MySQL table for a specific guild."""
from .database import connect
if not cursor:
database = connect()
cursor = database.cursor()
cursor.execute(
f"SELECT moderation_id FROM `moderation_{guild_id}` ORDER BY moderation_id DESC LIMIT 1"
)
result = cursor.fetchone()
return (result[0] + 1) if result else 1
async def log(interaction: Interaction, moderation_id: int, resolved: bool = False) -> None:
"""This function sends a message to the guild's configured logging channel when an infraction takes place."""
from ..models.moderation import Moderation
@ -134,7 +120,7 @@ async def log(interaction: Interaction, moderation_id: int, resolved: bool = Fal
logging_channel = interaction.guild.get_channel(logging_channel_id)
try:
moderation = Moderation.from_sql(interaction.client, moderation_id, interaction.guild_id)
moderation = Moderation.find_by_id(interaction.client, moderation_id, interaction.guild_id)
embed = await log_factory(
interaction=interaction, moderation=moderation, resolved=resolved
)
@ -159,7 +145,7 @@ async def send_evidenceformat(interaction: Interaction, moderation_id: int) -> N
if send_evidence_bool is False:
return
moderation = Moderation.from_sql(interaction.client, moderation_id, interaction.guild.id)
moderation = Moderation.find_by_id(interaction.client, moderation_id, interaction.guild.id)
content = await evidenceformat_factory(moderation=moderation)
await interaction.followup.send(content=content, ephemeral=True)