Feature/betheprofessional

general/betheprofessional/cog.py

@@ -1,7 +1,7 @@
 import string
-from typing import List
+from typing import List, Union
 
-from discord import Role, Guild, Member, Embed
+from discord import Member, Embed, Role
 from discord.ext import commands
 from discord.ext.commands import guild_only, Context, CommandError, UserInputError
 
@@ -10,9 +10,9 @@
 from PyDrocsid.database import db, select
 from PyDrocsid.embeds import send_long_embed
 from PyDrocsid.translations import t
-from PyDrocsid.util import calculate_edit_distance, check_role_assignable
+from PyDrocsid.util import calculate_edit_distance
 from .colors import Colors
-from .models import BTPRole
+from .models import BTPUser, BTPTopic
 from .permissions import BeTheProfessionalPermission
 from ...contributor import Contributor
 from ...pubsub import send_to_changelog
@@ -25,72 +25,53 @@ def split_topics(topics: str) -> List[str]:
     return [topic for topic in map(str.strip, topics.replace(";", ",").split(",")) if topic]
 
 
-async def parse_topics(guild: Guild, topics: str, author: Member) -> List[Role]:
-    roles: List[Role] = []
-    all_topics: List[Role] = await list_topics(guild)
-    for topic in split_topics(topics):
-        for role in guild.roles:
-            if role.name.lower() == topic.lower():
-                if role in all_topics:
-                    break
-                if not role.managed and role >= guild.me.top_role:
-                    raise CommandError(t.youre_not_the_first_one(topic, author.mention))
-        else:
-            if all_topics:
-
-                def dist(name: str) -> int:
-                    return calculate_edit_distance(name.lower(), topic.lower())
-
-                best_match = min([r.name for r in all_topics], key=dist)
+async def split_parents(topics: List[str]) -> List[tuple[str, str, Union[BTPTopic, None]]]:
+    result: List[tuple[str, str, Union[BTPTopic, None]]] = []
+    for topic in topics:
+        topic_tree = topic.split("/")
+        if len(topic_tree) > 3 or len(topic_tree) < 2:
+            raise CommandError(t.group_parent_format_help)
+        group = topic_tree[0]
+        query = select(BTPTopic).filter_by(name=topic_tree[1])
+        parent: Union[BTPTopic, None, CommandError] = (
+            (await db.first(query) if await db.exists(query) else CommandError(t.parent_not_exists(topic_tree[1])))
+            if len(topic_tree) > 2
+            else None
+        )
+        if isinstance(parent, CommandError):
+            raise parent
+        topic = topic_tree[-1]
+        result.append((topic, group, parent))
+    return result
+
+
+async def parse_topics(topics_str: str) -> List[BTPTopic]:
+    topics: List[BTPTopic] = []
+    all_topics: List[BTPTopic] = await get_topics()
+    for topic in split_topics(topics_str):
+        query = select(BTPTopic).filter_by(name=topic)
+        topic_db = await db.first(query)
+        if not (await db.exists(query)) and len(all_topics) > 0:
+
+            def dist(name: str) -> int:
+                return calculate_edit_distance(name.lower(), topic.lower())
+
+            best_match = min([r.name for r in all_topics], key=dist)
+            if best_match:
                 raise CommandError(t.topic_not_found_did_you_mean(topic, best_match))
-            raise CommandError(t.topic_not_found(topic))
-        roles.append(role)
-    return roles
-
-
-async def list_topics(guild: Guild) -> List[Role]:
-    roles: List[Role] = []
-    async for btp_role in await db.stream(select(BTPRole)):
-        if (role := guild.get_role(btp_role.role_id)) is None:
-            await db.delete(btp_role)
-        else:
-            roles.append(role)
-    return roles
-
-
-async def unregister_roles(ctx: Context, topics: str, *, delete_roles: bool):
-    guild: Guild = ctx.guild
-    roles: List[Role] = []
-    btp_roles: List[BTPRole] = []
-    names = split_topics(topics)
-    if not names:
-        raise UserInputError
-
-    for topic in names:
-        for role in guild.roles:
-            if role.name.lower() == topic.lower():
-                break
-        else:
-            raise CommandError(t.topic_not_registered(topic))
-        if (btp_role := await db.first(select(BTPRole).filter_by(role_id=role.id))) is None:
-            raise CommandError(t.topic_not_registered(topic))
-
-        roles.append(role)
-        btp_roles.append(btp_role)
+            else:
+                raise CommandError(t.topic_not_found(topic))
+        elif not (await db.exists(query)):
+            raise CommandError(t.no_topics_registered)
+        topics.append(topic_db)
+    return topics
 
-    for role, btp_role in zip(roles, btp_roles):
-        if delete_roles:
-            check_role_assignable(role)
-            await role.delete()
-        await db.delete(btp_role)
 
-    embed = Embed(title=t.betheprofessional, colour=Colors.BeTheProfessional)
-    embed.description = t.topics_unregistered(cnt=len(roles))
-    await send_to_changelog(
-        ctx.guild,
-        t.log_topics_unregistered(cnt=len(roles), topics=", ".join(f"`{r}`" for r in roles)),
-    )
-    await send_long_embed(ctx, embed)
+async def get_topics() -> List[BTPTopic]:
+    topics: List[BTPTopic] = []
+    async for topic in await db.stream(select(BTPTopic)):
+        topics.append(topic)
+    return topics
 
 
 class BeTheProfessionalCog(Cog, name="Self Assignable Topic Roles"):
@@ -104,7 +85,7 @@ async def list_topics(self, ctx: Context):
         """
 
         embed = Embed(title=t.available_topics_header, colour=Colors.BeTheProfessional)
-        out = [role.name for role in await list_topics(ctx.guild)]
+        out = [topic.name for topic in await get_topics()]
         if not out:
             embed.colour = Colors.error
             embed.description = t.no_topics_registered
@@ -123,16 +104,17 @@ async def assign_topics(self, ctx: Context, *, topics: str):
         """
 
         member: Member = ctx.author
-        roles: List[Role] = [r for r in await parse_topics(ctx.guild, topics, ctx.author) if r not in member.roles]
-
-        for role in roles:
-            check_role_assignable(role)
-
-        await member.add_roles(*roles)
-
+        topics: List[BTPTopic] = [
+            topic
+            for topic in await parse_topics(topics)
+            if (await db.exists(select(BTPTopic).filter_by(id=topic.id)))
+               and not (await db.exists(select(BTPUser).filter_by(user_id=member.id, topic=topic.id)))  # noqa: W503
+        ]
+        for topic in topics:
+            await BTPUser.create(member.id, topic.id)
         embed = Embed(title=t.betheprofessional, colour=Colors.BeTheProfessional)
-        embed.description = t.topics_added(cnt=len(roles))
-        if not roles:
+        embed.description = t.topics_added(cnt=len(topics))
+        if not topics:
             embed.colour = Colors.error
 
         await reply(ctx, embed=embed)
@@ -143,21 +125,21 @@ async def unassign_topics(self, ctx: Context, *, topics: str):
         """
         remove one or more topics (use * to remove all topics)
         """
-
         member: Member = ctx.author
         if topics.strip() == "*":
-            roles: List[Role] = await list_topics(ctx.guild)
+            topics: List[BTPTopic] = await get_topics()
         else:
-            roles: List[Role] = await parse_topics(ctx.guild, topics, ctx.author)
-        roles = [r for r in roles if r in member.roles]
-
-        for role in roles:
-            check_role_assignable(role)
+            topics: List[BTPTopic] = await parse_topics(topics)
+        affected_topics: List[BTPTopic] = []
+        for topic in topics:
+            if await db.exists(select(BTPUser).filter_by(user_id=member.id, topic=topic.id)):
+                affected_topics.append(topic)
 
-        await member.remove_roles(*roles)
+        for topic in affected_topics:
+            await db.delete(topic)
 
         embed = Embed(title=t.betheprofessional, colour=Colors.BeTheProfessional)
-        embed.description = t.topics_removed(cnt=len(roles))
+        embed.description = t.topics_removed(cnt=len(affected_topics))
         await reply(ctx, embed=embed)
 
     @commands.command(name="*")
@@ -168,43 +150,40 @@ async def register_topics(self, ctx: Context, *, topics: str):
         register one or more new topics
         """
 
-        guild: Guild = ctx.guild
         names = split_topics(topics)
-        if not names:
+        topics: List[tuple[str, str, Union[BTPTopic, None]]] = await split_parents(names)
+        if not names or not topics:
             raise UserInputError
 
         valid_chars = set(string.ascii_letters + string.digits + " !#$%&'()+-./:<=>?[\\]^_`{|}~")
-        to_be_created: List[str] = []
-        roles: List[Role] = []
-        for topic in names:
-            if any(c not in valid_chars for c in topic):
+        registered_topics: List[tuple[str, str, Union[BTPTopic, None]]] = []
+        for topic in topics:
+            if any(c not in valid_chars for c in topic[0]):
                 raise CommandError(t.topic_invalid_chars(topic))
 
-            for role in guild.roles:
-                if role.name.lower() == topic.lower():
-                    break
+            if await db.exists(select(BTPTopic).filter_by(name=topic[0])):
+                raise CommandError(
+                    t.topic_already_registered(f"{topic[1]}/{topic[2].name + '/' if topic[2] else ''}{topic[0]}"),
+                )
             else:
-                to_be_created.append(topic)
-                continue
+                registered_topics.append(topic)
 
-            if await db.exists(select(BTPRole).filter_by(role_id=role.id)):
-                raise CommandError(t.topic_already_registered(topic))
-
-            check_role_assignable(role)
-
-            roles.append(role)
-
-        for name in to_be_created:
-            roles.append(await guild.create_role(name=name, mentionable=True))
-
-        for role in roles:
-            await BTPRole.create(role.id)
+        for registered_topic in registered_topics:
+            await BTPTopic.create(
+                registered_topic[0],
+                None,
+                registered_topic[1],
+                registered_topic[2].id if registered_topic[2] is not None else None,
+            )
 
         embed = Embed(title=t.betheprofessional, colour=Colors.BeTheProfessional)
-        embed.description = t.topics_registered(cnt=len(roles))
+        embed.description = t.topics_registered(cnt=len(registered_topics))
         await send_to_changelog(
             ctx.guild,
-            t.log_topics_registered(cnt=len(roles), topics=", ".join(f"`{r}`" for r in roles)),
+            t.log_topics_registered(
+                cnt=len(registered_topics),
+                topics=", ".join(f"`{r[0]}`" for r in registered_topics),
+            ),
         )
         await reply(ctx, embed=embed)
 
@@ -216,14 +195,33 @@ async def delete_topics(self, ctx: Context, *, topics: str):
         delete one or more topics
         """
 
-        await unregister_roles(ctx, topics, delete_roles=True)
+        topics: List[str] = split_topics(topics)
 
-    @commands.command(name="%")
-    @BeTheProfessionalPermission.manage.check
-    @guild_only()
-    async def unregister_topics(self, ctx: Context, *, topics: str):
-        """
-        unregister one or more topics without deleting the roles
-        """
+        delete_topics: list[BTPTopic] = []
 
-        await unregister_roles(ctx, topics, delete_roles=False)
+        for topic in topics:
+            if not await db.exists(select(BTPTopic).filter_by(name=topic)):
+                raise CommandError(t.topic_not_registered(topic))
+            else:
+                btp_topic = await db.first(select(BTPTopic).filter_by(name=topic))
+                delete_topics.append(btp_topic)
+                for child_topic in await db.all(
+                        select(BTPTopic).filter_by(parent=btp_topic.id),
+                ):  # TODO Recursive? Fix more level childs
+                    delete_topics.insert(0, child_topic)
+        for topic in delete_topics:
+            if topic.role_id is not None:
+                role: Role = ctx.guild.get_role(topic.role_id)
+                await role.delete()
+            for user_topic in await db.all(select(BTPUser).filter_by(topic=topic.id)):
+                await db.delete(user_topic)
+                await db.commit()
+            await db.delete(topic)
+
+        embed = Embed(title=t.betheprofessional, colour=Colors.BeTheProfessional)
+        embed.description = t.topics_unregistered(cnt=len(delete_topics))
+        await send_to_changelog(
+            ctx.guild,
+            t.log_topics_unregistered(cnt=len(delete_topics), topics=", ".join(f"`{r}`" for r in delete_topics)),
+        )
+        await send_long_embed(ctx, embed)

general/betheprofessional/models.py

@@ -1,16 +1,39 @@
-from typing import Union
+from typing import Union, Optional
 
 from PyDrocsid.database import db
-from sqlalchemy import Column, BigInteger
+from sqlalchemy import Column, BigInteger, String, Integer, ForeignKey
 
 
-class BTPRole(db.Base):
-    __tablename__ = "btp_role"
+class BTPTopic(db.Base):
+    __tablename__ = "btp_topic"
 
-    role_id: Union[Column, int] = Column(BigInteger, primary_key=True, unique=True)
+    id: Union[Column, int] = Column(Integer, primary_key=True)
+    name: Union[Column, str] = Column(String(255))
+    parent: Union[Column, int] = Column(Integer)
+    role_id: Union[Column, int] = Column(BigInteger)
+    group: Union[Column, str] = Column(String(255))
 
     @staticmethod
-    async def create(role_id: int) -> "BTPRole":
-        row = BTPRole(role_id=role_id)
+    async def create(
+        name: str,
+        role_id: Union[int, None],
+        group: str,
+        parent: Optional[Union[int, None]],
+    ) -> "BTPTopic":
+        row = BTPTopic(name=name, role_id=role_id, parent=parent, group=group)
+        await db.add(row)
+        return row
+
+
+class BTPUser(db.Base):
+    __tablename__ = "btp_users"
+
+    id: Union[Column, int] = Column(Integer, primary_key=True)
+    user_id: Union[Column, int] = Column(BigInteger)
+    topic: Union[Column, int] = Column(Integer, ForeignKey(BTPTopic.id))
+
+    @staticmethod
+    async def create(user_id: int, topic: int) -> "BTPUser":
+        row = BTPUser(user_id=user_id, topic=topic)
         await db.add(row)
         return row

general/betheprofessional/translations/en.yml

@@ -38,3 +38,6 @@ topics_unregistered:
 log_topics_unregistered:
   one: "The **topic** {topics} has been **removed**."
   many: "{cnt} **topics** have been **removed**: {topics}"
+
+parent_not_exists: "Parent `{}` doesn't exists"
+group_parent_format_help: "Please write `Group-Name/[Parent-Name/]Topic-Name`"