Compare commits

..

18 commits

Author SHA1 Message Date
7af8e02462
chore(repo): flake formatting
Some checks failed
Actions / Lint Code (Ruff & Pylint) (push) Failing after 48s
Actions / Build Documentation (MkDocs) (push) Successful in 1m5s
2024-11-20 12:33:59 -05:00
a40f78e88f
chore(repo): use nixpkgs-python to get python version instead of setting languages.python.package
Some checks failed
Actions / Lint Code (Ruff & Pylint) (push) Failing after 47s
Actions / Build Documentation (MkDocs) (push) Successful in 1m3s
2024-11-20 12:27:59 -05:00
216fa74a53
chore(repo): add nix flake dev-shell for development environment
Some checks failed
Actions / Lint Code (Ruff & Pylint) (push) Failing after 43s
Actions / Build Documentation (MkDocs) (push) Successful in 56s
2024-11-20 12:22:57 -05:00
901c10a458
chore(repo): update issue templates
Some checks failed
Actions / Lint Code (Ruff & Pylint) (push) Failing after 29s
Actions / Build Documentation (MkDocs) (push) Successful in 30s
2024-11-01 11:21:38 -04:00
1150115171
remove poetry file and update workflow file to use correct meli secret
Some checks failed
Actions / Build Documentation (MkDocs) (push) Has been skipped
Actions / Lint Code (Ruff & Pylint) (push) Failing after 29s
2024-10-21 16:53:28 -04:00
8d4d61c017
chore(repo): add uv.lock file
Some checks failed
Actions / Build Documentation (MkDocs) (push) Successful in 28s
Actions / Lint Code (Ruff & Pylint) (push) Failing after 33s
2024-10-10 16:08:39 -04:00
23f2aaa2ee
feat(workflow): move to uv
Some checks failed
Actions / Build Documentation (MkDocs) (push) Successful in 32s
Actions / Lint Code (Ruff & Pylint) (push) Failing after 35s
2024-10-10 16:03:50 -04:00
8b6d269e1a
(repo): switch to uv
Some checks failed
Actions / Lint Code (Ruff & Pylint) (push) Failing after 4s
Actions / Build Documentation (MkDocs) (push) Failing after 7s
Signed-off-by: cswimr <seaswimmerthefsh@gmail.com>
2024-10-10 16:01:54 -04:00
8c323e36a9
docs(pterodactyl): link to the pelican-eggs repository instead of some random fork
Some checks failed
Actions / Lint Code (Ruff & Pylint) (push) Failing after 39s
Actions / Build Documentation (MkDocs) (push) Successful in 32s
Signed-off-by: cswimr <seaswimmerthefsh@gmail.com>
2024-09-25 12:07:53 -04:00
1392a9324a
chore(poetry): update deps
Some checks failed
Actions / Build Documentation (MkDocs) (push) Successful in 34s
Actions / Lint Code (Ruff & Pylint) (push) Failing after 44s
2024-08-28 17:01:58 -04:00
9cba1767e6
fix(repo): update all instances of SeaswimmerTheFsh to cswimr
Some checks failed
Actions / Build Documentation (MkDocs) (push) Failing after 22s
Actions / Lint Code (Ruff & Pylint) (push) Successful in 39s
2024-08-27 14:25:14 -04:00
cbfff6d211
fix(backup): update for red 3.5.13
All checks were successful
Actions / Lint Code (Ruff & Pylint) (push) Successful in 40s
Actions / Build Documentation (MkDocs) (push) Successful in 28s
2024-08-27 14:22:38 -04:00
56d1e14454
feat(repo): updated all cogs (except aurora) to use the help formatting present in the indev version of aurora 2024-08-27 14:22:19 -04:00
571be4e245
fix(pterodactyl): fixed a missing argument in one of the websocket events
All checks were successful
Actions / Build Documentation (MkDocs) (push) Successful in 26s
Actions / Lint Code (Ruff & Pylint) (push) Successful in 37s
2024-08-27 14:05:10 -04:00
6b2793e692
fix(actions): updated container tags
Some checks failed
Actions / Build Documentation (MkDocs) (push) Successful in 28s
Actions / Lint Code (Ruff & Pylint) (push) Failing after 38s
2024-08-27 14:03:15 -04:00
ad0bb7cf00
fix(pterodactyl): fixed join and leave listeners throwing errors
Some checks failed
Actions / Build Documentation (MkDocs) (push) Successful in 31s
Actions / Lint Code (Ruff & Pylint) (push) Failing after 56s
2024-08-26 19:43:57 -04:00
1244d5a941
fix(pterodactyl): don't depend on a website to host images when i can bundle image files in the cog itself
Some checks failed
Actions / Build Documentation (MkDocs) (push) Successful in 30s
Actions / Lint Code (Ruff & Pylint) (push) Failing after 54s
2024-08-26 17:46:42 -04:00
1c0f12660b
chore(pterodactyl): clean up some dirty code
All checks were successful
Actions / Build Documentation (MkDocs) (push) Successful in 30s
Actions / Lint Code (Ruff & Pylint) (push) Successful in 54s
2024-08-26 15:03:37 -04:00
59 changed files with 4318 additions and 6450 deletions

View file

@ -10,7 +10,7 @@ Aurora is a fully-featured moderation system. It is heavily inspired by Galactic
## Installation
```bash
[p]repo add seacogs https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs
[p]repo add seacogs https://www.coastalcommits.com/cswimr/SeaCogs
[p]cog install seacogs aurora
[p]cog load aurora
```

View file

@ -5,14 +5,14 @@ Backup allows you to export a JSON list of all of your installed repositories an
## Installation
```bash
[p]repo add seacogs https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs
[p]repo add seacogs https://www.coastalcommits.com/cswimr/SeaCogs
[p]cog install seacogs backup
[p]cog load backup
```
## Version Compatibility
As of commit [1edb08a](https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs/commit/1edb08a1271f12098ca0bed11a735f7162cedd14), the Backup cog no longer supports Red versions older than 3.5.6. If you want to use the cog on an earlier version (3.5.0 - 3.5.5), install the cog pinned to this commit: `43464db6a7c51bc69282b1ae3dc507a4aae851de`.
As of commit [1edb08a](https://www.coastalcommits.com/cswimr/SeaCogs/commit/1edb08a1271f12098ca0bed11a735f7162cedd14), the Backup cog no longer supports Red versions older than 3.5.6. If you want to use the cog on an earlier version (3.5.0 - 3.5.5), install the cog pinned to this commit: `43464db6a7c51bc69282b1ae3dc507a4aae851de`.
```bash
[p]cog installversion sea-cogs 43464db6a7c51bc69282b1ae3dc507a4aae851de backup

View file

@ -6,7 +6,7 @@ This cog does require an api key to work.
## Installation
```bash
[p]repo add seacogs https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs
[p]repo add seacogs https://www.coastalcommits.com/cswimr/SeaCogs
[p]cog install seacogs bible
[p]cog load bible
```

View file

@ -5,7 +5,7 @@ EmojiInfo allows you to retrieve information about an emoji.
## Installation
```bash
[p]repo add seacogs https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs
[p]repo add seacogs https://www.coastalcommits.com/cswimr/SeaCogs
[p]cog install seacogs emojiinfo
[p]cog load emojiinfo
```

View file

@ -5,7 +5,7 @@ Nerdify allows you to nerdify other people's text.
## Installation
```bash
[p]repo add seacogs https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs
[p]repo add seacogs https://www.coastalcommits.com/cswimr/SeaCogs
[p]cog install seacogs nerdify
[p]cog load nerdify
```

View file

@ -28,7 +28,7 @@ The Downloader cog allows you to add Git repositories to your bot in order to do
Now, use Downloader to add my repository to your bot:
```
[p]repo add sea-cogs https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs
[p]repo add sea-cogs https://www.coastalcommits.com/cswimr/SeaCogs
```
Now, install the Pterodactyl cog:

View file

@ -10,7 +10,7 @@ Pterodactyl allows for connecting to a Pterodactyl server through websockets. It
## Installation
```bash
[p]repo add seacogs https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs
[p]repo add seacogs https://www.coastalcommits.com/cswimr/SeaCogs
[p]cog install seacogs pterodactyl
[p]cog load aurora
```

View file

@ -10,7 +10,7 @@ There are a few caveats to running an instance of Red on Pterodactyl.
- You will not receive any support from the Red developers.
- The built-in Audio cog will not work.
- Depending on your host, you might have to request a [`tmpfs` size increase](https://github.com/ign-gg/Pterodactyl-Eggs/tree/master/bots/discord/redbot#additional-requirements).
- Depending on your host, you might have to request a [`tmpfs` size increase](https://github.com/pelican-eggs/eggs/tree/master/bots/discord/redbot#additional-requirements).
If these are unacceptable to you, you should [install Red normally](https://docs.discord.red/en/stable/install_guides/index.html).
///
@ -64,7 +64,7 @@ Red is quite a large bot, so I'll focus on the specifics of getting the bot work
```
2. Add my repository to the bot
```bash
[p]repo add sea-cogs https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs
[p]repo add sea-cogs https://www.coastalcommits.com/cswimr/SeaCogs
```
3. Install and load the Pterodactyl cog
```bash

10
.envrc Normal file
View file

@ -0,0 +1,10 @@
if ! has nix_direnv_version || ! nix_direnv_version 2.2.1; then
source_url "https://raw.githubusercontent.com/nix-community/nix-direnv/2.2.1/direnvrc" "sha256-zelF0vLbEl5uaqrfIzbgNzJWGmLzCmYAkInj/LNxvKs="
fi
watch_file flake.nix
watch_file flake.lock
if ! use flake . --no-pure-eval
then
echo "devenv could not be built. The devenv environment was not loaded. Make the necessary changes to devenv.nix and hit enter to try again." >&2
fi

View file

@ -1,8 +1,6 @@
name: Bug Report
about: File a bug report
title: "[Cog Name] "
labels: [bug]
ref: master
body:
- type: markdown
attributes:
@ -13,7 +11,7 @@ body:
attributes:
label: Please confirm that;
options:
- label: I have checked that this bug does not already have an opened/closed [issue](https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs/issues) or [pull request](https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs/pulls) associated with it.
- label: I have checked that this bug does not already have an opened/closed [issue](https://www.coastalcommits.com/cswimr/SeaCogs/issues) or [pull request](https://www.coastalcommits.com/cswimr/SeaCogs/pulls) associated with it.
required: true
- label: I have checked that I am on the latest version of [Red-DiscordBot](https://github.com/CogCreators/Red-DiscordBot), and SeaCogs.
required: true

View file

@ -1,8 +1,6 @@
name: Suggestion
about: Trying to suggest something for SeaCogs? Use this.
title: "[Cog Name] "
labels: enhancement
ref: master
labels: [enhancement]
body:
- type: markdown
attributes:
@ -13,7 +11,7 @@ body:
attributes:
label: What cog is your feature request for?
description: Specify the cog within the repository.
placeholder: E.g., ModerationCog
placeholder: E.g., Pterodactyl
validations:
required: true
- type: textarea

View file

@ -2,5 +2,5 @@
<!-- Create a new issue, if it doesn't exist yet -->
- [ ] By submitting this pull request, I permit SeaswimmerTheFsh to license my work under
the [Mozilla Public License Version 2.0](https://www.coastalcommits.com/SeaswimmerTheFsh/SeaCogs/src/branch/main/LICENSE).
- [ ] By submitting this pull request, I permit cswimr to license my work under
the [Mozilla Public License Version 2.0](https://www.coastalcommits.com/cswimr/SeaCogs/src/branch/main/LICENSE).

View file

@ -1,39 +1,46 @@
name: Actions
on:
push:
branches:
- 'main'
pull_request:
jobs:
Lint Code (Ruff & Pylint):
lint:
name: Lint Code (Ruff & Pylint)
runs-on: docker
container: www.coastalcommits.com/seaswimmerthefsh/actionscontainers-seacogs:latest
container: www.coastalcommits.com/cswimr/actions:uv
steps:
- name: Checkout
uses: actions/checkout@v3
- name: Install python
run: uv python install 3.11
- name: Install dependencies
run: poetry install --with dev --no-root
run: uv sync
- name: Analysing code with Ruff
run: ./.venv/bin/ruff check $(git ls-files '*.py')
run: uv run ruff check $(git ls-files '*.py')
continue-on-error: true
- name: Analysing code with Pylint
run: ./.venv/bin/pylint --rcfile=.forgejo/workflows/config/.pylintrc $(git ls-files '*.py')
run: uv run pylint --rcfile=.forgejo/workflows/config/.pylintrc $(git ls-files '*.py')
Build Documentation (MkDocs):
docs:
name: Build Documentation (MkDocs)
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
runs-on: docker
container: www.coastalcommits.com/seaswimmerthefsh/actionscontainers-seacogs:latest
container: www.coastalcommits.com/cswimr/actions:docs
steps:
- name: Checkout
uses: actions/checkout@v3
with:
fetch-depth: 0
- name: Install python
run: uv python install 3.11
- name: Install dependencies
run: poetry install --with docs --no-root
run: uv sync --no-dev --extra=documentation
- name: Set environment variables
uses: actions/env@v2
@ -42,7 +49,7 @@ jobs:
run: |
export SITE_URL="https://$CI_ACTION_REF_NAME_SLUG.seacogs.coastalcommits.com"
export EDIT_URI="src/branch/$CI_ACTION_REF_NAME/.docs"
./.venv/bin/mkdocs build -v
uv run mkdocs build -v
- name: Deploy documentation
run: |
@ -58,7 +65,7 @@ jobs:
npx -p "@getmeli/cli" meli upload ./site \
--url "https://pages.coastalcommits.com" \
--site "${{ vars.MELI_SITE_ID }}" \
--token "${{ secrets.MELI_SECRET }}" \
--token "${{ secrets.MELI_TOKEN }}" \
--release "$CI_ACTION_REF_NAME_SLUG/${{ env.GITHUB_SHA }}" \
--branch "$CI_ACTION_REF_NAME_SLUG"

2
.gitignore vendored
View file

@ -3,3 +3,5 @@
site
.venv
__pycache__
.direnv
.devenv

View file

@ -11,7 +11,7 @@ My assorted cogs for Red-DiscordBot.
To get started with a development environment, first clone this repository.
```sh
git clone https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs.git
git clone https://coastalcommits.com/cswimr/SeaCogs.git
```
Then, install Poetry.

View file

@ -9,14 +9,15 @@ import discord
from red_commons.logging import getLogger
from redbot.core import commands
from redbot.core.bot import Config, Red
from redbot.core.utils.chat_formatting import humanize_list
from redbot.core.utils.chat_formatting import bold, humanize_list
class AntiPolls(commands.Cog):
"""AntiPolls deletes messages that contain polls, with a configurable per-guild role and channel whitelist and support for default Discord permissions (Manage Messages)."""
__author__ = ["SeaswimmerTheFsh"]
__version__ = "1.0.0"
__author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
__git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
__version__ = "1.0.1"
__documentation__ = "https://seacogs.coastalcommits.com/antipolls/"
def __init__(self, bot: Red):
@ -38,9 +39,9 @@ class AntiPolls(commands.Cog):
n = "\n" if "\n\n" not in pre_processed else ""
text = [
f"{pre_processed}{n}",
f"Cog Version: **{self.__version__}**",
f"Author: {humanize_list(self.__author__)}",
f"Documentation: {self.__documentation__}",
f"{bold('Cog Version:')} [{self.__version__}]({self.__git__})",
f"{bold('Author:')} {humanize_list(self.__author__)}",
f"{bold('Documentation:')} {self.__documentation__}",
]
return "\n".join(text)

View file

@ -1,6 +1,6 @@
{
"author" : ["SeaswimmerTheFsh (seasw.)"],
"install_msg" : "Thank you for installing AntiPolls!\nYou can find the source code of this cog [here](https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs).",
"author" : ["cswimr"],
"install_msg" : "Thank you for installing AntiPolls!\nYou can find the source code of this cog [here](https://coastalcommits.com/cswimr/SeaCogs).",
"name" : "AntiPolls",
"short" : "AntiPolls deletes messages that contain polls.",
"description" : "AntiPolls deletes messages that contain polls, with a configurable per-guild role and channel whitelist and support for default Discord permissions (Manage Messages).",

File diff suppressed because it is too large Load diff

View file

@ -1,25 +1,20 @@
# pylint: disable=duplicate-code
import json
import os
from time import time
from typing import Dict, List
from datetime import timedelta
from typing import Dict
from discord import ButtonStyle, File, Interaction, Message, ui
from redbot.core import commands, data_manager
from redbot.core.utils.chat_formatting import warning
from discord import ButtonStyle, Interaction, Message, ui
from redbot.core import commands
from redbot.core.utils.chat_formatting import box, warning
from ..models.moderation import Moderation
from ..models.type import Type, type_registry
from ..utilities.json import dump
from ..utilities.utils import create_guild_table, timedelta_from_string
from ..utilities.database import connect, create_guild_table, mysql_log
class ImportAuroraView(ui.View):
def __init__(self, timeout, ctx, message, data: List[Dict[str, any]]):
def __init__(self, timeout, ctx, message):
super().__init__()
self.ctx: commands.Context = ctx
self.message: Message = message
self.data: List[Dict[str, any]] = data
@ui.button(label="Yes", style=ButtonStyle.success)
async def import_button_y(
@ -30,8 +25,14 @@ class ImportAuroraView(ui.View):
"Deleting original table...", ephemeral=True
)
database = connect()
cursor = database.cursor()
query = f"DROP TABLE IF EXISTS moderation_{self.ctx.guild.id};"
await Moderation.execute(query=query, return_obj=False)
cursor.execute(query)
cursor.close()
database.commit()
await interaction.edit_original_response(content="Creating new table...")
@ -39,101 +40,72 @@ class ImportAuroraView(ui.View):
await interaction.edit_original_response(content="Importing moderations...")
file = await self.ctx.message.attachments[0].read()
data: list[dict] = sorted(json.loads(file), key=lambda x: x["moderation_id"])
user_mod_types = ["NOTE", "WARN", "ADDROLE", "REMOVEROLE", "MUTE", "UNMUTE", "KICK", "TEMPBAN", "BAN", "UNBAN"]
channel_mod_types = ["SLOWMODE", "LOCKDOWN"]
failed_cases = []
for case in self.data:
for case in data:
if case["moderation_id"] == 0:
continue
moderation_type: Type = type_registry[case["moderation_type"].lower()]
if "target_type" not in case or not case["target_type"]:
if moderation_type.channel:
case["target_type"] = "channel"
if case["moderation_type"] in user_mod_types:
case["target_type"] = "USER"
elif case["moderation_type"] in channel_mod_types:
case["target_type"] = "CHANNEL"
else:
case["target_type"] = "user"
case["target_type"] = "USER"
if "role_id" not in case or not case["role_id"]:
case["role_id"] = None
else:
case["role_id"] = int(case["role_id"])
case["role_id"] = 0
case["target_id"] = int(case["target_id"])
case["moderator_id"] = int(case["moderator_id"])
changes = case.get("changes", None)
if not changes:
changes = []
else:
if not isinstance(changes, list):
changes = json.loads(changes)
if isinstance(changes, str):
changes: list[dict] = json.loads(changes)
for change in changes:
if "bot" in change:
del change["bot"]
if "changes" not in case or not case["changes"]:
case["changes"] = []
if "metadata" not in case:
metadata = {}
else:
if isinstance(case["metadata"], str):
metadata: Dict[str, any] = json.loads(case["metadata"])
else:
metadata = case["metadata"]
if not metadata.get("imported_from"):
metadata.update({"imported_from": "Aurora"})
metadata.update({"imported_timestamp": int(time())})
if case["duration"] != "NULL" and case["duration"] is not None:
duration = timedelta_from_string(case["duration"])
if moderation_type.key == "ban":
moderation_type = type_registry["tempban"]
if case["duration"] != "NULL":
hours, minutes, seconds = map(int, case["duration"].split(":"))
duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
else:
duration = None
duration = "NULL"
try:
await Moderation.log(
bot=interaction.client,
guild_id=self.ctx.guild.id,
moderator_id=case["moderator_id"],
moderation_type=moderation_type,
target_type=case["target_type"].lower(),
target_id=case["target_id"],
role_id=case["role_id"],
duration=duration,
reason=case["reason"],
await mysql_log(
self.ctx.guild.id,
case["moderator_id"],
case["moderation_type"],
case["target_type"],
case["target_id"],
case["role_id"],
duration,
case["reason"],
timestamp=case["timestamp"],
resolved=case["resolved"],
resolved_by=case["resolved_by"],
resolved_reason=case["resolve_reason"],
expired=case["expired"],
changes=changes,
changes=case["changes"],
metadata=metadata,
return_obj=False
database=database,
)
except Exception as e: # pylint: disable=broad-exception-caught
failed_cases.append(str(case["moderation_id"]) + f": {e}")
await interaction.edit_original_response(content="Import complete.")
if failed_cases:
filename = (
str(data_manager.cog_data_path(cog_instance=self))
+ str(os.sep)
+ f"failed_cases_{interaction.guild.id}.json"
)
with open(filename, "w", encoding="utf-8") as f:
dump(obj=failed_cases, fp=f, indent=2)
await interaction.channel.send(
await interaction.edit_original_response(
content="Import complete.\n"
+ warning("Failed to import the following cases:\n"),
file=File(
filename, f"failed_cases_{interaction.guild.id}.json"
+ warning("Failed to import the following cases:\n")
+ box(failed_cases)
)
)
os.remove(filename)
@ui.button(label="No", style=ButtonStyle.danger)
async def import_button_n(

View file

@ -1,21 +1,18 @@
# pylint: disable=duplicate-code
import json
from datetime import timedelta
from time import time
from discord import ButtonStyle, Interaction, Message, ui
from redbot.core import commands
from redbot.core.utils.chat_formatting import box, warning
from ..models.moderation import Change, Moderation
from ..utilities.utils import create_guild_table
from ..utilities.database import connect, create_guild_table, mysql_log
class ImportGalacticBotView(ui.View):
def __init__(self, timeout, ctx, message):
super().__init__()
self.ctx: commands.Context = ctx
self.timeout = timeout
self.message: Message = message
@ui.button(label="Yes", style=ButtonStyle.success)
@ -27,14 +24,14 @@ class ImportGalacticBotView(ui.View):
"Deleting original table...", ephemeral=True
)
database = await Moderation.connect()
cursor = await database.cursor()
database = connect()
cursor = database.cursor()
query = f"DROP TABLE IF EXISTS moderation_{self.ctx.guild.id};"
await cursor.execute(query)
cursor.execute(query)
await cursor.close()
await database.commit()
cursor.close()
database.commit()
await interaction.edit_original_response(content="Creating new table...")
@ -70,12 +67,12 @@ class ImportGalacticBotView(ui.View):
if case["duration"] is not None and float(case["duration"]) != 0:
duration = timedelta(seconds=round(float(case["duration"]) / 1000))
else:
duration = None
duration = "NULL"
except OverflowError:
failed_cases.append(case["case"])
continue
metadata = {"imported_from": "GalacticBot", "imported_timestamp": int(time())}
metadata = {"imported_from": "GalacticBot"}
if case["type"] == "SLOWMODE":
metadata["seconds"] = case["data"]["seconds"]
@ -101,37 +98,37 @@ class ImportGalacticBotView(ui.View):
if resolved_timestamp is None:
resolved_timestamp = timestamp
changes = [
Change.from_dict(interaction.client, {
{
"type": "ORIGINAL",
"reason": case["reason"],
"user_id": case["executor"],
"timestamp": timestamp,
}),
Change.from_dict(interaction.client, {
},
{
"type": "RESOLVE",
"reason": resolved_reason,
"user_id": resolved_by,
"timestamp": resolved_timestamp,
}),
},
]
else:
resolved = None
resolved_by = None
resolved_reason = None
changes = None
resolved = 0
resolved_by = "NULL"
resolved_reason = "NULL"
changes = []
if case["reason"] and case["reason"] != "N/A":
reason = case["reason"]
else:
reason = None
reason = "NULL"
await Moderation.log(
await mysql_log(
self.ctx.guild.id,
case["executor"],
case["type"],
case["targetType"],
case["target"],
None,
0,
duration,
reason,
timestamp=timestamp,

View file

@ -1,6 +1,6 @@
{
"author" : ["Seaswimmer (cswimr)"],
"install_msg" : "Thank you for installing Aurora!\nMost of this cog's functionality requires enabling slash commands.\nYou can find the source code of this cog [here](https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs).",
"author" : ["cswimr"],
"install_msg" : "Thank you for installing Aurora!\nMost of this cog's functionality requires enabling slash commands.\nYou can find the source code of this cog [here](https://coastalcommits.com/cswimr/SeaCogs).",
"name" : "Aurora",
"short" : "A full replacement for Red's core Mod cogs.",
"description" : "Aurora is a fully-featured moderation system. It is heavily inspired by GalacticBot, and is designed to be a more user-friendly alternative to Red's core Mod cogs. This cog stores all of its data in an SQLite database.",
@ -9,7 +9,6 @@
"disabled": false,
"min_bot_version": "3.5.0",
"min_python_version": [3, 10, 0],
"requirements": ["pydantic", "aiosqlite", "phx-class-registry==5.0.0"],
"tags": [
"mod",
"moderate",

View file

@ -1,24 +1,20 @@
from discord import ButtonStyle, Interaction, Message, ui
from discord.errors import NotFound
from redbot.core import commands
from redbot.core.utils.chat_formatting import error
from ..utilities.config import config
from ..utilities.factory import addrole_embed
from aurora.utilities.config import config
from aurora.utilities.factory import addrole_embed
class Addrole(ui.View):
def __init__(self, ctx: commands.Context, message: Message, timeout: int | None = None):
def __init__(self, ctx: commands.Context, message: Message, timeout: int = None):
super().__init__()
self.ctx = ctx
self.message = message
self.timeout = timeout
async def on_timeout(self):
try:
await self.message.edit(view=None)
except NotFound:
pass
@ui.select(cls=ui.RoleSelect, placeholder="Select a role", min_values=0, max_values=25)
async def addrole_select(self, interaction: Interaction, select: ui.RoleSelect):

View file

@ -1,24 +1,20 @@
from discord import ButtonStyle, Interaction, Message, ui
from discord.errors import NotFound
from redbot.core import commands
from ..utilities.config import config
from ..utilities.factory import guild_embed
from ..utilities.utils import create_pagesize_options
from aurora.utilities.config import config
from aurora.utilities.factory import guild_embed
from aurora.utilities.utils import create_pagesize_options
class Guild(ui.View):
def __init__(self, ctx: commands.Context, message: Message, timeout: int | None = None):
def __init__(self, ctx: commands.Context, message: Message, timeout: int = None):
super().__init__()
self.ctx = ctx
self.message = message
self.timeout = timeout
async def on_timeout(self):
try:
await self.message.edit(view=None)
except NotFound:
pass
@ui.button(label="Show Moderator", style=ButtonStyle.green, row=0)
async def show_moderator(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument

View file

@ -1,24 +1,20 @@
from discord import ButtonStyle, Interaction, Message, ui
from discord.errors import NotFound
from redbot.core import commands
from redbot.core.utils.chat_formatting import error
from ..utilities.config import config
from ..utilities.factory import immune_embed
from aurora.utilities.config import config
from aurora.utilities.factory import immune_embed
class Immune(ui.View):
def __init__(self, ctx: commands.Context, message: Message, timeout: int | None = None):
def __init__(self, ctx: commands.Context, message: Message, timeout: int = None):
super().__init__()
self.ctx = ctx
self.message = message
self.timeout = timeout
async def on_timeout(self):
try:
await self.message.edit(view=None)
except NotFound:
pass
@ui.select(cls=ui.RoleSelect, placeholder="Select a role", min_values=0, max_values=25)
async def immune_select(self, interaction: Interaction, select: ui.RoleSelect):

View file

@ -1,24 +1,20 @@
from discord import ButtonStyle, Interaction, Message, ui
from discord.errors import NotFound
from redbot.core import commands
from ..utilities.config import config
from ..utilities.factory import overrides_embed
from ..utilities.utils import create_pagesize_options
from aurora.utilities.config import config
from aurora.utilities.factory import overrides_embed
from aurora.utilities.utils import create_pagesize_options
class Overrides(ui.View):
def __init__(self, ctx: commands.Context, message: Message, timeout: int | None = None):
def __init__(self, ctx: commands.Context, message: Message, timeout: int = None):
super().__init__()
self.ctx = ctx
self.message = message
self.timeout = timeout
async def on_timeout(self):
try:
await self.message.edit(view=None)
except NotFound:
pass
@ui.button(label="Auto Evidence Format", style=ButtonStyle.green, row=0)
async def auto_evidenceformat(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument

View file

@ -1,71 +0,0 @@
from discord import ButtonStyle, Interaction, Message, ui
from discord.errors import NotFound
from redbot.core import commands
from ..models.type import Type
from ..utilities.config import config
from ..utilities.factory import type_embed
class Types(ui.View):
def __init__(self, ctx: commands.Context, message: Message, moderation_type: Type, timeout: int | None = None):
super().__init__()
self.ctx = ctx
self.message = message
self.type = moderation_type
self.timeout = timeout
async def on_timeout(self):
try:
await self.message.edit(view=None)
except NotFound:
pass
@ui.button(label="Show in History", style=ButtonStyle.green, row=0)
async def show_in_history(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
if not interaction.user.guild_permissions.manage_guild and not interaction.user.guild_permissions.administrator:
await interaction.response.send_message("You must have the manage guild permission to change this setting.", ephemeral=True)
return
await interaction.response.defer()
current_setting = await config.custom("types", interaction.guild.id, self.type.key).show_in_history()
await config.custom("types", interaction.guild.id, self.type.key).show_in_history.set(not current_setting)
await interaction.message.edit(embed=await type_embed(self.ctx, self.type))
@ui.button(label="Show Moderator", style=ButtonStyle.green, row=0)
async def show_moderator(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
if not interaction.user.guild_permissions.manage_guild and not interaction.user.guild_permissions.administrator:
await interaction.response.send_message("You must have the manage guild permission to change this setting.", ephemeral=True)
return
await interaction.response.defer()
current_setting = await config.custom("types", interaction.guild.id, self.type.key).show_moderator()
await config.custom("types", interaction.guild.id, self.type.key).show_moderator.set(not current_setting)
await interaction.message.edit(embed=await type_embed(self.ctx, self.type))
@ui.button(label="Use Discord Permissions", style=ButtonStyle.green, row=0)
async def use_discord_permissions(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
if not interaction.user.guild_permissions.manage_guild and not interaction.user.guild_permissions.administrator:
await interaction.response.send_message("You must have the manage guild permission to change this setting.", ephemeral=True)
return
await interaction.response.defer()
current_setting = await config.custom("types", interaction.guild.id, self.type.key).use_discord_permissions()
await config.custom("types", interaction.guild.id, self.type.key).use_discord_permissions.set(not current_setting)
await interaction.message.edit(embed=await type_embed(self.ctx, self.type))
@ui.button(label="DM Users", style=ButtonStyle.green, row=0)
async def dm_users(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
if not interaction.user.guild_permissions.manage_guild and not interaction.user.guild_permissions.administrator:
await interaction.response.send_message("You must have the manage guild permission to change this setting.", ephemeral=True)
return
await interaction.response.defer()
current_setting = await config.custom("types", interaction.guild.id, self.type.key).dm_users()
await config.custom("types", interaction.guild.id, self.type.key).dm_users.set(not current_setting)
await interaction.message.edit(embed=await type_embed(self.ctx, self.type))
@ui.button(label="Reset", style=ButtonStyle.red, row=1)
async def reset(self, interaction: Interaction, button: ui.Button): # pylint: disable=unused-argument
if not interaction.user.guild_permissions.manage_guild and not interaction.user.guild_permissions.administrator:
await interaction.response.send_message("You must have the manage guild permission to change this setting.", ephemeral=True)
return
await interaction.response.defer()
await config.custom("types", interaction.guild.id, self.type.key).clear()
await interaction.message.edit(embed=await type_embed(self.ctx, self.type))

View file

@ -1,2 +0,0 @@
from .moderation_types import * # noqa: F403
# This just imports all the built-in moderation types so they can be registered, as they aren't imported anywhere else.

View file

@ -1,30 +0,0 @@
from typing import Any, Optional
from discord import Guild
from pydantic import BaseModel, ConfigDict
from redbot.core.bot import Red
class AuroraBaseModel(BaseModel):
"""Base class for all models in Aurora."""
model_config = ConfigDict(ignored_types=(Red,), arbitrary_types_allowed=True)
bot: Red
def dump(self) -> dict:
return self.model_dump(exclude={"bot"})
def to_json(self, indent: int | None = None, file: Any | None = None, **kwargs) -> str:
from ..utilities.json import dump, dumps # pylint: disable=cyclic-import
return dump(self.dump(), file, indent=indent, **kwargs) if file else dumps(self.dump(), indent=indent, **kwargs)
class AuroraGuildModel(AuroraBaseModel):
"""Subclass of AuroraBaseModel that includes a guild_id attribute and a guild attribute."""
model_config = ConfigDict(ignored_types=(Red, Guild), arbitrary_types_allowed=True)
guild_id: int
guild: Optional[Guild] = None
def dump(self) -> dict:
return self.model_dump(exclude={"bot", "guild_id", "guild"})
def __repr__(self) -> str:
return f"<{self.__class__.__name__} guild_id={self.guild_id}>"

View file

@ -1,83 +0,0 @@
import json
from datetime import datetime, timedelta
from typing import Literal, Optional
from redbot.core.bot import Red
from ..utilities.utils import timedelta_from_string
from .base import AuroraBaseModel
from .partials import PartialUser
class Change(AuroraBaseModel):
type: Literal["ORIGINAL", "RESOLVE", "EDIT"]
timestamp: datetime
user_id: int
reason: Optional[str] = None
duration: Optional[timedelta] = None
end_timestamp: Optional[datetime] = None
@property
def unix_timestamp(self) -> int:
return int(self.timestamp.timestamp())
@property
def unix_end_timestamp(self) -> Optional[int]:
if self.end_timestamp:
return int(self.end_timestamp.timestamp())
return None
def __str__(self):
return f"{self.type} {self.user_id} {self.reason}"
def __repr__(self) -> str:
attrs = [
('type', self.type),
('timestamp', self.timestamp),
('user_id', self.user_id),
('reason', self.reason),
('duration', self.duration),
('end_timestamp', self.end_timestamp),
]
joined = ' '.join(f'{key}={value!r}' for key, value in attrs)
return f"<{self.__class__.__name__} {joined}>"
async def get_user(self) -> "PartialUser":
return await PartialUser.from_id(self.bot, self.user_id)
@classmethod
def from_dict(cls, bot: Red, data: dict) -> "Change":
if isinstance(data, str):
data = json.loads(data)
if data.get('duration') and not isinstance(data["duration"], timedelta) and not data["duration"] == "NULL":
duration = timedelta_from_string(data["duration"])
elif data.get('duration') and isinstance(data["duration"], timedelta):
duration = data["duration"]
else:
duration = None
if data.get('end_timestamp') and not isinstance(data["end_timestamp"], datetime):
end_timestamp = datetime.fromtimestamp(data["end_timestamp"])
elif data.get('end_timestamp') and isinstance(data["end_timestamp"], datetime):
end_timestamp = data["end_timestamp"]
else:
end_timestamp = None
if not isinstance(data["timestamp"], datetime):
timestamp = datetime.fromtimestamp(data["timestamp"])
else:
timestamp = data["timestamp"]
try:
data["user_id"] = int(data["user_id"])
except ValueError:
data["user_id"] = 0
data.update({
"timestamp": timestamp,
"end_timestamp": end_timestamp,
"duration": duration
})
if "bot" in data:
del data["bot"]
return cls(bot=bot, **data)

View file

@ -1,565 +0,0 @@
import json
import sqlite3
from datetime import datetime, timedelta
from time import time
from typing import Dict, Iterable, List, Optional, Tuple, Union
import discord
from aiosqlite import Connection, Cursor, OperationalError, Row
from aiosqlite import connect as aiosqlite_connect
from redbot.core import data_manager
from redbot.core.bot import Red
from ..utilities.logger import logger
from ..utilities.utils import timedelta_to_string
from .base import AuroraGuildModel
from .change import Change
from .partials import PartialChannel, PartialRole, PartialUser
from .type import Type, type_registry
class Moderation(AuroraGuildModel):
"""This class represents a moderation case in the database.
Attributes:
bot (Red): The bot instance.
guild (discord.Guild): The guild the case belongs to.
moderation_id (int): The ID of the moderation case.
timestamp (datetime): The timestamp of the case.
moderation_type (Type): The type of moderation case.
target_type (str): The type of target. Should be either `user` or `channel`.
target_id (int): The ID of the target.
moderator_id (int): The ID of the moderator who issued the case.
role_id (int): The ID of the role, if applicable.
duration (timedelta): The duration of the case, if applicable.
end_timestamp (datetime): The end timestamp of the case, if applicable.
reason (str): The reason for the case.
resolved (bool): Whether the case is resolved.
resolved_by (int): The ID of the user who resolved the case.
resolve_reason (str): The reason the case was resolved.
expired (bool): Whether the case is expired.
changes (List[Change]): A list of changes to the case.
metadata (Dict): A dictionary of metadata stored with the case.
Properties:
id (int): The ID of the case.
type (Type): The type of the case.
unix_timestamp (int): The timestamp of the case as a Unix timestamp.
Methods:
get_moderator: Gets the moderator who issued the case.
get_target: Gets the target of the case.
get_resolved_by: Gets the user who resolved the case.
get_role: Gets the role, if applicable.
resolve: Resolves the case.
update: Updates the case in the database.
Class Methods:
from_dict: Creates a `Moderation` object from a dictionary.
from_result: Creates a `Moderation` object from a database result.
execute: Executes a query on the database.
get_latest: Gets the latest cases from the database.
get_next_case_number: Gets the next case number to use.
find_by_id: Finds a case by its ID.
find_by_target: Finds cases by the target.
find_by_moderator: Finds cases by the moderator.
log: Logs a moderation case in the database.
Static Methods:
connect: Connects to the SQLite database.
"""
moderation_id: int
timestamp: datetime
moderation_type: Type
target_type: str
target_id: int
moderator_id: int
role_id: Optional[int] = None
duration: Optional[timedelta] = None
end_timestamp: Optional[datetime] = None
reason: Optional[str] = None
resolved: bool
resolved_by: Optional[int] = None
resolve_reason: Optional[str] = None
expired: bool
changes: List["Change"]
metadata: Dict
@property
def id(self) -> int:
return self.moderation_id
@property
def type(self) -> Type:
return self.moderation_type
@property
def unix_timestamp(self) -> int:
return int(self.timestamp.timestamp())
async def get_moderator(self) -> "PartialUser":
return await PartialUser.from_id(self.bot, self.moderator_id)
async def get_target(self) -> Union["PartialUser", "PartialChannel"]:
if self.target_type.lower() == "user":
return await PartialUser.from_id(self.bot, self.target_id)
return await PartialChannel.from_id(self.bot, self.target_id, self.guild)
async def get_resolved_by(self) -> Optional["PartialUser"]:
if self.resolved_by:
return await PartialUser.from_id(self.bot, self.resolved_by)
return None
async def get_role(self) -> Optional["PartialRole"]:
if self.role_id:
return await PartialRole.from_id(self.bot, self.guild, self.role_id)
return None
def __str__(self) -> str:
return f"{self.moderation_type} {self.target_type} {self.target_id} {self.reason}"
def __int__(self) -> int:
return self.moderation_id
def __repr__(self) -> str:
attrs = [
('guild_id', self.guild_id),
('moderation_id', self.moderation_id),
('timestamp', self.timestamp),
('type', self.type),
('target_type', self.target_type),
('target_id', self.target_id),
('moderator_id', self.moderator_id),
('role_id', self.role_id),
('duration', self.duration),
('end_timestamp', self.end_timestamp),
('reason', self.reason),
('resolved', self.resolved),
('resolved_by', self.resolved_by),
('resolve_reason', self.resolve_reason),
('expired', self.expired),
('changes', self.changes),
('metadata', self.metadata),
]
joined = ' '.join(f'{key}={value!r}' for key, value in attrs)
return f"<{self.__class__.__name__} {joined}>"
async def resolve(self, resolved_by: int, reason: str) -> Tuple[bool, str]:
if self.resolved:
raise ValueError("Case is already resolved!")
self.resolved = True
self.resolved_by = resolved_by
self.resolve_reason = reason
success, msg = await self.type.resolve_handler(moderation=self, reason=reason)
if not self.changes:
self.changes.append(Change.from_dict(self.bot, {
"type": "ORIGINAL",
"timestamp": self.timestamp,
"reason": self.reason,
"user_id": self.moderator_id,
"duration": self.duration,
"end_timestamp": self.end_timestamp,
}))
self.changes.append(Change.from_dict(self.bot, {
"type": "RESOLVE",
"timestamp": datetime.now(),
"reason": reason,
"user_id": resolved_by,
}))
await self.update()
return success, msg
async def update(self) -> None:
from ..utilities.json import dumps
query = f"UPDATE moderation_{self.guild_id} SET timestamp = ?, moderation_type = ?, target_type = ?, moderator_id = ?, role_id = ?, duration = ?, end_timestamp = ?, reason = ?, resolved = ?, resolved_by = ?, resolve_reason = ?, expired = ?, changes = ?, metadata = ? WHERE moderation_id = ?;"
await self.execute(query, (
self.timestamp.timestamp(),
self.moderation_type.key,
self.target_type,
self.moderator_id,
self.role_id,
timedelta_to_string(self.duration) if self.duration else None,
self.end_timestamp.timestamp() if self.end_timestamp else None,
self.reason,
self.resolved,
self.resolved_by,
self.resolve_reason,
self.expired,
dumps(self.changes),
dumps(self.metadata),
self.moderation_id,
))
logger.verbose("Row updated in moderation_%s!\n%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s",
self.moderation_id,
self.guild_id,
self.timestamp.timestamp(),
self.moderation_type.key,
self.target_type,
self.moderator_id,
self.role_id,
timedelta_to_string(self.duration) if self.duration else None,
self.end_timestamp.timestamp() if self.end_timestamp else None,
self.reason,
self.resolved,
self.resolved_by,
self.resolve_reason,
self.expired,
dumps(self.changes),
dumps(self.metadata),
)
@classmethod
async def from_dict(cls, bot: Red, data: dict) -> "Moderation":
if data.get("guild_id"):
try:
guild = bot.get_guild(data["guild_id"])
if not guild:
guild = await bot.fetch_guild(data["guild_id"])
except (discord.Forbidden, discord.HTTPException):
guild = None
data.update({"guild": guild})
return cls(bot=bot, **data)
@classmethod
async def from_result(cls, bot: Red, result: Iterable, guild_id: int) -> "Moderation":
if result[7] is not None and result[7] != "NULL":
try:
hours, minutes, seconds = map(int, result[7].split(':'))
duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
except ValueError as e:
logger.error("Error parsing duration for case %s: %s", result[0], result[7])
raise e
else:
duration = None
if result[14] is not None:
changes = json.loads(result[14])
change_obj_list = []
if changes:
for change in changes:
change_obj_list.append(Change.from_dict(bot=bot, data=change))
if result[15] is not None:
metadata = json.loads(result[15])
else:
metadata = {}
moderation_type = str.lower(result[2])
if moderation_type in type_registry:
moderation_type = type_registry[moderation_type]
else:
logger.error("Unknown moderation type in case %s: %s", result[0], result[2])
case = {
"moderation_id": int(result[0]),
"guild_id": int(guild_id),
"timestamp": datetime.fromtimestamp(result[1]),
"moderation_type": moderation_type,
"target_type": str(result[3]),
"target_id": int(result[4]),
"moderator_id": int(result[5]),
"role_id": int(result[6]) if result[6] is not None else None,
"duration": duration,
"end_timestamp": datetime.fromtimestamp(result[8]) if result[8] is not None else None,
"reason": result[9],
"resolved": bool(result[10]),
"resolved_by": result[11],
"resolve_reason": result[12],
"expired": bool(result[13]),
"changes": change_obj_list,
"metadata": metadata if metadata else {},
}
return await cls.from_dict(bot=bot, data=case)
@staticmethod
async def connect() -> Connection:
"""Connects to the SQLite database, and returns a connection object."""
try:
connection = await aiosqlite_connect(
database=data_manager.cog_data_path(raw_name="Aurora") / "aurora.db"
)
return connection
except OperationalError as e:
logger.error("Unable to access the SQLite database!\nError:\n%s", e.msg)
raise ConnectionRefusedError(
f"Unable to access the SQLite Database!\n{e.msg}"
) from e
@classmethod
async def execute(cls, query: str, parameters: tuple | None = None, bot: Red | None = None, guild_id: int | None = None, cursor: Cursor | None = None, return_obj: bool = True) -> Union[Tuple["Moderation"], Iterable[Row]]:
"""Executes a query on the database.
Arguments:
query (str): The query to execute.
parameters (tuple): The parameters to pass to the query.
bot (Red): The bot instance.
guild_id (int): The ID of the guild to execute the query on.
cursor (Cursor): The cursor to use for the query.
return_obj (bool): Whether to return the case object(s). Defaults to `True`. If `False`, returns a `Iterable` of `aiosqlite.Row` objects.
Returns: The result of the query, either as a `Tuple` of `Moderation` objects or an `Iterable` of `aiosqlite.Row` objects.
"""
logger.trace("Executing query: \"%s\" with parameters \"%s\"", query, parameters)
if not parameters:
parameters = ()
if not cursor:
no_cursor = True
database = await cls.connect()
cursor = await database.cursor()
else:
no_cursor = False
try:
await cursor.execute(query, parameters)
except OperationalError as e:
logger.error("Error executing query: \"%s\" with parameters \"%s\"\nError:\n%s",
query, parameters, e)
raise OperationalError(f"Error executing query: \"{query}\" with parameters \"{parameters}\"") from e
results = await cursor.fetchall()
await database.commit()
if no_cursor:
await cursor.close()
await database.close()
if results and return_obj and bot and guild_id:
cases = []
for result in results:
if result[0] == 0:
continue
case = await cls.from_result(bot=bot, result=result, guild_id=guild_id)
cases.append(case)
return tuple(cases)
return results
@classmethod
async def get_latest(cls, bot: Red, guild_id: int, before: datetime | None = None, after: datetime | None = None, limit: int | None = None, offset: int = 0, types: Iterable[Type] | None = None, expired: bool | None = None, cursor: Cursor | None = None) -> Tuple["Moderation"]:
params = []
query = f"SELECT * FROM moderation_{guild_id}"
conditions = []
if types:
conditions.append(f"moderation_type IN ({', '.join(['?' for _ in types])})")
params.extend([t.key for t in types])
if before:
conditions.append("timestamp < ?")
params.append(int(before.timestamp()))
if after:
conditions.append("timestamp > ?")
params.append(int(after.timestamp()))
if expired is not None:
conditions.append("expired = ?")
params.append(int(expired))
if conditions:
query += " WHERE " + " AND ".join(conditions)
query += " ORDER BY moderation_id DESC"
if limit:
query += " LIMIT ? OFFSET ?"
params.extend((limit, offset))
query += ";"
return await cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=tuple(params) if params else (), cursor=cursor)
@classmethod
async def get_next_case_number(cls, bot: Red, guild_id: int, cursor: Cursor | None = None) -> int:
result = await cls.get_latest(bot=bot, guild_id=guild_id, cursor=cursor, limit=1)
return (result[0].moderation_id + 1) if result else 1
@classmethod
async def find_by_id(cls, bot: Red, moderation_id: int, guild_id: int, cursor: Cursor | None = None) -> "Moderation":
query = f"SELECT * FROM moderation_{guild_id} WHERE moderation_id = ?;"
case = await cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=(moderation_id,), cursor=cursor)
if case:
return case[0]
raise ValueError(f"Case {moderation_id} not found in moderation_{guild_id}!")
@classmethod
async def find_by_target(cls, bot: Red, guild_id: int, target: int, before: datetime = None, after: datetime = None, types: Iterable[Type] | None = None, expired: bool | None = None, cursor: Cursor | None = None) -> Tuple["Moderation"]:
query = f"SELECT * FROM moderation_{guild_id} WHERE target_id = ?"
params = [target]
if types:
query += f" AND moderation_type IN ({', '.join(['?' for _ in types])})"
for t in types:
params.append(t.key)
if before:
query += " AND timestamp < ?"
params.append(int(before.timestamp()))
if after:
query += " AND timestamp > ?"
params.append(int(after.timestamp()))
if expired is not None:
query += " AND expired = ?"
params.append(int(expired))
query += " ORDER BY moderation_id DESC;"
return await cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=params, cursor=cursor)
@classmethod
async def find_by_moderator(cls, bot: Red, guild_id: int, moderator: int, before: datetime = None, after: datetime = None, types: Iterable[Type] | None = None, expired: bool | None = None, cursor: Cursor | None = None) -> Tuple["Moderation"]:
query = f"SELECT * FROM moderation_{guild_id} WHERE moderator_id = ?"
params = [moderator]
if types:
query += f" AND moderation_type IN ({', '.join(['?' for _ in types])})"
for t in types:
params.append(t.key)
if before:
query += " AND timestamp < ?"
params.append(int(before.timestamp()))
if after:
query += " AND timestamp > ?"
params.append(int(after.timestamp()))
if expired is not None:
query += " AND expired = ?"
params.append(int(expired))
query += " ORDER BY moderation_id DESC;"
return await cls.execute(bot=bot, guild_id=guild_id, query=query, parameters=params, cursor=cursor)
@classmethod
async def log(
cls,
bot: Red,
guild_id: int,
moderator_id: int,
moderation_type: Type,
target_type: str,
target_id: int,
role_id: int | None = None,
duration: timedelta | None = None,
reason: str | None = None,
database: sqlite3.Connection | None = None,
timestamp: datetime | None = None,
resolved: bool = False,
resolved_by: int | None = None,
resolved_reason: str | None = None,
expired: bool | None = None,
changes: list | None = None,
metadata: dict | None = None,
return_obj: bool = True,
) -> Union["Moderation", int]:
"""Logs a moderation case in the database.
Args:
bot (Red): The bot instance.
guild_id (int): The ID of the guild to log the case in.
moderator_id (int): The ID of the moderator who issued the case.
moderation_type (Type): The type of moderation case. See `aurora.models.moderation_types` for the built-in options.
target_type (str): The type of target. Should be either `user` or `channel`.
target_id (int): The ID of the target.
role_id (int): The ID of the role, if applicable.
duration (timedelta): The duration of the case, if applicable.
reason (str): The reason for the case.
database (sqlite3.Connection): The database connection to use to log the case. A connection will be automatically created if not provided.
timestamp (datetime): The timestamp of the case. Will be automatically generated if not provided.
resolved (bool): Whether the case is resolved.
resolved_by (int): The ID of the user who resolved the case.
resolved_reason (str): The reason the case was resolved.
expired (bool): Whether the case is expired.
changes (list): A list of changes to log. You usually shouldn't pass this, as it's automatically generated by the `/edit` and `/resolve` commands.
metadata (dict): A dictionary of metadata to store with the case.
return_obj (bool): Whether to return the case object. Defaults to `True`. If `False`, returns the case ID.
Returns:
Union[Moderation, int]: The `Moderation` object if `return_obj` is `True`, otherwise the case ID.
"""
from ..utilities.json import dumps
if not timestamp:
timestamp = datetime.fromtimestamp(time())
elif not isinstance(timestamp, datetime):
timestamp = datetime.fromtimestamp(timestamp)
if duration == "NULL":
duration = None
if duration is not None:
end_timestamp = timestamp + duration
else:
duration = None
end_timestamp = None
if not expired:
if end_timestamp:
expired = bool(timestamp > end_timestamp)
else:
expired = False
if reason == "NULL":
reason = None
if resolved_by in ["NULL", "?"]:
resolved_by = None
if resolved_reason == "NULL":
resolved_reason = None
if role_id == 0:
role_id = None
if not database:
database = await cls.connect()
close_db = True
else:
close_db = False
moderation_id = await cls.get_next_case_number(bot=bot, guild_id=guild_id)
case = {
"moderation_id": moderation_id,
"timestamp": timestamp.timestamp(),
"moderation_type": moderation_type.key,
"target_type": target_type,
"target_id": target_id,
"moderator_id": moderator_id,
"role_id": role_id,
"duration": timedelta_to_string(duration) if duration else None,
"end_timestamp": end_timestamp.timestamp() if end_timestamp else None,
"reason": reason,
"resolved": resolved,
"resolved_by": resolved_by,
"resolve_reason": resolved_reason,
"expired": expired,
"changes": dumps(changes),
"metadata": dumps(metadata)
}
sql = f"INSERT INTO `moderation_{guild_id}` (moderation_id, timestamp, moderation_type, target_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
await database.execute(sql, tuple(case.values()))
await database.commit()
if close_db:
await database.close()
logger.verbose(
"Row inserted into moderation_%s!\n%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s",
guild_id,
case["moderation_id"],
case["timestamp"],
case["moderation_type"],
case["target_type"],
case["target_id"],
case["moderator_id"],
case["role_id"],
case["duration"],
case["end_timestamp"],
case["reason"],
case["resolved"],
case["resolved_by"],
case["resolve_reason"],
case["expired"],
case["changes"],
case["metadata"],
)
if return_obj:
return await cls.find_by_id(bot=bot, moderation_id=moderation_id, guild_id=guild_id)
return moderation_id

File diff suppressed because it is too large Load diff

View file

@ -1,89 +0,0 @@
from discord import ChannelType, Forbidden, Guild, HTTPException, InvalidData, NotFound, Role, User
from discord.abc import Messageable
from redbot.core.bot import Red
from .base import AuroraBaseModel, AuroraGuildModel
class PartialUser(AuroraBaseModel):
id: int
username: str
discriminator: int
_obj: User | None
@property
def name(self):
return f"{self.username}#{self.discriminator}" if self.discriminator != 0 else self.username
def __str__(self):
return self.name
def __repr__(self):
return f"<{self.__class__.__name__} id={self.id}>"
@classmethod
async def from_id(cls, bot: Red, user_id: int) -> "PartialUser":
user = bot.get_user(user_id)
if not user:
try:
user = await bot.fetch_user(user_id)
return cls(bot=bot, id=user.id, username=user.name, discriminator=user.discriminator, _obj=user)
except NotFound:
return cls(bot=bot, id=user_id, username="Deleted User", discriminator=0, _obj=None)
return cls(bot=bot, id=user.id, username=user.name, discriminator=user.discriminator, _obj=user)
class PartialChannel(AuroraGuildModel):
id: int
name: str
type: ChannelType
_obj: Messageable | None
@property
def mention(self):
if self.name in ["Deleted Channel", "Forbidden Channel"]:
return self.name
return f"<#{self.id}>"
def __str__(self):
return self.mention
def __repr__(self):
return f"<{self.__class__.__name__} id={self.id} guild_id={self.guild_id}>"
@classmethod
async def from_id(cls, bot: Red, channel_id: int, guild: Guild) -> "PartialChannel":
channel = bot.get_channel(channel_id)
if not channel:
try:
channel = await bot.fetch_channel(channel_id)
return cls(bot=bot, guild_id=channel.guild.id, guild=guild, id=channel.id, name=channel.name, type=channel.type, _obj=channel)
except (NotFound, InvalidData, HTTPException, Forbidden) as e:
if e == Forbidden:
return cls(bot=bot, guild_id=0, id=channel_id, name="Forbidden Channel")
return cls(bot=bot, guild_id=0, id=channel_id, name="Deleted Channel", type=ChannelType.text, _obj=None)
return cls(bot=bot, guild_id=channel.guild.id, guild=guild, id=channel.id, name=channel.name, type=channel.type, _obj=channel)
class PartialRole(AuroraGuildModel):
id: int
name: str
_obj: Role | None
@property
def mention(self):
if self.name in ["Deleted Role", "Forbidden Role"]:
return self.name
return f"<@&{self.id}>"
def __str__(self):
return self.mention
def __repr__(self) -> str:
return f"<{self.__class__.__name__} id={self.id} guild_id={self.guild_id}>"
@classmethod
async def from_id(cls, bot: Red, guild: Guild, role_id: int) -> "PartialRole":
role = guild.get_role(role_id)
if not role:
return cls(bot=bot, guild_id=guild.id, id=role_id, name="Deleted Role", _obj=None)
return cls(bot=bot, guild_id=guild.id, id=role.id, name=role.name, _obj=role)

View file

@ -1,97 +0,0 @@
from abc import ABC, abstractmethod
from typing import Any, Dict, Tuple
from class_registry import ClassRegistry
from class_registry.base import AutoRegister
from discord import Interaction, Member, User
from discord.abc import Messageable
from redbot.core import commands
type_registry: Dict['str', 'Type'] = ClassRegistry(attr_name='key', unique=True)
class Type(AutoRegister(type_registry), ABC):
"""This is a base class for moderation types.
Attributes:
key (str): The key to use for this type. This should be unique, as this is how the type is registered internally. Changing this key will break existing cases with this type.
string (str): The string to display for this type.
verb (str): The verb to use for this type.
embed_desc (str): The string to use for embed descriptions.
channel (bool): Whether this type targets channels or users. If this is `true` in a subclass, its overridden handler methods should be typed with `discord.abc.Messageable` instead of `discord.Member | discord.User`.
removes_from_guild (bool): Whether this type's handler removes the target from the guild, or if the moderation is expected to occur whenever the user is not in the guild. This does not actually remove the target from the guild, the handler method is responsible for that.
Properties:
name (str): The string to display for this type. This is the same as the `string` attribute.
"""
key = "type"
string = "type"
verb = "typed"
embed_desc = "been "
channel = False
removes_from_guild = False
@abstractmethod
def void(self) -> Any:
"""This method should be overridden by any child classes. This is a placeholder to allow for automatic class registration."""
raise NotImplementedError
@property
def name(self) -> str:
"""Alias for the `string` attribute."""
return self.string
def __str__(self) -> str:
return self.string
def __repr__(self) -> str:
attrs = [
('key', self.key),
('channel', self.channel),
]
joined = ' '.join(f'{key}={value!r}' for key, value in attrs)
return f"<{self.__class__.__name__} {joined}>"
@classmethod
async def handler(cls, ctx: commands.Context, target: Member | User | Messageable, 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.
Arguments:
ctx (commands.Context): The context of the command.
target (discord.Member | discord.User | discord.abc.Messageable): The target of the moderation.
silent (bool): Whether details about the moderation should be DM'ed to the target of the moderation.
"""
raise NotImplementedError
@classmethod
async def resolve_handler(cls, moderation, reason: str) -> Tuple[bool, str]: # pylint: disable=unused-argument
"""This method should be overridden by any resolvable child classes, but should retain the same keyword arguments.
If your moderation type should not be resolvable, do not override this.
Arguments:
moderation (aurora.models.Moderation): The moderation to resolve.
reason (str): The reason for resolving the moderation.
"""
raise NotImplementedError
@classmethod
async def expiry_handler(cls, moderation) -> int: # pylint: disable=unused-argument
"""This method should be overridden by any expirable child classes, but should retain the same keyword arguments and return an integer.
If your moderation type should not expire, do not override this, but also do not set an `end_timestamp` when you log your moderation.
Arguments:
moderation (aurora.models.Moderation): The moderation that is expiring.
"""
raise NotImplementedError
@classmethod
async def duration_edit_handler(cls, interaction: Interaction, old_moderation, new_moderation) -> bool: # pylint: disable=unused-argument
"""This method should be overridden by any child classes with editable durations, but should retain the same keyword arguments and should return True if the duration was successfully modified, or False if it was not.
If your moderation type's duration should not be editable, do not override this.
Arguments:
interaction (discord.Interaction): The interaction that triggered the duration edit.
old_moderation (aurora.models.Moderation): The old moderation, from before the `/edit` command was invoked.
new_moderation (aurora.models.Moderation): The current state of the moderation.
"""
raise NotImplementedError

View file

@ -27,13 +27,3 @@ def register_config(config_obj: Config):
history_inline_pagesize=None,
auto_evidenceformat=None,
)
moderation_type = {
"show_in_history": True,
"show_moderator": None,
"use_discord_permissions": None,
"dm_users": None,
}
config_obj.init_custom("types", 2)
config_obj.register_custom("types", **moderation_type)

View file

@ -0,0 +1,219 @@
# pylint: disable=cyclic-import
import json
import sqlite3
import time
from datetime import datetime, timedelta
from discord import Guild
from redbot.core import data_manager
from .logger import logger
from .utils import (convert_timedelta_to_str, generate_dict,
get_next_case_number)
def connect() -> sqlite3.Connection:
"""Connects to the SQLite database, and returns a connection object."""
try:
connection = sqlite3.connect(
database=data_manager.cog_data_path(raw_name="Aurora") / "aurora.db"
)
return connection
except sqlite3.OperationalError as e:
logger.error("Unable to access the SQLite database!\nError:\n%s", e.msg)
raise ConnectionRefusedError(
f"Unable to access the SQLite Database!\n{e.msg}"
) from e
async def create_guild_table(guild: Guild):
database = connect()
cursor = database.cursor()
try:
cursor.execute(f"SELECT * FROM `moderation_{guild.id}`")
logger.debug("SQLite Table exists for server %s (%s)", guild.name, guild.id)
except sqlite3.OperationalError:
query = f"""
CREATE TABLE `moderation_{guild.id}` (
moderation_id INTEGER PRIMARY KEY,
timestamp INTEGER NOT NULL,
moderation_type TEXT NOT NULL,
target_type TEXT NOT NULL,
target_id TEXT NOT NULL,
moderator_id TEXT NOT NULL,
role_id TEXT,
duration TEXT,
end_timestamp INTEGER,
reason TEXT,
resolved INTEGER NOT NULL,
resolved_by TEXT,
resolve_reason TEXT,
expired INTEGER NOT NULL,
changes TEXT NOT NULL,
metadata TEXT NOT NULL
)
"""
cursor.execute(query)
index_query_1 = f"CREATE INDEX IF NOT EXISTS idx_target_id ON moderation_{guild.id}(target_id);"
cursor.execute(index_query_1)
index_query_2 = f"CREATE INDEX IF NOT EXISTS idx_moderator_id ON moderation_{guild.id}(moderator_id);"
cursor.execute(index_query_2)
index_query_3 = f"CREATE INDEX IF NOT EXISTS idx_moderation_id ON moderation_{guild.id}(moderation_id);"
cursor.execute(index_query_3)
insert_query = f"""
INSERT INTO `moderation_{guild.id}`
(moderation_id, timestamp, moderation_type, target_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""
insert_values = (
0,
0,
"NULL",
"NULL",
0,
0,
0,
"NULL",
0,
"NULL",
0,
"NULL",
"NULL",
0,
json.dumps([]),
json.dumps({}),
)
cursor.execute(insert_query, insert_values)
database.commit()
logger.debug(
"SQLite Table (moderation_%s) created for %s (%s)",
guild.id,
guild.name,
guild.id,
)
database.close()
async def mysql_log(
guild_id: str,
author_id: str,
moderation_type: str,
target_type: str,
target_id: int,
role_id: int,
duration: timedelta,
reason: str,
database: sqlite3.Connection = None,
timestamp: int = None,
resolved: bool = False,
resolved_by: str = None,
resolved_reason: str = None,
expired: bool = None,
changes: list = None,
metadata: dict = None,
) -> int:
if not timestamp:
timestamp = int(time.time())
if duration != "NULL":
end_timedelta = datetime.fromtimestamp(timestamp) + duration
end_timestamp = int(end_timedelta.timestamp())
duration = convert_timedelta_to_str(duration)
else:
end_timestamp = 0
if not expired:
if int(time.time()) > end_timestamp:
expired = 1
else:
expired = 0
if resolved_by is None:
resolved_by = "NULL"
if resolved_reason is None:
resolved_reason = "NULL"
if not database:
database = connect()
close_db = True
else:
close_db = False
cursor = database.cursor()
moderation_id = await get_next_case_number(guild_id=guild_id, cursor=cursor)
sql = f"INSERT INTO `moderation_{guild_id}` (moderation_id, timestamp, moderation_type, target_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
val = (
moderation_id,
timestamp,
moderation_type,
target_type,
target_id,
author_id,
role_id,
duration,
end_timestamp,
reason,
int(resolved),
resolved_by,
resolved_reason,
expired,
json.dumps(changes if changes else []),
json.dumps(metadata if metadata else {}),
)
cursor.execute(sql, val)
cursor.close()
database.commit()
if close_db:
database.close()
logger.debug(
"Row inserted into moderation_%s!\n%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s",
guild_id,
moderation_id,
timestamp,
moderation_type,
target_type,
target_id,
author_id,
role_id,
duration,
end_timestamp,
reason,
int(resolved),
resolved_by,
resolved_reason,
expired,
changes,
metadata,
)
return moderation_id
async def fetch_case(moderation_id: int, guild_id: str) -> dict:
"""This method fetches a case from the database and returns the case's dictionary."""
database = connect()
cursor = database.cursor()
query = f"SELECT * FROM moderation_{guild_id} WHERE moderation_id = ?;"
cursor.execute(query, (moderation_id,))
result = cursor.fetchone()
cursor.close()
database.close()
return generate_dict(result)

View file

@ -2,68 +2,78 @@
from datetime import datetime, timedelta
from typing import Union
from discord import Color, Embed, Guild, Interaction, Member, Message, Role, User
from discord import Color, Embed, Guild, Interaction, InteractionMessage, Member, Role, User
from redbot.core import commands
from redbot.core.bot import Red
from redbot.core.utils.chat_formatting import bold, box, error, humanize_timedelta, inline, warning
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
from aurora.utilities.config import config
from aurora.utilities.utils import fetch_channel_dict, fetch_user_dict, get_bool_emoji, get_next_case_number, get_pagesize_str
async def message_factory(
bot: Red,
color: Color,
guild: Guild,
reason: str,
moderation_type: Type,
moderator: Union[Member, User] | None = None,
duration: timedelta | None = None,
response: Message | None = None,
case: bool = True,
moderation_type: str,
moderator: Union[Member, User] = None,
duration: timedelta = None,
response: InteractionMessage = None,
role: Role = None,
) -> Embed:
"""This function creates a message from set parameters, meant for contacting the moderated user.
Args:
bot (Red): The bot instance.
color (Color): The color of the embed.
guild (Guild): The guild the moderation occurred in.
reason (str): The reason for the moderation.
moderation_type (Type): The type of moderation.
moderation_type (str): 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 (Message, optional): The response message. Defaults to None.
case (bool, optional): Whether the message is for a moderation case. Defaults to True.
response (InteractionMessage, 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 not moderation_type.removes_from_guild:
if response is not None and moderation_type not in [
"kicked",
"banned",
"tempbanned",
"unbanned",
]:
guild_name = f"[{guild.name}]({response.jump_url})"
else:
guild_name = guild.name
if duration:
title = moderation_type
if moderation_type in ["tempbanned", "muted"] and 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"
embed = Embed(
title=str.title(moderation_type.verb),
description=f"You have {moderation_type.embed_desc}{moderation_type.verb}{embed_duration} in {guild_name}.",
title=str.title(title),
description=f"You have {embed_desc} {moderation_type}{embed_duration} in {guild_name}.",
color=color,
timestamp=datetime.now(),
)
show_moderator = await config.custom("types", guild.id, moderation_type.key).show_moderator()
if show_moderator is None:
show_moderator = await config.guild(guild).show_moderator()
if show_moderator and moderator is not None:
if await config.guild(guild).show_moderator() and moderator is not None:
embed.add_field(
name="Moderator", value=f"`{moderator.name} ({moderator.id})`", inline=False
)
@ -75,211 +85,265 @@ async def message_factory(
else:
embed.set_author(name=guild.name)
if case:
embed.set_footer(
text=f"Case #{await Moderation.get_next_case_number(bot=bot, guild_id=guild.id):,}",
text=f"Case #{await get_next_case_number(guild.id):,}",
icon_url="attachment://arrow.png",
)
return embed
async def resolve_factory(moderation: Moderation, reason: str) -> Embed:
"""This function creates a resolved embed from set parameters, meant for contacting the moderated user.
Args:
moderation (aurora.models.Moderation): The moderation object.
reason (str): The reason for resolving the moderation.
Returns: `discord.Embed`
"""
embed = Embed(
title=str.title(moderation.type.name) + " Resolved",
description=f"Your {moderation.type.name} in {moderation.guild.name} has been resolved.",
color=await moderation.bot.get_embed_color(moderation.guild.channels[0]),
timestamp=datetime.now(),
)
embed.add_field(name="Reason", value=f"`{reason}`", inline=False)
if moderation.guild.icon.url is not None:
embed.set_author(name=moderation.guild.name, icon_url=moderation.guild.icon.url)
else:
embed.set_author(name=moderation.guild.name)
embed.set_footer(
text=f"Case #{moderation.id:,}",
icon_url="attachment://arrow.png",
)
return embed
async def log_factory(
ctx: commands.Context, moderation: Moderation, resolved: bool = False
interaction: Interaction, case_dict: dict, resolved: bool = False
) -> Embed:
"""This function creates a log embed from set parameters, meant for moderation logging.
Args:
ctx (commands.Context): The ctx object.
moderation (aurora.models.Moderation): The moderation object.
interaction (Interaction): The interaction object.
case_dict (dict): The case dictionary.
resolved (bool, optional): Whether the case is resolved or not. Defaults to False.
"""
target = await moderation.get_target()
moderator = await moderation.get_moderator()
if resolved:
embed = Embed(
title=f"📕 Case #{moderation.id:,} Resolved",
color=await ctx.bot.get_embed_color(ctx.channel),
if case_dict["target_type"] == "USER":
target_user = await fetch_user_dict(interaction.client, case_dict["target_id"])
target_name = (
f"`{target_user['name']}`"
if target_user["discriminator"] == "0"
else f"`{target_user['name']}#{target_user['discriminator']}`"
)
elif case_dict["target_type"] == "CHANNEL":
target_user = await fetch_channel_dict(interaction.guild, case_dict["target_id"])
if target_user["mention"]:
target_name = f"{target_user['mention']}"
else:
target_name = f"`{target_user['name']}`"
moderator_user = await fetch_user_dict(interaction.client, case_dict["moderator_id"])
moderator_name = (
f"`{moderator_user['name']}`"
if moderator_user["discriminator"] == "0"
else f"`{moderator_user['name']}#{moderator_user['discriminator']}`"
)
resolved_by = await moderation.get_resolved_by()
embed.description = f"**Type:** {str.title(moderation.type.string)}\n**Target:** {target.name} ({target.id})\n**Moderator:** {moderator.name} ({moderator.id})\n**Timestamp:** <t:{moderation.unix_timestamp}> | <t:{moderation.unix_timestamp}:R>"
embed = Embed(
title=f"📕 Case #{case_dict['moderation_id']:,} Resolved",
color=await interaction.client.get_embed_color(interaction.channel),
)
if moderation.duration is not 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_timedelta(timedelta=moderation.duration)} | <t:{moderation.end_timestamp}:R>"
if not moderation.expired
else str(humanize_timedelta(timedelta=moderation.duration))
f"{humanize_timedelta(timedelta=td)} | <t:{case_dict['end_timestamp']}:R>"
if case_dict["expired"] == "0"
else str(humanize_timedelta(timedelta=td))
)
embed.description = (
embed.description
+ f"\n**Duration:** {duration_embed}\n**Expired:** {moderation.expired}"
+ f"\n**Duration:** {duration_embed}\n**Expired:** {bool(case_dict['expired'])}"
)
if moderation.metadata.items():
for key, value in moderation.metadata.items():
embed.description += f"\n**{key.title()}:** {value}"
embed.add_field(name="Reason", value=box(moderation.reason), inline=False)
embed.add_field(name="Reason", value=box(case_dict["reason"]), inline=False)
resolved_user = await fetch_user_dict(interaction.client, 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_by.name}` ({resolved_by.id}) for:\n"
+ box(moderation.resolve_reason),
value=f"Resolved by `{resolved_name}` ({resolved_user['id']}) for:\n"
+ box(case_dict["resolve_reason"]),
inline=False,
)
else:
embed = Embed(
title=f"📕 Case #{moderation.id:,}",
color=await ctx.bot.get_embed_color(ctx.channel),
if case_dict["target_type"] == "USER":
target_user = await fetch_user_dict(interaction.client, case_dict["target_id"])
target_name = (
f"`{target_user['name']}`"
if target_user["discriminator"] == "0"
else f"`{target_user['name']}#{target_user['discriminator']}`"
)
embed.description = f"**Type:** {str.title(moderation.type.string)}\n**Target:** {target.name} ({target.id})\n**Moderator:** {moderator.name} ({moderator.id})\n**Timestamp:** <t:{moderation.unix_timestamp}> | <t:{moderation.unix_timestamp}:R>"
elif case_dict["target_type"] == "CHANNEL":
target_user = await fetch_channel_dict(interaction.guild, case_dict["target_id"])
if target_user["mention"]:
target_name = target_user["mention"]
else:
target_name = f"`{target_user['name']}`"
if moderation.duration:
moderator_user = await fetch_user_dict(interaction.client, case_dict["moderator_id"])
moderator_name = (
f"`{moderator_user['name']}`"
if moderator_user["discriminator"] == "0"
else f"`{moderator_user['name']}#{moderator_user['discriminator']}`"
)
embed = Embed(
title=f"📕 Case #{case_dict['moderation_id']:,}",
color=await interaction.client.get_embed_color(interaction.channel),
)
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_timedelta(timedelta=moderation.duration)} | <t:{moderation.unix_timestamp}:R>"
+ f"\n**Duration:** {humanize_timedelta(timedelta=td)} | <t:{case_dict['end_timestamp']}:R>"
)
if moderation.metadata.items():
for key, value in moderation.metadata.items():
embed.description += f"\n**{key.title()}:** {value}"
embed.add_field(name="Reason", value=box(moderation.reason), inline=False)
embed.add_field(name="Reason", value=box(case_dict["reason"]), inline=False)
return embed
async def case_factory(interaction: Interaction, moderation: Moderation) -> Embed:
async def case_factory(interaction: Interaction, case_dict: dict) -> Embed:
"""This function creates a case embed from set parameters.
Args:
interaction (discord.Interaction): The interaction object.
moderation (aurora.models.Moderation): The moderation object.
interaction (Interaction): The interaction object.
case_dict (dict): The case dictionary.
"""
target = await moderation.get_target()
moderator = await moderation.get_moderator()
if case_dict["target_type"] == "USER":
target_user = await fetch_user_dict(interaction.client, case_dict["target_id"])
target_name = (
f"`{target_user['name']}`"
if target_user["discriminator"] == "0"
else f"`{target_user['name']}#{target_user['discriminator']}`"
)
elif case_dict["target_type"] == "CHANNEL":
target_user = await fetch_channel_dict(interaction.guild, case_dict["target_id"])
if target_user["mention"]:
target_name = f"{target_user['mention']}"
else:
target_name = f"`{target_user['name']}`"
moderator_user = await fetch_user_dict(interaction.client, case_dict["moderator_id"])
moderator_name = (
f"`{moderator_user['name']}`"
if moderator_user["discriminator"] == "0"
else f"`{moderator_user['name']}#{moderator_user['discriminator']}`"
)
embed = Embed(
title=f"📕 Case #{moderation.id:,}",
title=f"📕 Case #{case_dict['moderation_id']:,}",
color=await interaction.client.get_embed_color(interaction.channel),
)
embed.description = f"**Type:** {str.title(moderation.type.string)}\n**Target:** `{target.name}` ({target.id})\n**Moderator:** `{moderator.name}` ({moderator.id})\n**Resolved:** {moderation.resolved}\n**Timestamp:** <t:{moderation.unix_timestamp}> | <t:{moderation.unix_timestamp}:R>"
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 moderation.duration:
duration_embed = (
f"{humanize_timedelta(timedelta=moderation.duration)} | <t:{moderation.unix_timestamp}:R>"
if moderation.expired is False
else str(humanize_timedelta(timedelta=moderation.duration))
if case_dict["duration"] != "NULL":
td = timedelta(
**{
unit: int(val)
for unit, val in zip(
["hours", "minutes", "seconds"], case_dict["duration"].split(":")
)
embed.description += f"\n**Duration:** {duration_embed}\n**Expired:** {moderation.expired}"
}
)
duration_embed = (
f"{humanize_timedelta(timedelta=td)} | <t:{case_dict['end_timestamp']}:R>"
if bool(case_dict["expired"]) is False
else str(humanize_timedelta(timedelta=td))
)
embed.description += f"\n**Duration:** {duration_embed}\n**Expired:** {bool(case_dict['expired'])}"
embed.description += (
f"\n**Changes:** {len(moderation.changes) - 1}"
if moderation.changes
f"\n**Changes:** {len(case_dict['changes']) - 1}"
if case_dict["changes"]
else "\n**Changes:** 0"
)
if moderation.role_id:
role = await moderation.get_role()
embed.description += f"\n**Role:** {role.name}"
if case_dict["role_id"]:
embed.description += f"\n**Role:** <@&{case_dict['role_id']}>"
if moderation.metadata:
if moderation.metadata.get("imported_from"):
if case_dict["metadata"]:
if case_dict["metadata"]["imported_from"]:
embed.description += (
f"\n**Imported From:** {moderation.metadata['imported_from']}"
f"\n**Imported From:** {case_dict['metadata']['imported_from']}"
)
moderation.metadata.pop("imported_from")
if moderation.metadata.get("imported_timestamp"):
embed.description += (
f"\n**Imported Timestamp:** <t:{moderation.metadata['imported_timestamp']}> | <t:{moderation.metadata['imported_timestamp']}:R>"
embed.add_field(name="Reason", value=box(case_dict["reason"]), inline=False)
if case_dict["resolved"] == 1:
resolved_user = await fetch_user_dict(interaction.client, case_dict["resolved_by"])
resolved_name = (
f"`{resolved_user['name']}`"
if resolved_user["discriminator"] == "0"
else f"`{resolved_user['name']}#{resolved_user['discriminator']}`"
)
moderation.metadata.pop("imported_timestamp")
if moderation.metadata.items():
for key, value in moderation.metadata.items():
embed.description += f"\n**{key.title()}:** {value}"
embed.add_field(name="Reason", value=box(moderation.reason), inline=False)
if moderation.resolved:
resolved_user = await moderation.get_resolved_by()
if not resolved_user:
resolved_user = PartialUser(bot=interaction.client, id=0, username="Deleted User", discriminator="0")
embed.add_field(
name="Resolve Reason",
value=f"Resolved by `{resolved_user.name or 'Deleted User'}` ({resolved_user.id or '0'}) for:\n{box(moderation.resolve_reason)}",
value=f"Resolved by {resolved_name} ({resolved_user['id']}) for:\n{box(case_dict['resolve_reason'])}",
inline=False,
)
return embed
async def changes_factory(interaction: Interaction, moderation: Moderation) -> Embed:
async def changes_factory(interaction: Interaction, case_dict: dict) -> Embed:
"""This function creates a changes embed from set parameters.
Args:
interaction (discord.Interaction): The interaction object.
moderation (aurora.models.Moderation): The moderation object.
interaction (Interaction): The interaction object.
case_dict (dict): The case dictionary.
"""
embed = Embed(
title=f"📕 Case #{moderation.id:,} Changes",
title=f"📕 Case #{case_dict['moderation_id']:,} Changes",
color=await interaction.client.get_embed_color(interaction.channel),
)
memory_dict = {}
if moderation.changes:
for change in moderation.changes:
if change.user_id not in memory_dict:
memory_dict[str(change.user_id)] = await change.get_user()
if case_dict["changes"]:
for change in case_dict["changes"]:
if change["user_id"] not in memory_dict:
memory_dict[str(change["user_id"])] = await fetch_user_dict(
interaction.client, change["user_id"]
)
user: PartialUser = memory_dict[str(change.user_id)]
user = memory_dict[str(change["user_id"])]
name = (
user["name"]
if user["discriminator"] == "0"
else f"{user['name']}#{user['discriminator']}"
)
timestamp = f"<t:{change.unix_timestamp}> | <t:{change.unix_timestamp}:R>"
end_timestamp = f"<t:{change.unix_end_timestamp}> | <t:{change.unix_end_timestamp}:R>" if change.end_timestamp else None
change_str = [
f"{bold('User:')} {inline(user.name)} ({user.id})",
f"{bold('Reason:')} {change.reason}" if change.reason else "",
f"{bold('Duration:')} {humanize_timedelta(timedelta=change.duration)}" if change.duration else "",
f"{bold('End Timestamp:')} {end_timestamp}" if end_timestamp else "",
f"{bold('Timestamp')} {timestamp}",
]
copy = change_str.copy()
for string in change_str:
if string == "":
copy.remove(string)
timestamp = f"<t:{change['timestamp']}> | <t:{change['timestamp']}:R>"
if change["type"] == "ORIGINAL":
embed.add_field(
name=change.type.title(),
value="\n".join(copy),
name="Original",
value=f"**User:** `{name}` ({user['id']})\n**Reason:** {change['reason']}\n**Timestamp:** {timestamp}",
inline=False,
)
elif change["type"] == "EDIT":
embed.add_field(
name="Edit",
value=f"**User:** `{name}` ({user['id']})\n**Reason:** {change['reason']}\n**Timestamp:** {timestamp}",
inline=False,
)
elif change["type"] == "RESOLVE":
embed.add_field(
name="Resolve",
value=f"**User:** `{name}` ({user['id']})\n**Reason:** {change['reason']}\n**Timestamp:** {timestamp}",
inline=False,
)
@ -289,29 +353,44 @@ async def changes_factory(interaction: Interaction, moderation: Moderation) -> E
return embed
async def evidenceformat_factory(moderation: Moderation) -> str:
async def evidenceformat_factory(interaction: Interaction, case_dict: dict) -> str:
"""This function creates a codeblock in evidence format from set parameters.
Args:
interaction (discord.Interaction): The interaction object.
moderation (aurora.models.Moderation): The moderation object.
interaction (Interaction): The interaction object.
case_dict (dict): The case dictionary.
"""
target = await moderation.get_target()
moderator = await moderation.get_moderator()
if case_dict["target_type"] == "USER":
target_user = await fetch_user_dict(interaction.client, case_dict["target_id"])
target_name = (
target_user["name"]
if target_user["discriminator"] == "0"
else f"{target_user['name']}#{target_user['discriminator']}"
)
content = f"Case: {moderation.id:,} ({str.title(moderation.type.string)})\nTarget: {target.name} ({target.id})\nModerator: {moderator.name} ({moderator.id})"
elif case_dict["target_type"] == "CHANNEL":
target_user = await fetch_channel_dict(interaction.guild, case_dict["target_id"])
target_name = target_user["name"]
if moderation.duration is not None:
content += f"\nDuration: {humanize_timedelta(timedelta=moderation.duration)}"
moderator_user = await fetch_user_dict(interaction.client, case_dict["moderator_id"])
moderator_name = (
moderator_user["name"]
if moderator_user["discriminator"] == "0"
else f"{moderator_user['name']}#{moderator_user['discriminator']}"
)
if moderation.role_id:
role = await moderation.get_role()
content += "\nRole: " + (role.name if role is not None else moderation.role_id)
content = f"Case: {case_dict['moderation_id']:,} ({str.title(case_dict['moderation_type'])})\nTarget: {target_name} ({target_user['id']})\nModerator: {moderator_name} ({moderator_user['id']})"
content += f"\nReason: {moderation.reason}"
if case_dict["role_id"] != "0":
role = interaction.guild.get_role(int(case_dict["role_id"]))
content += "\nRole: " + (role.name if role is not None else case_dict["role_id"])
for key, value in moderation.metadata.items():
content += f"\n{key.title()}: {value}"
if case_dict["duration"] != "NULL":
hours, minutes, seconds = map(int, case_dict["duration"].split(":"))
td = timedelta(hours=hours, minutes=minutes, seconds=seconds)
content += f"\nDuration: {humanize_timedelta(timedelta=td)}"
content += f"\nReason: {case_dict['reason']}"
return box(content, "prolog")
@ -543,41 +622,3 @@ async def immune_embed(ctx: commands.Context) -> Embed:
e.description += "\n\n" + immune_str
return e
async def type_embed(ctx: commands.Context, moderation_type = Type) -> Embed:
"""Generates a configuration menu field value for a guild's settings."""
type_settings = {
"show_in_history": await config.custom("types", ctx.guild.id, moderation_type.key).show_in_history(),
"show_moderator": await config.custom("types", ctx.guild.id, moderation_type.key).show_moderator(),
"use_discord_permissions": await config.custom("types", ctx.guild.id, moderation_type.key).use_discord_permissions(),
"dm_users": await config.custom("types", ctx.guild.id, moderation_type.key).dm_users(),
}
guild_str = [
"- "
+ bold("Show in History: ")
+ get_bool_emoji(type_settings["show_in_history"]),
"- "
+ bold("Show Moderator: ")
+ get_bool_emoji(type_settings["show_moderator"]),
"- "
+ bold("Use Discord Permissions: ")
+ get_bool_emoji(type_settings["use_discord_permissions"]),
"- "
+ bold("DM Users: ")
+ get_bool_emoji(type_settings["dm_users"]),
]
guild_str = "\n".join(guild_str)
e = await _config(ctx)
e.title += f": {moderation_type.string.title()} Configuration"
e.description = (
f"""
Use the buttons below to manage Aurora's configuration for the {bold(moderation_type.string)} moderation type.
If an option has a question mark (\N{BLACK QUESTION MARK ORNAMENT}\N{VARIATION SELECTOR-16}) next to it, Aurora will default to the guild level setting instead.
See `{ctx.prefix}aurora set guild` for more information.\n
"""
+ guild_str
)
return e

View file

@ -1,135 +0,0 @@
import json
from datetime import datetime, timedelta
from typing import Any
from redbot.core.bot import Red
from ..models.base import AuroraBaseModel
from ..models.type import Type
class JSONEncoder(json.JSONEncoder):
def default(self, o) -> Any:
match o:
case datetime():
return int(o.timestamp())
case timedelta():
from ..utilities.utils import timedelta_to_string
return timedelta_to_string(o)
case AuroraBaseModel():
return o.dump()
case Type():
return o.key
case Red():
return None
case _:
return super().default(o)
# This is a wrapper around the json module's dumps function that uses our custom JSONEncoder class
def dumps(obj, *, skipkeys=False, ensure_ascii=True, check_circular=True,
allow_nan=True, indent=None, separators=None,
default=None, sort_keys=False, **kw) -> str:
"""Serialize ``obj`` to a JSON formatted ``str``.
If ``skipkeys`` is true then ``dict`` keys that are not basic types
(``str``, ``int``, ``float``, ``bool``, ``None``) will be skipped
instead of raising a ``TypeError``.
If ``ensure_ascii`` is false, then the return value can contain non-ASCII
characters if they appear in strings contained in ``obj``. Otherwise, all
such characters are escaped in JSON strings.
If ``check_circular`` is false, then the circular reference check
for container types will be skipped and a circular reference will
result in an ``RecursionError`` (or worse).
If ``allow_nan`` is false, then it will be a ``ValueError`` to
serialize out of range ``float`` values (``nan``, ``inf``, ``-inf``) in
strict compliance of the JSON specification, instead of using the
JavaScript equivalents (``NaN``, ``Infinity``, ``-Infinity``).
If ``indent`` is a non-negative integer, then JSON array elements and
object members will be pretty-printed with that indent level. An indent
level of 0 will only insert newlines. ``None`` is the most compact
representation.
If specified, ``separators`` should be an ``(item_separator, key_separator)``
tuple. The default is ``(', ', ': ')`` if *indent* is ``None`` and
``(',', ': ')`` otherwise. To get the most compact JSON representation,
you should specify ``(',', ':')`` to eliminate whitespace.
``default(obj)`` is a function that should return a serializable version
of obj or raise TypeError. The default simply raises TypeError.
If *sort_keys* is true (default: ``False``), then the output of
dictionaries will be sorted by key.
"""
return json.dumps(
obj,
cls=JSONEncoder,
skipkeys=skipkeys,
ensure_ascii=ensure_ascii,
check_circular=check_circular,
allow_nan=allow_nan,
indent=indent,
separators=separators,
default=default,
sort_keys=sort_keys,
**kw
)
# This is a wrapper around the json module's dump function that uses our custom JSONEncoder class
def dump(obj, fp, *, skipkeys=False, ensure_ascii=True, check_circular=True,
allow_nan=True, indent=None, separators=None,
default=None, sort_keys=False, **kw) -> str:
"""Serialize ``obj`` as a JSON formatted stream to ``fp`` (a
``.write()``-supporting file-like object).
If ``skipkeys`` is true then ``dict`` keys that are not basic types
(``str``, ``int``, ``float``, ``bool``, ``None``) will be skipped
instead of raising a ``TypeError``.
If ``ensure_ascii`` is false, then the strings written to ``fp`` can
contain non-ASCII characters if they appear in strings contained in
``obj``. Otherwise, all such characters are escaped in JSON strings.
If ``check_circular`` is false, then the circular reference check
for container types will be skipped and a circular reference will
result in an ``RecursionError`` (or worse).
If ``allow_nan`` is false, then it will be a ``ValueError`` to
serialize out of range ``float`` values (``nan``, ``inf``, ``-inf``)
in strict compliance of the JSON specification, instead of using the
JavaScript equivalents (``NaN``, ``Infinity``, ``-Infinity``).
If ``indent`` is a non-negative integer, then JSON array elements and
object members will be pretty-printed with that indent level. An indent
level of 0 will only insert newlines. ``None`` is the most compact
representation.
If specified, ``separators`` should be an ``(item_separator, key_separator)``
tuple. The default is ``(', ', ': ')`` if *indent* is ``None`` and
``(',', ': ')`` otherwise. To get the most compact JSON representation,
you should specify ``(',', ':')`` to eliminate whitespace.
``default(obj)`` is a function that should return a serializable version
of obj or raise TypeError. The default simply raises TypeError.
If *sort_keys* is true (default: ``False``), then the output of
dictionaries will be sorted by key.
"""
return json.dump(
obj,
fp,
cls=JSONEncoder,
skipkeys=skipkeys,
ensure_ascii=ensure_ascii,
check_circular=check_circular,
allow_nan=allow_nan,
indent=indent,
separators=separators,
default=default,
sort_keys=sort_keys,
**kw
)

View file

@ -1,25 +1,23 @@
# pylint: disable=cyclic-import
from datetime import datetime, timedelta
from typing import Optional, Tuple, Union
import json
from datetime import datetime
from datetime import timedelta as td
from typing import Optional, Union
import aiosqlite
from dateutil.relativedelta import relativedelta as rd
from discord import File, Guild, Interaction, Member, SelectOption, TextChannel, User
from discord.errors import Forbidden
from discord import File, Guild, Interaction, Member, SelectOption, User
from discord.errors import Forbidden, NotFound
from redbot.core import commands, data_manager
from redbot.core.utils.chat_formatting import error
from ..models.type import Type
from ..utilities.config import config
from ..utilities.json import dumps
from ..utilities.logger import logger
from .config import config
def check_permissions(
user: User,
permissions: Tuple[str],
ctx: commands.Context | Interaction | None = None,
guild: Guild | None = None,
permissions: list,
ctx: Union[commands.Context, Interaction] = None,
guild: Guild = None,
) -> Union[bool, str]:
"""Checks if a user has a specific permission (or a list of permissions) in a channel."""
if ctx:
@ -44,17 +42,11 @@ def check_permissions(
async def check_moddable(
target: Union[User, Member, TextChannel], ctx: commands.Context, permissions: Tuple[str], moderation_type: Type,
target: Union[User, Member], interaction: Interaction, permissions: list
) -> bool:
"""Checks if a moderator can moderate a target."""
is_channel = isinstance(target, TextChannel)
use_discord_permissions = await config.custom("types", ctx.guild.id, moderation_type.key).use_discord_permissions()
if use_discord_permissions is None:
use_discord_permissions = await config.guild(ctx.guild).use_discord_permissions()
if check_permissions(ctx.bot.user, permissions, guild=ctx.guild):
await ctx.send(
if check_permissions(interaction.client.user, permissions, guild=interaction.guild):
await interaction.response.send_message(
error(
f"I do not have the `{permissions}` permission, required for this action."
),
@ -62,9 +54,9 @@ async def check_moddable(
)
return False
if use_discord_permissions is True:
if check_permissions(ctx.author, permissions, guild=ctx.guild):
await ctx.send(
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(
error(
f"You do not have the `{permissions}` permission, required for this action."
),
@ -72,21 +64,21 @@ async def check_moddable(
)
return False
if ctx.author.id == target.id:
await ctx.send(
if interaction.user.id == target.id:
await interaction.response.send_message(
content="You cannot moderate yourself!", ephemeral=True
)
return False
if not is_channel and target.bot:
await ctx.send(
if target.bot:
await interaction.response.send_message(
content="You cannot moderate bots!", ephemeral=True
)
return False
if isinstance(target, Member):
if ctx.author.top_role <= target.top_role and await config.guild(ctx.guild).respect_hierarchy() is True:
await ctx.send(
if interaction.user.top_role <= target.top_role and await config.guild(interaction.guild).respect_hierarchy() is True:
await interaction.response.send_message(
content=error(
"You cannot moderate members with a higher role than you!"
),
@ -94,17 +86,11 @@ async def check_moddable(
)
return False
if target.guild_permissions.administrator:
await ctx.send(
content="You cannot moderate members with the Administrator permission!", ephemeral=True
)
return False
if (
ctx.guild.get_member(ctx.bot.user.id).top_role
interaction.guild.get_member(interaction.client.user.id).top_role
<= target.top_role
):
await ctx.send(
await interaction.response.send_message(
content=error(
"You cannot moderate members with a role higher than the bot!"
),
@ -116,7 +102,7 @@ async def check_moddable(
for role in target.roles:
if role.id in immune_roles:
await ctx.send(
await interaction.response.send_message(
content=error("You cannot moderate members with an immune role!"),
ephemeral=True,
)
@ -125,55 +111,151 @@ async def check_moddable(
return True
async def log(ctx: commands.Context, moderation_id: int, resolved: bool = False) -> None:
async def get_next_case_number(guild_id: str, cursor=None) -> int:
"""This function returns the next case number from the MySQL table for a specific guild."""
from .database import connect
if not cursor:
database = connect()
cursor = database.cursor()
cursor.execute(
f"SELECT moderation_id FROM `moderation_{guild_id}` ORDER BY moderation_id DESC LIMIT 1"
)
result = cursor.fetchone()
return (result[0] + 1) if result else 1
def generate_dict(result) -> dict:
case = {
"moderation_id": result[0],
"timestamp": result[1],
"moderation_type": result[2],
"target_type": result[3],
"target_id": result[4],
"moderator_id": result[5],
"role_id": result[6],
"duration": result[7],
"end_timestamp": result[8],
"reason": result[9],
"resolved": result[10],
"resolved_by": result[11],
"resolve_reason": result[12],
"expired": result[13],
"changes": json.loads(result[14]),
"metadata": json.loads(result[15]),
}
return case
async def fetch_user_dict(client: commands.Bot, user_id: str) -> dict:
"""This function returns a dictionary containing either user information or a standard deleted user template."""
if user_id == "?":
user_dict = {"id": "?", "name": "Unknown User", "discriminator": "0"}
else:
try:
user = client.get_user(int(user_id))
if user is None:
user = await client.fetch_user(int(user_id))
user_dict = {
"id": user.id,
"name": user.name,
"discriminator": user.discriminator,
}
except NotFound:
user_dict = {
"id": user_id,
"name": "Deleted User",
"discriminator": "0",
}
return user_dict
async def fetch_channel_dict(guild: Guild, channel_id: int) -> dict:
"""This function returns a dictionary containing either channel information or a standard deleted channel template."""
try:
channel = guild.get_channel(int(channel_id))
if not channel:
channel = await guild.fetch_channel(channel_id)
channel_dict = {
"id": channel.id,
"name": channel.name,
"mention": channel.mention,
}
except NotFound:
channel_dict = {"id": channel_id, "name": "Deleted Channel", "mention": None}
return channel_dict
async def fetch_role_dict(guild: Guild, role_id: int) -> dict:
"""This function returns a dictionary containing either role information or a standard deleted role template."""
role = guild.get_role(int(role_id))
if not role:
role_dict = {"id": role_id, "name": "Deleted Role"}
role_dict = {"id": role.id, "name": role.name}
return role_dict
async def log(interaction: Interaction, moderation_id: int, resolved: bool = False) -> None:
"""This function sends a message to the guild's configured logging channel when an infraction takes place."""
from ..models.moderation import Moderation
from .database import fetch_case
from .factory import log_factory
logging_channel_id = await config.guild(ctx.guild).log_channel()
logging_channel_id = await config.guild(interaction.guild).log_channel()
if logging_channel_id != " ":
logging_channel = ctx.guild.get_channel(logging_channel_id)
logging_channel = interaction.guild.get_channel(logging_channel_id)
try:
moderation = await Moderation.find_by_id(ctx.bot, moderation_id, ctx.guild.id)
case = await fetch_case(moderation_id, interaction.guild.id)
if case:
embed = await log_factory(
ctx=ctx, moderation=moderation, resolved=resolved
interaction=interaction, case_dict=case, resolved=resolved
)
try:
await logging_channel.send(embed=embed)
except Forbidden:
return
except ValueError:
return
async def send_evidenceformat(ctx: commands.Context, moderation_id: int) -> None:
async def send_evidenceformat(interaction: Interaction, case_dict: dict) -> 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(ctx.author).auto_evidenceformat()
or await config.guild(guild=ctx.guild).auto_evidenceformat()
await config.user(interaction.user).auto_evidenceformat()
or await config.guild(interaction.guild).auto_evidenceformat()
or False
)
if send_evidence_bool is True:
moderation = await Moderation.find_by_id(ctx.bot, moderation_id, ctx.guild.id)
content = await evidenceformat_factory(moderation=moderation)
if not ctx.interaction:
await ctx.author.send(content=content)
else:
await ctx.send(content=content, ephemeral=True)
if send_evidence_bool is False:
return
content = await evidenceformat_factory(interaction=interaction, case_dict=case_dict)
await interaction.followup.send(content=content, ephemeral=True)
def convert_timedelta_to_str(timedelta: td) -> str:
"""This function converts a timedelta object to a string."""
total_seconds = int(timedelta.total_seconds())
hours = total_seconds // 3600
minutes = (total_seconds % 3600) // 60
seconds = total_seconds % 60
return f"{hours}:{minutes}:{seconds}"
def get_bool_emoji(value: Optional[bool]) -> str:
"""Returns a unicode emoji based on a boolean value."""
match value:
case True:
if value is True:
return "\N{WHITE HEAVY CHECK MARK}"
case False:
if value is False:
return "\N{NO ENTRY SIGN}"
case _:
return "\N{BLACK QUESTION MARK ORNAMENT}\N{VARIATION SELECTOR-16}"
@ -204,91 +286,13 @@ def create_pagesize_options() -> list[SelectOption]:
)
return options
def timedelta_from_relativedelta(relativedelta: rd) -> timedelta:
def timedelta_from_relativedelta(relativedelta: rd) -> td:
"""Converts a relativedelta object to a timedelta object."""
now = datetime.now()
then = now - relativedelta
return now - then
def timedelta_from_string(string: str) -> timedelta:
"""Converts a string to a timedelta object."""
hours, minutes, seconds = map(int, string.split(":"))
return timedelta(hours=hours, minutes=minutes, seconds=seconds)
def timedelta_to_string(td: timedelta) -> str:
"""Converts a timedelta object to a string."""
days = td.days * 24
hours, remainder = divmod(td.seconds, 3600)
minutes, seconds = divmod(remainder, 60)
return f"{days + hours}:{minutes:02}:{seconds:02}"
def get_footer_image(coginstance: commands.Cog) -> File:
"""Returns the footer image for the embeds."""
image_path = data_manager.bundled_data_path(coginstance) / "arrow.png"
return File(image_path, filename="arrow.png", description="arrow")
async def create_guild_table(guild: Guild) -> None:
from ..models.moderation import Moderation
try:
await Moderation.execute(f"SELECT * FROM `moderation_{guild.id}`", return_obj=False)
logger.trace("SQLite Table exists for server %s (%s)", guild.name, guild.id)
except aiosqlite.OperationalError:
query = f"""
CREATE TABLE `moderation_{guild.id}` (
moderation_id INTEGER PRIMARY KEY NOT NULL,
timestamp INTEGER NOT NULL,
moderation_type TEXT NOT NULL,
target_type TEXT NOT NULL,
target_id INTEGER NOT NULL,
moderator_id INTEGER NOT NULL,
role_id INTEGER,
duration TEXT,
end_timestamp INTEGER,
reason TEXT,
resolved INTEGER NOT NULL,
resolved_by TEXT,
resolve_reason TEXT,
expired INTEGER NOT NULL,
changes JSON NOT NULL,
metadata JSON NOT NULL
)
"""
await Moderation.execute(query=query, return_obj=False)
index_query_1 = f"CREATE INDEX IF NOT EXISTS idx_target_id ON moderation_{guild.id}(target_id);"
await Moderation.execute(query=index_query_1, return_obj=False)
index_query_2 = f"CREATE INDEX IF NOT EXISTS idx_moderator_id ON moderation_{guild.id}(moderator_id);"
await Moderation.execute(query=index_query_2, return_obj=False)
index_query_3 = f"CREATE INDEX IF NOT EXISTS idx_moderation_id ON moderation_{guild.id}(moderation_id);"
await Moderation.execute(query=index_query_3, return_obj=False)
insert_query = f"""
INSERT INTO `moderation_{guild.id}`
(moderation_id, timestamp, moderation_type, target_type, target_id, moderator_id, role_id, duration, end_timestamp, reason, resolved, resolved_by, resolve_reason, expired, changes, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""
insert_values = (
0,
0,
"NULL",
"NULL",
0,
0,
None,
None,
None,
None,
0,
None,
None,
0,
dumps([]),
dumps({}),
)
await Moderation.execute(query=insert_query, parameters=insert_values, return_obj=False)
logger.trace("SQLite Table created for server %s (%s)", guild.name, guild.id)

View file

@ -14,15 +14,16 @@ from redbot.cogs.downloader import errors
from redbot.cogs.downloader.converters import InstalledCog
from redbot.core import commands
from redbot.core.bot import Red
from redbot.core.utils.chat_formatting import error, humanize_list, text_to_file
from redbot.core.utils.chat_formatting import bold, error, humanize_list, text_to_file
# pylint: disable=protected-access
class Backup(commands.Cog):
"""A utility to make reinstalling repositories and cogs after migrating the bot far easier."""
__author__ = ["SeaswimmerTheFsh"]
__version__ = "1.1.0"
__author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
__git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
__version__ = "1.1.1"
__documentation__ = "https://seacogs.coastalcommits.com/backup/"
def __init__(self, bot: Red):
@ -35,9 +36,9 @@ class Backup(commands.Cog):
n = "\n" if "\n\n" not in pre_processed else ""
text = [
f"{pre_processed}{n}",
f"Cog Version: **{self.__version__}**",
f"Author: {humanize_list(self.__author__)}",
f"Documentation: {self.__documentation__}",
f"{bold('Cog Version:')} [{self.__version__}]({self.__git__})",
f"{bold('Author:')} {humanize_list(self.__author__)}",
f"{bold('Documentation:')} {self.__documentation__}",
]
return "\n".join(text)
@ -196,7 +197,7 @@ class Backup(commands.Cog):
cog_modules = []
for cog in cogs:
# If you're forking this cog, make sure to change these strings!
if cog["name"] == "backup" and "SeaswimmerTheFsh/SeaCogs" in url:
if cog["name"] == "backup" and "cswimr/SeaCogs" in url:
continue
try:
cog_module = await InstalledCog.convert(ctx, cog["name"])
@ -232,7 +233,7 @@ class Backup(commands.Cog):
commit = None
# If you're forking this cog, make sure to change these strings!
if cog_name == "backup" and "SeaswimmerTheFsh/SeaCogs" in url:
if cog_name == "backup" and "cswimr/SeaCogs" in url:
continue
async with repository.checkout(

View file

@ -1,6 +1,6 @@
{
"author" : ["SeaswimmerTheFsh (seasw.)"],
"install_msg" : "Thank you for installing Backup!\nYou can find the source code of this cog [here](https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs).",
"author" : ["cswimr"],
"install_msg" : "Thank you for installing Backup!\nYou can find the source code of this cog [here](https://coastalcommits.com/cswimr/SeaCogs).",
"name" : "Backup",
"short" : "A utility to make reinstalling repositories and cogs after migrating the bot far easier.",
"description" : "A utility to make reinstalling repositories and cogs after migrating the bot far easier.",
@ -8,7 +8,7 @@
"hidden": false,
"disabled": false,
"min_bot_version": "3.5.6",
"max_bot_version": "3.5.12",
"max_bot_version": "3.5.13",
"min_python_version": [3, 9, 0],
"tags": [
"utility",

View file

@ -15,7 +15,7 @@ from PIL import Image
from red_commons.logging import getLogger
from redbot.core import Config, commands, data_manager
from redbot.core.bot import Red
from redbot.core.utils.chat_formatting import error, humanize_list
from redbot.core.utils.chat_formatting import bold, error, humanize_list
import bible.errors
from bible.models import Version
@ -24,9 +24,10 @@ from bible.models import Version
class Bible(commands.Cog):
"""Retrieve Bible verses from the API.bible API."""
__author__ = ["SeaswimmerTheFsh"]
__version__ = "1.1.0"
__documentation__ = "https://seacogs.coastalcommits.com/bible/"
__author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
__git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
__version__ = "1.1.1"
__documentation__ = "https://seacogs.coastalcommits.com/pterodactyl/"
def __init__(self, bot: Red):
super().__init__()
@ -44,12 +45,13 @@ class Bible(commands.Cog):
n = "\n" if "\n\n" not in pre_processed else ""
text = [
f"{pre_processed}{n}",
f"Cog Version: **{self.__version__}**",
f"Author: {humanize_list(self.__author__)}",
f"Documentation: {self.__documentation__}",
f"{bold('Cog Version:')} [{self.__version__}]({self.__git__})",
f"{bold('Author:')} {humanize_list(self.__author__)}",
f"{bold('Documentation:')} {self.__documentation__}",
]
return "\n".join(text)
def get_icon(self, color: Colour) -> File:
"""Get the docs.api.bible favicon with a given color."""
image_path = data_manager.bundled_data_path(self) / "api.bible-logo.png"

View file

@ -1,6 +1,6 @@
{
"author" : ["SeaswimmerTheFsh (seasw.)"],
"install_msg" : "Thank you for installing Bible!\nThis cog requires setting an API key for API.Bible. Please read the [documentation](https://seacogs.coastalcommits.com/bible/#setup) for more information.\nYou can find the source code of this cog [here](https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs).",
"author" : ["cswimr"],
"install_msg" : "Thank you for installing Bible!\nThis cog requires setting an API key for API.Bible. Please read the [documentation](https://seacogs.coastalcommits.com/bible/#setup) for more information.\nYou can find the source code of this cog [here](https://coastalcommits.com/cswimr/SeaCogs).",
"name" : "Bible",
"short" : "Retrieve Bible verses from API.Bible.",
"description" : "Retrieve Bible verses from the API.Bible API. This cog requires an API.Bible api key.",

View file

@ -1,5 +1,4 @@
import io
from typing import Any, Literal
import aiohttp
import discord
@ -15,9 +14,10 @@ from .model import PartialEmoji
class EmojiInfo(commands.Cog):
"""Retrieve information about emojis."""
__author__: list[str] = ["SeaswimmerTheFsh"]
__version__: str = "1.0.0"
__documentation__: str = "https://seacogs.coastalcommits.com/emojiinfo/"
__author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
__git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
__version__ = "1.0.1"
__documentation__ = "https://seacogs.coastalcommits.com/emojiinfo/"
def __init__(self, bot: Red) -> None:
super().__init__()
@ -25,16 +25,17 @@ class EmojiInfo(commands.Cog):
self.logger: RedTraceLogger = getLogger(name="red.SeaCogs.Emoji")
def format_help_for_context(self, ctx: commands.Context) -> str:
pre_processed: Any | Literal[''] = super().format_help_for_context(ctx) or ""
n: Literal['\n'] | Literal[''] = "\n" if "\n\n" not in pre_processed else ""
text: list[str] = [
pre_processed = super().format_help_for_context(ctx) or ""
n = "\n" if "\n\n" not in pre_processed else ""
text = [
f"{pre_processed}{n}",
f"Cog Version: **{self.__version__}**",
f"Author: {humanize_list(items=self.__author__)}",
f"Documentation: {self.__documentation__}",
f"{bold('Cog Version:')} [{self.__version__}]({self.__git__})",
f"{bold('Author:')} {humanize_list(self.__author__)}",
f"{bold('Documentation:')} {self.__documentation__}",
]
return "\n".join(text)
async def fetch_twemoji(self, unicode_emoji) -> str:
base_url = "https://cdn.jsdelivr.net/gh/jdecked/twemoji@latest/assets/72x72/"
emoji_codepoint = "-".join([hex(ord(char))[2:] for char in unicode_emoji])

View file

@ -1,5 +1,5 @@
{
"author" : ["SeaswimmerTheFsh (seasw.)"],
"author" : ["cswimr"],
"install_msg" : "Thank you for installing Emoji!",
"name" : "Emoji",
"short" : "Retrieve information about emojis.",

303
flake.lock Normal file
View file

@ -0,0 +1,303 @@
{
"nodes": {
"cachix": {
"inputs": {
"devenv": [
"devenv"
],
"flake-compat": [
"devenv"
],
"git-hooks": [
"devenv"
],
"nixpkgs": "nixpkgs"
},
"locked": {
"lastModified": 1728672398,
"narHash": "sha256-KxuGSoVUFnQLB2ZcYODW7AVPAh9JqRlD5BrfsC/Q4qs=",
"owner": "cachix",
"repo": "cachix",
"rev": "aac51f698309fd0f381149214b7eee213c66ef0a",
"type": "github"
},
"original": {
"owner": "cachix",
"ref": "latest",
"repo": "cachix",
"type": "github"
}
},
"devenv": {
"inputs": {
"cachix": "cachix",
"flake-compat": "flake-compat",
"git-hooks": "git-hooks",
"nix": "nix",
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"lastModified": 1732121232,
"narHash": "sha256-CmJt7aeSCJnJYGtYpyslRI+pC28RPVD43PD/7kkIVuM=",
"owner": "cachix",
"repo": "devenv",
"rev": "6ff1e5f92c0d74bbb12f7454a239ca2f02e05ea1",
"type": "github"
},
"original": {
"owner": "cachix",
"repo": "devenv",
"type": "github"
}
},
"flake-compat": {
"flake": false,
"locked": {
"lastModified": 1696426674,
"narHash": "sha256-kvjfFW7WAETZlt09AgDn1MrtKzP7t90Vf7vypd3OL1U=",
"owner": "edolstra",
"repo": "flake-compat",
"rev": "0f9255e01c2351cc7d116c072cb317785dd33b33",
"type": "github"
},
"original": {
"owner": "edolstra",
"repo": "flake-compat",
"type": "github"
}
},
"flake-compat_2": {
"flake": false,
"locked": {
"lastModified": 1696426674,
"narHash": "sha256-kvjfFW7WAETZlt09AgDn1MrtKzP7t90Vf7vypd3OL1U=",
"owner": "edolstra",
"repo": "flake-compat",
"rev": "0f9255e01c2351cc7d116c072cb317785dd33b33",
"type": "github"
},
"original": {
"owner": "edolstra",
"repo": "flake-compat",
"type": "github"
}
},
"flake-parts": {
"inputs": {
"nixpkgs-lib": [
"devenv",
"nix",
"nixpkgs"
]
},
"locked": {
"lastModified": 1712014858,
"narHash": "sha256-sB4SWl2lX95bExY2gMFG5HIzvva5AVMJd4Igm+GpZNw=",
"owner": "hercules-ci",
"repo": "flake-parts",
"rev": "9126214d0a59633752a136528f5f3b9aa8565b7d",
"type": "github"
},
"original": {
"owner": "hercules-ci",
"repo": "flake-parts",
"type": "github"
}
},
"git-hooks": {
"inputs": {
"flake-compat": [
"devenv"
],
"gitignore": "gitignore",
"nixpkgs": [
"devenv",
"nixpkgs"
],
"nixpkgs-stable": [
"devenv"
]
},
"locked": {
"lastModified": 1730302582,
"narHash": "sha256-W1MIJpADXQCgosJZT8qBYLRuZls2KSiKdpnTVdKBuvU=",
"owner": "cachix",
"repo": "git-hooks.nix",
"rev": "af8a16fe5c264f5e9e18bcee2859b40a656876cf",
"type": "github"
},
"original": {
"owner": "cachix",
"repo": "git-hooks.nix",
"type": "github"
}
},
"gitignore": {
"inputs": {
"nixpkgs": [
"devenv",
"git-hooks",
"nixpkgs"
]
},
"locked": {
"lastModified": 1709087332,
"narHash": "sha256-HG2cCnktfHsKV0s4XW83gU3F57gaTljL9KNSuG6bnQs=",
"owner": "hercules-ci",
"repo": "gitignore.nix",
"rev": "637db329424fd7e46cf4185293b9cc8c88c95394",
"type": "github"
},
"original": {
"owner": "hercules-ci",
"repo": "gitignore.nix",
"type": "github"
}
},
"libgit2": {
"flake": false,
"locked": {
"lastModified": 1697646580,
"narHash": "sha256-oX4Z3S9WtJlwvj0uH9HlYcWv+x1hqp8mhXl7HsLu2f0=",
"owner": "libgit2",
"repo": "libgit2",
"rev": "45fd9ed7ae1a9b74b957ef4f337bc3c8b3df01b5",
"type": "github"
},
"original": {
"owner": "libgit2",
"repo": "libgit2",
"type": "github"
}
},
"nix": {
"inputs": {
"flake-compat": [
"devenv"
],
"flake-parts": "flake-parts",
"libgit2": "libgit2",
"nixpkgs": "nixpkgs_2",
"nixpkgs-23-11": [
"devenv"
],
"nixpkgs-regression": [
"devenv"
],
"pre-commit-hooks": [
"devenv"
]
},
"locked": {
"lastModified": 1727438425,
"narHash": "sha256-X8ES7I1cfNhR9oKp06F6ir4Np70WGZU5sfCOuNBEwMg=",
"owner": "domenkozar",
"repo": "nix",
"rev": "f6c5ae4c1b2e411e6b1e6a8181cc84363d6a7546",
"type": "github"
},
"original": {
"owner": "domenkozar",
"ref": "devenv-2.24",
"repo": "nix",
"type": "github"
}
},
"nixpkgs": {
"locked": {
"lastModified": 1730531603,
"narHash": "sha256-Dqg6si5CqIzm87sp57j5nTaeBbWhHFaVyG7V6L8k3lY=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "7ffd9ae656aec493492b44d0ddfb28e79a1ea25d",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixos-unstable",
"repo": "nixpkgs",
"type": "github"
}
},
"nixpkgs-python": {
"inputs": {
"flake-compat": "flake-compat_2",
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"lastModified": 1730716553,
"narHash": "sha256-n4cibCp/ggDlSacCTnP8dVnywclQKYcHy6PRfe35Hk0=",
"owner": "cachix",
"repo": "nixpkgs-python",
"rev": "8fcdb8ec34a1c2bae3f5326873a41b310e948ccc",
"type": "github"
},
"original": {
"owner": "cachix",
"repo": "nixpkgs-python",
"type": "github"
}
},
"nixpkgs_2": {
"locked": {
"lastModified": 1717432640,
"narHash": "sha256-+f9c4/ZX5MWDOuB1rKoWj+lBNm0z0rs4CK47HBLxy1o=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "88269ab3044128b7c2f4c7d68448b2fb50456870",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "release-24.05",
"repo": "nixpkgs",
"type": "github"
}
},
"nixpkgs_3": {
"locked": {
"lastModified": 1731676054,
"narHash": "sha256-OZiZ3m8SCMfh3B6bfGC/Bm4x3qc1m2SVEAlkV6iY7Yg=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "5e4fbfb6b3de1aa2872b76d49fafc942626e2add",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixos-unstable",
"repo": "nixpkgs",
"type": "github"
}
},
"root": {
"inputs": {
"devenv": "devenv",
"nixpkgs": "nixpkgs_3",
"nixpkgs-python": "nixpkgs-python",
"systems": "systems"
}
},
"systems": {
"locked": {
"lastModified": 1681028828,
"narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
"owner": "nix-systems",
"repo": "default",
"rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
"type": "github"
},
"original": {
"owner": "nix-systems",
"repo": "default",
"type": "github"
}
}
},
"root": "root",
"version": 7
}

46
flake.nix Normal file
View file

@ -0,0 +1,46 @@
{
inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
nixpkgs-python.url = "github:cachix/nixpkgs-python";
nixpkgs-python.inputs = { nixpkgs.follows = "nixpkgs"; };
systems.url = "github:nix-systems/default";
devenv.url = "github:cachix/devenv";
devenv.inputs.nixpkgs.follows = "nixpkgs";
};
nixConfig = {
extra-trusted-public-keys =
"devenv.cachix.org-1:w1cLUi8dv3hnoSPGAuibQv+f9TZLr6cv/Hm9XgU50cw=";
extra-substituters = "https://devenv.cachix.org";
};
outputs = { self, nixpkgs, devenv, systems, ... }@inputs:
let forEachSystem = nixpkgs.lib.genAttrs (import systems);
in {
packages = forEachSystem (system: {
devenv-up = self.devShells.${system}.default.config.procfileScript;
devenv-test = self.devShells.${system}.default.config.test;
});
devShells = forEachSystem (system:
let pkgs = nixpkgs.legacyPackages.${system};
in {
default = devenv.lib.mkShell {
inherit inputs pkgs;
modules = [{
languages.python = {
enable = true;
version = "3.11";
uv = {
enable = true;
sync = {
enable = true;
allExtras = true;
};
};
};
}];
};
});
};
}

View file

@ -1,9 +1,9 @@
{
"author": [
"SeaswimmerTheFsh (seasw.)"
"cswimr"
],
"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/SeaCogs/issues) or join my [Discord Server](https://discord.gg/eMUMe77Yb8 ).",
"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/SeaCogs/issues) or join my [Discord Server](https://discord.gg/eMUMe77Yb8 ).",
"index_name": "sea-cogs",
"short": "Various cogs for Red, by SeaswimmerTheFsh (seasw.)",
"description": "Various cogs for Red, by SeaswimmerTheFsh (seasw.)"
"short": "Various cogs for Red, by cswimr",
"description": "Various cogs for Red, by cswimr"
}

View file

@ -1,12 +1,12 @@
site_name: SeaCogs Documentation
site_url: !ENV [SITE_URL, 'https://seacogs.coastalcommits.com']
repo_name: CoastalCommits
repo_url: https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs
repo_url: https://coastalcommits.com/cswimr/SeaCogs
edit_uri: !ENV [EDIT_URI, 'src/branch/main/.docs']
copyright: Copyright &copy; 2023-2024, SeaswimmerTheFsh
copyright: Copyright &copy; 2023-2024, cswimr
docs_dir: .docs
site_author: SeaswimmerTheFsh
site_author: cswimr
site_description: Documentation for my Red-DiscordBot Cogs.
nav:
@ -30,7 +30,7 @@ nav:
plugins:
- git-authors
- search
#- social
- social
- git-revision-date-localized:
enable_creation_date: true
type: timeago
@ -113,3 +113,5 @@ watch:
- ./bible
- ./nerdify
- ./pterodactyl
- ./emojiinfo
- ./antipolls

View file

@ -1,6 +1,6 @@
{
"author" : ["SeaswimmerTheFsh (seasw.)"],
"install_msg" : "Thank you for installing Nerdify!\nYou can find the source code of this cog [here](https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs). Based off of PhasecoreX's [UwU](<https://github.com/PhasecoreX/PCXCogs/tree/master/uwu>) cog.",
"author" : ["cswimr"],
"install_msg" : "Thank you for installing Nerdify!\nYou can find the source code of this cog [here](https://coastalcommits.com/cswimr/SeaCogs). Based off of PhasecoreX's [UwU](<https://github.com/PhasecoreX/PCXCogs/tree/master/uwu>) cog.",
"name" : "Nerdify",
"short" : "Nerdify your text!",
"description" : "Nerdify your text!",

View file

@ -12,13 +12,15 @@ from typing import Any, Optional, Union
import discord
from redbot.core import commands
from redbot.core.utils import chat_formatting, common_filters
from redbot.core.utils.chat_formatting import bold, humanize_list
class Nerdify(commands.Cog):
"""Nerdify your text."""
__author__ = ["SeaswimmerTheFsh"]
__version__ = "1.3.4"
__author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
__git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
__version__ = "1.3.5"
__documentation__ = "https://seacogs.coastalcommits.com/nerdify/"
def __init__(self, bot):
@ -29,12 +31,13 @@ class Nerdify(commands.Cog):
n = "\n" if "\n\n" not in pre_processed else ""
text = [
f"{pre_processed}{n}",
f"Cog Version: **{self.__version__}**",
f"Author: {chat_formatting.humanize_list(self.__author__)}",
f"Documentation: {self.__documentation__}"
f"{bold('Cog Version:')} [{self.__version__}]({self.__git__})",
f"{bold('Author:')} {humanize_list(self.__author__)}",
f"{bold('Documentation:')} {self.__documentation__}",
]
return "\n".join(text)
@commands.command(aliases=["nerd"])
async def nerdify(
self, ctx: commands.Context, *, text: Optional[str] = None

2850
poetry.lock generated

File diff suppressed because it is too large Load diff

Binary file not shown.

After

Width:  |  Height:  |  Size: 65 KiB

View file

@ -1,6 +1,6 @@
{
"author" : ["SeaswimmerTheFsh (seasw.)"],
"install_msg" : "Thank you for installing Pterodactyl!\nYou can find the source code of this cog [here](https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs).\nDocumentation can be found [here](https://seacogs.coastalcommits.com/pterodactyl ).",
"author" : ["cswimr"],
"install_msg" : "Thank you for installing Pterodactyl!\nYou can find the source code of this cog [here](https://coastalcommits.com/cswimr/SeaCogs).\nDocumentation can be found [here](https://seacogs.coastalcommits.com/pterodactyl ).",
"name" : "Pterodactyl",
"short" : "Interface with Pterodactyl through websockets.",
"description" : "Interface with Pterodactyl through websockets.",
@ -9,7 +9,7 @@
"disabled": false,
"min_bot_version": "3.5.0",
"min_python_version": [3, 8, 0],
"requirements": ["git+https://github.com/SeaswimmerTheFsh/pydactyl", "websockets"],
"requirements": ["git+https://github.com/cswimr/pydactyl", "websockets"],
"tags": [
"pterodactyl",
"minecraft",

View file

@ -9,7 +9,7 @@ from pydactyl import PterodactylClient
from redbot.core import app_commands, commands
from redbot.core.app_commands import Choice
from redbot.core.bot import Red
from redbot.core.utils.chat_formatting import box, error, humanize_list
from redbot.core.utils.chat_formatting import bold, box, error, humanize_list
from redbot.core.utils.views import ConfirmView
from pterodactyl import mcsrvstatus
@ -20,8 +20,9 @@ from pterodactyl.logger import logger
class Pterodactyl(commands.Cog):
"""Pterodactyl allows you to manage your Pterodactyl Panel from Discord."""
__author__ = ["SeaswimmerTheFsh"]
__version__ = "2.0.0"
__author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
__git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
__version__ = "2.0.4"
__documentation__ = "https://seacogs.coastalcommits.com/pterodactyl/"
def __init__(self, bot: Red):
@ -39,12 +40,29 @@ class Pterodactyl(commands.Cog):
n = "\n" if "\n\n" not in pre_processed else ""
text = [
f"{pre_processed}{n}",
f"Cog Version: **{self.__version__}**",
f"Author: {humanize_list(self.__author__)}",
f"Documentation: {self.__documentation__}",
f"{bold('Cog Version:')} [{self.__version__}]({self.__git__})",
f"{bold('Author:')} {humanize_list(self.__author__)}",
f"{bold('Documentation:')} {self.__documentation__}",
]
return "\n".join(text)
async def cog_load(self) -> None:
pterodactyl_keys = await self.bot.get_shared_api_tokens("pterodactyl")
api_key = pterodactyl_keys.get("api_key")
if api_key is None:
self.task.cancel()
raise ValueError("Pterodactyl API key not set. Please set it using `[p]set api`.")
base_url = await config.base_url()
if base_url is None:
self.task.cancel()
raise ValueError("Pterodactyl base URL not set. Please set it using `[p]pterodactyl config url`.")
server_id = await config.server_id()
if server_id is None:
self.task.cancel()
raise ValueError("Pterodactyl server ID not set. Please set it using `[p]pterodactyl config serverid`.")
self.client = PterodactylClient(base_url, api_key).client
async def cog_unload(self) -> None:
self.update_topic.cancel()
self.task.cancel()
@ -175,65 +193,36 @@ class Pterodactyl(commands.Cog):
async def power(self, ctx: Union[discord.Interaction, commands.Context], action: str, action_ing: str, warning: str = '') -> None:
if isinstance(ctx, discord.Interaction):
author = ctx.user
else:
author = ctx.author
ctx = await self.bot.get_context(ctx)
current_status = await config.current_status()
if current_status == action_ing:
if isinstance(ctx, discord.Interaction):
return await ctx.response.send_message(f"Server is already {action_ing}.", ephemeral=True)
return await ctx.send(f"Server is already {action_ing}.")
return await ctx.send(f"Server is already {action_ing}.", ephemeral=True)
if current_status in ["starting", "stopping"] and action != "kill":
if isinstance(ctx, discord.Interaction):
return await ctx.response.send_message("Another power action is already in progress.", ephemeral=True)
return await ctx.send("Another power action is already in progress.")
return await ctx.send("Another power action is already in progress.", ephemeral=True)
view = ConfirmView(author, disable_buttons=True)
view = ConfirmView(ctx.author, disable_buttons=True)
if isinstance(ctx, discord.Interaction):
await ctx.response.send_message(f"{warning}Are you sure you want to {action} the server?", view=view)
else:
message = await ctx.send(f"{warning}Are you sure you want to {action} the server?", view=view)
await view.wait()
if view.result is True:
if isinstance(ctx, discord.Interaction):
await ctx.edit_original_response(content=f"Sending websocket command to {action} server...", view=None)
else:
await message.edit(content=f"Sending websocket command to {action} server...", view=None)
await self.websocket.send(json.dumps({"event": "set state", "args": [action]}))
if isinstance(ctx, discord.Interaction):
await ctx.edit_original_response(content=f"Server {action_ing}", view=None)
else:
await message.edit(content=f"Server {action_ing}", view=None)
else:
if isinstance(ctx, discord.Interaction):
await ctx.edit_original_response(content="Cancelled.", view=None)
else:
await message.edit(content="Cancelled.", view=None)
async def send_command(self, ctx: Union[discord.Interaction, commands.Context], command: str):
channel = self.bot.get_channel(await config.console_channel())
if isinstance(ctx, discord.Interaction):
if channel:
await channel.send(f"Received console command from {ctx.user.id}: {command[:1900]}", allowed_mentions=discord.AllowedMentions.none())
try:
await self.websocket.send(json.dumps({"event": "send command", "args": [command]}))
await ctx.response.send_message(f"Command sent to server. {box(command, 'json')}", ephemeral=True)
except websockets.exceptions.ConnectionClosed as e:
logger.error("WebSocket connection closed: %s", e)
await ctx.response.send_message(error("WebSocket connection closed."))
self.task.cancel()
self.retry_counter = 0
self.task = self.get_task()
else:
ctx = await self.bot.get_context(ctx)
if channel:
await channel.send(f"Received console command from {ctx.author.id}: {command[:1900]}", allowed_mentions=discord.AllowedMentions.none())
try:

View file

@ -1,12 +1,14 @@
# pylint: disable=cyclic-import
import json
import re
from typing import Optional, Union
from pathlib import Path
from typing import Optional, Tuple, Union
import aiohttp
import discord
import websockets
from pydactyl import PterodactylClient
from redbot.core.data_manager import bundled_data_path
from redbot.core.utils.chat_formatting import bold, pagify
from pterodactyl.config import config
@ -78,7 +80,12 @@ async def establish_websocket_connection(coginstance: Pterodactyl) -> None:
if join_message:
if chat_channel is not None:
if coginstance.bot.embed_requested(chat_channel):
await chat_channel.send(embed=await generate_join_leave_embed(join_message, True))
embed, img = await generate_join_leave_embed(coginstance=coginstance, username=join_message,join=True)
if img:
with open(img, 'rb') as file:
await chat_channel.send(embed=embed, file=file)
else:
await chat_channel.send(embed=embed)
else:
await chat_channel.send(f"{join_message} joined the game", allowed_mentions=discord.AllowedMentions.none())
@ -86,7 +93,12 @@ async def establish_websocket_connection(coginstance: Pterodactyl) -> None:
if leave_message:
if chat_channel is not None:
if coginstance.bot.embed_requested(chat_channel):
await chat_channel.send(embed=await generate_join_leave_embed(leave_message, False))
embed, img = await generate_join_leave_embed(coginstance=coginstance, username=leave_message,join=False)
if img:
with open(img, 'rb') as file:
await chat_channel.send(embed=embed, file=file)
else:
await chat_channel.send(embed=embed)
else:
await chat_channel.send(f"{leave_message} left the game", allowed_mentions=discord.AllowedMentions.none())
@ -94,7 +106,7 @@ async def establish_websocket_connection(coginstance: Pterodactyl) -> None:
if achievement_message:
if chat_channel is not None:
if coginstance.bot.embed_requested(chat_channel):
await chat_channel.send(embed=await generate_achievement_embed(achievement_message['username'], achievement_message['achievement'], achievement_message['challenge']))
await chat_channel.send(embed=await generate_achievement_embed(coginstance, achievement_message['username'], achievement_message['achievement'], achievement_message['challenge']))
else:
await chat_channel.send(f"{achievement_message['username']} has {'completed the challenge' if achievement_message['challenge'] else 'made the advancement'} {achievement_message['achievement']}")
@ -153,7 +165,7 @@ async def check_if_server_message(text: str) -> Union[bool, str]:
regex = await config.server_regex()
match: Optional[re.Match[str]] = re.match(regex, text)
if match:
logger.debug("Message is a server message")
logger.trace("Message is a server message")
return match.group(1)
return False
@ -162,7 +174,7 @@ async def check_if_chat_message(text: str) -> Union[bool, dict]:
match: Optional[re.Match[str]] = re.match(regex, text)
if match:
groups = {"username": match.group(1), "message": match.group(2)}
logger.debug("Message is a chat message\n%s", json.dumps(groups))
logger.trace("Message is a chat message\n%s", json.dumps(groups))
return groups
return False
@ -170,7 +182,7 @@ async def check_if_join_message(text: str) -> Union[bool, str]:
regex = await config.join_regex()
match: Optional[re.Match[str]] = re.match(regex, text)
if match:
logger.debug("Message is a join message")
logger.trace("Message is a join message")
return match.group(1)
return False
@ -178,7 +190,7 @@ async def check_if_leave_message(text: str) -> Union[bool, str]:
regex = await config.leave_regex()
match: Optional[re.Match[str]] = re.match(regex, text)
if match:
logger.debug("Message is a leave message")
logger.trace("Message is a leave message")
return match.group(1)
return False
@ -191,23 +203,23 @@ async def check_if_achievement_message(text: str) -> Union[bool, dict]:
groups["challenge"] = True
else:
groups["challenge"] = False
logger.debug("Message is an achievement message")
logger.trace("Message is an achievement message")
return groups
return False
async def get_info(username: str) -> Optional[dict]:
logger.debug("Retrieving player info for %s", username)
logger.verbose("Retrieving player info for %s", username)
endpoint = await config.api_endpoint()
async with aiohttp.ClientSession() as session:
async with session.get(f"https://playerdb.co/api/player/{endpoint}/{username}") as response:
if response.status == 200:
logger.debug("Player info retrieved for %s", username)
logger.verbose("Player info retrieved for %s", username)
return await response.json()
logger.error("Failed to retrieve player info for %s: %s", username, response.status)
logger.warning("Failed to retrieve player info for %s: %s", username, response.status)
return None
async def send_chat_discord(coginstance: Pterodactyl, username: str, message: str, avatar_url: str) -> None:
logger.debug("Sending chat message to Discord")
logger.trace("Sending chat message to Discord")
channel = coginstance.bot.get_channel(await config.chat_channel())
if channel is not None:
webhooks = await channel.webhooks()
@ -215,33 +227,37 @@ async def send_chat_discord(coginstance: Pterodactyl, username: str, message: st
if webhook is None:
webhook = await channel.create_webhook(name="Pterodactyl Chat")
await webhook.send(content=message, username=username, avatar_url=avatar_url, allowed_mentions=discord.AllowedMentions(everyone=False, roles=False, users=True))
logger.debug("Chat message sent to Discord")
logger.trace("Chat message sent to Discord")
else:
logger.warning("Chat channel not set. Skipping sending chat message to Discord")
async def generate_join_leave_embed(username: str, join: bool) -> discord.Embed:
async def generate_join_leave_embed(coginstance: Pterodactyl, username: str, join: bool) -> Tuple[discord.Embed, Optional[Union[str, Path]]]:
embed = discord.Embed()
embed.color = discord.Color.green() if join else discord.Color.red()
embed.description = await config.join_msg() if join else await config.leave_msg()
info = await get_info(username)
if info:
img = None
embed.set_author(name=username, icon_url=info['data']['player']['avatar'])
else:
embed.set_author(name=username, icon_url='https://seafsh.cc/u/j3AzqQ.png')
img = bundled_data_path(coginstance) / "unknown.png"
embed.set_author(name=username, icon_url='attachment://unknown.png')
embed.timestamp = discord.utils.utcnow()
return embed
return embed, img
async def generate_achievement_embed(username: str, achievement: str, challenge: bool) -> discord.Embed:
async def generate_achievement_embed(coginstance: Pterodactyl, username: str, achievement: str, challenge: bool) -> Tuple[discord.Embed, Optional[Union[str, Path]]]:
embed = discord.Embed()
embed.color = discord.Color.from_str('#a800a7') if challenge else discord.Color.from_str('#54fb54')
embed.description = f"{bold(username)} has {'completed the challenge' if challenge else 'made the advancement'} {bold(achievement)}"
info = await get_info(username)
if info:
img = None
embed.set_author(name=username, icon_url=info['data']['player']['avatar'])
else:
embed.set_author(name=username, icon_url='https://seafsh.cc/u/j3AzqQ.png')
img = bundled_data_path(coginstance) / "unknown.png"
embed.set_author(name=username, icon_url='attachment://unknown.png')
embed.timestamp = discord.utils.utcnow()
return embed
return embed, img
def mask_ip(string: str) -> str:
def check(match: re.Match[str]):

View file

@ -1,49 +1,44 @@
[tool.poetry]
[project]
name = "seacogs"
version = "0.1.0"
description = "My assorted cogs for Red-DiscordBot."
authors = ["SeaswimmerTheFsh"]
license = "MPL 2"
authors = [{name = "cswimr", email = "seaswimmerthefsh@gmail.com"}]
license = {file="LICENSE"}
readme = "README.md"
package-mode = false
requires-python = ">=3.11"
dependencies = [
"aiosqlite>=0.20.0",
"beautifulsoup4>=4.12.3",
"colorthief>=0.2.1",
"markdownify>=0.13.1",
"numpy>=2.1.2",
"phx-class-registry>=5.0.0",
"pillow>=10.4.0",
"py-dactyl",
"pydantic>=2.9.2",
"red-discordbot>=3.5.13",
"websockets>=13.1",
]
[tool.poetry.dependencies]
python = ">=3.11,<3.12"
Red-DiscordBot = "^3.5.9"
py-dactyl = "^2.0.4"
websockets = "^12.0"
pillow = "^10.3.0"
numpy = "^1.26.4"
pydantic = "^2.7.1"
colorthief = "^0.2.1"
beautifulsoup4 = "^4.12.3"
markdownify = "^0.12.1"
aiosqlite = "^0.20.0"
phx-class-registry = "^5.0.0"
[project.optional-dependencies]
documentation = [
"mkdocs>=1.6.1",
"mkdocs-git-authors-plugin>=0.9.0",
"mkdocs-git-revision-date-localized-plugin>=1.2.9",
"mkdocs-material[imaging]>=9.5.40",
"mkdocstrings[python]>=0.26.1",
"mkdocs-redirects>=1.2.1",
]
[tool.poetry.group.dev]
optional = true
[tool.uv]
dev-dependencies = [
"pylint>=3.3.1",
"ruff>=0.6.9",
"sqlite-web>=0.6.4",
]
[tool.poetry.group.dev.dependencies]
ruff = "^0.3.1"
pylint = "^3.1.0"
pipx = "^1.5.0"
sqlite-web = "^0.6.4"
[tool.poetry.group.docs]
optional = true
[tool.poetry.group.docs.dependencies]
mkdocs = "1.5.3"
mkdocstrings = {extras = ["python"], version = "0.24.0"}
mkdocs-git-authors-plugin = "0.7.2"
mkdocs-git-revision-date-localized-plugin = "1.2.2"
mkdocs-material = {extras = ["imaging"], version = "^9.5.2"}
mkdocs-redirects = "^1.2.1"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
[tool.uv.sources]
py-dactyl = { git = "https://github.com/cswimr/pydactyl" }
[tool.ruff]
# Exclude a variety of commonly ignored directories.

View file

@ -1,6 +1,6 @@
{
"author" : ["SeaswimmerTheFsh (seasw.)"],
"install_msg" : "Thank you for installing SeaUtils!\nYou can find the source code of this cog [here](https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs).",
"author" : ["cswimr"],
"install_msg" : "Thank you for installing SeaUtils!\nYou can find the source code of this cog [here](https://coastalcommits.com/cswimr/SeaCogs).",
"name" : "SeaUtils",
"short" : "A collection of useful utilities.",
"description" : "A collection of useful utilities.",

View file

@ -38,22 +38,26 @@ def format_rfc_text(text: str, number: int) -> str:
class SeaUtils(commands.Cog):
"""A collection of random utilities."""
__author__ = ["SeaswimmerTheFsh"]
__version__ = "1.0.0"
__author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
__git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
__version__ = "1.0.1"
__documentation__ = "https://seacogs.coastalcommits.com/seautils/"
def __init__(self, bot: Red) -> None:
self.bot = bot
def format_help_for_context(self, ctx: commands.Context) -> str:
pre_processed = super().format_help_for_context(ctx=ctx) or ""
pre_processed = super().format_help_for_context(ctx) or ""
n = "\n" if "\n\n" not in pre_processed else ""
text = [
f"{pre_processed}{n}",
f"Cog Version: **{self.__version__}**",
f"Author: {cf.humanize_list(items=self.__author__)}"
f"{cf.bold('Cog Version:')} [{self.__version__}]({self.__git__})",
f"{cf.bold('Author:')} {cf.humanize_list(self.__author__)}",
f"{cf.bold('Documentation:')} {self.__documentation__}",
]
return "\n".join(text)
def format_src(self, obj: Any) -> str:
"""A large portion of this code is repurposed from Zephyrkul's RTFS cog.
https://github.com/Zephyrkul/FluffyCogs/blob/master/rtfs/rtfs.py"""

1826
uv.lock Normal file

File diff suppressed because it is too large Load diff