Enhance AutoMod configuration by introducing event-specific settings and thresholds. Refactor Automod schemas to include event configurations, thresholds, and limits for ignored roles and channels. Update database interactions to support new structure and improve handling of AutoMod rules in cogs, ensuring dynamic thresholds for various events. Update dashboard components to accommodate new configurations and improve user experience in managing AutoMod settings.
Some checks failed
CI / Bot (Python) (push) Successful in 12s
CI / Dashboard (Next.js) (push) Failing after 9s

This commit is contained in:
TheOnlyMace
2026-07-21 23:05:05 +02:00
parent d24b810164
commit 36b32de34c
17 changed files with 842 additions and 165 deletions

View File

@@ -17,7 +17,7 @@ from api.dependencies import get_bot, limiter
from api.discord_auth import get_manageable_guild_ids, get_discord_headers
from api.db_manager import db_manager
from api.schemas import (
GuildSummary, GuildDetails, PrefixConfig, AutomodConfig,
GuildSummary, GuildDetails, PrefixConfig, AutomodConfig, AutomodEventConfig,
TicketConfig, LevelingConfig, LoggingConfig, TicketEmbed,
TicketCategory, LevelingEmbedStyle, PrefixUpdate,
AutomodUpdate, LevelingUpdate, LoggingUpdate, TicketUpdate,
@@ -30,11 +30,23 @@ from api.schemas import (
RRConfig, RRUpdate, ReactionRoleEntry,
InviteStat, InvitesLeaderboard
)
from utils.automod_settings import (
EVENT_ID_TO_DB,
THRESHOLD_META,
MAX_IGNORED,
NSFW_KEYWORDS,
normalize_event_id,
normalize_punishment,
ensure_thresholds_table,
get_all_thresholds,
set_thresholds,
)
from typing import TYPE_CHECKING, List, Optional
import math
import aiosqlite
import json
import os
import discord
if TYPE_CHECKING:
from core.axiom import Axiom
@@ -121,78 +133,264 @@ async def get_guild_automod(guild_id: int):
Retrieves the AutoMod configuration for a specific guild.
"""
db = await db_manager.get_connection(db_path('automod.db'))
# Check enabled status
await ensure_thresholds_table(db)
cursor = await db.execute("SELECT enabled FROM automod WHERE guild_id = ?", (guild_id,))
enabled_row = await cursor.fetchone()
enabled = bool(enabled_row[0]) if enabled_row else False
# Get punishments
cursor = await db.execute("SELECT event, punishment FROM automod_punishments WHERE guild_id = ?", (guild_id,))
punishments = {row[0]: row[1] for row in await cursor.fetchall()}
cursor = await db.execute(
"SELECT event, punishment FROM automod_punishments WHERE guild_id = ?",
(guild_id,),
)
raw_punishments = {row[0]: row[1] for row in await cursor.fetchall()}
# Get ignored items
cursor = await db.execute("SELECT type, id FROM automod_ignored WHERE guild_id = ?", (guild_id,))
events: dict[str, AutomodEventConfig] = {}
punishments_out: dict[str, str] = {}
for event_id, db_name in EVENT_ID_TO_DB.items():
present = db_name in raw_punishments
pun = raw_punishments.get(db_name)
if event_id == "anti_nsfw_link":
events[event_id] = AutomodEventConfig(
enabled=present,
punishment="block_message",
readonly_punishment=True,
)
if present:
punishments_out[event_id] = "block_message"
else:
norm = normalize_punishment(pun) if present else None
events[event_id] = AutomodEventConfig(
enabled=present,
punishment=norm or ("Mute" if present else None),
readonly_punishment=False,
)
if present and norm:
punishments_out[event_id] = norm
cursor = await db.execute(
"SELECT type, id FROM automod_ignored WHERE guild_id = ?", (guild_id,)
)
ignored_items = await cursor.fetchall()
ignored_roles = [row[1] for row in ignored_items if row[0] == 'role']
ignored_channels = [row[1] for row in ignored_items if row[0] == 'channel']
ignored_roles = [row[1] for row in ignored_items if row[0] == "role"]
ignored_channels = [row[1] for row in ignored_items if row[0] == "channel"]
# Get logging channel
cursor = await db.execute("SELECT log_channel FROM automod_logging WHERE guild_id = ?", (guild_id,))
cursor = await db.execute(
"SELECT log_channel FROM automod_logging WHERE guild_id = ?", (guild_id,)
)
logging_row = await cursor.fetchone()
logging_channel = logging_row[0] if logging_row else None
thresholds = await get_all_thresholds(guild_id)
return AutomodConfig(
guild_id=guild_id,
enabled=enabled,
punishments=punishments,
punishments=punishments_out,
events=events,
thresholds=thresholds,
threshold_meta=THRESHOLD_META,
ignored_roles=ignored_roles,
ignored_channels=ignored_channels,
logging_channel=logging_channel
logging_channel=logging_channel,
limits={"max_ignored_roles": MAX_IGNORED, "max_ignored_channels": MAX_IGNORED},
)
async def _sync_nsfw_rule(
bot: "Axiom",
guild_id: int,
*,
enabled: bool,
ignored_roles: list[int],
ignored_channels: list[int],
) -> None:
guild = bot.get_guild(guild_id)
if not guild:
return
existing = None
try:
rules = await guild.fetch_automod_rules()
existing = next((r for r in rules if r.name == "Anti NSFW Links"), None)
except (discord.Forbidden, discord.HTTPException):
return
if not enabled:
if existing:
try:
await existing.delete(reason="Automod NSFW disabled via dashboard")
except (discord.Forbidden, discord.HTTPException):
pass
return
exempt_roles = [discord.Object(i) for i in ignored_roles[:MAX_IGNORED]]
exempt_channels = [discord.Object(i) for i in ignored_channels[:MAX_IGNORED]]
try:
if existing:
await existing.edit(
exempt_roles=exempt_roles,
exempt_channels=exempt_channels,
enabled=True,
reason="Automod NSFW updated via dashboard",
)
else:
await guild.create_automod_rule(
name="Anti NSFW Links",
event_type=discord.AutoModRuleEventType.message_send,
trigger=discord.AutoModTrigger(
type=discord.AutoModRuleTriggerType.keyword,
keyword_filter=NSFW_KEYWORDS,
),
actions=[
discord.AutoModRuleAction(type=discord.AutoModRuleActionType.block_message),
],
enabled=True,
exempt_roles=exempt_roles,
exempt_channels=exempt_channels,
reason="Automod NSFW enabled via dashboard",
)
except (discord.Forbidden, discord.HTTPException) as e:
print(f"Automod NSFW sync error for guild {guild_id}: {e}")
@router.patch("/{guild_id}/automod", summary="Update AutoMod config", description="Partially updates the AutoMod configuration components.")
async def patch_guild_automod(guild_id: int, data: AutomodUpdate):
async def patch_guild_automod(
guild_id: int,
data: AutomodUpdate,
bot: "Axiom" = Depends(get_bot),
):
"""
Updates parts of the AutoMod configuration for a specific guild.
"""
db = await db_manager.get_connection(db_path('automod.db'))
await ensure_thresholds_table(db)
if data.enabled is not None:
await db.execute(
"INSERT OR REPLACE INTO automod (guild_id, enabled) VALUES (?, ?)",
(guild_id, 1 if data.enabled else 0)
(guild_id, 1 if data.enabled else 0),
)
if data.punishments is not None:
for event, punishment in data.punishments.items():
await db.execute(
"INSERT OR REPLACE INTO automod_punishments (guild_id, event, punishment) VALUES (?, ?, ?)",
(guild_id, event, punishment)
)
# Build canonical enabled events from events{} and/or legacy punishments{}
events_payload = data.events
if events_payload is None and data.punishments is not None:
events_payload = {}
for raw_key, raw_pun in data.punishments.items():
eid = normalize_event_id(raw_key)
if not eid:
continue
if eid == "anti_nsfw_link":
events_payload[eid] = AutomodEventConfig(
enabled=True, punishment="block_message", readonly_punishment=True
)
else:
pun = normalize_punishment(raw_pun) or "Mute"
events_payload[eid] = AutomodEventConfig(enabled=True, punishment=pun)
if data.ignored_roles is not None:
await db.execute("DELETE FROM automod_ignored WHERE guild_id = ? AND type = 'role'", (guild_id,))
for role_id in data.ignored_roles:
nsfw_enabled: bool | None = None
if events_payload is not None:
await db.execute(
"DELETE FROM automod_punishments WHERE guild_id = ?", (guild_id,)
)
for raw_id, cfg in events_payload.items():
eid = normalize_event_id(raw_id)
if not eid or not cfg.enabled:
if eid == "anti_nsfw_link":
nsfw_enabled = False
continue
db_name = EVENT_ID_TO_DB[eid]
if eid == "anti_nsfw_link":
await db.execute(
"INSERT INTO automod_punishments (guild_id, event, punishment) VALUES (?, ?, ?)",
(guild_id, db_name, "Block message"),
)
nsfw_enabled = True
else:
pun = normalize_punishment(cfg.punishment) or "Mute"
await db.execute(
"INSERT INTO automod_punishments (guild_id, event, punishment) VALUES (?, ?, ?)",
(guild_id, db_name, pun),
)
# Full-replace: omitted/disabled NSFW must drop the Discord rule
if nsfw_enabled is None:
nsfw_enabled = False
if data.thresholds is not None:
await set_thresholds(db, guild_id, data.thresholds)
ignored_roles = data.ignored_roles
ignored_channels = data.ignored_channels
if ignored_roles is not None:
ignored_roles = list(dict.fromkeys(ignored_roles))[:MAX_IGNORED]
await db.execute(
"DELETE FROM automod_ignored WHERE guild_id = ? AND type = 'role'",
(guild_id,),
)
for role_id in ignored_roles:
await db.execute(
"INSERT OR REPLACE INTO automod_ignored (guild_id, type, id) VALUES (?, 'role', ?)",
(guild_id, role_id)
(guild_id, role_id),
)
if data.ignored_channels is not None:
await db.execute("DELETE FROM automod_ignored WHERE guild_id = ? AND type = 'channel'", (guild_id,))
for channel_id in data.ignored_channels:
if ignored_channels is not None:
ignored_channels = list(dict.fromkeys(ignored_channels))[:MAX_IGNORED]
await db.execute(
"DELETE FROM automod_ignored WHERE guild_id = ? AND type = 'channel'",
(guild_id,),
)
for channel_id in ignored_channels:
await db.execute(
"INSERT OR REPLACE INTO automod_ignored (guild_id, type, id) VALUES (?, 'channel', ?)",
(guild_id, channel_id)
(guild_id, channel_id),
)
if data.logging_channel is not None:
if data.clear_logging:
await db.execute(
"INSERT OR REPLACE INTO automod_logging (guild_id, log_channel) VALUES (?, ?)",
(guild_id, data.logging_channel)
"DELETE FROM automod_logging WHERE guild_id = ?", (guild_id,)
)
elif data.logging_channel is not None:
if data.logging_channel == 0:
await db.execute(
"DELETE FROM automod_logging WHERE guild_id = ?", (guild_id,)
)
else:
await db.execute(
"INSERT OR REPLACE INTO automod_logging (guild_id, log_channel) VALUES (?, ?)",
(guild_id, data.logging_channel),
)
await db.commit()
# Resolve current ignore lists for NSFW sync if not in this payload
if ignored_roles is None or ignored_channels is None:
cursor = await db.execute(
"SELECT type, id FROM automod_ignored WHERE guild_id = ?", (guild_id,)
)
items = await cursor.fetchall()
if ignored_roles is None:
ignored_roles = [r[1] for r in items if r[0] == "role"]
if ignored_channels is None:
ignored_channels = [r[1] for r in items if r[0] == "channel"]
if nsfw_enabled is None and (data.ignored_roles is not None or data.ignored_channels is not None):
# Keep Discord rule exempts in sync when ignores change
cursor = await db.execute(
"SELECT 1 FROM automod_punishments WHERE guild_id = ? AND event = ?",
(guild_id, EVENT_ID_TO_DB["anti_nsfw_link"]),
)
nsfw_enabled = bool(await cursor.fetchone())
if nsfw_enabled is not None:
await _sync_nsfw_rule(
bot,
guild_id,
enabled=nsfw_enabled,
ignored_roles=ignored_roles or [],
ignored_channels=ignored_channels or [],
)
return {"status": "success", "guild_id": guild_id}
@router.get("/{guild_id}/tickets", response_model=TicketConfig, summary="Get Ticket config", description="Retrieves the support ticket system setup, categories, and staff roles.")

View File

@@ -12,7 +12,7 @@
# ╚══════════════════════════════════════════════════════════════════╝
from pydantic import BaseModel, HttpUrl
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional
# --- Bot Schemas ---
@@ -67,13 +67,24 @@ class PrefixConfig(BaseModel):
guild_id: int
prefix: str
class AutomodEventConfig(BaseModel):
enabled: bool = False
punishment: Optional[str] = None # Mute | Kick | Ban | block_message (NSFW)
readonly_punishment: bool = False
class AutomodConfig(BaseModel):
guild_id: int
enabled: bool
punishments: Dict[str, str]
ignored_roles: List[int]
ignored_channels: List[int]
logging_channel: Optional[int]
# Legacy flat map (API event ids or bot names) — kept for older clients
punishments: Dict[str, str] = {}
events: Dict[str, AutomodEventConfig] = {}
thresholds: Dict[str, Dict[str, float]] = {}
threshold_meta: Dict[str, List[Dict[str, Any]]] = {}
ignored_roles: List[int] = []
ignored_channels: List[int] = []
logging_channel: Optional[int] = None
limits: Dict[str, int] = {"max_ignored_roles": 10, "max_ignored_channels": 10}
class TicketCategory(BaseModel):
name: str
@@ -299,10 +310,14 @@ class TicketUpdate(BaseModel):
class AutomodUpdate(BaseModel):
enabled: Optional[bool] = None
punishments: Optional[Dict[str, str]] = None
punishments: Optional[Dict[str, str]] = None # legacy
events: Optional[Dict[str, AutomodEventConfig]] = None
thresholds: Optional[Dict[str, Dict[str, float]]] = None
ignored_roles: Optional[List[int]] = None
ignored_channels: Optional[List[int]] = None
# Use a sentinel-friendly optional: omit = no change; null = clear logging
logging_channel: Optional[int] = None
clear_logging: Optional[bool] = None
class LevelingUpdate(BaseModel):
enabled: Optional[bool] = None

View File

@@ -11,6 +11,7 @@
# ║ ║
# ╚══════════════════════════════════════════════════════════════════╝
from utils.db_paths import db_path
from utils.automod_settings import get_thresholds
import discord
from utils.emoji import TICK
@@ -23,7 +24,6 @@ import asyncio
class AntiEmojiSpam(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.emoji_threshold = 5
async def is_automod_enabled(self, guild_id):
async with aiosqlite.connect(db_path("automod.db")) as db:
@@ -118,17 +118,20 @@ class AntiEmojiSpam(commands.Cog):
emoji_count = len(emoji_pattern.findall(message.content))
th = await get_thresholds(guild_id, "anti_emoji_spam")
emoji_threshold = int(th.get("max_emojis", 5))
mute_minutes = float(th.get("mute_minutes", 1))
if emoji_count > self.emoji_threshold:
if emoji_count > emoji_threshold:
punishment = await self.get_punishment(guild_id)
action_taken = None
reason = f"Emoji Spam ({emoji_count} emojis)"
try:
if punishment == "Mute":
timeout_duration = discord.utils.utcnow() + timedelta(minutes=1)
timeout_duration = discord.utils.utcnow() + timedelta(minutes=mute_minutes)
await user.edit(timed_out_until=timeout_duration, reason=reason)
action_taken = "Muted for 1 minute"
action_taken = f"Muted for {int(mute_minutes)} minutes"
elif punishment == "Kick":
await user.kick(reason=reason)
action_taken = "Kicked"

View File

@@ -11,6 +11,7 @@
# ║ ║
# ╚══════════════════════════════════════════════════════════════════╝
from utils.db_paths import db_path
from utils.automod_settings import get_thresholds
import discord
from utils.emoji import TICK
@@ -105,15 +106,17 @@ class AntiInvite(commands.Cog):
if any(invite.code == invite_code for invite in invite):
return
th = await get_thresholds(guild_id, "anti_invites")
mute_minutes = float(th.get("mute_minutes", 12))
punishment = await self.get_punishment(guild_id)
action_taken = None
reason = "Posted an invite link"
try:
if punishment == "Mute":
timeout_duration = discord.utils.utcnow() + timedelta(minutes=12)
timeout_duration = discord.utils.utcnow() + timedelta(minutes=mute_minutes)
await user.edit(timed_out_until=timeout_duration, reason="Posted an invite link")
action_taken = "Muted for 12 minutes"
action_taken = f"Muted for {int(mute_minutes)} minutes"
elif punishment == "Kick":
await user.kick(reason="Posted an invite link")
action_taken = "Kicked"

View File

@@ -11,6 +11,7 @@
# ║ ║
# ╚══════════════════════════════════════════════════════════════════╝
from utils.db_paths import db_path
from utils.automod_settings import get_thresholds
import discord
from utils.emoji import TICK
@@ -22,7 +23,6 @@ import asyncio
class AntiMassMention(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.mass_mention_threshold = 5
async def is_automod_enabled(self, guild_id):
async with aiosqlite.connect(db_path("automod.db")) as db:
@@ -97,17 +97,21 @@ class AntiMassMention(commands.Cog):
return
th = await get_thresholds(guild_id, "anti_mass_mention")
mention_threshold = int(th.get("max_mentions", 5))
mute_minutes = float(th.get("mute_minutes", 3))
mention_count = message.content.count("<@")
if mention_count >= self.mass_mention_threshold:
if mention_count >= mention_threshold:
punishment = await self.get_punishment(guild_id)
action_taken = None
reason = f"Mass Mention ({mention_count} mentions)"
try:
if punishment == "Mute":
timeout_duration = discord.utils.utcnow() + timedelta(minutes=3)
timeout_duration = discord.utils.utcnow() + timedelta(minutes=mute_minutes)
await user.edit(timed_out_until=timeout_duration, reason=reason)
action_taken = "Muted for 3 minutes"
action_taken = f"Muted for {int(mute_minutes)} minutes"
elif punishment == "Kick":
await user.kick(reason=reason)
action_taken = "Kicked"

View File

@@ -11,6 +11,7 @@
# ║ ║
# ╚══════════════════════════════════════════════════════════════════╝
from utils.db_paths import db_path
from utils.automod_settings import get_thresholds
import discord
from utils.emoji import TICK
@@ -22,8 +23,6 @@ from datetime import timedelta
class AntiCaps(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.caps_threshold = 70
self.mute_duration = 2 * 60
async def is_automod_enabled(self, guild_id):
async with aiosqlite.connect(db_path("automod.db")) as db:
@@ -74,9 +73,6 @@ class AntiCaps(commands.Cog):
@commands.Cog.listener()
async def on_message(self, message):
if len(message.content) < 45:
return
if message.author.bot:
return
@@ -85,6 +81,14 @@ class AntiCaps(commands.Cog):
channel = message.channel
guild_id = guild.id
th = await get_thresholds(guild_id, "anti_caps")
min_length = int(th.get("min_length", 45))
caps_threshold = float(th.get("percent", 70))
mute_minutes = float(th.get("mute_minutes", 1))
if len(message.content) < min_length:
return
if not await self.is_automod_enabled(guild_id) or not await self.is_anti_caps_enabled(guild_id):
return
@@ -103,16 +107,16 @@ class AntiCaps(commands.Cog):
caps_count = sum(1 for c in message.content if c.isupper())
caps_percentage = (caps_count / len(message.content)) * 100
if caps_percentage > self.caps_threshold:
if caps_percentage > caps_threshold:
punishment = await self.get_punishment(guild_id)
action_taken = None
reason = "Excessive Caps"
try:
if punishment == "Mute":
timeout_duration = discord.utils.utcnow() + timedelta(minutes=1)
timeout_duration = discord.utils.utcnow() + timedelta(minutes=mute_minutes)
await user.edit(timed_out_until=timeout_duration, reason="Excessive Caps")
action_taken = "Muted for 1 minutes"
action_taken = f"Muted for {int(mute_minutes)} minutes"
elif punishment == "Kick":
await user.kick(reason="Excessive Caps")
action_taken = "Kicked"

View File

@@ -11,6 +11,7 @@
# ║ ║
# ╚══════════════════════════════════════════════════════════════════╝
from utils.db_paths import db_path
from utils.automod_settings import get_thresholds
import discord
from utils.emoji import TICK
@@ -107,15 +108,17 @@ class AntiLink(commands.Cog):
if self.spotify_pattern.search(message.content):
return
th = await get_thresholds(guild_id, "anti_link")
mute_minutes = float(th.get("mute_minutes", 7))
punishment = await self.get_punishment(guild_id)
action_taken = None
reason = "Posted a link"
try:
if punishment == "Mute":
timeout_duration = discord.utils.utcnow() + timedelta(minutes=7)
timeout_duration = discord.utils.utcnow() + timedelta(minutes=mute_minutes)
await user.edit(timed_out_until=timeout_duration, reason=reason)
action_taken = "Muted for 7 minutes"
action_taken = f"Muted for {int(mute_minutes)} minutes"
elif punishment == "Kick":
await user.kick(reason=reason)
action_taken = "Kicked"

View File

@@ -11,6 +11,7 @@
# ║ ║
# ╚══════════════════════════════════════════════════════════════════╝
from utils.db_paths import db_path
from utils.automod_settings import get_thresholds
import discord
from utils.emoji import TICK
@@ -22,8 +23,6 @@ from datetime import timedelta
class AntiSpam(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.spam_threshold = 5
self.mute_duration = 12 * 60
self.recent_messages = {}
async def is_automod_enabled(self, guild_id):
@@ -99,21 +98,26 @@ class AntiSpam(commands.Cog):
return
current_time = message.created_at.timestamp()
th = await get_thresholds(guild_id, "anti_spam")
window = float(th.get("window_seconds", 10))
spam_threshold = int(th.get("max_messages", 5))
mute_minutes = float(th.get("mute_minutes", 12))
user_messages = self.recent_messages.get(user.id, [])
user_messages = [msg for msg in user_messages if current_time - msg < 10]
user_messages = [msg for msg in user_messages if current_time - msg < window]
user_messages.append(current_time)
self.recent_messages[user.id] = user_messages
if len(user_messages) > self.spam_threshold:
if len(user_messages) > spam_threshold:
punishment = await self.get_punishment(guild_id)
action_taken = None
reason = "Spamming"
try:
if punishment == "Mute":
timeout_duration = discord.utils.utcnow() + timedelta(minutes=12)
timeout_duration = discord.utils.utcnow() + timedelta(minutes=mute_minutes)
await user.edit(timed_out_until=timeout_duration, reason="Spamming")
action_taken = "Muted for 12 minutes"
action_taken = f"Muted for {int(mute_minutes)} minutes"
elif punishment == "Kick":
await user.kick(reason="Spamming")
action_taken = "Kicked"

View File

@@ -181,6 +181,15 @@ class Automod(commands.Cog):
PRIMARY KEY (guild_id)
)
""")
await db.execute("""
CREATE TABLE IF NOT EXISTS automod_thresholds (
guild_id INTEGER,
event TEXT,
key TEXT,
value REAL,
PRIMARY KEY (guild_id, event, key)
)
""")
await db.commit()
@commands.hybrid_group(

Binary file not shown.

View File

@@ -0,0 +1,187 @@
# ╔══════════════════════════════════════════════════════════════════╗
# ║ ║
# ║ +-+-+-+-+-+-+-+-+ ║
# ║ |H|e|x|a|H|o|s|t| ║
# ║ +-+-+-+-+-+-+-+-+ ║
# ║ ║
# ║ © 2026 HexaHost — All Rights Reserved ║
# ║ ║
# ║ discord ── https://discord.gg/hexahost ║
# ║ github ── https://github.com/theoneandonlymace ║
# ║ ║
# ╚══════════════════════════════════════════════════════════════════╝
"""Shared AutoMod settings: event IDs, punishments, thresholds."""
from __future__ import annotations
from typing import Any
import aiosqlite
from utils.db_paths import db_path
# API id -> bot DB event name
EVENT_ID_TO_DB: dict[str, str] = {
"anti_spam": "Anti spam",
"anti_caps": "Anti caps",
"anti_link": "Anti link",
"anti_invites": "Anti invites",
"anti_mass_mention": "Anti mass mention",
"anti_emoji_spam": "Anti emoji spam",
"anti_nsfw_link": "Anti NSFW link",
}
# Also accept legacy UI ids
EVENT_ID_ALIASES: dict[str, str] = {
"anti_links": "anti_link",
"anti_mentions": "anti_mass_mention",
}
DB_EVENT_TO_ID: dict[str, str] = {v: k for k, v in EVENT_ID_TO_DB.items()}
VALID_PUNISHMENTS = frozenset({"Mute", "Kick", "Ban"})
PUNISHMENT_ALIASES: dict[str, str] = {
"mute": "Mute",
"kick": "Kick",
"ban": "Ban",
"Mute": "Mute",
"Kick": "Kick",
"Ban": "Ban",
"delete": "Mute", # legacy dashboard — map to Mute
"warn": "Mute",
}
MAX_IGNORED = 10
NSFW_KEYWORDS = [
"porn", "xxx", "adult", "sex", "nsfw", "xnxx", "onlyfans", "brazzers",
"xhamster", "xvideos", "pornhub", "redtube", "livejasmin", "youporn",
"tube8", "pornhat", "swxvid", "ixxx",
]
DEFAULT_THRESHOLDS: dict[str, dict[str, float]] = {
"anti_spam": {"max_messages": 5, "window_seconds": 10, "mute_minutes": 12},
"anti_caps": {"percent": 70, "min_length": 45, "mute_minutes": 1},
"anti_link": {"mute_minutes": 7},
"anti_invites": {"mute_minutes": 12},
"anti_mass_mention": {"max_mentions": 5, "mute_minutes": 3},
"anti_emoji_spam": {"max_emojis": 5, "mute_minutes": 1},
"anti_nsfw_link": {},
}
THRESHOLD_META: dict[str, list[dict[str, Any]]] = {
"anti_spam": [
{"key": "max_messages", "label": "Max messages", "min": 2, "max": 20, "step": 1},
{"key": "window_seconds", "label": "Window (seconds)", "min": 3, "max": 60, "step": 1},
{"key": "mute_minutes", "label": "Mute minutes", "min": 1, "max": 1440, "step": 1},
],
"anti_caps": [
{"key": "percent", "label": "Caps percent", "min": 40, "max": 100, "step": 1},
{"key": "min_length", "label": "Min message length", "min": 10, "max": 200, "step": 1},
{"key": "mute_minutes", "label": "Mute minutes", "min": 1, "max": 1440, "step": 1},
],
"anti_link": [
{"key": "mute_minutes", "label": "Mute minutes", "min": 1, "max": 1440, "step": 1},
],
"anti_invites": [
{"key": "mute_minutes", "label": "Mute minutes", "min": 1, "max": 1440, "step": 1},
],
"anti_mass_mention": [
{"key": "max_mentions", "label": "Max mentions", "min": 2, "max": 20, "step": 1},
{"key": "mute_minutes", "label": "Mute minutes", "min": 1, "max": 1440, "step": 1},
],
"anti_emoji_spam": [
{"key": "max_emojis", "label": "Max emojis", "min": 2, "max": 30, "step": 1},
{"key": "mute_minutes", "label": "Mute minutes", "min": 1, "max": 1440, "step": 1},
],
}
def normalize_event_id(raw: str) -> str | None:
if raw in EVENT_ID_TO_DB:
return raw
if raw in EVENT_ID_ALIASES:
return EVENT_ID_ALIASES[raw]
if raw in DB_EVENT_TO_ID:
return DB_EVENT_TO_ID[raw]
return None
def normalize_punishment(raw: str | None) -> str | None:
if raw is None:
return None
return PUNISHMENT_ALIASES.get(raw) or PUNISHMENT_ALIASES.get(raw.strip())
async def ensure_thresholds_table(db: aiosqlite.Connection) -> None:
await db.execute(
"""
CREATE TABLE IF NOT EXISTS automod_thresholds (
guild_id INTEGER,
event TEXT,
key TEXT,
value REAL,
PRIMARY KEY (guild_id, event, key)
)
"""
)
async def get_thresholds(guild_id: int, event_id: str) -> dict[str, float]:
"""Return merged defaults + DB overrides for an API event id."""
defaults = dict(DEFAULT_THRESHOLDS.get(event_id, {}))
db_event = EVENT_ID_TO_DB.get(event_id)
if not db_event:
return defaults
async with aiosqlite.connect(db_path("automod.db")) as db:
await ensure_thresholds_table(db)
cursor = await db.execute(
"SELECT key, value FROM automod_thresholds WHERE guild_id = ? AND event = ?",
(guild_id, db_event),
)
for key, value in await cursor.fetchall():
defaults[str(key)] = float(value)
return defaults
async def get_all_thresholds(guild_id: int) -> dict[str, dict[str, float]]:
out: dict[str, dict[str, float]] = {
eid: dict(vals) for eid, vals in DEFAULT_THRESHOLDS.items()
}
async with aiosqlite.connect(db_path("automod.db")) as db:
await ensure_thresholds_table(db)
cursor = await db.execute(
"SELECT event, key, value FROM automod_thresholds WHERE guild_id = ?",
(guild_id,),
)
for event, key, value in await cursor.fetchall():
eid = DB_EVENT_TO_ID.get(event)
if not eid:
continue
out.setdefault(eid, {})[str(key)] = float(value)
return out
async def set_thresholds(
db: aiosqlite.Connection,
guild_id: int,
thresholds: dict[str, dict[str, float]],
) -> None:
await ensure_thresholds_table(db)
for event_id, pairs in thresholds.items():
eid = normalize_event_id(event_id)
if not eid or eid == "anti_nsfw_link":
continue
db_event = EVENT_ID_TO_DB[eid]
allowed = DEFAULT_THRESHOLDS.get(eid, {})
for key, value in pairs.items():
if key not in allowed:
continue
await db.execute(
"""
INSERT OR REPLACE INTO automod_thresholds (guild_id, event, key, value)
VALUES (?, ?, ?, ?)
""",
(guild_id, db_event, key, float(value)),
)