Rewrite the whole thing

minearchy_bot/__init__.py

@@ -0,0 +1 @@
+from .core import *

minearchy_bot/__main__.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from json import load as parse_json
+from os import environ as env
+from pathlib import Path
+
+from uvloop import install as install_uvloop
+
+from . import MinearchyBot
+
+install_uvloop()
+
+with (Path(__file__).parent.parent / "config.json").open() as f:
+    config = parse_json(f)
+
+for key in ("HIDE", "NO_UNDERSCORE"):
+    env[f"JISHAKU_{key}"] = "True"
+
+bot = MinearchyBot(token=config["BOT_TOKEN"], webhook_url=config["WEBHOOK_URL"])
+
+bot.run()

minearchy_bot/cogs/error_handler.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from contextlib import suppress as suppress_error
+from traceback import format_exception as format_exit
+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 ..core import MinearchyBot
+
+
+class ErrorHandler(Cog):
+    def __init__(self, bot: MinearchyBot) -> 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_exit(type(error), error, error.__traceback__))
+            print(f"Ignoring exception in command {ctx.command}:\n{trace}")
+            await self.bot.log_webhook.send(f"<@512640455834337290>```{trace}```")
+
+
+async def setup(bot: MinearchyBot) -> None:
+    await bot.add_cog(ErrorHandler(bot))

minearchy_bot/cogs/minecraft_server.py

@@ -0,0 +1,155 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from discord.ext.commands import Cog, command, group
+from discord.ui import Button, View
+
+if TYPE_CHECKING:
+    from discord.ext.commands import Context
+
+    from ..core import MinearchyBot
+
+
+class MinecraftServer(
+    Cog,
+    name="Minecraft Server",
+    description="Utilities for the Minecraft server.",
+):
+    def __init__(self, bot: MinearchyBot) -> None:
+        self.bot = bot
+
+    @group(
+        invoke_without_command=True,
+        brief="Sends the server IP.",
+        help="Sends the server IP.",
+    )
+    async def ip(self, ctx: Context) -> None:
+        await ctx.reply(
+            f"Java edition IP: `{self.bot.server.java.ip}`\nBedrock edition IP:"
+            f" `{self.bot.server.bedrock.ip}`\nNote: Minecraft 1.19 is required to join."
+        )
+
+    @ip.command(brief="Sends the Java edition IP.", help="Sends the Java edition IP.")
+    async def java(self, ctx: Context) -> None:
+        await ctx.reply(
+            "The IP to connect on Minecraft Java edition is"
+            f" `{self.bot.server.java.ip}`\nNote: Minecraft 1.19 is required to join."
+        )
+
+    @ip.command(
+        brief="Sends the Bedrock edition IP.",
+        help="Sends the Bedrock edition IP.",
+    )
+    async def bedrock(self, ctx: Context) -> None:
+        await ctx.reply(
+            "The IP to connect on Minecraft Bedrock edition is"
+            f" `{self.bot.server.bedrock.ip}`\nNote: Minecraft 1.19"
+            " is required to join."
+        )
+
+    @command(
+        brief="Shows information about the Minecraft server.",
+        help="Shows the total player count, the Minecraft server IP and the server latency.",
+    )
+    async def status(self, ctx: Context) -> None:
+        status = await self.bot.server.status()
+        await ctx.reply(f"The Minecraft server has {status.players.online} players online.")
+
+    @command(brief="Sends the link to the wiki.", help="Sends the link to the wiki.")
+    async def wiki(self, ctx: Context) -> None:
+        view = View()
+        view.add_item(
+            Button(
+                label="Go to the wiki!",
+                url="https://landsofminearchy.com/wiki",
+            )
+        )
+        await ctx.reply(view=view)
+
+    @command(
+        brief="Sends the link to the store.",
+        help="Sends the link to the store.",
+    )
+    async def store(self, ctx: Context) -> None:
+        view = View()
+        view.add_item(
+            Button(
+                label="Go to the store!",
+                url="https://landsofminearchy.com/store",
+            )
+        )
+        await ctx.reply(view=view)
+
+    @command(
+        aliases=("forums",),
+        brief="Sends the link to the forum.",
+        help="Sends the link to the forum.",
+    )
+    async def forum(self, ctx: Context) -> None:
+        view = View()
+        view.add_item(
+            Button(
+                label="Go to the forum!",
+                url="https://landsofminearchy.com/forum",
+            )
+        )
+        await ctx.reply(view=view)
+
+    @command(
+        aliases=("map",),
+        brief="Sends the link to the dynmap.",
+        help="Sends the link to the dynmap.",
+    )
+    async def dynmap(self, ctx: Context) -> None:
+        view = View()
+        view.add_item(
+            Button(
+                label="Go to the dynmap!",
+                url="https://landsofminearchy.com/dynmap",
+            )
+        )
+        await ctx.reply(
+            content="The dynmap is an interactive, live map of our Minecraft server.", view=view
+        )
+
+    @command(
+        brief="Sends the links you can use to vote for the Minecraft server.",
+        help="Sends the links you can use to vote for the Minecraft server.",
+    )
+    async def vote(self, ctx: Context) -> None:
+        view = View()
+        view.add_item(
+            Button(
+                label="Vote for the Minecraft server!",
+                url="https://landsofminearchy.com/vote",
+            )
+        )
+        await ctx.reply(view=view)
+
+    @command(
+        aliases=(
+            "apply",
+            "staffapply",
+            "applystaff",
+            "applyforstaff",
+            "staff-application",
+            "staff-applications",
+            "staff_applications",
+        ),
+        brief="Sends the link to the staff application.",
+        help="Sends the link to the staff application.",
+    )
+    async def staff_application(self, ctx: Context) -> None:
+        view = View()
+        view.add_item(
+            Button(
+                label="Apply for staff!",
+                url="https://docs.google.com/forms/d/1I7Rh_e-ZTXm5L51XoKZsOAk7NAJcHomUUCuOlQcARvY/viewform",
+            )
+        )
+        await ctx.reply(view=view)
+
+
+async def setup(bot: MinearchyBot) -> None:
+    await bot.add_cog(MinecraftServer(bot))

minearchy_bot/cogs/misc.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+from collections import defaultdict as DefaultDict, deque as Deque
+from datetime import timedelta as TimeDelta
+from inspect import cleandoc as strip_doc
+from platform import python_version
+from time import monotonic as ping_time, time as current_time
+from typing import TYPE_CHECKING
+
+from discord import Color, Embed, TextChannel
+from discord.ext.commands import Cog, command, has_permissions
+from discord.utils import escape_markdown
+
+from ..util import override
+
+if TYPE_CHECKING:
+    from discord import Message
+    from discord.ext.commands import Context
+
+    from ..core import MinearchyBot
+
+
+class Miscellaneous(
+    Cog,
+    name="Miscellaneous",
+    description="Various utilities.",
+):
+    def __init__(self, bot: MinearchyBot) -> None:
+        self.bot = bot
+        self.bot.help_command.cog = self
+        self.sniped = DefaultDict(Deque)
+
+    @override
+    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 = ping_time()
+        message = await ctx.reply("Pong!")
+        ts = ping_time() - ts
+        await message.edit(content=f"Pong! `{int(ts * 1000)}ms`")
+
+    @command(brief="Sends info about the bot.", help="Sends info about the bot.")
+    async def info(self, ctx: Context) -> None:
+        await ctx.reply(
+            strip_doc(
+                f"""
+            __**Bot Info**__
+            **Python Version:** v{python_version()}
+            **Uptime:** `{TimeDelta(seconds=int(current_time() - self.bot.ready_timestamp))}`
+            """
+            )
+        )
+
+    @command(brief="Hello!", help="Hello!")
+    async def hello(self, ctx: Context) -> None:
+        await ctx.reply(f"Hi {escape_markdown(ctx.author.name)}, yes the bot is running :)")
+
+    @command(
+        brief="Sends the total members in the server.",
+        help="Sends the total members in the server.",
+    )
+    async def members(self, ctx: Context) -> None:
+        await ctx.reply(f"There are `{ctx.guild.member_count}` users in this server.")
+
+    @Cog.listener()
+    async def on_message_delete(self, message: Message) -> None:
+        if not message.guild:
+            return
+
+        self.sniped[message.channel.id].appendleft((message, int(current_time())))
+
+        self.sniped[message.channel.id] = self.sniped[message.channel.id][:5]  # type: ignore
+
+    @command(
+        brief="Sends the latest deleted messages.",
+        help=(
+            "Sends the last 5 deleted messages in a specified channel.\nIf the channel"
+            " isn't specified, it uses the current channel."
+        ),
+    )
+    @has_permissions(manage_messages=True)  # needs to be able to delete messages to run the command
+    async def snipe(self, ctx: Context, channel: TextChannel = None) -> None:  # type: ignore
+        if channel is None:
+            channel = ctx.channel
+
+        logs = self.sniped[channel.id]
+
+        if not logs:
+            await ctx.reply(
+                "There are no messages to be sniped in"
+                f" {'this channel.' if channel.id == ctx.channel.id else channel.mention}"
+            )
+            return
+
+        embed = Embed(
+            title=(
+                "Showing last 5 deleted messages for"
+                f" {'the current channel' if ctx.channel.id == channel.id else channel}"
+            ),
+            description="The lower the number is, the more recent it got deleted.",
+            color=Color.random(),
+        )
+
+        zwsp = "\uFEFF"
+
+        for i, log in reversed(list(enumerate(logs))):
+            message, ts = log
+
+            embed.add_field(
+                name=str(i) + ("" if i else " (latest)"),
+                value=strip_doc(
+                    f"""
+                Author: {message.author.mention} (ID: {message.author.id}, Plain: {escape_markdown(str(message.author))})
+                Deleted at: <t:{ts}:F> (Relative: <t:{ts}:R>)
+                Content:
+                ```
+                {message.content.replace('`', f'{zwsp}`{zwsp}')}
+                ```
+                """
+                ),
+                inline=False,
+            )
+
+        await ctx.reply(embed=embed)
+
+
+async def setup(bot: MinearchyBot) -> None:
+    await bot.add_cog(Miscellaneous(bot))

minearchy_bot/cogs/moderation.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from datetime import timedelta as TimeDelta
+from typing import TYPE_CHECKING
+
+from discord import Member
+from discord.ext.commands import Cog, Context, command, has_permissions
+
+if TYPE_CHECKING:
+    from ..core import MinearchyBot
+
+
+class Moderation(Cog):
+    def __init__(self, bot: MinearchyBot) -> None:
+        self.bot = bot
+        self.time_values = {
+            "d": "days",
+            "h": "hours",
+            "m": "minutes",
+            "s": "seconds",
+        }
+
+    @command(aliases=("mute",), brief="Times out a user.", help="Times out a user.")
+    @has_permissions(manage_messages=True)
+    async def timeout(self, ctx: Context, member: Member, duration: str = "1d") -> None:
+        if duration[-1] not in self.time_values or len(duration) < 2:
+            await ctx.reply("Invalid duration. Valid durations are: d, h, m, s.")
+            return
+
+        try:
+            time = int(duration[:-1])
+        except ValueError:
+            await ctx.reply("Invalid time.")
+            return
+
+        # days, hours, minutes, seconds
+        clean_time_name = self.time_values[duration[-1]]
+
+        # this is so cursed but works
+        await member.timeout(
+            TimeDelta(**{clean_time_name: time}), reason=f"Timed out by moderator {ctx.author}"
+        )
+
+        await ctx.reply(f"Timed out {member.mention} for {time} {clean_time_name}.")
+
+
+async def setup(bot: MinearchyBot) -> None:
+    await bot.add_cog(Moderation(bot))

minearchy_bot/core/__init__.py

@@ -0,0 +1 @@
+from .minearchy_bot import *

minearchy_bot/core/minearchy_bot.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+__all__ = ("MinearchyBot",)
+
+from asyncio import run as run_coro
+from inspect import cleandoc as strip_doc
+from itertools import chain as chain_iter
+from pathlib import Path
+from time import time as current_time
+from traceback import format_exc as format_exit
+
+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,
+)
+
+from ..minecraft_server import GeyserServer
+from ..util import override
+
+
+class MinearchyBot(CommandsBot):
+    ready_timestamp: float
+    log_webhook: Webhook
+
+    def __init__(self, *, token: str, webhook_url: str) -> None:
+        self.token = token
+        self.webhook_url = webhook_url
+
+        self.server = GeyserServer(
+            java_ip="play.landsofminearchy.com",
+            bedrock_ip="bedrock.landsofminearchy.com",
+        )
+
+        super().__init__(
+            command_prefix=when_mentioned_or("="),
+            strip_after_prefix=True,
+            case_insensitive=True,
+            status=Game("on play.landsofminearchy.com"),
+            owner_ids={512640455834337290, 160087716757897216},
+            allowed_mentions=AllowedMentions.none(),
+            max_messages=100,
+            intents=Intents(
+                guilds=True,
+                members=True,
+                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.",
+            ),
+        )
+
+    @override
+    async def on_ready(self) -> None:
+        print(
+            strip_doc(
+                f"""
+            Connected to Discord!
+            User: {self.user}
+            ID: {self.user.id}
+            """
+            )
+        )
+
+        self.ready_timestamp = current_time()
+        await self.log_webhook.send("Bot is now online!")
+
+    async def load_extensions(self) -> None:
+        cogs = Path(__file__).parent.parent / "cogs"
+        for file_name in chain_iter(
+            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_exit()}")
+
+    @override
+    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, reconnect=True)
+
+        try:
+            run_coro(runner())
+        except KeyboardInterrupt:
+            pass

minearchy_bot/minecraft_server/__init__.py

@@ -0,0 +1 @@
+from .geyser_server import *

minearchy_bot/minecraft_server/geyser_server.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+__all__ = ("GeyserServer",)
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from mcstatus import JavaServer
+
+if TYPE_CHECKING:
+    from mcstatus.pinger import PingResponse
+
+
+@dataclass
+class ServerInfo:
+    ip: str
+    port: int
+
+
+class GeyserServer:
+    def __init__(
+        self,
+        *,
+        java_ip: str,
+        java_port: int = 25565,
+        bedrock_ip: str,
+        bedrock_port: int = 19132,
+    ) -> None:
+        self.__server = JavaServer.lookup(java_ip, java_port)
+        self.java = ServerInfo(java_ip, java_port)
+        self.bedrock = ServerInfo(bedrock_ip, bedrock_port)
+
+    async def status(self) -> PingResponse:
+        return await self.__server.async_status()

minearchy_bot/util/__init__.py

@@ -0,0 +1 @@
+from .override import *

minearchy_bot/util/override.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+__all__ = ("override",)
+
+from typing import Callable, TypeVar
+
+T = TypeVar("T", bound=Callable)
+
+
+def override(function: T) -> T:
+    """Basically Java's @Override annotation. Makes stuff less ambiguous."""
+    return function