WIP: Refactor Aurora (3.0.0) #29

Draft
cswimr wants to merge 342 commits from aurora-pydantic into main
10 changed files with 321 additions and 172 deletions
Showing only changes of commit 9f068bba6f - Show all commits

View file

@ -29,11 +29,14 @@ from .menus.immune import Immune
from .menus.overrides import Overrides
from .models.change import Change
from .models.moderation import Moderation
from .models.moderation_types import Ban, Tempban
from .utilities.config import config, register_config
from .utilities.factory import addrole_embed, case_factory, changes_factory, evidenceformat_factory, guild_embed, immune_embed, message_factory, overrides_embed
from .utilities.json import dump
from .utilities.logger import logger
from .utilities.utils import check_moddable, check_permissions, create_guild_table, get_footer_image, log, send_evidenceformat, timedelta_from_relativedelta
from .utilities.moderate import moderate
from .utilities.registry import type_registry
from .utilities.utils import check_moddable, check_permissions, create_guild_table, get_footer_image, log, send_evidenceformat
class Aurora(commands.Cog):
@ -42,7 +45,7 @@ class Aurora(commands.Cog):
This cog stores all of its data in an SQLite database."""
__author__ = ["SeaswimmerTheFsh"]
__version__ = "2.3.0"
__version__ = "2.4.0"
__documentation__ = "https://seacogs.coastalcommits.com/aurora/"
async def red_delete_data_for_user(self, *, requester, user_id: int):
@ -72,6 +75,7 @@ class Aurora(commands.Cog):
def __init__(self, bot: Red) -> None:
super().__init__()
self.bot = bot
self.type_registry = type_registry
register_config(config)
self.handle_expiry.start()
# If we don't override aiosqlite's logging level, it will spam the console with dozens of debug messages per query.
@ -775,124 +779,28 @@ class Aurora(commands.Cog):
How many days of messages to delete?
silent: bool
Should the user be messaged?"""
if not await check_moddable(target, interaction, ["ban_members"]):
return
if delete_messages is None:
delete_messages_seconds = 0
else:
delete_messages_seconds = delete_messages.value
try:
await interaction.guild.fetch_ban(target)
await interaction.response.send_message(
content=error(f"{target.mention} is already banned!"), ephemeral=True
)
return
except discord.errors.NotFound:
pass
if duration:
parsed_time = parse_relativedelta(duration)
if parsed_time is None:
await interaction.response.send_message(
content=error("Please provide a valid duration!"), ephemeral=True
)
return
try:
parsed_time = timedelta_from_relativedelta(parsed_time)
except ValueError:
await interaction.response.send_message(
content=error("Please provide a valid duration!"), ephemeral=True
)
return
await interaction.response.send_message(
content=f"{target.mention} has been banned for {humanize_timedelta(timedelta=parsed_time)}!\n**Reason** - `{reason}`"
)
try:
embed = await message_factory(
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
moderation_type="tempbanned",
response=await interaction.original_response(),
duration=parsed_time,
)
await target.send(embed=embed, file=get_footer_image(self))
except discord.errors.HTTPException:
pass
await interaction.guild.ban(
await moderate(
interaction,
target,
reason=f"Tempbanned by {interaction.user.id} for: {reason} (Duration: {parsed_time})",
delete_message_seconds=delete_messages_seconds,
silent,
["ban_members"],
Tempban,
reason=reason,
duration=duration,
delete_messages=delete_messages,
)
moderation = await Moderation.log(
interaction.client,
interaction.guild.id,
interaction.user.id,
"TEMPBAN",
"USER",
target.id,
None,
parsed_time,
reason,
)
await interaction.edit_original_response(
content=f"{target.mention} has been banned for {humanize_timedelta(timedelta=parsed_time)}! (Case `#{moderation.id}`)\n**Reason** - `{reason}`"
)
await log(interaction, moderation.id)
await send_evidenceformat(interaction, moderation.id)
else:
await interaction.response.send_message(
content=f"{target.mention} has been banned!\n**Reason** - `{reason}`"
)
if silent is None:
silent = not await config.guild(interaction.guild).dm_users()
if silent is False:
try:
embed = await message_factory(
bot=interaction.client,
color=await self.bot.get_embed_color(interaction.channel),
guild=interaction.guild,
moderator=interaction.user,
reason=reason,
moderation_type="banned",
response=await interaction.original_response(),
)
await target.send(embed=embed, file=get_footer_image(self))
except discord.errors.HTTPException:
pass
await interaction.guild.ban(
await moderate(
interaction,
target,
reason=f"Banned by {interaction.user.id} for: {reason}",
delete_message_seconds=delete_messages_seconds,
silent,
["ban_members"],
Ban,
reason=reason,
delete_messages=delete_messages,
)
moderation = await Moderation.log(
interaction.client,
interaction.guild.id,
interaction.user.id,
"BAN",
"USER",
target.id,
0,
"NULL",
reason,
)
await interaction.edit_original_response(
content=f"{target.mention} has been banned! (Case `#{moderation.id:,}`)\n**Reason** - `{reason}`"
)
await log(interaction, moderation.id)
await send_evidenceformat(interaction, moderation.id)
@app_commands.command(name="unban")
async def unban(
self,

View file

@ -9,7 +9,7 @@
"disabled": false,
"min_bot_version": "3.5.0",
"min_python_version": [3, 10, 0],
"requirements": ["pydantic", "aiosqlite"],
"requirements": ["pydantic", "aiosqlite", "class-registry"],
"tags": [
"mod",
"moderate",

View file

@ -0,0 +1,164 @@
from discord import File, Guild, Member, User
from discord.errors import HTTPException, NotFound
from redbot.core import app_commands, commands
from redbot.core.bot import Red
from redbot.core.commands.converter import parse_relativedelta
from redbot.core.utils.chat_formatting import bold, error, humanize_timedelta, inline
from ..utilities.factory import message_factory
from ..utilities.registry import type_registry
from ..utilities.utils import get_footer_image, log, send_evidenceformat, timedelta_from_relativedelta
from .moderation import Moderation
from .type import Type
def get_icon(bot: Red) -> File:
cog = bot.get_cog("Aurora")
if cog:
return get_footer_image(cog)
raise ValueError("Aurora cog not found. How was this managed?")
@type_registry.register(key="ban")
class Ban(Type):
def __init__(self) -> None:
self.type="ban"
self.verb="banned"
@classmethod
async def handler(cls, ctx: commands.Context, target: Member | User, silent: bool, reason: str = None, delete_messages: app_commands.Choice | None = None) -> 'Ban':
"""Ban a user."""
bot = ctx.bot
try:
await ctx.guild.fetch_ban(target)
await ctx.send(content=error(f"{target.mention} is already banned!"), ephemeral=True)
except NotFound:
pass
if delete_messages is None:
delete_messages_seconds = 0
else:
delete_messages_seconds = delete_messages.value
response_message = await ctx.send(f"{target.mention} has been {cls.verb}!\n{bold('Reason:')} {inline(reason)}")
if silent is True:
try:
embed = await message_factory(
bot,
await bot.get_embed_color(ctx.channel),
ctx.guild,
reason,
cls.type,
ctx.author,
None,
response_message
)
await target.send(embed=embed, file=get_icon(bot))
except HTTPException:
pass
await ctx.guild.ban(target, reason=f"Banned by {ctx.author.id} for: {reason}", delete_message_seconds=delete_messages_seconds)
moderation = await Moderation.log(
bot,
ctx.guild.id,
ctx.author.id,
cls.type,
'USER',
target.id,
None,
None,
reason
)
await response_message.edit(content=f"{target.mention} has been {cls.verb}! (Case {inline(f'#{moderation.id}')})\n{bold('Reason:')} {inline(reason)}")
await log(ctx, moderation.id)
await send_evidenceformat(ctx, moderation.id)
return cls
@classmethod
async def resolve_handler(cls, bot: Red, guild: Guild, target: Member, reason: str):
try:
await guild.fetch_ban(user=target)
except NotFound:
return
await guild.unban(user=target, reason=reason)
try:
embed = await message_factory(
bot,
await bot.get_embed_color(guild.channels[0]),
guild,
reason,
'unban',
None,
None,
None
)
await target.send(embed=embed, file=get_icon(bot))
except HTTPException:
pass
@type_registry.register(key="tempban")
class Tempban(Ban):
def __init__(self) -> None:
super().__init__()
self.type="tempban"
self.verb="tempbanned"
@classmethod
async def handler(cls, ctx: commands.Context, target: Member | User, silent: bool, duration: str, reason: str = None, delete_messages: app_commands.Choice | None = None) -> 'Ban':
"""Ban a user."""
bot = ctx.bot
try:
await ctx.guild.fetch_ban(target)
await ctx.send(content=error(f"{target.mention} is already banned!"), ephemeral=True)
except NotFound:
pass
if delete_messages is None:
delete_messages_seconds = 0
else:
delete_messages_seconds = delete_messages.value
parsed_time = parse_relativedelta(duration)
if not parsed_time:
await ctx.send(content=error("Please provide a valid duration!"), ephemeral=True)
try:
parsed_time = timedelta_from_relativedelta(parsed_time)
except ValueError:
await ctx.send(content=error("Please provide a valid duration!"), ephemeral=True)
response_message = await ctx.send(f"{target.mention} has been {cls.verb} for {humanize_timedelta(parsed_time)}!\n{bold('Reason:')} {inline(reason)}")
if silent is True:
try:
embed = await message_factory(
bot,
await bot.get_embed_color(ctx.channel),
ctx.guild,
reason,
cls.type,
ctx.author,
parsed_time,
response_message
)
await target.send(embed=embed, file=get_icon(bot))
except HTTPException:
pass
await ctx.guild.ban(target, reason=f"Tempbanned by {ctx.author.id} for: {reason} (Duration: {parsed_time})", delete_message_seconds=delete_messages_seconds)
moderation = await Moderation.log(
bot,
ctx.guild.id,
ctx.author.id,
cls.type,
'USER',
target.id,
None,
parsed_time,
reason
)
await response_message.edit(content=f"{target.mention} has been {cls.verb} for {humanize_timedelta(parsed_time)}! (Case {inline(f'#{moderation.id}')})\n{bold('Reason:')} {inline(reason)}")
await log(ctx, moderation.id)
await send_evidenceformat(ctx, moderation.id)
return cls

18
aurora/models/type.py Normal file
View file

@ -0,0 +1,18 @@
from discord import Member, User
from redbot.core import commands
class Type(object):
def __init__(self) -> None:
self.type = None
self.verb = None
self.embed_desc = "been"
def __str__(self) -> str:
return self.type
@classmethod
async def handler(cls, ctx: commands.Context, target: Member | User, silent: bool, **kwargs) -> 'Type': # pylint: disable=unused-argument
"""This method should be overridden by any child classes, but should retain the same starting keyword arguments."""
raise NotImplementedError

View file

@ -2,13 +2,14 @@
from datetime import datetime, timedelta
from typing import Union
from discord import Color, Embed, Guild, Interaction, InteractionMessage, Member, Role, User
from discord import Color, Embed, Guild, Interaction, Member, Message, 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 ..models.type import Type
from .config import config
from .utils import get_bool_emoji, get_pagesize_str
@ -18,10 +19,10 @@ async def message_factory(
color: Color,
guild: Guild,
reason: str,
moderation_type: str,
moderation_type: Type,
moderator: Union[Member, User] | None = None,
duration: timedelta | None = None,
response: InteractionMessage | None = None,
response: Message | None = None,
role: Role | None = None,
) -> Embed:
"""This function creates a message from set parameters, meant for contacting the moderated user.
@ -31,49 +32,47 @@ async def message_factory(
color (Color): The color of the embed.
guild (Guild): The guild the moderation occurred in.
reason (str): The reason for the moderation.
moderation_type (str): The type of moderation.
moderation_type (Type): The type of moderation.
moderator (Union[Member, User], optional): The moderator who performed the moderation. Defaults to None.
duration (timedelta, optional): The duration of the moderation. Defaults to None.
response (InteractionMessage, optional): The response message. Defaults to None.
response (Message, optional): The response message. Defaults to None.
role (Role, optional): The role that was added or removed. Defaults to None.
Returns:
embed: The message embed.
"""
if response is not None and moderation_type not in [
"kicked",
"banned",
"tempbanned",
"unbanned",
if response is not None and moderation_type.type not in [
"kick",
"ban",
"tempban",
"unban",
]:
guild_name = f"[{guild.name}]({response.jump_url})"
else:
guild_name = guild.name
title = moderation_type
if moderation_type in ["tempbanned", "muted"] and duration:
if duration:
embed_duration = f" for {humanize_timedelta(timedelta=duration)}"
else:
embed_duration = ""
if moderation_type == "note":
embed_desc = "received a"
elif moderation_type == "addrole":
embed_desc = f"received the {role.name} role"
title = "Role Added"
moderation_type = ""
elif moderation_type == "removerole":
embed_desc = f"lost the {role.name} role"
title = "Role Removed"
moderation_type = ""
else:
embed_desc = "been"
# if moderation_type.type == "note":
# embed_desc = "received a"
# elif moderation_type.type == "addrole":
# embed_desc = f"received the {role.name} role"
# title = "Role Added"
# verb = ""
# elif moderation_type.type == "removerole":
# embed_desc = f"lost the {role.name} role"
# title = "Role Removed"
# verb = ""
# else:
# embed_desc = "been"
embed = Embed(
title=str.title(title),
description=f"You have {embed_desc} {moderation_type}{embed_duration} in {guild_name}.",
title=str.title(moderation_type.type),
description=f"You have {moderation_type.embed_desc} {moderation_type.verb}{embed_duration} in {guild_name}.",
color=color,
timestamp=datetime.now(),
)
@ -99,12 +98,12 @@ async def message_factory(
async def log_factory(
interaction: Interaction, moderation: Moderation, resolved: bool = False
ctx: commands.Context, moderation: Moderation, resolved: bool = False
) -> Embed:
"""This function creates a log embed from set parameters, meant for moderation logging.
Args:
interaction (discord.Interaction): The interaction object.
ctx (commands.Context): The ctx object.
moderation (aurora.models.Moderation): The moderation object.
resolved (bool, optional): Whether the case is resolved or not. Defaults to False.
"""
@ -113,7 +112,7 @@ async def log_factory(
if resolved:
embed = Embed(
title=f"📕 Case #{moderation.id:,} Resolved",
color=await interaction.client.get_embed_color(interaction.channel),
color=await ctx.bot.get_embed_color(ctx.channel),
)
resolved_by = await moderation.get_resolved_by()
@ -145,7 +144,7 @@ async def log_factory(
else:
embed = Embed(
title=f"📕 Case #{moderation.id:,}",
color=await interaction.client.get_embed_color(interaction.channel),
color=await ctx.bot.get_embed_color(ctx.channel),
)
embed.description = f"**Type:** {str.title(moderation.type)}\n**Target:** {target.name} ({target.id})\n**Moderator:** {moderator.name} ({moderator.id})\n**Timestamp:** <t:{moderation.unix_timestamp}> | <t:{moderation.unix_timestamp}:R>"

View file

@ -0,0 +1,41 @@
from typing import List, Union
import discord
from redbot.core import app_commands, commands
from ..models.moderation_types import Type
from .config import config
from .registry import type_registry
from .utils import check_moddable
async def moderate(ctx: Union[commands.Context, discord.Interaction], target: discord.Member, silent: bool | None, permissions: List[str], moderation_type: Type | str, **kwargs) -> None | Type:
"""This function is used to moderate users.
It checks if the target can be moderated, then calls the handler method of the moderation type specified.
Args:
bot (Red): The bot instance.
ctx (Union[commands.Context, discord.Interaction]): The context of the command. If this is a `discord.Interaction` object, it will be converted to a `commands.Context` object. Additionally, if the interaction orignated from a context menu, the `ctx.author` attribute will be overriden to `interaction.user`.
target (discord.Member): The target user to moderate.
silent (bool | None): Whether to send the moderation action to the target.
permissions (List[str]): The permissions required to moderate the target.
moderation_type (Type): The moderation type (handler) to use. See `aurora.models.moderation_types` for some examples.
**kwargs: The keyword arguments to pass to the handler method.
"""
if not await check_moddable(target, ctx, permissions):
return
if silent is None:
silent = not await config.guild(ctx.guild).dm_users()
if isinstance(moderation_type, str):
moderation_type = type_registry[str.lower(moderation_type)]
if isinstance(ctx, discord.Interaction):
interaction = ctx
ctx = await commands.Context.from_interaction(interaction)
if isinstance(interaction.command, app_commands.ContextMenu):
ctx.author = interaction.user
return await moderation_type.handler(
ctx,
target,
silent,
**kwargs
)

View file

@ -0,0 +1,3 @@
from class_registry import ClassRegistry
type_registry = ClassRegistry()

View file

@ -17,7 +17,7 @@ from ..utilities.logger import logger
def check_permissions(
user: User,
permissions: Tuple[str],
ctx: Union[commands.Context, Interaction] | None = None,
ctx: commands.Context | Interaction | None = None,
guild: Guild | None = None,
) -> Union[bool, str]:
"""Checks if a user has a specific permission (or a list of permissions) in a channel."""
@ -43,12 +43,12 @@ def check_permissions(
async def check_moddable(
target: Union[User, Member, TextChannel], interaction: Interaction, permissions: Tuple[str]
target: Union[User, Member, TextChannel], ctx: commands.Context, permissions: Tuple[str]
) -> bool:
"""Checks if a moderator can moderate a target."""
is_channel = isinstance(target, TextChannel)
if check_permissions(interaction.client.user, permissions, guild=interaction.guild):
await interaction.response.send_message(
if check_permissions(ctx.bot.user, permissions, guild=ctx.guild):
await ctx.send(
error(
f"I do not have the `{permissions}` permission, required for this action."
),
@ -56,9 +56,9 @@ async def check_moddable(
)
return False
if await config.guild(interaction.guild).use_discord_permissions() is True:
if check_permissions(interaction.user, permissions, guild=interaction.guild):
await interaction.response.send_message(
if await config.guild(ctx.guild).use_discord_permissions() is True:
if check_permissions(ctx.author, permissions, guild=ctx.guild):
await ctx.send(
error(
f"You do not have the `{permissions}` permission, required for this action."
),
@ -66,21 +66,21 @@ async def check_moddable(
)
return False
if interaction.user.id == target.id:
await interaction.response.send_message(
if ctx.author.id == target.id:
await ctx.send(
content="You cannot moderate yourself!", ephemeral=True
)
return False
if not is_channel and target.bot:
await interaction.response.send_message(
await ctx.send(
content="You cannot moderate bots!", ephemeral=True
)
return False
if isinstance(target, Member):
if interaction.user.top_role <= target.top_role and await config.guild(interaction.guild).respect_hierarchy() is True:
await interaction.response.send_message(
if ctx.author.top_role <= target.top_role and await config.guild(ctx.guild).respect_hierarchy() is True:
await ctx.send(
content=error(
"You cannot moderate members with a higher role than you!"
),
@ -89,10 +89,10 @@ async def check_moddable(
return False
if (
interaction.guild.get_member(interaction.client.user.id).top_role
ctx.guild.get_member(ctx.bot.user.id).top_role
<= target.top_role
):
await interaction.response.send_message(
await ctx.send(
content=error(
"You cannot moderate members with a role higher than the bot!"
),
@ -104,7 +104,7 @@ async def check_moddable(
for role in target.roles:
if role.id in immune_roles:
await interaction.response.send_message(
await ctx.send(
content=error("You cannot moderate members with an immune role!"),
ephemeral=True,
)
@ -113,19 +113,19 @@ async def check_moddable(
return True
async def log(interaction: Interaction, moderation_id: int, resolved: bool = False) -> None:
async def log(ctx: commands.Context, 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
from .factory import log_factory
logging_channel_id = await config.guild(interaction.guild).log_channel()
logging_channel_id = await config.guild(ctx.guild).log_channel()
if logging_channel_id != " ":
logging_channel = interaction.guild.get_channel(logging_channel_id)
logging_channel = ctx.guild.get_channel(logging_channel_id)
try:
moderation = await Moderation.find_by_id(interaction.client, moderation_id, interaction.guild_id)
moderation = await Moderation.find_by_id(ctx.bot, moderation_id, ctx.guild_id)
embed = await log_factory(
interaction=interaction, moderation=moderation, resolved=resolved
ctx=ctx, moderation=moderation, resolved=resolved
)
try:
await logging_channel.send(embed=embed)
@ -135,22 +135,22 @@ async def log(interaction: Interaction, moderation_id: int, resolved: bool = Fal
return
async def send_evidenceformat(interaction: Interaction, moderation_id: int) -> None:
async def send_evidenceformat(ctx: commands.Context, moderation_id: int) -> None:
"""This function sends an ephemeral message to the moderator who took the moderation action, with a pre-made codeblock for use in the mod-evidence channel."""
from ..models.moderation import Moderation
from .factory import evidenceformat_factory
send_evidence_bool = (
await config.user(interaction.user).auto_evidenceformat()
or await config.guild(interaction.guild).auto_evidenceformat()
await config.user(ctx.author).auto_evidenceformat()
or await config.guild(guild=ctx.guild).auto_evidenceformat()
or False
)
if send_evidence_bool is False:
return
moderation = await Moderation.find_by_id(interaction.client, moderation_id, interaction.guild.id)
moderation = await Moderation.find_by_id(ctx.bot, moderation_id, ctx.guild.id)
content = await evidenceformat_factory(moderation=moderation)
await interaction.followup.send(content=content, ephemeral=True)
await ctx.send(content=content, ephemeral=True)
def get_bool_emoji(value: Optional[bool]) -> str:

19
poetry.lock generated
View file

@ -1,4 +1,4 @@
# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand.
# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand.
[[package]]
name = "aiohttp"
@ -599,6 +599,20 @@ files = [
{file = "charset_normalizer-3.3.2-py3-none-any.whl", hash = "sha256:3e4d1f6587322d2788836a99c69062fbb091331ec940e02d12d179c1d53e25fc"},
]
[[package]]
name = "class-registry"
version = "2.1.2"
description = "Factory+Registry pattern for Python classes."
optional = false
python-versions = "*"
files = [
{file = "class-registry-2.1.2.tar.gz", hash = "sha256:678bdb0322566c07a4d8905140d364bd34a73baf46bf7580fc2e06fa994d4e7e"},
{file = "class_registry-2.1.2-py2.py3-none-any.whl", hash = "sha256:cfb855514753e2edfe8d88b14a6e449820682fe0983efe61b83df28b688b3e5a"},
]
[package.dependencies]
six = "*"
[[package]]
name = "click"
version = "8.1.7"
@ -1199,6 +1213,7 @@ optional = false
python-versions = ">=3.6"
files = [
{file = "mkdocs-redirects-1.2.1.tar.gz", hash = "sha256:9420066d70e2a6bb357adf86e67023dcdca1857f97f07c7fe450f8f1fb42f861"},
{file = "mkdocs_redirects-1.2.1-py3-none-any.whl", hash = "sha256:497089f9e0219e7389304cffefccdfa1cac5ff9509f2cb706f4c9b221726dffb"},
]
[package.dependencies]
@ -2673,4 +2688,4 @@ multidict = ">=4.0"
[metadata]
lock-version = "2.0"
python-versions = ">=3.11,<3.12"
content-hash = "22b824824f73dc3dc1a9a0a01060371ee1f6414e5bef39cb7455d21121988b47"
content-hash = "bf7dd1ef2ebf8aedeb3295201cf04b53e5cd04cca488fd1e7e0257cbe9597513"

View file

@ -19,6 +19,7 @@ colorthief = "^0.2.1"
beautifulsoup4 = "^4.12.3"
markdownify = "^0.12.1"
aiosqlite = "^0.20.0"
class-registry = "^2.1.2"
[tool.poetry.group.dev]
optional = true