Close previous QOTD threads

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)

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
 

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]