Compare commits

..

3 commits

Author SHA1 Message Date
9f86c20df4
feat(moderation): updated prisma_log (mysql_log) and get_next_case_number to use prisma, added get_next_global_case_number
Some checks failed
Pylint / Pylint (3.10) (push) Failing after 1m0s
BREAKING: changed database schema
2023-10-27 10:40:18 -04:00
9ec42f5d60
feat(moderation): adding prisma schema
Some checks failed
Pylint / Pylint (3.10) (push) Failing after 58s
2023-10-26 21:05:01 -04:00
b8c3efdedc
feat(moderation): added required config values for prisma connections
Some checks failed
Pylint / Pylint (3.10) (push) Failing after 58s
2023-10-26 20:45:05 -04:00
29 changed files with 2553 additions and 1250 deletions

View file

@ -1,20 +1,30 @@
name: Actions name: Pylint
on: on: [push]
push:
branches:
- 'main'
pull_request:
jobs: jobs:
Lint Code (Pylint): Pylint:
runs-on: docker runs-on: docker
container: www.coastalcommits.com/cswimr/actions:galaxycogs strategy:
matrix:
python-version: ["3.10"]
container: catthehacker/ubuntu:act-latest
steps: steps:
- name: Checkout - name: Checkout
uses: actions/checkout@v3 uses: actions/checkout@v3.6.0
with:
token: ${{ secrets.COASTALCOMMITSTOKEN}}
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install Poetry
run: curl -sSL https://cdn.seaswimmer.cc/go/poetry | python${{ matrix.python-version }} -
- name: Install dependencies - name: Install dependencies
run: poetry install --with dev --no-root run: |
export PATH="$HOME/.local/bin:$PATH"
- name: Analysing code with Pylint poetry env use ${{ matrix.python-version }}
run: pylint --rcfile .forgejo/workflows/config/.pylintrc $(git ls-files '*.py') poetry install --with dev
- name: Analysing the code with Pylint
run: |
export PATH="$HOME/.local/bin:$PATH"
poetry run pylint $(git ls-files '*.py')

1
.gitignore vendored
View file

@ -1,2 +1,3 @@
galaxy/slashtag arguments.txt galaxy/slashtag arguments.txt
.venv .venv
*.db

View file

@ -1,6 +1,5 @@
[MESSAGES CONTROL] [MESSAGES CONTROL]
disable= disable=
too-many-lines,
missing-module-docstring, missing-module-docstring,
missing-function-docstring, missing-function-docstring,
missing-class-docstring, missing-class-docstring,

View file

@ -1,13 +1,11 @@
import asyncio import asyncio
import os import os
import discord import discord
from redbot.core import Config, checks, commands, data_manager from redbot.core import Config, checks, commands, data_manager
class ExportChannels(commands.Cog): class ExportChannels(commands.Cog):
"""Custom cog to export channels to JSON and HTML formats using Discord Chat Exporter. """Custom cog to export channels to JSON and HTML formats using Discord Chat Exporter.
Developed by cswimr and yname.""" Developed by SeaswimmerTheFsh and yname."""
def __init__(self, bot): def __init__(self, bot):
self.bot = bot self.bot = bot

View file

@ -5,7 +5,7 @@ from redbot.core import Config, commands
class Forums(commands.Cog): class Forums(commands.Cog):
"""Custom cog intended for use on the Galaxy discord server. """Custom cog intended for use on the Galaxy discord server.
Developed by cswimr.""" Developed by SeaswimmerTheFsh."""
def __init__(self, bot): def __init__(self, bot):
self.bot = bot self.bot = bot

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr"], "author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing Forums!\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs", "install_msg" : "Thank you for installing Forums!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs",
"name" : "Forums", "name" : "Forums",
"short" : "Custom cog intended for use on the Galaxy discord server.", "short" : "Custom cog intended for use on the Galaxy discord server.",
"description" : "Custom cog intended for use on the Galaxy discord server.", "description" : "Custom cog intended for use on the Galaxy discord server.",

View file

@ -1,15 +1,15 @@
import re
from datetime import datetime
from random import randint from random import randint
import re
import subprocess
from datetime import datetime
import discord import discord
from redbot.core import Config, app_commands, commands from redbot.core import Config, app_commands, checks, commands
from redbot.core.app_commands import Choice from redbot.core.app_commands import Choice
class Galaxy(commands.Cog): class Galaxy(commands.Cog):
"""Custom cog intended for use on the Galaxy discord server. """Custom cog intended for use on the Galaxy discord server.
Developed by cswimr.""" Developed by SeaswimmerTheFsh."""
def __init__(self, bot): def __init__(self, bot):
self.bot = bot self.bot = bot
@ -19,6 +19,18 @@ class Galaxy(commands.Cog):
autoreact_emoji = '💀' autoreact_emoji = '💀'
) )
@commands.command()
@checks.is_owner()
async def nslookup(self, ctx: commands.Context, *, website: str):
"""This command uses `nslookup` to check the IP Address of any given website."""
try:
result = subprocess.run(['nslookup', website], capture_output=True, text=True, check=True)
await ctx.send(f"```\n{result.stdout}\n```")
except subprocess.CalledProcessError as e:
await ctx.send(f"Error executing `nslookup`: `{e}`")
except FileNotFoundError:
await ctx.send("`nslookup` command not found. Make sure you have `nslookup` installed and it's in your system PATH.")
@commands.command() @commands.command()
async def carnagerefund(self, ctx: commands.Context, message_id: str): async def carnagerefund(self, ctx: commands.Context, message_id: str):
"""This command generates a link to refund carnage of killed ships.""" """This command generates a link to refund carnage of killed ships."""

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr"], "author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing Galaxy!\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs", "install_msg" : "Thank you for installing Galaxy!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs",
"name" : "Galaxy", "name" : "Galaxy",
"short" : "Custom cog intended for use on the Galaxy discord server.", "short" : "Custom cog intended for use on the Galaxy discord server.",
"description" : "Custom cog intended for use on the Galaxy discord server.", "description" : "Custom cog intended for use on the Galaxy discord server.",

View file

@ -1,8 +1,8 @@
{ {
"author": [ "author": [
"cswimr, yname, meelyman" "SeaswimmerTheFsh, yname, meelyman"
], ],
"install_msg": "Thanks for installing my repo!\n\nIf you have any issues with any of the cogs, please create an issue here: https://coastalcommits.com/cswimr/GalaxyCogs/issues.", "install_msg": "Thanks for installing my repo!\n\nIf you have any issues with any of the cogs, please create an issue here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs/issues.",
"name": "Galaxy", "name": "Galaxy",
"short": "Cogs intended for use on the Galaxy discord server.", "short": "Cogs intended for use on the Galaxy discord server.",
"description": "Custom cogs/cog modifications intended for the Galaxy discord server." "description": "Custom cogs/cog modifications intended for the Galaxy discord server."

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr"], "author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing Info!\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs", "install_msg" : "Thank you for installing Info!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs",
"name" : "Info", "name" : "Info",
"short" : "Provides information on Discord objects.", "short" : "Provides information on Discord objects.",
"description" : "Provides information on Discord objects. Most of this code is shamelessly ripped from <https://github.com/Cog-Creators/Red-DiscordBot/tree/V3/develop/redbot/cogs>.", "description" : "Provides information on Discord objects. Most of this code is shamelessly ripped from <https://github.com/Cog-Creators/Red-DiscordBot/tree/V3/develop/redbot/cogs>.",

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr"], "author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing Issues!\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs", "install_msg" : "Thank you for installing Issues!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs",
"name" : "Issues", "name" : "Issues",
"short" : "This cog allows you to create Gitea issues through a Discord modal.", "short" : "This cog allows you to create Gitea issues through a Discord modal.",
"description" : "This cog allows you to create Gitea issues through a Discord modal.", "description" : "This cog allows you to create Gitea issues through a Discord modal.",

View file

@ -1,12 +1,10 @@
import discord import discord
from redbot.core import Config, app_commands, checks, commands from redbot.core import Config, app_commands, commands, checks
from . import modals from . import modals
class Issues(commands.Cog): class Issues(commands.Cog):
"""This cog allows you to create Gitea issues through a Discord modal. """This cog allows you to create Gitea issues through a Discord modal.
Developed by cswimr.""" Developed by SeaswimmerTheFsh."""
def __init__(self, bot): def __init__(self, bot):
self.bot = bot self.bot = bot
@ -33,8 +31,8 @@ class Issues(commands.Cog):
"""Found a bug or have a suggestion for the Galaxy bot? Use this command.""" """Found a bug or have a suggestion for the Galaxy bot? Use this command."""
color = await self.bot.get_embed_color(None) color = await self.bot.get_embed_color(None)
embed = discord.Embed(title="Issue Reporting & Suggestions", color=await self.bot.get_embed_color(None), description="Have a problem or a suggestion for the Galaxy bot or GalaxyCogs? Read this!") embed = discord.Embed(title="Issue Reporting & Suggestions", color=await self.bot.get_embed_color(None), description="Have a problem or a suggestion for the Galaxy bot or GalaxyCogs? Read this!")
embed.add_field(name="Bot Issues & Suggestions", value="If you'd like to submit a suggestion or a bug report to the developers of the Galaxy bot, please do so with the buttons below or by going [here](https://coastalcommits.com/cswimr/GalaxyCogs/issues/new/choose).\n**Please make sure whatever you're suggesting or reporting doesn't have an existing issue! If it does, you can comment on that issue with additional details if necessary.**") embed.add_field(name="Bot Issues & Suggestions", value="If you'd like to submit a suggestion or a bug report to the developers of the Galaxy bot, please do so with the buttons below or by going [here](https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs/issues/new/choose).\n**Please make sure whatever you're suggesting or reporting doesn't have an existing issue! If it does, you can comment on that issue with additional details if necessary.**")
embed.add_field(name="Cog Issues & Suggestions", value="If you'd like to submit a suggestion or a bug report to the developers of GalaxyCogs, please do so with the buttons below or by going [here](https://coastalcommits.com/cswimr/GalaxyCogs/issues/new/choose).\n**Please make sure whatever you're suggesting or reporting doesn't have an existing issue! If it does, you can comment on that issue with additional details if necessary.**") embed.add_field(name="Cog Issues & Suggestions", value="If you'd like to submit a suggestion or a bug report to the developers of GalaxyCogs, please do so with the buttons below or by going [here](https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs/issues/new/choose).\n**Please make sure whatever you're suggesting or reporting doesn't have an existing issue! If it does, you can comment on that issue with additional details if necessary.**")
await interaction.response.send_message(embed=embed, view=self.IssueButtons(color, self, interaction), ephemeral=True) await interaction.response.send_message(embed=embed, view=self.IssueButtons(color, self, interaction), ephemeral=True)
async def submit_issue_request(self, interaction: discord.Interaction, original_interaction: discord.Interaction, embed: discord.Embed): async def submit_issue_request(self, interaction: discord.Interaction, original_interaction: discord.Interaction, embed: discord.Embed):

5
moderation/__init__.py Normal file
View file

@ -0,0 +1,5 @@
from .moderation import Moderation
async def setup(bot):
await bot.add_cog(Moderation(bot))

11
moderation/info.json Normal file
View file

@ -0,0 +1,11 @@
{
"author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing Moderation!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs",
"name" : "Moderation",
"short" : "Custom cog intended for use on the Galaxy discord server.",
"description" : "Custom cog intended for use on the Galaxy discord server.",
"end_user_data_statement" : "This cog does not store any End User Data.",
"requirements": ["mysql-connector-python", "humanize", "pytimeparse2"],
"hidden": false,
"disabled": false
}

983
moderation/moderation.py Normal file
View file

@ -0,0 +1,983 @@
import logging
import time
from datetime import datetime, timedelta, timezone
from typing import Union
import discord
import humanize
from prisma import Prisma
from discord.ext import tasks
from pytimeparse2 import disable_dateutil, parse
from redbot.core import app_commands, checks, Config, commands
from redbot.core.app_commands import Choice
class Moderation(commands.Cog):
"""Custom moderation cog.
Developed by SeaswimmerTheFsh."""
def __init__(self, bot):
self.bot = bot
self.config = Config.get_conf(self, identifier=481923957134912)
self.config.register_global(
database_provider = "sqlite",
database_address= "file:moderation.db",
database_port = " ",
database_name = " ",
database_username = " ",
database_password = " "
)
self.config.register_guild(
ignore_other_bots = True,
dm_users = True,
log_channel = " "
)
disable_dateutil()
self.handle_expiry.start() # pylint: disable=no-member
self.logger = logging.getLogger('red.seaswimmerthefsh.moderation')
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:
self.logger.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:
if not await self.bot.cog_disabled_in_guild(self, guild):
await self.create_guild_table(guild)
except ConnectionRefusedError:
return
async def cog_unload(self):
self.handle_expiry.cancel() # pylint: disable=no-member
@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."""
if not await self.bot.cog_disabled_in_guild(self, guild):
conf = await self.check_conf([
'mysql_address',
'mysql_database',
'mysql_username',
'mysql_password'
])
if conf:
self.logger.error("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 not await self.bot.cog_disabled_in_guild(self, entry.guild):
if await self.config.guild(entry.guild.id).ignore_other_bots() is True:
if entry.user.bot or entry.target.bot:
return
else:
if entry.user.id == self.bot.user.id:
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 database, and returns a connection object."""
provider = await self.config.database_provider()
address = await self.config.database_address()
if provider == "sqlite" and address == "file:moderation.db":
return await Prisma().connect()
else:
return await Prisma(
provider=provider,
address=address,
port=await self.config.database_port(),
database=await self.config.database_name(),
username=await self.config.database_username(),
password=await self.config.database_password()
).connect()
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
def check_permissions(self, user: discord.User, permissions: list, ctx: Union[commands.Context, discord.Interaction] = None, guild: discord.Guild = None):
"""Checks if a user has a specific permission (or a list of permissions) in a channel."""
if ctx:
member = ctx.guild.get_member(user.id)
resolved_permissions = ctx.channel.permissions_for(member)
elif guild:
member = guild.get_member(user.id)
resolved_permissions = member.guild_permissions
else:
raise(KeyError)
for permission in permissions:
if not getattr(resolved_permissions, permission, False) and not resolved_permissions.administrator is True:
return permission
return False
async def prisma_log(self, guild_id: str, author_id: str, moderation_type: str, target_id: int, duration, reason: str, role_id = None):
timestamp = int(time.time())
if duration != "NULL":
end_timedelta = datetime.fromtimestamp(timestamp) + duration
end_timestamp = int(end_timedelta.timestamp())
else:
end_timestamp = 0
if not role_id:
role_id = "NULL"
db = await self.connect()
global_id = await self.get_next_global_case_number(database=db)
moderation_id = await self.get_next_case_number(guild_id=guild_id, database=db)
await db.Case.create(
data={
'globalId': global_id,
'guildId': guild_id,
'moderationId': moderation_id,
'timestamp': timestamp,
'moderationType': moderation_type,
'targetId': target_id,
'moderatorId': author_id,
'roleId': role_id,
'duration': duration,
'endTimestamp': end_timestamp,
'reason': reason,
'resolved': False,
'resolvedBy': None,
'resolveReason': None,
'expired': False
}
)
await db.disconnect()
return moderation_id
async def get_next_case_number(self, guild_id: str, database = None):
"""This method returns the next case number from the database table for a specific guild."""
if not database:
database = await self.connect()
db_not_provided = True
else:
db_not_provided = False
result = await database.Case.find_first(
select={"moderationId": True},
where={"guildId": guild_id},
order=[{"moderationId": "desc"}],
)
if db_not_provided:
await database.disconnect()
return result.moderationId + 1 if result else 1
async def get_next_global_case_number(self, database = None):
"""This method returns the next case number from the database table."""
if not database:
database = await self.connect()
db_not_provided = True
else:
db_not_provided = False
result = await database.Case.find_first(
select={"globalId": True},
order=[{"globalId": "desc"}],
)
if db_not_provided:
await database.disconnect()
return result.globalId + 1 if result else 1
def generate_dict(self, result):
case: dict = {
"moderation_id": result[0],
"timestamp": result[1],
"moderation_type": result[2],
"target_id": result[3],
"moderator_id": result[4],
"duration": result[5],
"end_timestamp": result[6],
"reason": result[7],
"resolved": result[8],
"resolved_by": result[9],
"resolve_reason": result[10],
"expired": result[11]
}
return case
async def fetch_user_dict(self, interaction: discord.Interaction, user_id: str):
"""This method returns a dictionary containing either user information or a standard deleted user template."""
try:
user = await interaction.client.fetch_user(user_id)
user_dict = {
'id': user.id,
'name': user.name,
'discriminator': user.discriminator
}
except discord.errors.NotFound:
user_dict = {
'id': user_id,
'name': 'Deleted User',
'discriminator': '0'
}
return user_dict
async def embed_factory(self, embed_type: str, /, interaction: discord.Interaction = None, case_dict: dict = None, guild: discord.Guild = None, reason: str = None, moderation_type: str = None, response: discord.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.
Valid arguments for 'embed_type':
- 'message'
- 'log' - WIP
- 'case'
Required arguments for 'message':
- guild
- reason
- moderation_type
- response
- duration (optional)
Required arguments for 'log':
- interaction
- case_dict
- resolved (optional)
Required arguments for 'case':
- interaction
- case_dict"""
if embed_type == 'message':
if moderation_type in ["kicked", "banned", "tempbanned", "unbanned"]:
guild_name = guild.name
else:
guild_name = f"[{guild.name}]({response.jump_url})"
if moderation_type in ["tempbanned", "muted"] and duration:
embed_duration = f" for {humanize.precisedelta(duration)}"
else:
embed_duration = ""
if moderation_type == "note":
embed_desc = "recieved a"
else:
embed_desc = "been"
embed = discord.Embed(title=str.title(moderation_type), description=f"You have {embed_desc} {moderation_type}{embed_duration} in {guild_name}.", color=await self.bot.get_embed_color(None), timestamp=datetime.now())
embed.add_field(name='Reason', value=f"`{reason}`")
embed.set_author(name=guild.name, icon_url=guild.icon.url)
embed.set_footer(text=f"Case #{await self.get_next_case_number(guild.id)}", icon_url="https://cdn.discordapp.com/attachments/1070822161389994054/1159469476773904414/arrow-right-circle-icon-512x512-2p1e2aaw.png?ex=65312319&is=651eae19&hm=3cebdd28e805c13a79ec48ef87c32ca532ffa6b9ede2e48d0cf8e5e81f3a6818&")
return embed
if embed_type == 'case':
target_user = await self.fetch_user_dict(interaction, case_dict['target_id'])
moderator_user = await self.fetch_user_dict(interaction, case_dict['moderator_id'])
target_name = f"`{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']}"
embed = discord.Embed(title=f"📕 Case #{case_dict['moderation_id']}", color=await self.bot.get_embed_color(None))
embed.description = f"**Type:** {str.title(case_dict['moderation_type'])}\n**Target:** {target_name} ({target_user['id']})\n**Moderator:** {moderator_name} ({moderator_user['id']})\n**Resolved:** {bool(case_dict['resolved'])}\n**Timestamp:** <t:{case_dict['timestamp']}> | <t:{case_dict['timestamp']}:R>"
if case_dict['duration'] != 'NULL':
td = timedelta(**{unit: int(val) for unit, val in zip(["hours", "minutes", "seconds"], case_dict["duration"].split(":"))})
duration_embed = f"{humanize.precisedelta(td)} | <t:{case_dict['end_timestamp']}:R>" if case_dict["expired"] == '0' else str(humanize.precisedelta(td))
embed.description = embed.description + f"\n**Duration:** {duration_embed}\n**Expired:** {bool(case_dict['expired'])}"
embed.add_field(name='Reason', value=f"```{case_dict['reason']}```", inline=False)
if case_dict['resolved'] == 1:
resolved_user = await self.fetch_user_dict(interaction, case_dict['resolved_by'])
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)
return embed
if embed_type == 'log':
if resolved:
target_user = await self.fetch_user_dict(interaction, case_dict['target_id'])
moderator_user = await self.fetch_user_dict(interaction, case_dict['moderator_id'])
target_name = f"`{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']}"
embed = discord.Embed(title=f"📕 Case #{case_dict['moderation_id']} Resolved", color=await self.bot.get_embed_color(None))
embed.description = f"**Type:** {str.title(case_dict['moderation_type'])}\n**Target:** {target_name} ({target_user['id']})\n**Moderator:** {moderator_name} ({moderator_user['id']})\n**Timestamp:** <t:{case_dict['timestamp']}> | <t:{case_dict['timestamp']}:R>"
if case_dict['duration'] != 'NULL':
td = timedelta(**{unit: int(val) for unit, val in zip(["hours", "minutes", "seconds"], case_dict["duration"].split(":"))})
duration_embed = f"{humanize.precisedelta(td)} | <t:{case_dict['end_timestamp']}:R>" if case_dict["expired"] == '0' else str(humanize.precisedelta(td))
embed.description = embed.description + f"\n**Duration:** {duration_embed}\n**Expired:** {bool(case_dict['expired'])}"
embed.add_field(name='Reason', value=f"```{case_dict['reason']}```", inline=False)
resolved_user = await self.fetch_user_dict(interaction, case_dict['resolved_by'])
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 self.fetch_user_dict(interaction, case_dict['target_id'])
moderator_user = await self.fetch_user_dict(interaction, case_dict['moderator_id'])
target_name = f"`{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']}"
embed = discord.Embed(title=f"📕 Case #{case_dict['moderation_id']}", color=await self.bot.get_embed_color(None))
embed.description = f"**Type:** {str.title(case_dict['moderation_type'])}\n**Target:** {target_name} ({target_user['id']})\n**Moderator:** {moderator_name} ({moderator_user['id']})\n**Timestamp:** <t:{case_dict['timestamp']}> | <t:{case_dict['timestamp']}:R>"
if case_dict['duration'] != 'NULL':
td = timedelta(**{unit: int(val) for unit, val in zip(["hours", "minutes", "seconds"], case_dict["duration"].split(":"))})
embed.description = embed.description + f"\n**Duration:** {humanize.precisedelta(td)} | <t:{case_dict['end_timestamp']}:R>"
embed.add_field(name='Reason', value=f"```{case_dict['reason']}```", inline=False)
return embed
raise(TypeError("'type' argument is invalid!"))
async def fetch_case(self, moderation_id: int, guild_id: str):
"""This method fetches a case from the database and returns the case's dictionary."""
database = await self.connect()
cursor = database.cursor()
query = "SELECT * FROM moderation_%s WHERE moderation_id = %s;"
cursor.execute(query, (guild_id, moderation_id))
result = cursor.fetchone()
cursor.close()
database.close()
return self.generate_dict(result)
async def log(self, interaction: discord.Interaction, moderation_id: int, resolved: bool = False):
"""This method sends a message to the guild's configured logging channel when an infraction takes place."""
logging_channel_id = await self.config.guild(interaction.guild).log_channel()
if logging_channel_id != " ":
logging_channel = interaction.guild.get_channel(logging_channel_id)
case = await self.fetch_case(moderation_id, interaction.guild.id)
if case:
embed = await self.embed_factory('log', interaction=interaction, case_dict=case, resolved=resolved)
try:
await logging_channel.send(embed=embed)
except discord.errors.Forbidden:
return
@app_commands.command(name="note")
async def note(self, interaction: discord.Interaction, target: discord.User, reason: str, silent: bool = None):
"""Add a note to a user.
Parameters
-----------
target: discord.User
Who are you noting?
reason: str
Why are you noting this user?
silent: bool
Should the user be messaged?"""
if interaction.guild.get_member(target.id):
target_member = interaction.guild.get_member(target.id)
if interaction.guild.get_member(interaction.client.user.id).top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a role higher than the bot!", ephemeral=True)
return
if interaction.user.top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a higher role than you!", ephemeral=True)
return
await interaction.response.send_message(content=f"{target.mention} has recieved a note!\n**Reason** - `{reason}`")
if silent is None:
silent = not await self.config.guild(interaction.guild).dm_users()
if silent is False:
try:
embed = await self.embed_factory('message', guild=interaction.guild, reason=reason, moderation_type='note', response=await interaction.original_response())
await target.send(embed=embed)
except discord.errors.HTTPException:
pass
moderation_id = await self.mysql_log(interaction.guild.id, interaction.user.id, 'NOTE', target.id, 'NULL', reason)
await self.log(interaction, moderation_id)
@app_commands.command(name="warn")
async def warn(self, interaction: discord.Interaction, target: discord.Member, reason: str, silent: bool = None):
"""Warn a user.
Parameters
-----------
target: discord.Member
Who are you warning?
reason: str
Why are you warning this user?
silent: bool
Should the user be messaged?"""
if interaction.guild.get_member(target.id):
target_member = interaction.guild.get_member(target.id)
if interaction.guild.get_member(interaction.client.user.id).top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a role higher than the bot!", ephemeral=True)
return
if interaction.user.top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a higher role than you!", ephemeral=True)
return
await interaction.response.send_message(content=f"{target.mention} has been warned!\n**Reason** - `{reason}`")
if silent is None:
silent = not await self.config.guild(interaction.guild).dm_users()
if silent is False:
try:
embed = await self.embed_factory('message', guild=interaction.guild, reason=reason, moderation_type='warned', response=await interaction.original_response())
await target.send(embed=embed)
except discord.errors.HTTPException:
pass
moderation_id = await self.mysql_log(interaction.guild.id, interaction.user.id, 'WARN', target.id, 'NULL', reason)
await self.log(interaction, moderation_id)
@app_commands.command(name="mute")
async def mute(self, interaction: discord.Interaction, target: discord.Member, duration: str, reason: str, silent: bool = None):
"""Mute a user.
Parameters
-----------
target: discord.Member
Who are you unbanning?
duration: str
How long are you muting this user for?
reason: str
Why are you unbanning this user?
silent: bool
Should the user be messaged?"""
if interaction.guild.get_member(target.id):
target_member = interaction.guild.get_member(target.id)
if interaction.guild.get_member(interaction.client.user.id).top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a role higher than the bot!", ephemeral=True)
return
if interaction.user.top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a higher role than you!", ephemeral=True)
return
permissions = self.check_permissions(interaction.client.user, ['moderate_members'], interaction)
if permissions:
await interaction.response.send_message(f"I do not have the `{permissions}` permission, required for this action.", ephemeral=True)
return
if target.is_timed_out() is True:
await interaction.response.send_message(f"{target.mention} is already muted!", allowed_mentions=discord.AllowedMentions(users=False), ephemeral=True)
return
try:
parsed_time = parse(sval=duration, as_timedelta=True, raise_exception=True)
except ValueError:
await interaction.response.send_message("Please provide a valid duration!", ephemeral=True)
return
if parsed_time.total_seconds() / 1000 > 2419200000:
await interaction.response.send_message("Please provide a duration that is less than 28 days.")
return
await target.timeout(parsed_time, reason=f"Muted by {interaction.user.id} for: {reason}")
await interaction.response.send_message(content=f"{target.mention} has been muted for {humanize.precisedelta(parsed_time)}!\n**Reason** - `{reason}`")
if silent is None:
silent = not await self.config.guild(interaction.guild).dm_users()
if silent is False:
try:
embed = await self.embed_factory('message', guild=interaction.guild, reason=reason, moderation_type='muted', response=await interaction.original_response(), duration=parsed_time)
await target.send(embed=embed)
except discord.errors.HTTPException:
pass
moderation_id = await self.mysql_log(interaction.guild.id, interaction.user.id, 'MUTE', target.id, parsed_time, reason)
await self.log(interaction, moderation_id)
@app_commands.command(name="unmute")
async def unmute(self, interaction: discord.Interaction, target: discord.Member, reason: str = None, silent: bool = None):
"""Unmute a user.
Parameters
-----------
target: discord.user
Who are you unmuting?
reason: str
Why are you unmuting this user?
silent: bool
Should the user be messaged?"""
if interaction.guild.get_member(target.id):
target_member = interaction.guild.get_member(target.id)
if interaction.guild.get_member(interaction.client.user.id).top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a role higher than the bot!", ephemeral=True)
return
if interaction.user.top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a higher role than you!", ephemeral=True)
return
permissions = self.check_permissions(interaction.client.user, ['moderate_members'], interaction)
if permissions:
await interaction.response.send_message(f"I do not have the `{permissions}` permission, required for this action.", ephemeral=True)
return
if target.is_timed_out() is False:
await interaction.response.send_message(f"{target.mention} is not muted!", allowed_mentions=discord.AllowedMentions(users=False), ephemeral=True)
return
if reason:
await target.timeout(None, reason=f"Unmuted by {interaction.user.id} for: {reason}")
else:
await target.timeout(None, reason=f"Unbanned by {interaction.user.id}")
reason = "No reason given."
await interaction.response.send_message(content=f"{target.mention} has been unmuted!\n**Reason** - `{reason}`")
if silent is None:
silent = not await self.config.guild(interaction.guild).dm_users()
if silent is False:
try:
embed = await self.embed_factory('message', guild=interaction.guild, reason=reason, moderation_type='unmuted', response=await interaction.original_response())
await target.send(embed=embed)
except discord.errors.HTTPException:
pass
moderation_id = await self.mysql_log(interaction.guild.id, interaction.user.id, 'UNMUTE', target.id, 'NULL', reason)
await self.log(interaction, moderation_id)
@app_commands.command(name="kick")
async def kick(self, interaction: discord.Interaction, target: discord.Member, reason: str, silent: bool = None):
"""Kick a user.
Parameters
-----------
target: discord.user
Who are you kicking?
reason: str
Why are you kicking this user?
silent: bool
Should the user be messaged?"""
if interaction.guild.get_member(target.id):
target_member = interaction.guild.get_member(target.id)
if interaction.guild.get_member(interaction.client.user.id).top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a role higher than the bot!", ephemeral=True)
return
if interaction.user.top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a higher role than you!", ephemeral=True)
return
permissions = self.check_permissions(interaction.client.user, ['kick_members'], interaction)
if permissions:
await interaction.response.send_message(f"I do not have the `{permissions}` permission, required for this action.", ephemeral=True)
return
await interaction.response.send_message(content=f"{target.mention} has been kicked!\n**Reason** - `{reason}`")
if silent is None:
silent = not await self.config.guild(interaction.guild).dm_users()
if silent is False:
try:
embed = await self.embed_factory('message', guild=interaction.guild, reason=reason, moderation_type='kicked', response=await interaction.original_response())
await target.send(embed=embed)
except discord.errors.HTTPException:
pass
await target.kick(f"Kicked by {interaction.user.id} for: {reason}")
moderation_id = await self.mysql_log(interaction.guild.id, interaction.user.id, 'KICK', target.id, 'NULL', reason)
await self.log(interaction, moderation_id)
@app_commands.command(name="ban")
@app_commands.choices(delete_messages=[
Choice(name="None", value=0),
Choice(name='1 Hour', value=3600),
Choice(name='12 Hours', value=43200),
Choice(name='1 Day', value=86400),
Choice(name='3 Days', value=259200),
Choice(name='7 Days', value=604800),
])
async def ban(self, interaction: discord.Interaction, target: discord.User, reason: str, duration: str = None, delete_messages: Choice[int] = 0, silent: bool = None):
"""Ban a user.
Parameters
-----------
target: discord.user
Who are you banning?
duration: str
How long are you banning this user for?
reason: str
Why are you banning this user?
delete_messages: Choices[int]
How many days of messages to delete?
silent: bool
Should the user be messaged?"""
if interaction.guild.get_member(target.id):
target_member = interaction.guild.get_member(target.id)
if interaction.guild.get_member(interaction.client.user.id).top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a role higher than the bot!", ephemeral=True)
return
if interaction.user.top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a higher role than you!", ephemeral=True)
return
permissions = self.check_permissions(interaction.client.user, ['ban_members'], interaction)
if permissions:
await interaction.response.send_message(f"I do not have the `{permissions}` permission, required for this action.", ephemeral=True)
return
try:
await interaction.guild.fetch_ban(target)
await interaction.response.send_message(content=f"{target.mention} is already banned!", ephemeral=True)
return
except discord.errors.NotFound:
pass
if duration:
try:
parsed_time = parse(sval=duration, as_timedelta=True, raise_exception=True)
except ValueError:
await interaction.response.send_message("Please provide a valid duration!", ephemeral=True)
return
await interaction.response.send_message(content=f"{target.mention} has been banned for {humanize.precisedelta(parsed_time)}!\n**Reason** - `{reason}`")
try:
embed = await self.embed_factory('message', guild=interaction.guild, reason=reason, moderation_type='tempbanned', response=await interaction.original_response(), duration=parsed_time)
await target.send(embed=embed)
except discord.errors.HTTPException:
pass
await interaction.guild.ban(target, reason=f"Tempbanned by {interaction.user.id} for: {reason} (Duration: {parsed_time})", delete_message_seconds=delete_messages)
await self.mysql_log(interaction.guild.id, interaction.user.id, 'TEMPBAN', target.id, parsed_time, reason)
else:
await interaction.response.send_message(content=f"{target.mention} has been banned!\n**Reason** - `{reason}`")
if silent is None:
silent = not await self.config.guild(interaction.guild).dm_users()
if silent is False:
try:
embed = embed = await self.embed_factory('message', guild=interaction.guild, reason=reason, moderation_type='banned', response=await interaction.original_response())
await target.send(embed=embed)
except discord.errors.HTTPException:
pass
await interaction.guild.ban(target, reason=f"Banned by {interaction.user.id} for: {reason}", delete_message_seconds=delete_messages)
moderation_id = await self.mysql_log(interaction.guild.id, interaction.user.id, 'BAN', target.id, 'NULL', reason)
await self.log(interaction, moderation_id)
@app_commands.command(name="unban")
async def unban(self, interaction: discord.Interaction, target: discord.User, reason: str = None, silent: bool = None):
"""Unban a user.
Parameters
-----------
target: discord.user
Who are you unbanning?
reason: str
Why are you unbanning this user?
silent: bool
Should the user be messaged?"""
if interaction.guild.get_member(target.id):
target_member = interaction.guild.get_member(target.id)
if interaction.guild.get_member(interaction.client.user.id).top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a role higher than the bot!", ephemeral=True)
return
if interaction.user.top_role <= target_member.top_role:
await interaction.response.send_message(content="You cannot moderate members with a higher role than you!", ephemeral=True)
return
permissions = self.check_permissions(interaction.client.user, ['ban_members'], interaction)
if permissions:
await interaction.response.send_message(f"I do not have the `{permissions}` permission, required for this action.", ephemeral=True)
return
try:
await interaction.guild.fetch_ban(target)
except discord.errors.NotFound:
await interaction.response.send_message(content=f"{target.mention} is not banned!", ephemeral=True)
return
if reason:
await interaction.guild.unban(target, reason=f"Unbanned by {interaction.user.id} for: {reason}")
else:
await interaction.guild.unban(target, reason=f"Unbanned by {interaction.user.id}")
reason = "No reason given."
await interaction.response.send_message(content=f"{target.mention} has been unbanned!\n**Reason** - `{reason}`")
if silent is None:
silent = not await self.config.guild(interaction.guild).dm_users()
if silent is False:
try:
embed = await self.embed_factory('message', guild=interaction.guild, reason=reason, moderation_type='unbanned', response=await interaction.original_response())
await target.send(embed=embed)
except discord.errors.HTTPException:
pass
moderation_id = await self.mysql_log(interaction.guild.id, interaction.user.id, 'UNBAN', target.id, 'NULL', reason)
await self.log(interaction, moderation_id)
@app_commands.command(name="history")
async def history(self, interaction: discord.Interaction, target: discord.User = None, moderator: discord.User = None, pagesize: app_commands.Range[int, 1, 25] = 5, page: int = 1, epheremal: bool = False):
"""List previous infractions.
Parameters
-----------
target: discord.User
User whose infractions to query, overrides moderator if both are given
moderator: discord.User
Query by moderator
pagesize: app_commands.Range[int, 1, 25]
Amount of infractions to list per page
page: int
Page to select
epheremal: bool
Hide the command response"""
permissions = self.check_permissions(interaction.client.user, ['embed_links'], interaction)
if permissions:
await interaction.response.send_message(f"I do not have the `{permissions}` permission, required for this action.", ephemeral=True)
return
database = await self.connect()
cursor = database.cursor()
if target:
query = """SELECT *
FROM moderation_%s
WHERE target_id = %s
ORDER BY moderation_id DESC;"""
cursor.execute(query, (interaction.guild.id, target.id))
elif moderator:
query = """SELECT *
FROM moderation_%s
WHERE moderator_id = %s
ORDER BY moderation_id DESC;"""
cursor.execute(query, (interaction.guild.id, moderator.id))
else:
query = """SELECT *
FROM moderation_%s
ORDER BY moderation_id DESC;"""
cursor.execute(query, (interaction.guild.id,))
results = cursor.fetchall()
result_dict_list = []
for result in results:
case_dict = self.generate_dict(result)
result_dict_list.append(case_dict)
if target or moderator:
case_quantity = len(result_dict_list)
else:
case_quantity = len(result_dict_list) - 1 # account for case 0 technically existing
page_quantity = round(case_quantity / pagesize)
start_index = (page - 1) * pagesize
end_index = page * pagesize
embed = discord.Embed(color=await self.bot.get_embed_color(None))
embed.set_author(icon_url=interaction.guild.icon.url, name='Infraction History')
embed.set_footer(text=f"Page {page}/{page_quantity} | {case_quantity} Results")
for case in result_dict_list[start_index:end_index]:
if str(case['moderation_id']) == '0':
continue
target_user = await self.fetch_user_dict(interaction, case['target_id'])
moderator_user = await self.fetch_user_dict(interaction, case['moderator_id'])
target_name = f"`{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']}"
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']})\n**Reason:** `{str(case['reason'])[:150]}`"
if case['duration'] != 'NULL':
td = timedelta(**{unit: int(val) for unit, val in zip(["hours", "minutes", "seconds"], case["duration"].split(":"))})
duration_embed = f"{humanize.precisedelta(td)} | <t:{case['end_timestamp']}:R>" if case["expired"] == '0' else f"{humanize.precisedelta(td)} | Expired"
field_value = field_value + f"\n**Duration:** {duration_embed}"
if bool(case['resolved']):
field_value = field_value + "\n**Resolved:** True"
embed.add_field(name=field_name, value=field_value, inline=False)
await interaction.response.send_message(embed=embed, ephemeral=epheremal)
@app_commands.command(name="resolve")
async def resolve(self, interaction: discord.Interaction, case_number: int, reason: str = None):
"""Resolve a specific case.
Parameters
-----------
case_number: int
Case number of the case you're trying to resolve
reason: str
Reason for resolving case"""
permissions = self.check_permissions(interaction.client.user, ['embed_links', 'moderate_members', 'ban_members'], interaction)
if permissions:
await interaction.response.send_message(f"I do not have the `{permissions}` permission, required for this action.", ephemeral=True)
return
conf = await self.check_conf(['mysql_database'])
if conf:
raise(LookupError)
database = await self.connect()
cursor = database.cursor()
db = await self.config.mysql_database()
query_1 = "SELECT * FROM moderation_%s WHERE moderation_id = %s;"
cursor.execute(query_1, (interaction.guild.id, case_number))
result_1 = cursor.fetchone()
if result_1 is None or case_number == 0:
await interaction.response.send_message(content=f"There is no moderation with a case number of {case_number}.", ephemeral=True)
return
query_2 = "SELECT * FROM moderation_%s WHERE moderation_id = %s AND resolved = 0;"
cursor.execute(query_2, (interaction.guild.id, case_number))
result_2 = cursor.fetchone()
if result_2 is None:
await interaction.response.send_message(content=f"This moderation has already been resolved!\nUse `/case {case_number}` for more information.", ephemeral=True)
return
case = self.generate_dict(result_2)
if reason is None:
reason = "No reason given."
if case['moderation_type'] in ['UNMUTE', 'UNBAN']:
await interaction.response.send_message(content="You cannot resolve this type of moderation!", ephemeral=True)
if case['moderation_type'] in ['MUTE', 'TEMPBAN', 'BAN']:
if case['moderation_type'] == 'MUTE':
try:
member = await interaction.guild.fetch_member(case['target_id'])
await member.timeout(None, reason=f"Case #{case_number} resolved by {interaction.user.id}")
except discord.NotFound:
pass
if case['moderation_type'] in ['TEMPBAN', 'BAN']:
try:
user = await interaction.client.fetch_user(case['target_id'])
await interaction.guild.unban(user, reason=f"Case #{case_number} resolved by {interaction.user.id}")
except discord.NotFound:
pass
resolve_query = f"UPDATE `{db}`.`moderation_{interaction.guild.id}` SET resolved = 1, expired = 1, resolved_by = %s, resolve_reason = %s WHERE moderation_id = %s"
else:
resolve_query = f"UPDATE `{db}`.`moderation_{interaction.guild.id}` SET resolved = 1, resolved_by = %s, resolve_reason = %s WHERE moderation_id = %s"
cursor.execute(resolve_query, (interaction.user.id, reason, case_number))
database.commit()
response_query = "SELECT * FROM moderation_%s WHERE moderation_id = %s;"
cursor.execute(response_query, (interaction.guild.id, case_number))
result = cursor.fetchone()
case_dict = self.generate_dict(result)
embed = await self.embed_factory('case', interaction=interaction, case_dict=case_dict)
await interaction.response.send_message(content=f"✅ Moderation #{case_number} resolved!", embed=embed)
await self.log(interaction, case_number, True)
cursor.close()
database.close()
@app_commands.command(name="case")
async def case(self, interaction: discord.Interaction, case_number: int, ephemeral: bool = False):
"""Check the details of a specific case.
Parameters
-----------
case_number: int
What case are you looking up?
ephemeral: bool
Hide the command response"""
permissions = self.check_permissions(interaction.client.user, ['embed_links'], interaction)
if permissions:
await interaction.response.send_message(f"I do not have the `{permissions}` permission, required for this action.", ephemeral=True)
return
if case_number != 0:
case = await self.fetch_case(case_number, interaction.guild.id)
if case:
embed = await self.embed_factory('case', interaction=interaction, case_dict=case)
await interaction.response.send_message(embed=embed, ephemeral=ephemeral)
return
await interaction.response.send_message(content=f"No case with case number `{case_number}` found.", ephemeral=True)
@tasks.loop(minutes=1)
async def handle_expiry(self):
conf = await self.check_conf(['mysql_database'])
if conf:
raise(LookupError)
database = await self.connect()
cursor = database.cursor()
db = await self.config.mysql_database()
guilds: list[discord.Guild] = self.bot.guilds
for guild in guilds:
if not await self.bot.cog_disabled_in_guild(self, guild):
tempban_query = f"SELECT target_id, moderation_id FROM moderation_{guild.id} WHERE end_timestamp != 0 AND end_timestamp <= %s AND moderation_type = 'TEMPBAN' AND expired = 0"
try:
cursor.execute(tempban_query, (time.time(),))
result = cursor.fetchall()
except mysql.connector.errors.ProgrammingError:
continue
target_ids = [row[0] for row in result]
moderation_ids = [row[1] for row in result]
for target_id, moderation_id in zip(target_ids, moderation_ids):
user: discord.User = await self.bot.fetch_user(target_id)
await guild.unban(user, reason=f"Automatic unban from case #{moderation_id}")
embed = await self.embed_factory('message', guild, f'Automatic unban from case #{moderation_id}', 'unbanned')
try:
await user.send(embed=embed)
except discord.errors.HTTPException:
pass
expiry_query = f"UPDATE `{db}`.`moderation_{guild.id}` SET expired = 1 WHERE (end_timestamp != 0 AND end_timestamp <= %s AND expired = 0) OR (expired = 0 AND resolved = 1)"
cursor.execute(expiry_query, (time.time(),))
database.commit()
cursor.close()
database.close()
@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):
"""Toggle if the cog should ignore other bots' moderations."""
await self.config.guild(ctx.guild).ignore_other_bots.set(not await self.config.guild(ctx.guild).ignore_other_bots())
await ctx.send(f"Ignore bots setting set to {await self.config.guild(ctx.guild).ignore_other_bots()}")
@moderationset.command(name="dm")
@checks.admin()
async def moderationset_dm(self, ctx: commands.Context):
"""Toggle automatically messaging moderated users.
This option can be overridden by specifying the `silent` argument in any moderation command."""
await self.config.guild(ctx.guild).dm_users.set(not await self.config.guild(ctx.guild).dm_users())
await ctx.send(f"DM users setting set to {await self.config.guild(ctx.guild).dm_users()}")
@moderationset.command(name="logchannel")
@checks.admin()
async def moderationset_logchannel(self, ctx: commands.Context, channel: discord.TextChannel = None):
"""Set a channel to log infractions to."""
if channel:
await self.config.guild(ctx.guild).log_channel.set(channel.id)
await ctx.send(f"Logging channel set to {channel.mention}.")
else:
await self.config.guild(ctx.guild).log_channel.set(" ")
await ctx.send("Logging channel disabled.")
@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)
@commands.command(aliases=["tdc"])
async def timedeltaconvert(self, ctx: commands.Context, *, duration: str):
"""This command converts a duration to a [`timedelta`](https://docs.python.org/3/library/datetime.html#datetime.timedelta) Python object.
**Example usage**
`[p]timedeltaconvert 1 day 15hr 82 minutes 52s`
**Output**
`1 day, 16:22:52`"""
try:
parsed_time = parse(duration, as_timedelta=True, raise_exception=True)
await ctx.send(f"`{str(parsed_time)}`")
except ValueError:
await ctx.send("Please provide a convertible value!")

View file

@ -0,0 +1,30 @@
datasource db {
provider = "sqlite"
url = "file:moderation.db"
}
generator client {
provider = "prisma-client-py"
interface = "asyncio"
recursive_type_depth = 5
}
model Case {
globalId Int @id @unique
guildId BigInt
moderationId Int
timestamp DateTime
moderationType String
targetId BigInt
moderatorId BigInt
roleId BigInt?
duration String?
endTimestamp DateTime?
reason String?
resolved Boolean
resolvedBy BigInt?
resolveReason String?
expired Boolean
@@index([guildId, moderationId, targetId, moderatorId])
}

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr"], "author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing Podcast!\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs", "install_msg" : "Thank you for installing Podcast!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs",
"name" : "Podcast", "name" : "Podcast",
"short" : "Provides a questions submission system.", "short" : "Provides a questions submission system.",
"description" : "Provies a questions submission system.", "description" : "Provies a questions submission system.",

View file

@ -1,9 +1,8 @@
from redbot.core import Config, checks, commands from redbot.core import commands, checks, Config
class Podcast(commands.Cog): class Podcast(commands.Cog):
"""Provides a questions submission system for podcasts. """Provides a questions submission system for podcasts.
Developed by cswimr.""" Developed by SeaswimmerTheFsh."""
def __init__(self, bot): def __init__(self, bot):
self.bot = bot self.bot = bot

2613
poetry.lock generated

File diff suppressed because it is too large Load diff

View file

@ -5,7 +5,6 @@ description = "Custom cogs/cog modifications intended for the Galaxy discord ser
authors = ["Galaxy Discord Management Team"] authors = ["Galaxy Discord Management Team"]
license = "MPL 2" license = "MPL 2"
readme = "README.md" readme = "README.md"
package-mode = false
[tool.poetry.dependencies] [tool.poetry.dependencies]
python = ">=3.9,<3.12" python = ">=3.9,<3.12"
@ -15,7 +14,6 @@ prisma = "^0.10.0"
mysql-connector-python = "^8.1.0" mysql-connector-python = "^8.1.0"
humanize = "^4.8.0" humanize = "^4.8.0"
pytube = "^15.0.0" pytube = "^15.0.0"
ruff = "^0.3.6"
[tool.poetry.group.dev] [tool.poetry.group.dev]
optional = true optional = true

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr"], "author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing Send!\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs", "install_msg" : "Thank you for installing Send!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs",
"name" : "Send", "name" : "Send",
"short" : "Allows you to send messages as the bot user!", "short" : "Allows you to send messages as the bot user!",
"description" : "Allows you to send messages as the bot user!.", "description" : "Allows you to send messages as the bot user!.",

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr"], "author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing Shortmute!\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs", "install_msg" : "Thank you for installing Shortmute!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs",
"name" : "Shortmute", "name" : "Shortmute",
"short" : "Allows staff members to shortmute individuals for up to 30 minutes.", "short" : "Allows staff members to shortmute individuals for up to 30 minutes.",
"description" : "Allows staff members to shortmute individuals for up to 30 minutes, using Discord's Timeouts feature.", "description" : "Allows staff members to shortmute individuals for up to 30 minutes, using Discord's Timeouts feature.",

View file

@ -37,7 +37,7 @@ class Shortmute(commands.Cog):
An image file used as evidence for the shortmute, do not use with evidence_link An image file used as evidence for the shortmute, do not use with evidence_link
skip_confirmation: bool = None skip_confirmation: bool = None
This allows you skip the confirmation prompt and immediately shortmute the user. This allows you skip the confirmation prompt and immediately shortmute the user.
""" """
disable_dateutil() disable_dateutil()
timedelta = parse(f'{duration} minutes', as_timedelta=True) timedelta = parse(f'{duration} minutes', as_timedelta=True)
if evidence_image and evidence_link: if evidence_image and evidence_link:
@ -56,7 +56,7 @@ class Shortmute(commands.Cog):
"timedelta": timedelta, "timedelta": timedelta,
"reason": reason, "reason": reason,
"interaction": interaction, "interaction": interaction,
"color": await self.bot.get_embed_color(interaction.guild), "color": await self.bot.get_embed_color(None),
"evidence": evidence "evidence": evidence
} }
blacklisted_users_list = await self.config.guild(interaction.guild).blacklisted_users() blacklisted_users_list = await self.config.guild(interaction.guild).blacklisted_users()
@ -91,13 +91,13 @@ class Shortmute(commands.Cog):
await interaction.response.send_message(content=f"Please shortmute the user for longer than {readable_duration}! The maximum duration is 30 minutes.", ephemeral=True) await interaction.response.send_message(content=f"Please shortmute the user for longer than {readable_duration}! The maximum duration is 30 minutes.", ephemeral=True)
return return
if skip_confirmation is False: if skip_confirmation is False:
embed = discord.Embed(title="Are you sure?", description=f"**Moderator:** {interaction.user.mention}\n**Target:** {target.mention}\n**Duration:** {readable_duration}\n**Reason:** `{reason}`", color=await self.bot.get_embed_color(interaction.guild)) embed = discord.Embed(title="Are you sure?", description=f"**Moderator:** {interaction.user.mention}\n**Target:** {target.mention}\n**Duration:** {readable_duration}\n**Reason:** `{reason}`", color=await self.bot.get_embed_color(None))
embed.set_footer(text="/shortmute") embed.set_footer(text="/shortmute")
if evidence: if evidence:
embed.set_image(url=evidence) embed.set_image(url=evidence)
await interaction.response.send_message(embed=embed, view=self.ShortmuteButtons(timeout=180, passed_info=passed_info), ephemeral=True) await interaction.response.send_message(embed=embed, view=self.ShortmuteButtons(timeout=180, passed_info=passed_info), ephemeral=True)
elif skip_confirmation is True: elif skip_confirmation is True:
edit_embed = discord.Embed(title="Shortmute confirmed!", description=f"**Moderator:** {interaction.user.mention}\n**Target:** {target.mention}\n**Duration:** {readable_duration}\n**Reason:** `{reason}`", color=await self.bot.get_embed_color(interaction.guild)) edit_embed = discord.Embed(title="Shortmute confirmed!", description=f"**Moderator:** {interaction.user.mention}\n**Target:** {target.mention}\n**Duration:** {readable_duration}\n**Reason:** `{reason}`", color=await self.bot.get_embed_color(None))
edit_embed.set_footer(text="/shortmute") edit_embed.set_footer(text="/shortmute")
if evidence: if evidence:
edit_embed.set_image(url=evidence) edit_embed.set_image(url=evidence)
@ -105,7 +105,7 @@ class Shortmute(commands.Cog):
await target.timeout(timedelta, reason=f"User shortmuted for {readable_duration} by {interaction.user.name} ({interaction.user.id}) for: {reason}") await target.timeout(timedelta, reason=f"User shortmuted for {readable_duration} by {interaction.user.name} ({interaction.user.id}) for: {reason}")
shortmute_msg = await interaction.channel.send(content=f"{target.mention} was shortmuted for {readable_duration} by {interaction.user.mention} for: `{reason}`") shortmute_msg = await interaction.channel.send(content=f"{target.mention} was shortmuted for {readable_duration} by {interaction.user.mention} for: `{reason}`")
if await self.config.guild(interaction.guild).dm() is True: if await self.config.guild(interaction.guild).dm() is True:
dm_embed = discord.Embed(title=f"You've been shortmuted in {interaction.guild.name}!", description=f"**Moderator:** {interaction.user.mention}\n**Target:** {target.mention}\n**Message:** {shortmute_msg.jump_url}\n**Duration:** {readable_duration}\n**Reason:** `{reason}`", color=await self.bot.get_embed_color(interaction.guild)) dm_embed = discord.Embed(title=f"You've been shortmuted in {interaction.guild.name}!", description=f"**Moderator:** {interaction.user.mention}\n**Target:** {target.mention}\n**Message:** {shortmute_msg.jump_url}\n**Duration:** {readable_duration}\n**Reason:** `{reason}`", color=await self.bot.get_embed_color(None))
dm_embed.set_footer(text="/shortmute") dm_embed.set_footer(text="/shortmute")
if evidence: if evidence:
dm_embed.set_image(url=evidence) dm_embed.set_image(url=evidence)
@ -115,7 +115,7 @@ class Shortmute(commands.Cog):
await message.edit(content="Could not message the target, user most likely has Direct Messages disabled.") await message.edit(content="Could not message the target, user most likely has Direct Messages disabled.")
logging_channels_list = await self.config.guild(interaction.guild).logging_channels() logging_channels_list = await self.config.guild(interaction.guild).logging_channels()
if logging_channels_list: if logging_channels_list:
logging_embed = discord.Embed(title="User Shortmuted", description=f"**Moderator:** {interaction.user.mention} ({interaction.user.id})\n**Target:** {target.mention} ({target.id})\n**Message:** {shortmute_msg.jump_url}\n**Duration:** {readable_duration}\n**Reason:** `{reason}`\n**Confirmation Skipped:** True", color=await self.bot.get_embed_color(interaction.guild)) logging_embed = discord.Embed(title="User Shortmuted", description=f"**Moderator:** {interaction.user.mention} ({interaction.user.id})\n**Target:** {target.mention} ({target.id})\n**Message:** {shortmute_msg.jump_url}\n**Duration:** {readable_duration}\n**Reason:** `{reason}`\n**Confirmation Skipped:** True", color=await self.bot.get_embed_color(None))
logging_embed.set_footer(text="/shortmute") logging_embed.set_footer(text="/shortmute")
if evidence: if evidence:
logging_embed.set_image(url=evidence) logging_embed.set_image(url=evidence)

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr, meelyman"], "author" : ["SeaswimmerTheFsh, meelyman"],
"install_msg" : "Thank you for installing Suggestions!\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs\nYou can find the original source code from SauriCogs here: <https://github.com/elijabesu/SauriCogs>", "install_msg" : "Thank you for installing Suggestions!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs\nYou can find the original source code from SauriCogs here: <https://github.com/elijabesu/SauriCogs>",
"name" : "Suggestions", "name" : "Suggestions",
"short" : "Simple suggestions system.", "short" : "Simple suggestions system.",
"description" : "Per guild suggestions system.", "description" : "Per guild suggestions system.",

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr"], "author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing SugonCredit!\n**Please load the Economy cog before loading this cog.**\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs.", "install_msg" : "Thank you for installing SugonCredit!\n**Please load the Economy cog before loading this cog.**\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs.",
"name" : "SugonCredit", "name" : "SugonCredit",
"short" : "Simple points system.", "short" : "Simple points system.",
"description" : "Implements a way for moderators to give out social-credit like points, dubbed 'sugoncredits' by the community.", "description" : "Implements a way for moderators to give out social-credit like points, dubbed 'sugoncredits' by the community.",

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr"], "author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing World Zero!\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs", "install_msg" : "Thank you for installing World Zero!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs",
"name" : "World Zero", "name" : "World Zero",
"short" : "This cog is meant to provide random functions for my crippling World Zero addiction!", "short" : "This cog is meant to provide random functions for my crippling World Zero addiction!",
"description" : "This cog is meant to provide random functions for my crippling World Zero addiction!", "description" : "This cog is meant to provide random functions for my crippling World Zero addiction!",

View file

@ -1,10 +1,9 @@
import discord
from redbot.core import commands from redbot.core import commands
import discord
class WorldZero(commands.Cog): class WorldZero(commands.Cog):
"""This cog is meant to provide random functions for my crippling World Zero addiction! """This cog is meant to provide random functions for my crippling World Zero addiction!
Developed by cswimr.""" Developed by SeaswimmerTheFsh."""
def __init__(self, bot): def __init__(self, bot):
self.bot = bot self.bot = bot

View file

@ -1,6 +1,6 @@
{ {
"author" : ["cswimr"], "author" : ["SeaswimmerTheFsh"],
"install_msg" : "Thank you for installing YouTubeDownloader!\nYou can find the source code of this cog here: https://coastalcommits.com/cswimr/GalaxyCogs", "install_msg" : "Thank you for installing YouTubeDownloader!\nYou can find the source code of this cog here: https://coastalcommits.com/SeaswimmerTheFsh/GalaxyCogs",
"name" : "YouTubeDownloader", "name" : "YouTubeDownloader",
"short" : "Custom cog intended for use on the Galaxy discord server.", "short" : "Custom cog intended for use on the Galaxy discord server.",
"description" : "Custom cog intended for use on the Galaxy discord server.", "description" : "Custom cog intended for use on the Galaxy discord server.",

View file

@ -23,7 +23,7 @@ class YouTubeDownloader(commands.Cog):
blacklisted_users = await self.config.blacklisted_users() blacklisted_users = await self.config.blacklisted_users()
for blacklisted_user_id in blacklisted_users: for blacklisted_user_id in blacklisted_users:
if blacklisted_user_id == user_id: if blacklisted_user_id == user_id:
raise self.UserBlacklisted return self.UserBlacklisted
@commands.command() @commands.command()
@checks.is_owner() @checks.is_owner()
@ -43,7 +43,7 @@ class YouTubeDownloader(commands.Cog):
await ctx.send("The path you've provided doesn't exist!") await ctx.send("The path you've provided doesn't exist!")
@commands.command(aliases=["dl"]) @commands.command(aliases=["dl"])
async def download(self, ctx: commands.Context, url: str, audio_only: bool = True, delete: bool = True, *, subfolder: str = None): async def download(self, ctx: commands.Context, url: str, audio_only: bool = True, delete: bool = False, *, subfolder: str = None):
"""This command downloads a YouTube Video as an `m4a` (or `mp4`) and uploads the file to discord. """This command downloads a YouTube Video as an `m4a` (or `mp4`) and uploads the file to discord.
If you're considered a bot owner, you will be able to save downloaded files to the data path set in the `[p]change_data_path` command. If you're considered a bot owner, you will be able to save downloaded files to the data path set in the `[p]change_data_path` command.
@ -58,15 +58,14 @@ class YouTubeDownloader(commands.Cog):
- The `subfolder` argument only does anything if `delete` is set to False, but it allows you to save to a subfolder in the data path you've set previously without having to change said data path manually.""" - The `subfolder` argument only does anything if `delete` is set to False, but it allows you to save to a subfolder in the data path you've set previously without having to change said data path manually."""
try: try:
await self.blacklist_checker(ctx.author.id) self.blacklist_checker(ctx.author.id)
except self.UserBlacklisted: except self.UserBlacklisted:
await ctx.send("You are blacklisted from running this command!") await ctx.send("You are blacklisted from running this command!")
return return
def youtube_download(url: str, path: str): def youtube_download(url: str, path: str):
"""This function does the actual downloading of the YouTube Video.""" """This function does the actual downloading of the YouTube Video."""
yt = YouTube(url=url) yt = YouTube(url=url)
translation_table = dict.fromkeys(map(ord, r'<>:"/\|?*'), None) filename = f"{yt.title} ({yt.video_id})"
filename = f"{yt.title.translate(translation_table)} ({yt.video_id})"
if audio_only: if audio_only:
stream = yt.streams.filter(only_audio=True, mime_type='audio/mp4') stream = yt.streams.filter(only_audio=True, mime_type='audio/mp4')
stream = stream.order_by('abr') stream = stream.order_by('abr')
@ -121,7 +120,7 @@ class YouTubeDownloader(commands.Cog):
file.close() file.close()
if delete is True or await self.bot.is_owner(ctx.author) is False: if delete is True or await self.bot.is_owner(ctx.author) is False:
if output[1] is False: if output[1] is False:
os.remove(output[0]) os.remove(output)
await complete_message.edit(content="YouTube Downloader completed!\nFile has been deleted.\nDownloaded file:") await complete_message.edit(content="YouTube Downloader completed!\nFile has been deleted.\nDownloaded file:")
if output[1] is True: if output[1] is True:
await complete_message.edit(content="YouTube Downloader completed!\nFile has not been deleted, as it was previously downloaded and saved.\nDownloaded file:") await complete_message.edit(content="YouTube Downloader completed!\nFile has not been deleted, as it was previously downloaded and saved.\nDownloaded file:")