Fix private tag storage

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/tags.py

@@ -1,99 +1,129 @@
-import re
-from nextcord import Embed, Interaction, SlashOption, slash_command, Attachment
+from nextcord import Attachment, Embed, Interaction, SlashOption, slash_command
 from nextcord.ext import commands
-from bot_base import APBot  # Ensure this is imported correctly
-from config_handler import Config
-config_path = "config.json"
-conf = Config(config_path)
+from bot_base import APBot
+from app_config import get_command_guild_ids, load_optional_config
+
+conf = load_optional_config()
+COMMAND_GUILD_IDS = get_command_guild_ids(conf)
+IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg", ".gif", ".webp")
+
+
+def normalize_tag_name(name: str) -> str:
+    return " ".join(name.strip().lower().split())
+
+
+def has_role(member, role_name: str) -> bool:
+    return any(getattr(role, "name", None) == role_name for role in getattr(member, "roles", []))
+
+
+def can_upload_tag_image(member) -> bool:
+    return has_role(member, "Honorable")
+
+
+def is_image_attachment(attachment: Attachment) -> bool:
+    content_type = getattr(attachment, "content_type", None)
+    filename = getattr(attachment, "filename", "") or ""
+
+    if content_type and content_type.startswith("image/"):
+        return True
+
+    return filename.lower().endswith(IMAGE_EXTENSIONS)
 
-# Only these roles can run the commands
-REQUIRED_ROLES = {"Honorable", "Chat Moderator", "Admin"}
 
 class Tags(commands.Cog):
     def __init__(self, bot: APBot) -> None:
         self.bot = bot
 
-    def _has_required_role(self, inter: Interaction) -> bool:
-        """Check if the user has at least one of the required roles."""
-        return any(role.name in REQUIRED_ROLES for role in inter.user.roles)
+    @slash_command(name="tag", description="Manage private tags", guild_ids=COMMAND_GUILD_IDS)
+    async def tag(self, inter: Interaction):
+        pass
 
-    @slash_command(name="tag", description="Manage tags", guild_ids=[conf.get("guild_id")])
-    async def _tag(self, inter: Interaction):
-        pass  # Root command
-
-    @_tag.subcommand(name="create", description="Create a new tag")
-    async def _tag_create(
+    @tag.subcommand(name="create", description="Create a private tag")
+    async def tag_create(
         self,
         inter: Interaction,
-        name: str = SlashOption("name", description="Name of the tag", required=True),
-        content: Attachment = SlashOption("content", description="Content of the tag as a file or image", required=True)
+        name: str = SlashOption(name="name", description="Name of the tag", required=True),
+        content: str = SlashOption(name="content", description="Text content for the tag", required=True),
+        image: Attachment = SlashOption(name="image", description="Optional image for Honorable users", required=False),
     ) -> None:
-        if not self._has_required_role(inter):
-            return await inter.send("You don't have permission to use this command.", ephemeral=True)
+        tag_name = normalize_tag_name(name)
+
+        if not tag_name:
+            await inter.send("Tag name cannot be empty.", ephemeral=True)
+            return
+
+        if not content.strip():
+            await inter.send("Tag content cannot be empty.", ephemeral=True)
+            return
+
+        image_url = None
+        if image is not None:
+            if not can_upload_tag_image(inter.user):
+                await inter.send("Only members with the Honorable role can upload images with tags.", ephemeral=True)
+                return
+
+            if not is_image_attachment(image):
+                await inter.send("Tag image must be an image file.", ephemeral=True)
+                return
 
-        content_url = content.url
-        if await self.bot.db.tags.exists(inter.guild.id, name):
-            return await inter.send("A tag with this name already exists.", ephemeral=True)
+            image_url = image.url
 
-        await self.bot.db.tags.create(inter.guild.id, inter.user.id, name, content_url)
-        await inter.send(f"Tag '{name}' created successfully!", ephemeral=False)
+        if await self.bot.db.tags.exists(inter.guild.id, inter.user.id, tag_name):
+            await inter.send("You already have a tag with this name.", ephemeral=True)
+            return
 
-    @_tag.subcommand(name="delete", description="Delete an existing tag")
-    async def _tag_delete(
+        await self.bot.db.tags.create(inter.guild.id, inter.user.id, tag_name, content.strip(), image_url)
+        await inter.send(f"Tag `{tag_name}` created successfully.", ephemeral=True)
+
+    @tag.subcommand(name="delete", description="Delete one of your private tags")
+    async def tag_delete(
         self,
         inter: Interaction,
-        name: str = SlashOption("name", description="Name of the tag", required=True)
+        name: str = SlashOption(name="name", description="Name of the tag", required=True),
     ) -> None:
-        if not self._has_required_role(inter):
-            return await inter.send("You don't have permission to use this command.", ephemeral=True)
+        tag_name = normalize_tag_name(name)
 
-        if not await self.bot.db.tags.exists(inter.guild.id, name):
-            return await inter.send("Tag not found.", ephemeral=True)
+        if not await self.bot.db.tags.exists(inter.guild.id, inter.user.id, tag_name):
+            await inter.send("Tag not found.", ephemeral=True)
+            return
 
-        await self.bot.db.tags.delete(inter.guild.id, name)
-        await inter.send(f"Tag '{name}' deleted successfully!", ephemeral=False)
+        await self.bot.db.tags.delete(inter.guild.id, inter.user.id, tag_name)
+        await inter.send(f"Tag `{tag_name}` deleted successfully.", ephemeral=True)
 
-    @_tag.subcommand(name="list", description="List all tags")
-    async def _tag_list(self, inter: Interaction) -> None:
-        if not self._has_required_role(inter):
-            return await inter.send("You don't have permission to use this command.", ephemeral=True)
+    @tag.subcommand(name="list", description="List your private tags")
+    async def tag_list(self, inter: Interaction) -> None:
+        tags = await self.bot.db.tags.get_all(inter.guild.id, inter.user.id)
 
-        tags = await self.bot.db.tags.get_all(inter.guild.id)
         if not tags:
-            return await inter.send("No tags available in this server.", ephemeral=False)
+            await inter.send("You do not have any saved tags.", ephemeral=True)
+            return
 
         tag_list = "\n".join(f"- {tag['name']}" for tag in tags)
-        await inter.send(embed=Embed(title="Available Tags", description=tag_list), ephemeral=False)
+        await inter.send(embed=Embed(title="Your Tags", description=tag_list), ephemeral=True)
 
-    @_tag.subcommand(name="display", description="Display a tag's content")
-    async def _tag_display(
+    @tag.subcommand(name="display", description="Display one of your private tags")
+    async def tag_display(
         self,
         inter: Interaction,
-        name: str = SlashOption("name", description="Name of the tag", required=True)
+        name: str = SlashOption(name="name", description="Name of the tag", required=True),
     ) -> None:
-        if not self._has_required_role(inter):
-            return await inter.send("You don't have permission to use this command.", ephemeral=True)
-
-        tag = await self.bot.db.tags.get_tag(inter.guild.id, name)
-        if not tag:
-            return await inter.send("Tag not found.", ephemeral=False)
-
-        tag_content = tag['content']
-        url_pattern = r"(https?://[^\s]+)"
-        urls = re.findall(url_pattern, tag_content)
-        text_content = re.sub(url_pattern, "", tag_content).strip()
-
-        if urls:
-            if text_content:
-                await inter.send(f"Tag '{name}' content:\n{text_content}", ephemeral=False)
-            for url in urls:
-                await inter.send(url, ephemeral=False)
-        else:
-            await inter.send(f"Tag '{name}' content:\n{text_content}", ephemeral=False)
+        tag_name = normalize_tag_name(name)
+        tag_data = await self.bot.db.tags.get_tag(inter.guild.id, inter.user.id, tag_name)
+
+        if not tag_data:
+            await inter.send("Tag not found.", ephemeral=True)
+            return
+
+        embed = Embed(title=tag_data["name"], description=tag_data.get("content") or "")
+        if tag_data.get("image_url"):
+            embed.set_image(url=tag_data["image_url"])
+
+        await inter.send(embed=embed, ephemeral=True)
 
     @commands.Cog.listener("on_ready")
-    async def _tag_on_ready(self) -> None:
+    async def tag_on_ready(self) -> None:
         print("Tag cog is ready.")
 
+
 def setup(bot: APBot):
     bot.add_cog(Tags(bot))

src/database_handler.py

@@ -704,31 +704,38 @@ class TagsDatabase(BaseDatabase):
     def __init__(self, conf=None):
         super().__init__(conf)
 
-    async def exists(self, guild_id: int, name: str) -> bool:
-        tag = await self.tags.find_one({"guild_id": guild_id, "name": name})
+    async def exists(self, guild_id: int, user_id: int, name: str) -> bool:
+        tag = await self.tags.find_one({"guild_id": guild_id, "user_id": user_id, "name": name})
         return tag is not None
 
-    async def create(self, guild_id: int, user_id: int, name: str, content: str) -> None:
-        tag_dict = await self.tags.find_one({"guild_id": guild_id, "name": name})
-        if tag_dict is None:
-            tag_dict = {"guild_id": guild_id, "user_id": user_id, "name": name, "content": content}
-            await self.tags.insert_one(tag_dict)
-        else:
+    async def create(self, guild_id: int, user_id: int, name: str, content: str, image_url: Optional[str] = None) -> None:
+        tag_dict = await self.tags.find_one({"guild_id": guild_id, "user_id": user_id, "name": name})
+        if tag_dict is not None:
             raise ValueError("Tag with this name already exists.")
 
-    async def delete(self, guild_id: int, name: str) -> None:
-        await self.tags.delete_one({"guild_id": guild_id, "name": name})
+        await self.tags.insert_one({
+            "guild_id": guild_id,
+            "user_id": user_id,
+            "name": name,
+            "content": content,
+            "image_url": image_url,
+        })
+
+    async def delete(self, guild_id: int, user_id: int, name: str) -> None:
+        await self.tags.delete_one({"guild_id": guild_id, "user_id": user_id, "name": name})
 
-    async def update(self, guild_id: int, name: str, new_content: str) -> None:
-        await self.tags.update_one({"guild_id": guild_id, "name": name}, {"$set": {"content": new_content}})
+    async def update(self, guild_id: int, user_id: int, name: str, new_content: str, image_url: Optional[str] = None) -> None:
+        await self.tags.update_one(
+            {"guild_id": guild_id, "user_id": user_id, "name": name},
+            {"$set": {"content": new_content, "image_url": image_url}},
+        )
 
-    async def get_all(self, guild_id: int) -> list:
-        tags = await self.tags.find({"guild_id": guild_id}).to_list(length=None)
-        return tags
+    async def get_all(self, guild_id: int, user_id: int) -> list:
+        tags = await self.tags.find({"guild_id": guild_id, "user_id": user_id}).to_list(length=None)
+        return sorted(tags, key=lambda tag: tag.get("name", ""))
 
-    async def get_tag(self, guild_id: int, name: str) -> Optional[dict]:
-        tag = await self.tags.find_one({"guild_id": guild_id, "name": name})
-        return tag if tag else None
+    async def get_tag(self, guild_id: int, user_id: int, name: str) -> Optional[dict]:
+        return await self.tags.find_one({"guild_id": guild_id, "user_id": user_id, "name": name})
 
     async def clear_all_tags(self) -> None:
         await self.tags.delete_many({})

tests/test_tags.py

@@ -0,0 +1,125 @@
+import asyncio
+import copy
+from types import SimpleNamespace
+
+from cogs.tags import can_upload_tag_image, is_image_attachment, normalize_tag_name
+from database_handler import TagsDatabase
+
+
+class FakeCursor:
+    def __init__(self, docs):
+        self.docs = docs
+
+    async def to_list(self, length=None):
+        return copy.deepcopy(self.docs)
+
+
+class FakeCollection:
+    def __init__(self):
+        self.docs = []
+
+    def matches(self, doc, query):
+        return all(doc.get(key) == value for key, value in query.items())
+
+    async def find_one(self, query):
+        for doc in self.docs:
+            if self.matches(doc, query):
+                return copy.deepcopy(doc)
+        return None
+
+    async def insert_one(self, doc):
+        self.docs.append(copy.deepcopy(doc))
+
+    async def delete_one(self, query):
+        self.docs = [doc for doc in self.docs if not self.matches(doc, query)]
+
+    async def update_one(self, query, update):
+        for doc in self.docs:
+            if self.matches(doc, query):
+                doc.update(update.get("$set", {}))
+                return
+
+    def find(self, query):
+        return FakeCursor([doc for doc in self.docs if self.matches(doc, query)])
+
+    async def delete_many(self, query):
+        self.docs = [doc for doc in self.docs if not self.matches(doc, query)]
+
+
+def make_tags_db():
+    db = object.__new__(TagsDatabase)
+    db.tags = FakeCollection()
+    return db
+
+
+def test_normalize_tag_name():
+    assert normalize_tag_name("  Unit   Circle  ") == "unit circle"
+
+
+def test_can_upload_tag_image_requires_honorable():
+    member = SimpleNamespace(roles=[SimpleNamespace(name="Honorable")])
+    other = SimpleNamespace(roles=[SimpleNamespace(name="Student")])
+
+    assert can_upload_tag_image(member) is True
+    assert can_upload_tag_image(other) is False
+
+
+def test_is_image_attachment_checks_content_type_or_extension():
+    assert is_image_attachment(SimpleNamespace(content_type="image/png", filename="file.bin")) is True
+    assert is_image_attachment(SimpleNamespace(content_type=None, filename="graph.PNG")) is True
+    assert is_image_attachment(SimpleNamespace(content_type="text/plain", filename="notes.txt")) is False
+
+
+def test_tags_database_keeps_user_tags_private():
+    db = make_tags_db()
+
+    asyncio.run(db.create(1, 10, "z tag", "second", None))
+    asyncio.run(db.create(1, 10, "unit circle", "mine", None))
+    asyncio.run(db.create(1, 20, "unit circle", "theirs", "https://example.com/a.png"))
+
+    my_tag = asyncio.run(db.get_tag(1, 10, "unit circle"))
+    their_tag = asyncio.run(db.get_tag(1, 20, "unit circle"))
+    my_tags = asyncio.run(db.get_all(1, 10))
+
+    assert my_tag["content"] == "mine"
+    assert their_tag["content"] == "theirs"
+    assert [tag["name"] for tag in my_tags] == ["unit circle", "z tag"]
+    assert my_tags[0]["user_id"] == 10
+
+    asyncio.run(db.delete(1, 10, "unit circle"))
+
+    assert asyncio.run(db.get_tag(1, 10, "unit circle")) is None
+    assert asyncio.run(db.get_tag(1, 20, "unit circle")) is not None
+
+
+def test_tags_database_blocks_duplicate_names_for_same_user():
+    db = make_tags_db()
+
+    asyncio.run(db.create(1, 10, "unit circle", "mine", None))
+
+    try:
+        asyncio.run(db.create(1, 10, "unit circle", "duplicate", None))
+    except ValueError:
+        pass
+    else:
+        raise AssertionError("Expected duplicate tag creation to fail")
+
+    tags = asyncio.run(db.get_all(1, 10))
+    assert len(tags) == 1
+    assert tags[0]["content"] == "mine"
+
+
+def test_tags_database_update_is_user_scoped():
+    db = make_tags_db()
+
+    asyncio.run(db.create(1, 10, "unit circle", "mine", None))
+    asyncio.run(db.create(1, 20, "unit circle", "theirs", None))
+
+    asyncio.run(db.update(1, 10, "unit circle", "updated", "https://example.com/new.png"))
+
+    my_tag = asyncio.run(db.get_tag(1, 10, "unit circle"))
+    their_tag = asyncio.run(db.get_tag(1, 20, "unit circle"))
+
+    assert my_tag["content"] == "updated"
+    assert my_tag["image_url"] == "https://example.com/new.png"
+    assert their_tag["content"] == "theirs"