Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
279 changes: 205 additions & 74 deletions src/cogs/moderation/infraction.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import nextcord
import re
from nextcord import slash_command, Permissions, Interaction, User, Embed, Member, TextChannel, Object, Color, SlashOption, SlashOption, SlashOption
import nextcord
from nextcord import slash_command, Permissions, Interaction, Embed, Member, TextChannel, Color, SlashOption
from nextcord.ext import commands
from typing import Optional
from bot_base import APBot
Expand All @@ -10,18 +10,76 @@

conf = load_optional_config()
COMMAND_GUILD_IDS = get_command_guild_ids(conf)
DISCORD_EPOCH_MS = 1420070400000
IMAGE_URL_EXTENSIONS = (".png", ".jpg", ".jpeg", ".gif", ".webp")


def to_snowflake(value):
def clip_text(value, limit: int = 1000) -> str:
text = str(value or "")
return text if len(text) <= limit else f"{text[:limit - 3]}..."


def is_possible_snowflake(value: int) -> bool:
text = str(value)
if not 17 <= len(text) <= 20:
return False

created_ms = (value >> 22) + DISCORD_EPOCH_MS
now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
return DISCORD_EPOCH_MS <= created_ms <= now_ms + 86400000


def add_snowflake_candidate(candidates: list[int], seen: set[int], value: str) -> None:
try:
candidate = int(value)
except (TypeError, ValueError):
return

if candidate in seen:
return

if not is_possible_snowflake(candidate):
return

seen.add(candidate)
candidates.append(candidate)


def snowflake_candidates(value) -> list[int]:
candidates: list[int] = []
seen: set[int] = set()

if value is None:
return None
if isinstance(value, int):
return value
if isinstance(value, str):
m = re.search(r"\d{17,20}", value)
if m:
return int(m.group(0))
return None
return candidates

if hasattr(value, "id"):
value = value.id

text = str(value)

for match in re.finditer(r"<@!?(\d{17,20})>", text):
add_snowflake_candidate(candidates, seen, match.group(1))

for match in re.finditer(r"(?<!\d)(\d{17,20})(?!\d)", text):
add_snowflake_candidate(candidates, seen, match.group(1))

for match in re.finditer(r"\d{18,}", text):
digits = match.group(0)

for size in range(min(20, len(digits) - 1), 16, -1):
add_snowflake_candidate(candidates, seen, digits[:size])
add_snowflake_candidate(candidates, seen, digits[-size:])

for size in range(min(20, len(digits) - 1), 16, -1):
for start in range(1, len(digits) - size):
add_snowflake_candidate(candidates, seen, digits[start:start + size])

return candidates


def to_snowflake(value):
candidates = snowflake_candidates(value)
return candidates[0] if candidates else None


def format_update_date(value):
Expand All @@ -41,6 +99,71 @@ def format_update_date(value):
return f"<t:{int(dt.timestamp())}:R>"


def format_action_name(actiontype: str) -> str:
action = actiontype or "unknown"

if action == "note":
return "Internal Note"

return action.replace("-", " ").title()


def format_duration(value) -> Optional[str]:
if not value:
return None

try:
duration_seconds = int(value.total_seconds()) if isinstance(value, timedelta) else int(value)
except (ValueError, TypeError, AttributeError):
return None

if duration_seconds <= 0:
return None

days, rem = divmod(duration_seconds, 86400)
hours, rem = divmod(rem, 3600)
minutes, seconds = divmod(rem, 60)

parts = []
if days:
parts.append(f"{days}d")
if hours:
parts.append(f"{hours}h")
if minutes:
parts.append(f"{minutes}m")
if seconds:
parts.append(f"{seconds}s")

return " ".join(parts) if parts else "0s"


def is_http_url(value) -> bool:
return str(value or "").lower().startswith(("http://", "https://"))


def can_embed_image(value) -> bool:
if not is_http_url(value):
return False

url = str(value).split("?", 1)[0].split("#", 1)[0].lower()
return url.endswith(IMAGE_URL_EXTENSIONS)


def format_attachment_line(value) -> str:
if not value:
return ""

attachment = str(value).strip()

if not attachment:
return ""

if is_http_url(attachment):
return f"Evidence: [View attachment]({attachment})\n"

return f"Evidence: `{clip_text(attachment, 200)}`\n"


def format_infraction_updates(updates):
if not updates:
return ""
Expand All @@ -58,7 +181,7 @@ def format_infraction_updates(updates):
moderator = update.get("moderator", "Unknown moderator")
moderator_id = to_snowflake(moderator)
moderator_text = f"<@{moderator_id}>" if moderator_id else str(moderator)
note = update.get("update") or update.get("note") or "No note provided"
note = clip_text(update.get("update") or update.get("note") or "No note provided", 500)
date_text = format_update_date(update.get("date"))
lines.append(f"- {note} — {moderator_text}, {date_text}")

Expand All @@ -68,6 +191,65 @@ class Infraction(commands.Cog):
def __init__(self, bot: APBot) -> None:
self.bot = bot

async def fetch_moderator_from_id(self, guild, moderator_id: int):
if guild is not None:
member = guild.get_member(moderator_id)

if member is not None:
return member

try:
member = await guild.fetch_member(moderator_id)

if member is not None:
return member
except (AttributeError, nextcord.NotFound, nextcord.Forbidden, nextcord.HTTPException):
pass

try:
cached_user = self.bot.get_user(moderator_id)

if cached_user is not None:
return cached_user
except AttributeError:
pass

try:
return await self.bot.fetch_user(moderator_id)
except (AttributeError, nextcord.NotFound, nextcord.Forbidden, nextcord.HTTPException):
return None

async def resolve_moderator(self, guild, raw_moderator):
candidates = snowflake_candidates(raw_moderator)

for moderator_id in candidates:
moderator = await self.fetch_moderator_from_id(guild, moderator_id)

if moderator is not None:
return moderator, moderator_id

fallback_id = candidates[0] if candidates else None
return None, fallback_id

def format_moderator(self, raw_moderator, moderator, moderator_id: Optional[int]) -> str:
if moderator is not None:
display = (
getattr(moderator, "global_name", None)
or getattr(moderator, "display_name", None)
or getattr(moderator, "name", None)
or str(moderator)
)
mention = getattr(moderator, "mention", f"<@{moderator_id}>")
return f"{mention} ({display})"

if moderator_id is not None:
return f"<@{moderator_id}> (not found)"

if raw_moderator:
return f"`{raw_moderator}` (invalid legacy moderator ID)"

return "Unknown moderator"

def has_mod_role(self, member: Member) -> bool:
allowed_role_names = {"Trial Chat Moderator", "Chat Moderator", "Admin"}
allowed_role_ids = {
Expand Down Expand Up @@ -136,44 +318,12 @@ async def warnings(self, inter: Interaction, member: Member):
total = len(infractions)

for index, inf in enumerate(infractions, start=1):
action = (
"Internal Note"
if inf.actiontype == "note"
else (inf.actiontype or "unknown").capitalize().replace("-", " ")
)
reason = inf.reason or "No reason provided"
action = format_action_name(inf.actiontype)
reason = clip_text(inf.reason or "No reason provided", 1400)

# ---- Moderator lookup (fixed) ----
raw_mod = getattr(inf, "moderator", None)
moderator_id = to_snowflake(raw_mod)

mod = None
if moderator_id:
# Try guild cache first (only if in a guild)
if inter.guild is not None:
mod = inter.guild.get_member(moderator_id)

# Fallback to API fetch
if mod is None:
try:
mod = await self.bot.fetch_user(moderator_id)
except nextcord.HTTPException:
mod = None

# Build moderator display:
# - If we found the user => mention + display name
# - If not => still mention the ID if we have one, else show whatever was stored
if mod:
display = getattr(mod, "global_name", None) or mod.name
mod_line = f"{mod.mention} ({display})"
else:
if moderator_id:
mod_line = f"<@{moderator_id}> (unknown)"
else:
mod_line = f"{raw_mod} (unknown)" if raw_mod else "Unknown moderator"
# -------------------------------


mod, moderator_id = await self.resolve_moderator(inter.guild, raw_mod)
mod_line = self.format_moderator(raw_mod, mod, moderator_id)

time_val = getattr(inf, "actiontime", None)

Expand All @@ -194,46 +344,27 @@ async def warnings(self, inter: Interaction, member: Member):
unix = int(time_val.timestamp())
timestamp = f"<t:{unix}:F> (<t:{unix}:R>)"


# ---- Duration ----
duration_line = ""
if getattr(inf, "duration", None):
try:
dur = inf.duration
duration_seconds = int(dur.total_seconds()) if isinstance(dur, timedelta) else int(dur)

days, rem = divmod(duration_seconds, 86400)
hours, rem = divmod(rem, 3600)
minutes, seconds = divmod(rem, 60)

parts = []
if days:
parts.append(f"{days}d")
if hours:
parts.append(f"{hours}h")
if minutes:
parts.append(f"{minutes}m")
if seconds:
parts.append(f"{seconds}s")

duration_line = f"Duration: {' '.join(parts) if parts else '0s'}\n"
except (ValueError, TypeError, AttributeError):
pass

duration = format_duration(getattr(inf, "duration", None))
duration_line = f"Duration: {duration}\n" if duration else ""
attachment_url = getattr(inf, "attachment_url", None)
attachment_line = format_attachment_line(attachment_url)
update_line = format_infraction_updates(getattr(inf, "update", []))
embed = Embed(
title=f"#{index} - {action}",
description=(
f"Index: {index}\n"
f"Reason: {reason}\n"
f"{duration_line}"
f"{attachment_line}"
f"Responsible Mod: {mod_line}\n"
f"{update_line}"
f"{index}/{total} infractions • {timestamp}"
),
color=color_map.get(getattr(inf, "actiontype", ""), Color.gold()),
)

if can_embed_image(attachment_url):
embed.set_image(url=attachment_url)

await inter.channel.send(embed=embed)

Expand Down
Loading
Loading