274 lines
13 KiB
Python
274 lines
13 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
import logging
|
|
import time
|
|
import discord
|
|
import mysql.connector
|
|
from pytimeparse2 import disable_dateutil, parse
|
|
from redbot.core import Config, checks, commands
|
|
|
|
class Moderation(commands.Cog):
|
|
"""Custom cog moderation cog, meant to copy GalacticBot.
|
|
Developed by SeaswimmerTheFsh."""
|
|
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
self.config = Config.get_conf(self, identifier=481923957134912)
|
|
self.config.register_global(
|
|
mysql_address= " ",
|
|
mysql_database = " ",
|
|
mysql_username = " ",
|
|
mysql_password = " ",
|
|
ignore_other_bots = True
|
|
)
|
|
disable_dateutil()
|
|
|
|
async def cog_load(self):
|
|
"""This method prepares the database schema for all of the guilds the bot is currently in."""
|
|
conf = await self.check_conf([
|
|
'mysql_address',
|
|
'mysql_database',
|
|
'mysql_username',
|
|
'mysql_password'
|
|
])
|
|
if conf:
|
|
logging.fatal("Failed to create tables, due to MySQL connection configuration being unset.")
|
|
return
|
|
guilds: list[discord.Guild] = self.bot.guilds
|
|
try:
|
|
for guild in guilds:
|
|
await self.create_guild_table(guild)
|
|
except ConnectionRefusedError:
|
|
return
|
|
|
|
@commands.Cog.listener('on_guild_join')
|
|
async def db_generate_guild_join(self, guild: discord.Guild):
|
|
"""This method prepares the database schema whenever the bot joins a guild."""
|
|
conf = await self.check_conf([
|
|
'mysql_address',
|
|
'mysql_database',
|
|
'mysql_username',
|
|
'mysql_password'
|
|
])
|
|
if conf:
|
|
logging.fatal("Failed to create a table for %s, due to MySQL connection configuration being unset.", guild.id)
|
|
return
|
|
try:
|
|
await self.create_guild_table(guild)
|
|
except ConnectionRefusedError:
|
|
return
|
|
|
|
@commands.Cog.listener('on_audit_log_entry_create')
|
|
async def autologger(self, entry: discord.AuditLogEntry):
|
|
"""This method automatically logs moderations done by users manually ("right clicks")."""
|
|
if await self.config.ignore_other_bots() is True:
|
|
if entry.user.bot or entry.target.bot:
|
|
return
|
|
duration = "NULL"
|
|
if entry.reason:
|
|
reason = entry.reason + " (This action was performed without the bot.)"
|
|
else:
|
|
reason = "This action was performed without the bot."
|
|
if entry.action == discord.AuditLogAction.kick:
|
|
moderation_type = 'KICK'
|
|
elif entry.action == discord.AuditLogAction.ban:
|
|
moderation_type = 'BAN'
|
|
elif entry.action == discord.AuditLogAction.unban:
|
|
moderation_type = 'UNBAN'
|
|
elif entry.action == discord.AuditLogAction.member_update:
|
|
if entry.after.timed_out_until is not None:
|
|
timed_out_until_aware = entry.after.timed_out_until.replace(tzinfo=timezone.utc)
|
|
duration_datetime = timed_out_until_aware - datetime.now(tz=timezone.utc)
|
|
minutes = round(duration_datetime.total_seconds() / 60)
|
|
duration = timedelta(minutes=minutes)
|
|
moderation_type = 'MUTE'
|
|
else:
|
|
moderation_type = 'UNMUTE'
|
|
else:
|
|
return
|
|
await self.mysql_log(entry.guild.id, entry.user.id, moderation_type, entry.target.id, duration, reason)
|
|
|
|
async def connect(self):
|
|
"""Connects to the MySQL database, and returns a connection object."""
|
|
conf = await self.check_conf([
|
|
'mysql_address',
|
|
'mysql_database',
|
|
'mysql_username',
|
|
'mysql_password'
|
|
])
|
|
if conf:
|
|
raise LookupError("MySQL connection details not set properly!")
|
|
try:
|
|
connection = mysql.connector.connect(
|
|
host=await self.config.mysql_address(),
|
|
user=await self.config.mysql_username(),
|
|
password=await self.config.mysql_password(),
|
|
database=await self.config.mysql_database()
|
|
)
|
|
return connection
|
|
except mysql.connector.ProgrammingError as e:
|
|
logging.fatal("Unable to access the MySQL database!\nError:\n%s", e.msg)
|
|
raise ConnectionRefusedError(f"Unable to access the MySQL Database!\n{e.msg}") from e
|
|
|
|
async def create_guild_table(self, guild: discord.Guild):
|
|
database = await self.connect()
|
|
cursor = database.cursor()
|
|
try:
|
|
cursor.execute(f"SELECT * FROM `{guild.id}_moderation`")
|
|
logging.info("MySQL Table exists for server %s (%s)", guild.name, guild.id)
|
|
except mysql.connector.errors.ProgrammingError:
|
|
query = f"""
|
|
CREATE TABLE `{guild.id}_moderation` (
|
|
moderation_id INT UNIQUE PRIMARY KEY NOT NULL,
|
|
timestamp INT NOT NULL,
|
|
moderation_type LONGTEXT NOT NULL,
|
|
target_id LONGTEXT NOT NULL,
|
|
moderator_id LONGTEXT NOT NULL,
|
|
duration LONGTEXT,
|
|
end_timestamp INT,
|
|
reason LONGTEXT,
|
|
resolved BOOL NOT NULL,
|
|
resolve_reason LONGTEXT,
|
|
expired BOOL NOT NULL
|
|
)
|
|
"""
|
|
cursor.execute(query)
|
|
insert_query = f"""
|
|
INSERT INTO `{guild.id}_moderation`
|
|
(moderation_id, timestamp, moderation_type, target_id, moderator_id, duration, end_timestamp, reason, resolved, resolve_reason, expired)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
"""
|
|
insert_values = (0, 0, "NULL", 0, 0, "NULL", 0, "NULL", 0, "NULL", 0)
|
|
cursor.execute(insert_query, insert_values)
|
|
database.commit()
|
|
database.close()
|
|
logging.info("MySQL Table created for %s (%s)\n%s_moderation", guild.name, guild.id, guild.id)
|
|
else:
|
|
database.close()
|
|
return
|
|
|
|
async def check_conf(self, config: list):
|
|
"""Checks if any required config options are not set."""
|
|
not_found_list = []
|
|
for item in config:
|
|
if await self.config.item() == " ":
|
|
not_found_list.append(item)
|
|
return not_found_list
|
|
|
|
async def mysql_log(self, guild_id: str, author_id: str, moderation_type: str, target_id: int, duration, reason: str):
|
|
timestamp = int(time.time())
|
|
if duration != "NULL":
|
|
end_timedelta = datetime.fromtimestamp(timestamp) + duration
|
|
end_timestamp = int(end_timedelta.timestamp())
|
|
else:
|
|
end_timestamp = 0
|
|
database = await self.connect()
|
|
cursor = database.cursor()
|
|
cursor.execute(f"SELECT moderation_id FROM `{guild_id}_moderation` ORDER BY moderation_id DESC LIMIT 1")
|
|
moderation_id = cursor.fetchone()[0] + 1
|
|
sql = f"INSERT INTO `{guild_id}_moderation` (moderation_id, timestamp, moderation_type, target_id, moderator_id, duration, end_timestamp, reason, resolved, resolve_reason, expired) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
|
|
val = (moderation_id, timestamp, moderation_type, target_id, author_id, duration, end_timestamp, f"{reason}", 0, "NULL", 0)
|
|
cursor.execute(sql, val)
|
|
database.commit()
|
|
database.close()
|
|
logging.debug("MySQL row inserted into %s_moderation!\n%s, %s, %s, %s, %s, %s, %s, %s, 0, NULL", guild_id, moderation_id, timestamp, moderation_type, target_id, author_id, duration, end_timestamp, reason)
|
|
|
|
@commands.hybrid_command(name="mute", aliases=['timeout'])
|
|
@commands.mod()
|
|
async def mute(self, ctx: commands.Context, target: discord.Member, duration: str, reason: str):
|
|
"""Mute a user."""
|
|
try:
|
|
parsed_time = parse(sval=duration, as_timedelta=True, raise_exception=True)
|
|
except ValueError:
|
|
await ctx.message.reply(f"Please provide a valid duration!\nSee `{ctx.prefix}tdc`")
|
|
return
|
|
await target.timeout(parsed_time)
|
|
response = await ctx.send(content=f"{target.mention} has been muted for {str(parsed_time)}!\n**Reason** - `{reason}`")
|
|
try:
|
|
embed = discord.Embed(title="Muted", description=f"You have been muted for `{str(parsed_time)}` in [{ctx.guild.name}]({response.jump_url}).", color=await self.bot.get_embed_color(None))
|
|
embed.add_field(name='Reason', value=f"`{reason}`")
|
|
await target.send(embed=embed)
|
|
except discord.errors.HTTPException:
|
|
await response.edit(content=f"{response.content}\n*Failed to send DM, user likely has the bot blocked.*")
|
|
await self.mysql_log(ctx.guild.id, ctx.author.id, 'MUTE', target.id, parsed_time, reason)
|
|
|
|
@commands.group(autohelp=True)
|
|
@checks.admin()
|
|
async def moderationset(self, ctx: commands.Context):
|
|
"""Manage moderation commands."""
|
|
|
|
@moderationset.command(name="ignorebots")
|
|
@checks.admin()
|
|
async def moderationset_ignorebots(self, ctx: commands.Context):
|
|
await self.config.ignore_other_bots.set(not await self.config.ignore_other_bots())
|
|
await ctx.send(f"Ignore bots setting set to {await self.config.ignore_other_bots()}")
|
|
|
|
@moderationset.command(name="mysql")
|
|
@checks.is_owner()
|
|
async def moderationset_mysql(self, ctx: commands.Context):
|
|
"""Configure MySQL connection details."""
|
|
await ctx.message.add_reaction("✅")
|
|
await ctx.author.send(content="Click the button below to configure your MySQL connection details.", view=self.ConfigButtons(60))
|
|
|
|
class ConfigButtons(discord.ui.View):
|
|
def __init__(self, timeout):
|
|
super().__init__()
|
|
self.config = Config.get_conf(None, cog_name='Moderation', identifier=481923957134912)
|
|
|
|
@discord.ui.button(label="Edit", style=discord.ButtonStyle.success)
|
|
async def config_button(self, interaction: discord.Interaction, button: discord.ui.Button): # pylint: disable=unused-argument
|
|
await interaction.response.send_modal(Moderation.MySQLConfigModal(self.config))
|
|
|
|
class MySQLConfigModal(discord.ui.Modal, title="MySQL Database Configuration"):
|
|
def __init__(self, config):
|
|
super().__init__()
|
|
self.config = config
|
|
address = discord.ui.TextInput(
|
|
label="Address",
|
|
placeholder="Input your MySQL address here.",
|
|
style=discord.TextStyle.short,
|
|
required=False,
|
|
max_length=300
|
|
)
|
|
database = discord.ui.TextInput(
|
|
label="Database",
|
|
placeholder="Input the name of your database here.",
|
|
style=discord.TextStyle.short,
|
|
required=False,
|
|
max_length=300
|
|
)
|
|
username = discord.ui.TextInput(
|
|
label="Username",
|
|
placeholder="Input your MySQL username here.",
|
|
style=discord.TextStyle.short,
|
|
required=False,
|
|
max_length=300
|
|
)
|
|
password = discord.ui.TextInput(
|
|
label="Password",
|
|
placeholder="Input your MySQL password here.",
|
|
style=discord.TextStyle.short,
|
|
required=False,
|
|
max_length=300
|
|
)
|
|
|
|
async def on_submit(self, interaction: discord.Interaction):
|
|
message = ""
|
|
if self.address.value != "":
|
|
await self.config.mysql_address.set(self.address.value)
|
|
message += f"- Address set to\n - `{self.address.value}`\n"
|
|
if self.database.value != "":
|
|
await self.config.mysql_database.set(self.database.value)
|
|
message += f"- Database set to\n - `{self.database.value}`\n"
|
|
if self.username.value != "":
|
|
await self.config.mysql_username.set(self.username.value)
|
|
message += f"- Username set to\n - `{self.username.value}`\n"
|
|
if self.password.value != "":
|
|
await self.config.mysql_password.set(self.password.value)
|
|
trimmed_password = self.password.value[:8]
|
|
message += f"- Password set to\n - `{trimmed_password}` - Trimmed for security\n"
|
|
if message == "":
|
|
trimmed_password = str(await self.config.mysql_password())[:8]
|
|
send = f"No changes were made.\nCurrent configuration:\n- Address:\n - `{await self.config.mysql_address()}`\n- Database:\n - `{await self.config.mysql_database()}`\n- Username:\n - `{await self.config.mysql_username()}`\n- Password:\n - `{trimmed_password}` - Trimmed for security"
|
|
else:
|
|
send = f"Configuration changed:\n{message}"
|
|
await interaction.response.send_message(send, ephemeral=True)
|