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
100 changes: 82 additions & 18 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
from nextcord import slash_command, Permissions, Interaction, User, Embed, Member, TextChannel, Object, Color, SlashOption, SlashOption, SlashOption
from nextcord.ext import commands
from typing import Optional
from bot_base import APBot
Expand All @@ -23,6 +23,47 @@ def to_snowflake(value):
return int(m.group(0))
return None


def format_update_date(value):
if isinstance(value, datetime):
dt = value
elif isinstance(value, str):
try:
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
return "unknown time"
else:
return "unknown time"

if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)

return f"<t:{int(dt.timestamp())}:R>"


def format_infraction_updates(updates):
if not updates:
return ""

lines = []
for update in updates[-5:]:
if isinstance(update, str):
lines.append(f"- {update}")
continue

if not isinstance(update, dict):
lines.append(f"- {update}")
continue

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"
date_text = format_update_date(update.get("date"))
lines.append(f"- {note} — {moderator_text}, {date_text}")

return "Notes:\n" + "\n".join(lines) + "\n"

class Infraction(commands.Cog):
def __init__(self, bot: APBot) -> None:
self.bot = bot
Expand Down Expand Up @@ -179,12 +220,15 @@ async def warnings(self, inter: Interaction, member: Member):
except (ValueError, TypeError, AttributeError):
pass

update_line = format_infraction_updates(getattr(inf, "update", []))
embed = Embed(
title=action,
title=f"#{index} - {action}",
description=(
f"Index: {index}\n"
f"Reason: {reason}\n"
f"{duration_line}"
f"Responsible Mod: {mod_line}\n"
f"{update_line}"
f"{index}/{total} infractions • {timestamp}"
),
color=color_map.get(getattr(inf, "actiontype", ""), Color.gold()),
Expand Down Expand Up @@ -240,27 +284,47 @@ async def editip(self, interaction: Interaction, member: Member, change: int):
)
)

# Work on later
# @slash_command(name="update", description="Update a member's infraction history.", default_member_permissions=Permissions(moderate_members=True))
# async def update(self, interaction: Interaction, user: User, infraction: int, update: str):
# member_config = await self.bot.db.base_db.read_user_config(user.id)
# infractions = member_config["infractions"]
# infraction_update = infractions[infraction]
@slash_command(
name="update",
description="Update a member's infraction history.",
default_member_permissions=Permissions(moderate_members=True),
guild_ids=COMMAND_GUILD_IDS,
)
async def update(
self,
interaction: Interaction,
member: Member,
index: int,
action: str = SlashOption(choices=["reason", "delete", "note"], description="Choose what to update."),
text: Optional[str] = SlashOption(description="New reason or note text. Leave empty only for delete.", required=False),
):
if not self.has_mod_role(interaction.user):
await interaction.response.send_message("You do not have permission to use this command.", ephemeral=True)
return

# if infraction_update.get('update') is None:
# infraction_update['update'] = []
if index < 1:
await interaction.response.send_message("Use the index shown in /warnings, starting at 1.", ephemeral=True)
return

# update_dict = {
# "moderator": interaction.user.mention,
# "update": update,
# "date": datetime.utcnow()
# }
if action in {"reason", "note"} and not text:
await interaction.response.send_message("Text is required for this update type.", ephemeral=True)
return

# infraction_update['update'].append(update_dict)
if action == "reason":
updated = await self.bot.db.base_db.update_infraction_reason(member.id, index, text)
response = f"Updated reason for infraction #{index} on {member.mention}."
elif action == "delete":
updated = await self.bot.db.base_db.delete_infraction(member.id, index)
response = f"Deleted infraction #{index} from {member.mention}'s history. Infraction points were not changed."
else:
updated = await self.bot.db.base_db.add_infraction_note(member.id, index, interaction.user.id, text)
response = f"Added a note to infraction #{index} on {member.mention}."

# await self.bot.db.base_db.update_user_config(user.id, member_config) # Save changes
if not updated:
await interaction.response.send_message("That infraction index does not exist.", ephemeral=True)
return

# await interaction.response.send_message("Infraction updated successfully.", ephemeral=True)
await interaction.response.send_message(response, ephemeral=True)

@slash_command(name="userip", description="View a member's infraction points.", default_member_permissions=Permissions(moderate_members=True), guild_ids=COMMAND_GUILD_IDS)
async def userip(self, interaction: Interaction, member: Member):
Expand Down
53 changes: 53 additions & 0 deletions src/database_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,59 @@ async def add_infraction(self, user_id: int, infraction: Infraction):




async def update_infraction_reason(self, user_id: int, index: int, reason: str) -> bool:
user_config = await self.read_user_config(user_id)
infractions = user_config.get("infractions", [])

if index < 1 or index > len(infractions):
return False

infraction = infractions[index - 1]
infraction["reason"] = reason
infraction["mute_reason"] = reason
await self.update_user_config(user_id, user_config)
return True

async def delete_infraction(self, user_id: int, index: int) -> bool:
user_config = await self.read_user_config(user_id)
infractions = user_config.get("infractions", [])

if index < 1 or index > len(infractions):
return False

del infractions[index - 1]
user_config["infractions"] = infractions
await self.update_user_config(user_id, user_config)
return True

async def add_infraction_note(self, user_id: int, index: int, moderator_id: int, note: str, actiontime=None) -> bool:
user_config = await self.read_user_config(user_id)
infractions = user_config.get("infractions", [])

if index < 1 or index > len(infractions):
return False

if actiontime is None:
actiontime = datetime.now(timezone.utc)

if isinstance(actiontime, datetime):
if actiontime.tzinfo is None:
actiontime = actiontime.replace(tzinfo=timezone.utc)
actiontime = actiontime.astimezone(timezone.utc).isoformat()

infraction = infractions[index - 1]
updates = infraction.get("update") or []
updates.append({
"moderator": moderator_id,
"update": note,
"date": actiontime,
})
infraction["update"] = updates

await self.update_user_config(user_id, user_config)
return True

async def get_user_infractions(self, user_id: int) -> list:
user_config = await self.read_user_config(user_id)
infractions = user_config.get("infractions", [])
Expand Down
88 changes: 58 additions & 30 deletions tests/test_database_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -212,43 +212,71 @@ def test_get_user_infractions_normalizes_old_mute_records():
assert infractions[0].attachment_url == "https://example.com/old.png"
assert infractions[0].actiontime == datetime(2024, 2, 3, 10, 15, 0, tzinfo=timezone.utc)


def test_wordle_start_season_deactivates_existing_active_season():
db = make_wordle_db([
{"guild_id": 1, "channel_id": 10, "start_date": "2026-05-01", "end_date": "2026-05-31", "active": True},
def test_update_infraction_reason_updates_new_and_legacy_reason_fields():
db = make_base_db([
{
"_id": 1,
"user_id": 5,
"infraction_points": 0,
"infractions": [{"type": "mute", "mute_reason": "old"}],
}
])

asyncio.run(db.start_season(1, 20, "2026-06-01", "2026-06-30"))

assert db.wordle_seasons.docs[0]["active"] is False
assert db.wordle_seasons.docs[1]["active"] is True
assert db.wordle_seasons.docs[1]["channel_id"] == 20


def test_wordle_leaderboard_sums_scores_and_sorts_lowest_first():
db = make_wordle_db()
result = asyncio.run(db.update_infraction_reason(5, 1, "new reason"))
stored = asyncio.run(db.read_user_config(5))

asyncio.run(db.save_result(1, 10, 111, "Push", 1801, 4, False, False, 4, "2026-05-01", 100, "2026-05-01", "2026-05-31", "user_share"))
asyncio.run(db.save_result(1, 10, 111, "Push", 1802, 3, False, True, 2, "2026-05-02", 101, "2026-05-01", "2026-05-31", "user_share"))
asyncio.run(db.save_result(1, 10, 222, "Other", 1801, None, True, False, 7, "2026-05-01", 102, "2026-05-01", "2026-05-31", "user_share"))
assert result is True
assert stored["infractions"][0]["reason"] == "new reason"
assert stored["infractions"][0]["mute_reason"] == "new reason"

rows = asyncio.run(db.get_leaderboard(1, "2026-05-01", "2026-05-31"))

assert rows[0]["user_id"] == 111
assert rows[0]["total_score"] == 6
assert rows[0]["games"] == 2
assert rows[0]["hard_games"] == 1
assert rows[1]["user_id"] == 222
assert rows[1]["failures"] == 1
def test_add_infraction_note_appends_update_entry():
db = make_base_db([
{
"_id": 1,
"user_id": 5,
"infraction_points": 0,
"infractions": [{"actiontype": "warn", "reason": "old"}],
}
])
date = datetime(2024, 1, 2, 3, 4, 5, tzinfo=timezone.utc)

result = asyncio.run(db.add_infraction_note(5, 1, 999, "extra context", date))
stored = asyncio.run(db.read_user_config(5))

assert result is True
assert stored["infractions"][0]["update"][0]["moderator"] == 999
assert stored["infractions"][0]["update"][0]["update"] == "extra context"
assert stored["infractions"][0]["update"][0]["date"] == "2024-01-02T03:04:05+00:00"


def test_delete_infraction_removes_only_selected_index():
db = make_base_db([
{
"_id": 1,
"user_id": 5,
"infraction_points": 10,
"infractions": [
{"actiontype": "warn", "reason": "first"},
{"actiontype": "mute", "reason": "second"},
],
}
])

result = asyncio.run(db.delete_infraction(5, 1))
stored = asyncio.run(db.read_user_config(5))

def test_wordle_summary_does_not_downgrade_hard_user_share():
db = make_wordle_db()
assert result is True
assert len(stored["infractions"]) == 1
assert stored["infractions"][0]["reason"] == "second"
assert stored["infraction_points"] == 10

asyncio.run(db.save_result(1, 10, 111, "Push", 1801, 3, False, True, 2, "2026-05-01", 100, "2026-05-01", "2026-05-31", "user_share"))
asyncio.run(db.save_result(1, 10, 111, "Push", None, 3, False, False, 3, "2026-05-01", 101, "2026-05-01", "2026-05-31", "wordle_bot_summary"))

rows = asyncio.run(db.get_leaderboard(1, "2026-05-01", "2026-05-31"))
def test_infraction_update_helpers_reject_bad_index():
db = make_base_db([
{"_id": 1, "user_id": 5, "infraction_points": 0, "infractions": []}
])

assert rows[0]["total_score"] == 2
assert rows[0]["hard_games"] == 1
assert asyncio.run(db.update_infraction_reason(5, 1, "reason")) is False
assert asyncio.run(db.add_infraction_note(5, 1, 999, "note")) is False
assert asyncio.run(db.delete_infraction(5, 1)) is False
46 changes: 13 additions & 33 deletions tests/test_moderation_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,39 +28,19 @@ def test_to_snowflake_parses_mentions_and_ints():
assert to_snowflake("not-a-user") is None
assert to_snowflake(None) is None

from cogs.moderation.commands import (
current_timeout_ends_after,
format_duration_seconds,
validate_selfmute_duration,
)


def test_validate_selfmute_duration_accepts_bounds():
assert validate_selfmute_duration("10m") == (600, None)
assert validate_selfmute_duration("168h") == (604800, None)
assert validate_selfmute_duration("1w") == (604800, None)


def test_validate_selfmute_duration_rejects_out_of_range():
assert validate_selfmute_duration("9m")[0] is None
assert validate_selfmute_duration("169h")[0] is None


def test_current_timeout_ends_after_prevents_shortening_existing_timeout():
unmute_time = datetime(2026, 5, 25, 12, 0, tzinfo=timezone.utc)
member = SimpleNamespace(communication_disabled_until=datetime(2026, 5, 25, 13, 0, tzinfo=timezone.utc))

assert current_timeout_ends_after(member, unmute_time) is True


def test_current_timeout_ends_after_allows_longer_selfmute():
unmute_time = datetime(2026, 5, 25, 12, 0, tzinfo=timezone.utc)
member = SimpleNamespace(communication_disabled_until=datetime(2026, 5, 25, 11, 0, tzinfo=timezone.utc))
from datetime import datetime, timezone
from cogs.moderation.infraction import format_infraction_updates

assert current_timeout_ends_after(member, unmute_time) is False

def test_format_infraction_updates_shows_note_moderator_and_date():
text = format_infraction_updates([
{
"moderator": 123456789012345678,
"update": "extra context",
"date": datetime(2024, 1, 1, tzinfo=timezone.utc),
}
])

def test_format_duration_seconds():
assert format_duration_seconds(600) == "10m"
assert format_duration_seconds(604800) == "7d"
assert format_duration_seconds(90061) == "1d 1h 1m 1s"
assert "Notes:" in text
assert "extra context" in text
assert "<@123456789012345678>" in text
Loading