[BoostUtils] new cog (WIP)

README.md

@@ -31,6 +31,7 @@ To add the cogs to your instance please do: `[p]repo add Seina-Cogs https://gith
 | StableDiffusion | 0.1.0            | <details><summary>Stable Diffusion using the Replicate API.</summary>Stable Diffusion using the Replicate API.</details>        |
 | ModManager  | 0.1.0 | <details><summary>Force ban/kick users.</summary>Force ban/kick users so that they stay in the ban/kick list even if someone tries to manually unban them.</details>  |
 | AutoDelete  | 0.1.0 | <details><summary>Auto delete messages in specific channels with fancy html logging.</summary>Auto delete messages in specific channels with fancy html logging.</details>  |
+| BoostUtils  | 0.1.0 | <details><summary>Booster Utilities. (WIP)</summary>Various nitro boosting utilities. (WORK IN PROGRESS)</details>                          |
 
 
 Any questions you can find [Melon](https://discord.com/oauth2/authorize?client_id=808706062013825036&scope=bot&permissions=1099511627767%20applications.commands) and myself over on [the support server](https://discord.gg/mXfYuMy92r)

boostutils/_tagscript.py

@@ -0,0 +1,53 @@
+from typing import Any, Dict, Final, List, final
+
+import TagScriptEngine as tse
+from redbot.core import commands
+from redbot.core.utils.chat_formatting import humanize_number
+
+boosted: Final[str] = "{member(mention)} just boosted the server."
+unboosted: Final[str] = "{member(mention)} just unboosted the server."
+
+TAGSCRIPT_LIMIT: Final[int] = 10_000
+
+blocks: List[tse.Block] = [
+    tse.LooseVariableGetterBlock(),
+    tse.AssignmentBlock(),
+    tse.CommandBlock(),
+    tse.EmbedBlock(),
+]
+
+engine: tse.Interpreter = tse.Interpreter(blocks)
+
+
+def process_tagscript(content: str, seed_variables: Dict[str, tse.Adapter] = {}) -> Dict[str, Any]:
+    output: tse.Response = engine.process(content, seed_variables)
+    kwargs: Dict[str, Any] = {}
+    if output.body:
+        kwargs["content"] = output.body[:2000]
+    if embed := output.actions.get("embed"):
+        kwargs["embed"] = embed
+    return kwargs
+
+
+class TagError(Exception):
+    """Base exception class."""
+
+
+@final
+class TagCharacterLimitReached(TagError):
+    """Raised when the TagScript character limit is reached."""
+
+    def __init__(self, limit: int, length: int):
+        super().__init__(
+            f"TagScript cannot be longer than {humanize_number(limit)} (**{humanize_number(length)}**)."
+        )
+
+
+@final
+class TagscriptConverter(commands.Converter[str]):
+    async def convert(self, ctx: commands.Context, argument: str) -> str:  # type: ignore
+        try:
+            await ctx.cog.validate_tagscript(argument)  # type: ignore
+        except TagError as e:
+            raise commands.BadArgument(str(e))
+        return argument

boostutils/abc.py

@@ -0,0 +1,24 @@
+from abc import ABC, ABCMeta, abstractmethod
+from typing import Any
+
+from redbot.core.bot import Red
+from redbot.core import commands, Config
+
+
+class CompositeMetaClass(commands.CogMeta, ABCMeta):
+    pass
+
+
+class MixinMeta(ABC, metaclass=CompositeMetaClass):
+    bot: Red
+    config: Config
+
+    def __init__(self, *_args: Any) -> None:
+        super().__init__(*_args)
+
+    @staticmethod
+    @abstractmethod
+    async def validate_tagscript(tagscript: str) -> bool: ...
+
+    @abstractmethod
+    def format_help_for_context(self, ctx: commands.Context) -> str: ...

boostutils/commands/message.py

@@ -0,0 +1,172 @@
+from typing import Dict, List, Literal, Optional, Union
+
+import discord
+from redbot.core import commands
+from redbot.core.utils.views import SimpleMenu
+from redbot.core.utils.chat_formatting import box, humanize_list
+
+from ..abc import MixinMeta, CompositeMetaClass
+from .._tagscript import TagscriptConverter
+from ..utils import group_embeds_by_fields
+
+
+class MessageCommands(MixinMeta, metaclass=CompositeMetaClass):
+    @commands.guild_only()
+    @commands.group(name="boostmessage", aliases=["boostmsg", "bmsg"])  # type: ignore
+    async def _message(self, _: commands.GuildContext):
+        """Configuration commands for boost messages."""
+
+    @_message.command(name="settings", aliases=["show", "showsettings", "ss"])
+    async def _message_settings(self, ctx: commands.GuildContext):
+        """See the boost messages settings configured for your server."""
+        cutoff = lambda x: x if len(x) < 1024 else x[:1021] + "..."  # noqa: E731
+        config: Dict[str, Union[bool, List[int], str]] = await self.config.guild(
+            ctx.guild
+        ).boost_message()
+        channels: List[int] = config["channels"]  # type: ignore
+        description: str = "**Enable:** {}\n\n".format(config["toggle"])
+        embed: discord.Embed = discord.Embed(
+            title="Boost Messages Settings for **__{}__**".format(ctx.guild.name),
+            description=description,
+            color=await ctx.embed_color(),
+        )
+        fields: List[Dict[str, Union[str, bool]]] = [
+            dict(
+                name="**Boost Message:**",
+                value=box(cutoff(config["boosted"])),
+                inline=False,
+            ),
+            dict(
+                name="**Unboost Message:**",
+                value=box(cutoff(config["unboosted"])),
+                inline=False,
+            ),
+            dict(
+                name="**Channels:**",
+                value=(
+                    humanize_list(
+                        [
+                            chan.mention
+                            for channel in channels
+                            if (chan := ctx.guild.get_channel(channel))
+                        ]
+                    )
+                    if channels
+                    else "No channels configured."
+                ),
+                inline=False,
+            ),
+        ]
+        embeds: List[discord.Embed] = [
+            embed,
+            *await group_embeds_by_fields(
+                *fields,
+                per_embed=3,
+                title="Boost Messages Settings for **__{}__**".format(ctx.guild.name),
+                color=await ctx.embed_color(),
+            ),
+        ]
+        await SimpleMenu(embeds).start(ctx)  # type: ignore
+
+    @_message.command(name="toggle")
+    @commands.admin_or_permissions(manage_guild=True)
+    async def _message_toggle(
+        self,
+        ctx: commands.GuildContext,
+        true_or_false: Optional[bool] = None,
+    ):
+        """
+        Enable or disable boost messages.
+
+        - Running the command with no arguments will disable the boost messages.
+        """
+        if true_or_false is None:
+            await self.config.guild(ctx.guild).boost_message.toggle.clear()  # type: ignore
+            await ctx.send(
+                "Boost message is now untoggled.",
+                reference=ctx.message.to_reference(fail_if_not_exists=False),
+                allowed_mentions=discord.AllowedMentions(replied_user=False),
+            )
+            return
+        await self.config.guild(ctx.guild).boost_message.toggle.set(true_or_false)  # type: ignore
+        await ctx.send(
+            f"Boost message is now {'toggle' if true_or_false else 'untoggled'}.",
+            reference=ctx.message.to_reference(fail_if_not_exists=False),
+            allowed_mentions=discord.AllowedMentions(replied_user=False),
+        )
+
+    @_message.command(name="channels")
+    @commands.admin_or_permissions(manage_guild=True)
+    async def _message_channels(
+        self,
+        ctx: commands.GuildContext,
+        add_or_remove: Literal["add", "remove"],
+        channels: commands.Greedy[discord.TextChannel],
+    ):
+        """Add or remove the channels for boost messages."""
+        async with self.config.guild(ctx.guild).boost_message.channels() as c:  # type: ignore
+            for channel in channels:
+                if add_or_remove.lower() == "add":
+                    if channel.id not in c:
+                        c.append(channel.id)
+                elif add_or_remove.lower() == "remove":
+                    if channel.id in c:
+                        c.remove(channel.id)
+        await ctx.send(
+            f"Successfully {'added' if add_or_remove.lower() == 'add' else 'removed'} {len(channels)} channels.",
+            reference=ctx.message.to_reference(fail_if_not_exists=False),
+            allowed_mentions=discord.AllowedMentions(replied_user=False),
+        )
+
+    @_message.group(name="message")
+    @commands.admin_or_permissions(manage_guild=True)
+    async def _message_message(self, _: commands.GuildContext):
+        """Configure boost and unboost messages."""
+
+    @_message_message.command(name="boosted", aliases=["boost"])
+    async def _message_boosted(
+        self, ctx: commands.GuildContext, *, message: Optional[TagscriptConverter] = None
+    ):
+        """
+        Configure the boost message.
+
+        - Running the command with no arguments will reset the boost message.
+        """
+        if message is None:
+            await self.config.guild(ctx.guild).boost_message.boosted.clear()  # type: ignore
+            await ctx.send(
+                "Cleared the boosted message.",
+                reference=ctx.message.to_reference(fail_if_not_exists=False),
+                allowed_mentions=discord.AllowedMentions(replied_user=False),
+            )
+            return
+        await self.config.guild(ctx.guild).boost_message.boosted.set(message)  # type: ignore
+        await ctx.send(
+            f"Changed the boosted message:\n{box(str(message), lang='json')}",
+            reference=ctx.message.to_reference(fail_if_not_exists=False),
+            allowed_mentions=discord.AllowedMentions(replied_user=False),
+        )
+
+    @_message_message.command(name="unboosted", aliases=["unboost"])
+    async def _message_unboosted(
+        self, ctx: commands.GuildContext, *, message: Optional[TagscriptConverter] = None
+    ):
+        """
+        Configure the unboost message.
+
+        - Running the command with no arguments will reset the unboost message.
+        """
+        if message is None:
+            await self.config.guild(ctx.guild).boost_message.unboosted.clear()  # type: ignore
+            await ctx.send(
+                "Cleared the unboosted message.",
+                reference=ctx.message.to_reference(fail_if_not_exists=False),
+                allowed_mentions=discord.AllowedMentions(replied_user=False),
+            )
+            return
+        await self.config.guild(ctx.guild).boost_message.unboosted.set(message)  # type: ignore
+        await ctx.send(
+            f"Changed the unboosted message:\n{box(str(message), lang='json')}",
+            reference=ctx.message.to_reference(fail_if_not_exists=False),
+            allowed_mentions=discord.AllowedMentions(replied_user=False),
+        )

boostutils/core.py

@@ -0,0 +1,57 @@
+from typing import Any, Dict, Final, List, Union
+
+from redbot.core.bot import Red
+from redbot.core import commands, Config
+from redbot.core.utils.chat_formatting import humanize_list
+
+from .abc import CompositeMetaClass
+from .events import EventMixin
+from .commands.message import MessageCommands
+from ._tagscript import boosted, unboosted, TAGSCRIPT_LIMIT, TagCharacterLimitReached
+
+
+class BoostUtils(
+    commands.Cog,
+    EventMixin,
+    MessageCommands,
+    metaclass=CompositeMetaClass,
+):
+    """Nitro Boost Utilities."""
+
+    __author__: Final[List[str]] = ["inthedark.org"]
+    __version__: Final[str] = "0.1.0"
+
+    def __init__(self, bot: Red, _args: Any) -> None:
+        super().__init__(*_args)
+        self.bot: Red = bot
+        self.config: Config = Config.get_conf(
+            self,
+            identifier=69_666_420,
+            force_registration=True,
+        )
+        default_guild: Dict[str, Dict[str, Union[bool, List[int], str]]] = {
+            "boost_message": {
+                "toggle": False,
+                "channels": [],
+                "boosted": boosted,
+                "unboosted": unboosted,
+            }
+        }
+        self.config.register_guild(**default_guild)
+
+    def format_help_for_context(self, ctx: commands.Context) -> str:
+        pre_processed: str = super().format_help_for_context(ctx)
+        n: str = "\n" if "\n\n" not in pre_processed else ""
+        text = [
+            f"{pre_processed}{n}",
+            f"**Author:** {humanize_list(self.__author__)}",
+            f"**Version:** {str(self.__version__)}",
+        ]
+        return "\n".join(text)
+
+    @staticmethod
+    async def validate_tagscript(tagscript: str) -> bool:
+        length = len(tagscript)
+        if length > TAGSCRIPT_LIMIT:
+            raise TagCharacterLimitReached(TAGSCRIPT_LIMIT, length)
+        return True