Add repl.py

v_repl_bot/__init__.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+__all__ = ("ReplBot",)
+
+import asyncio
+from inspect import cleandoc as strip
+from itertools import chain
+from pathlib import Path
+from time import time as get_time
+from traceback import format_exc as format_exception
+
+from aiohttp import ClientSession as AIOHTTPSession
+from discord import AllowedMentions, Game, Intents, Webhook
+from discord.ext.commands import (
+    Bot as CommandsBot,
+    ExtensionFailed,
+    NoEntryPointError,
+    when_mentioned_or,
+)
+
+
+class ReplBot(CommandsBot):
+    ready_timestamp: float
+    log_webhook: Webhook
+    session: AIOHTTPSession
+
+    def __init__(self, *, token: str, webhook_url: str) -> None:
+        self.token = token
+        self.webhook_url = webhook_url
+
+        super().__init__(
+            command_prefix = when_mentioned_or("&"),
+            strip_after_prefix = True,
+            case_insensitive = True,
+            status = Game("on play.minearchy.com"),
+            owner_ids = { 512640455834337290 },
+            allowed_mentions = AllowedMentions.none(),
+            max_messages = 100,
+            intents = Intents(
+                messages = True,
+                message_content = True,
+            ),
+            help_attrs = dict(
+                brief = "Sends help.",
+                help = "Sends all the commands of the bot, or help of a specific command or module.",
+            ),
+        )
+
+    async def on_ready(self) -> None:
+        print(
+            strip(
+                f"""
+                Connected to Discord!
+                User: {self.user}
+                ID: {self.user.id}
+                """
+            )
+        )
+
+        self.ready_timestamp = get_time()
+        await self.log_webhook.send("Bot is now online!")
+
+    async def load_extensions(self) -> None:
+        cogs = Path(__file__).parent / "cogs"
+        for file_name in chain(
+            map(
+                lambda file_path: ".".join(file_path.relative_to(cogs.parent.parent).parts)[:-3],
+                cogs.rglob("*.py"),
+            ),
+            ("jishaku",),
+        ):
+            try:
+                await self.load_extension(file_name)
+                print(f"Loaded {file_name}")
+            except (ExtensionFailed, NoEntryPointError):
+                print(f"Couldn't load {file_name}:\n{format_exception()}")
+
+    def run(self) -> None:
+        async def runner() -> None:
+            async with self, AIOHTTPSession() as self.session:
+                self.log_webhook = Webhook.from_url(
+                    self.webhook_url, session = self.session, bot_token = self.token
+                )
+                await self.load_extensions()
+                await self.start(self.token)
+
+        try:
+            asyncio.run(runner())
+        except KeyboardInterrupt:
+            pass

v_repl_bot/__main__.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import json
+from os import environ as env
+from pathlib import Path
+
+import uvloop
+
+from . import ReplBot
+
+
+def main() -> None:
+    uvloop.install()
+
+    config = json.loads(
+        (
+            Path(__file__).parent / "config.json"
+        ).read_text()
+    )
+
+    env[f"JISHAKU_HIDE"] = "True"
+    env[f"JISHAKU_NO_UNDERSCORE"] = "True"
+
+    bot = ReplBot(
+        token = config["BOT_TOKEN"],
+        webhook_url = config["WEBHOOK_URL"]
+    )
+
+    bot.run()

v_repl_bot/cogs/error_handler.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import asyncio
+from contextlib import suppress as suppress_error
+from traceback import format_exception
+from typing import TYPE_CHECKING
+
+from discord import HTTPException
+from discord.ext.commands import (
+    ChannelNotFound,
+    Cog,
+    CommandNotFound,
+    MissingPermissions,
+    MissingRequiredArgument,
+    NoPrivateMessage,
+    NotOwner,
+    TooManyArguments,
+)
+
+if TYPE_CHECKING:
+    from discord.ext.commands import CommandError, Context
+
+    from .. import ReplBot
+
+
+class ErrorHandler(Cog):
+    def __init__(self, bot: ReplBot) -> None:
+        self.bot = bot
+
+    @Cog.listener()
+    async def on_command_error(self, ctx: Context, error: CommandError) -> None:
+        if hasattr(ctx.command, "on_error"):
+            return
+
+        if cog := ctx.cog:
+            if cog._get_overridden_method(cog.cog_command_error) is not None:
+                return
+
+        ignored = (CommandNotFound,)
+        error = getattr(error, "original", error)
+
+        if isinstance(error, ignored):
+            return
+
+        elif isinstance(error, NoPrivateMessage):
+            with suppress_error(HTTPException):
+                await ctx.author.send(
+                    f"The command `{ctx.command.qualified_name}` cannot be used in DMs."
+                )
+
+        elif isinstance(error, (MissingPermissions, NotOwner)):
+            await ctx.reply("You can't use this command!")
+
+        elif isinstance(error, MissingRequiredArgument):
+            await ctx.reply(f"Missing a required argument: `{error.param.name}`.")
+
+        elif isinstance(error, TooManyArguments):
+            await ctx.reply("Too many arguments.")
+
+        elif isinstance(error, ChannelNotFound):
+            await ctx.reply("Invalid channel.")
+
+        else:
+            trace = "".join(format_exception(type(error), error, error.__traceback__))
+            print(f"Ignoring exception in command {ctx.command}:\n{trace}")
+
+            await asyncio.gather(
+                self.bot.log_webhook.send(f"<@512640455834337290>```{trace}```"),
+                ctx.reply(
+                    "An error occurred while executing the command. The error has been reported."
+                ),
+            )
+
+
+async def setup(bot: ReplBot) -> None:
+    await bot.add_cog(ErrorHandler(bot))

v_repl_bot/cogs/miscellanious.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from datetime import timedelta as TimeDelta
+from inspect import cleandoc as strip
+from platform import python_version
+from time import monotonic as get_monotonic, time as get_time
+from typing import TYPE_CHECKING
+
+from discord.ext.commands import Cog, command
+
+if TYPE_CHECKING:
+    from discord.ext.commands import Context
+
+    from .. import ReplBot
+
+
+class Miscellaneous(
+    Cog,
+    name = "Miscellaneous",
+    description = "Miscellaneous commands.",
+):
+    def __init__(self, bot: ReplBot) -> None:
+        self.bot = bot
+        self.bot.help_command.cog = self
+
+    def cog_unload(self) -> None:
+        self.bot.help_command.cog = None
+        self.bot.help_command.hidden = True
+
+    @command(
+        brief = "Sends the bots ping.",
+        help = "Sends the bots ping."
+    )
+    async def ping(self, ctx: Context) -> None:
+        ts = get_monotonic()
+        message = await ctx.reply("Pong!")
+        ts = get_monotonic() - ts
+        await message.edit(content = f"Pong! `{int(ts * 1000)}ms`")
+
+    @command(
+        brief = "Sends the GitHub repository link for the bot.",
+        help = "Sends the GitHub repository link for the bot.",
+    )
+    async def github(self, ctx: Context) -> None:
+        # Not a button since I want the embed.
+        await ctx.reply("https://github.com/RGBCube/v-repl-bot")
+
+    @command(
+        brief = "Sends info about the bot.",
+        help = "Sends info about the bot."
+    )
+    async def info(self, ctx: Context) -> None:
+        await ctx.reply(
+            strip(
+                f"""
+            __**Bot Info**__
+            **Python Version:** v{python_version()}
+            **Uptime:** `{TimeDelta(seconds = int(get_time() - self.bot.ready_timestamp))}`
+            """
+            )
+        )
+
+
+async def setup(bot: ReplBot) -> None:
+    await bot.add_cog(Miscellaneous(bot))

v_repl_bot/cogs/repl.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from discord.ext import commands
+from discord.ext.commands import Cog, command
+from jishaku.codeblocks import Codeblock, codeblock_converter
+
+if TYPE_CHECKING:
+    from discord.ext.commands import Context
+
+    from .. import ReplBot
+
+
+class REPL(
+    Cog,
+    name = "REPL",
+    description = "REPL (Read, Eval, Print, Loop) commands.",
+):
+    def __init__(self, bot: ReplBot) -> None:
+        self.bot = bot
+
+    @command(
+        aliases = ("run", "repl"),
+        brief = "Runs V code.",
+        help = "Runs V code."
+    )
+    async def eval(
+        self,
+        ctx: Context,
+        code: Codeblock | None = commands.param(converter = codeblock_converter, default = None)
+    ) -> None:
+        if code is None:
+            await ctx.reply("No code provided.")
+
+        with self.bot.session.post(
+            "https://vlang.io/play",
+            data = { "code": code.content },
+        ) as response:
+            await ctx.reply(
+                "```\n" + response.text.replace("`", "\u200B`\u200B") + "\n```"
+            )  # Zero-width space.
+
+
+async def setup(bot: ReplBot) -> None:
+    await bot.add_cog(REPL(bot))

v_repl_bot/config.example.json

@@ -0,0 +1,4 @@
+{
+  "BOT_TOKEN": "",
+  "WEBHOOK_URL": ""
+}