diff --git a/src/cogs/moderation/infraction.py b/src/cogs/moderation/infraction.py index d6c4f1c..c7d1096 100644 --- a/src/cogs/moderation/infraction.py +++ b/src/cogs/moderation/infraction.py @@ -30,8 +30,12 @@ def __init__(self, bot: APBot) -> None: def has_mod_role(self, member: Member) -> bool: allowed_role_names = {"Trial Chat Moderator", "Chat Moderator", "Admin"} allowed_role_ids = { - conf.get("bot_staff_role_id"), - conf.get("special_perms_role_id"), + role_id + for role_id in ( + conf.get("bot_staff_role_id"), + conf.get("special_perms_role_id"), + ) + if role_id is not None } perms = getattr(member, "guild_permissions", None) diff --git a/src/cogs/threads.py b/src/cogs/threads.py index 0549694..09b9529 100644 --- a/src/cogs/threads.py +++ b/src/cogs/threads.py @@ -3,11 +3,28 @@ from nextcord import Interaction, Embed, Object, ui, Color from nextcord.ext import commands from nextcord import slash_command -from config_handler import Config -config_path = "config.json" -conf = Config(config_path) blue = Color.teal() +QOTD_CHANNEL_NAME = "qotd" +QOTD_CURATOR_ROLE_NAME = "QOTD Curator" +QOTD_THREAD_CONFIG_NAME = "qotd_thread_check" + + +def is_qotd_thread(thread: nextcord.Thread) -> bool: + parent = getattr(thread, "parent", None) + return (getattr(parent, "name", "") or "").lower() == QOTD_CHANNEL_NAME + + +def has_qotd_curator_role(member) -> bool: + return any(getattr(role, "name", None) == QOTD_CURATOR_ROLE_NAME for role in getattr(member, "roles", [])) + + +def thread_created_at(thread: nextcord.Thread): + return getattr(thread, "created_at", None) or getattr(thread, "archive_timestamp", None) + + +def is_thread_closed(thread: nextcord.Thread) -> bool: + return bool(getattr(thread, "archived", False)) """ Under work for later updates """ class PingHelpers(ui.View): @@ -73,6 +90,121 @@ class Threads(commands.Cog): def __init__(self, bot: commands.Bot): self.bot = bot + async def get_thread_owner(self, thread: nextcord.Thread): + owner = getattr(thread, "owner", None) + if owner: + return owner + + owner_id = getattr(thread, "owner_id", None) + guild = getattr(thread, "guild", None) + if not owner_id or guild is None: + return None + + member = guild.get_member(owner_id) + if member: + return member + + try: + return await guild.fetch_member(owner_id) + except (nextcord.NotFound, nextcord.Forbidden, nextcord.HTTPException): + return None + + async def qotd_initial_check_done(self) -> bool: + config = await self.bot.db.base_db.read_bot_config(QOTD_THREAD_CONFIG_NAME) + return bool(config and config.get("initial_check_done")) + + async def mark_qotd_initial_check_done(self) -> None: + config = await self.bot.db.base_db.read_bot_config(QOTD_THREAD_CONFIG_NAME) or { + "name": QOTD_THREAD_CONFIG_NAME, + } + config["initial_check_done"] = True + await self.bot.db.base_db.update_bot_config(config) + + async def close_thread_if_open(self, thread: nextcord.Thread) -> bool: + if is_thread_closed(thread): + return False + + try: + await thread.edit(archived=True) + return True + except (nextcord.Forbidden, nextcord.HTTPException): + return False + + async def active_qotd_threads(self, parent, current_thread: nextcord.Thread) -> list[nextcord.Thread]: + return [ + candidate + for candidate in getattr(parent, "threads", []) + if candidate.id != current_thread.id + ] + + async def archived_qotd_threads( + self, + parent, + current_thread: nextcord.Thread, + *, + limit: int | None, + ) -> list[nextcord.Thread]: + if not hasattr(parent, "archived_threads"): + return [] + + threads = [] + try: + async for candidate in parent.archived_threads(limit=limit): + if candidate.id != current_thread.id: + threads.append(candidate) + except (nextcord.Forbidden, nextcord.HTTPException): + return [] + + return threads + + def dedupe_threads(self, threads: list[nextcord.Thread]) -> list[nextcord.Thread]: + deduped = {} + for thread in threads: + deduped[thread.id] = thread + return list(deduped.values()) + + async def get_all_previous_qotd_threads(self, thread: nextcord.Thread) -> list[nextcord.Thread]: + parent = thread.parent + candidates = await self.active_qotd_threads(parent, thread) + candidates.extend(await self.archived_qotd_threads(parent, thread, limit=None)) + return self.dedupe_threads(candidates) + + async def get_most_recent_previous_qotd_thread(self, thread: nextcord.Thread): + parent = thread.parent + candidates = await self.active_qotd_threads(parent, thread) + + if not candidates: + candidates.extend(await self.archived_qotd_threads(parent, thread, limit=1)) + + candidates = self.dedupe_threads(candidates) + if not candidates: + return None + + return max(candidates, key=lambda candidate: thread_created_at(candidate) or candidate.id) + + async def close_qotd_threads_for_new_thread(self, thread: nextcord.Thread) -> None: + if await self.qotd_initial_check_done(): + previous_thread = await self.get_most_recent_previous_qotd_thread(thread) + if previous_thread: + await self.close_thread_if_open(previous_thread) + return + + previous_threads = await self.get_all_previous_qotd_threads(thread) + for previous_thread in previous_threads: + await self.close_thread_if_open(previous_thread) + + await self.mark_qotd_initial_check_done() + + async def handle_qotd_thread_create(self, thread: nextcord.Thread) -> bool: + if not is_qotd_thread(thread): + return False + + owner = await self.get_thread_owner(thread) + if owner and has_qotd_curator_role(owner): + await self.close_qotd_threads_for_new_thread(thread) + + return True + @slash_command(name="resolve", description="Mark a thread as resolved and archive it.") async def resolve(self, interaction: Interaction): """ @@ -119,6 +251,9 @@ async def on_thread_create(self, thread: nextcord.Thread): - Sends initial message in forum post when created with option to dismiss. - After 10 minutes, provides an option to ping helpers. """ + if await self.handle_qotd_thread_create(thread): + return + if thread.parent.name in ["modmail", "important-updates"]: return diff --git a/tests/test_threads.py b/tests/test_threads.py new file mode 100644 index 0000000..8cf47e1 --- /dev/null +++ b/tests/test_threads.py @@ -0,0 +1,135 @@ +import asyncio +from datetime import datetime, timezone +from types import SimpleNamespace + +from cogs.threads import ( + QOTD_THREAD_CONFIG_NAME, + Threads, + has_qotd_curator_role, + is_qotd_thread, +) + + +class FakeThread: + def __init__(self, thread_id, parent, *, archived=False, created_at=None, owner=None): + self.id = thread_id + self.parent = parent + self.archived = archived + self.created_at = created_at or datetime(2026, 5, 25, tzinfo=timezone.utc) + self.owner = owner + self.edit_calls = [] + + async def edit(self, **kwargs): + self.edit_calls.append(kwargs) + if "archived" in kwargs: + self.archived = kwargs["archived"] + + +class FakeParent: + def __init__(self, name="qotd"): + self.name = name + self.threads = [] + self.archived = [] + self.archived_limits = [] + + async def archived_threads(self, limit=50): + self.archived_limits.append(limit) + count = 0 + for thread in self.archived: + if limit is not None and count >= limit: + break + count += 1 + yield thread + + +class FakeBaseDB: + def __init__(self, config=None): + self.config = config + self.updated_config = None + + async def read_bot_config(self, name): + if name == QOTD_THREAD_CONFIG_NAME: + return self.config + return None + + async def update_bot_config(self, config): + self.config = dict(config) + self.updated_config = dict(config) + + +def make_cog(config=None): + db = SimpleNamespace(base_db=FakeBaseDB(config)) + bot = SimpleNamespace(db=db) + return Threads(bot), db.base_db + + +def role(name): + return SimpleNamespace(name=name) + + +def test_is_qotd_thread_checks_parent_name(): + qotd_parent = SimpleNamespace(name="qotd") + other_parent = SimpleNamespace(name="math") + + assert is_qotd_thread(SimpleNamespace(parent=qotd_parent)) is True + assert is_qotd_thread(SimpleNamespace(parent=other_parent)) is False + + +def test_has_qotd_curator_role(): + assert has_qotd_curator_role(SimpleNamespace(roles=[role("QOTD Curator")])) is True + assert has_qotd_curator_role(SimpleNamespace(roles=[role("Student")])) is False + + +def test_first_qotd_run_closes_all_previous_threads_and_sets_flag(): + cog, db = make_cog(config=None) + parent = FakeParent() + curator = SimpleNamespace(roles=[role("QOTD Curator")]) + + new_thread = FakeThread(3, parent, owner=curator) + old_open = FakeThread(1, parent, archived=False) + old_closed = FakeThread(2, parent, archived=True) + parent.threads = [new_thread, old_open] + parent.archived = [old_closed] + + asyncio.run(cog.handle_qotd_thread_create(new_thread)) + + assert old_open.archived is True + assert old_open.edit_calls == [{"archived": True}] + assert old_closed.edit_calls == [] + assert db.config["initial_check_done"] is True + assert parent.archived_limits == [None] + + +def test_later_qotd_runs_only_close_most_recent_previous_thread(): + cog, _ = make_cog(config={"name": QOTD_THREAD_CONFIG_NAME, "initial_check_done": True}) + parent = FakeParent() + curator = SimpleNamespace(roles=[role("QOTD Curator")]) + + older = FakeThread(1, parent, archived=False, created_at=datetime(2026, 5, 20, tzinfo=timezone.utc)) + newest_previous = FakeThread(2, parent, archived=False, created_at=datetime(2026, 5, 24, tzinfo=timezone.utc)) + new_thread = FakeThread(3, parent, owner=curator, created_at=datetime(2026, 5, 25, tzinfo=timezone.utc)) + parent.threads = [older, newest_previous, new_thread] + + asyncio.run(cog.handle_qotd_thread_create(new_thread)) + + assert older.archived is False + assert older.edit_calls == [] + assert newest_previous.archived is True + assert newest_previous.edit_calls == [{"archived": True}] + assert parent.archived_limits == [] + + +def test_later_qotd_runs_check_one_archived_thread_when_no_active_previous_thread(): + cog, _ = make_cog(config={"name": QOTD_THREAD_CONFIG_NAME, "initial_check_done": True}) + parent = FakeParent() + curator = SimpleNamespace(roles=[role("QOTD Curator")]) + + new_thread = FakeThread(3, parent, owner=curator) + archived_previous = FakeThread(2, parent, archived=True) + parent.threads = [new_thread] + parent.archived = [archived_previous] + + asyncio.run(cog.handle_qotd_thread_create(new_thread)) + + assert archived_previous.edit_calls == [] + assert parent.archived_limits == [1]