Mcdoc command

commanderbot/ext/mcdoc/__init__.py

@@ -0,0 +1,9 @@
+from discord.ext.commands import Bot
+
+from commanderbot.core.utils import is_commander_bot
+from commanderbot.ext.mcdoc.mcdoc_cog import McdocCog
+
+
+async def setup(bot: Bot):
+    assert is_commander_bot(bot)
+    await bot.add_configured_cog(__name__, McdocCog)

commanderbot/ext/mcdoc/mcdoc_cog.py

@@ -0,0 +1,129 @@
+from typing import Optional
+import aiohttp
+
+from discord import Embed, Emoji, Interaction
+from discord.app_commands import (
+    allowed_installs,
+    command,
+    describe,
+)
+from discord.ext import tasks
+from discord.ext.commands import Cog
+
+from commanderbot.core.commander_bot import CommanderBot
+from commanderbot.ext.mcdoc.mcdoc_symbols import McdocSymbols
+from commanderbot.ext.mcdoc.mcdoc_types import McdocContext
+from commanderbot.ext.mcdoc.mcdoc_exceptions import InvalidVersionError, RequestSymbolsError, RequestVersionError, EmojiNotFoundError
+from commanderbot.ext.mcdoc.mcdoc_options import McdocOptions
+from commanderbot.lib import constants, AllowedMentions
+from commanderbot.lib.predicates import is_convertable_to
+
+
+class McdocCog(Cog, name="commanderbot.ext.mcdoc"):
+    def __init__(self, bot: CommanderBot, **options):
+        self.bot: CommanderBot = bot
+        self.options = McdocOptions.from_data(options)
+
+        self._latest_version: Optional[str] = None
+        self._symbols: Optional[McdocSymbols] = None
+        self._etag: Optional[str] = None
+
+    async def cog_load(self):
+        self._fetch_latest_version.start()
+
+    async def cog_unload(self):
+        self._fetch_latest_version.stop()
+
+    async def _fetch_symbols(self) -> McdocSymbols:
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(
+                    self.options.symbols_url,
+                    headers={
+                        "User-Agent": constants.USER_AGENT,
+                        "If-None-Match": self._etag or "",
+                    },
+                ) as response:
+                    # Use cached symbols if they are still valid
+                    if response.status == 304 and self._symbols:
+                        return self._symbols
+
+                    if response.status != 200:
+                        raise RequestSymbolsError()
+
+                    # Store symbol data and ETag header
+                    data: dict = await response.json()
+                    self._symbols = McdocSymbols(data)
+                    self.etag = response.headers.get("ETag")
+
+                    return self._symbols
+
+        except aiohttp.ClientError:
+            raise RequestSymbolsError()
+
+    @tasks.loop(hours=1)
+    async def _fetch_latest_version(self) -> str:
+        try:
+            # Try to update the version
+            async with aiohttp.ClientSession(raise_for_status=True) as session:
+                async with session.get(self.options.manifest_url, headers={
+                    "User-Agent": constants.USER_AGENT,
+                }) as response:
+                    data: dict = await response.json()
+                    release: str = data["latest"]["release"]
+                    self._latest_version = release
+                    return self._latest_version
+
+        except aiohttp.ClientError:
+            raise RequestVersionError()
+
+    async def _get_latest_version(self, override: Optional[str]) -> str:
+        if override:
+            return override
+        if self._latest_version:
+            return self._latest_version
+        return await self._fetch_latest_version()
+
+    def _get_emoji(self, type: str) -> Emoji:
+        name = self.options.emoji_prefix + type
+        emoji = self.bot.application_emojis.get(name)
+        if emoji is None:
+            raise EmojiNotFoundError(name)
+        return emoji
+
+    @command(name="mcdoc", description="Query vanilla mcdoc types")
+    @describe(
+        query="The mcdoc identifier",
+        version="The Minecraft game version (defaults to the latest release)",
+    )
+    @allowed_installs(guilds=True, users=True)
+    async def cmd_mcdoc(self, interaction: Interaction, query: str, version: Optional[str]):
+        # Respond to the interaction with a defer since the web request may take a while
+        await interaction.response.defer()
+
+        try:
+            # Fetch the vanilla-mcdoc symbols and search for a symbol
+            symbols = await self._fetch_symbols()
+            symbol = symbols.search(query)
+
+            # Use the version override or get the cached latest version
+            version = await self._get_latest_version(version)
+
+            # Validate that the version number can be used to compare
+            if not version.startswith("1.") or not is_convertable_to(version[2:], float):
+                raise InvalidVersionError(version)
+        except Exception as ex:
+            await interaction.delete_original_response()
+            raise ex
+
+        # Create a context object used for rendering
+        ctx = McdocContext(version, symbols, self._get_emoji)
+
+        embed: Embed = Embed(
+            title=symbol.title(ctx),
+            description=symbol.body(ctx),
+            color=0x2783E3, # Spyglass blue
+        )
+        embed.set_footer(text=f"vanilla-mcdoc · {version}", icon_url=self.options.icon_url)
+
+        await interaction.followup.send(embed=embed, allowed_mentions=AllowedMentions.none())

commanderbot/ext/mcdoc/mcdoc_exceptions.py

@@ -0,0 +1,33 @@
+from commanderbot.lib import ResponsiveException
+
+
+class McdocException(ResponsiveException):
+    pass
+
+
+class RequestSymbolsError(McdocException):
+    def __init__(self):
+        super().__init__(f"😵 Unable to fetch vanilla-mcdoc symbol data")
+
+
+class RequestVersionError(McdocException):
+    def __init__(self):
+        super().__init__(f"😵 Unable to fetch the latest version number")
+
+
+class InvalidVersionError(McdocException):
+    def __init__(self, version: str):
+        self.version: str = version
+        super().__init__(f"😬 Invalid version format `{self.version}`. Only release versions are allowed.")
+
+
+class QueryReturnedNoResults(McdocException):
+    def __init__(self, query: str):
+        self.query: str = query
+        super().__init__(f"😔 Could not find any symbols matching `{self.query}`")
+
+
+class EmojiNotFoundError(McdocException):
+    def __init__(self, name: str):
+        self.name: str = name
+        super().__init__(f"😵 Unable to get application emoji `{self.name}`")

commanderbot/ext/mcdoc/mcdoc_options.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass
+from typing import Any, Optional, Self
+
+from commanderbot.lib import FromDataMixin
+
+@dataclass
+class McdocOptions(FromDataMixin):
+    symbols_url: str
+    manifest_url: str
+    emoji_prefix: str
+    icon_url: Optional[str] = None
+
+    @classmethod
+    def try_from_data(cls, data: Any) -> Optional[Self]:
+        if isinstance(data, dict):
+            return cls(
+                symbols_url=data["symbols_url"],
+                manifest_url=data["manifest_url"],
+                emoji_prefix=data["emoji_prefix"],
+                icon_url=data.get("icon_url"),
+            )

commanderbot/ext/mcdoc/mcdoc_symbols.py

@@ -0,0 +1,125 @@
+from collections import defaultdict
+from dataclasses import dataclass
+from typing import Optional, Union
+import re
+
+from commanderbot.ext.mcdoc.mcdoc_exceptions import QueryReturnedNoResults
+from commanderbot.ext.mcdoc.mcdoc_types import McdocContext, McdocType, deserialize_mcdoc
+
+
+@dataclass
+class SymbolResult:
+    identifier: str
+    typeDef: McdocType
+
+    def title(self, ctx: McdocContext):
+        name = ctx.symbols.compact_path(self.identifier)
+        return self.typeDef.title(name, ctx)
+
+    def body(self, ctx: McdocContext):
+        return self.typeDef.render(ctx)
+
+
+@dataclass
+class DispatchResult:
+    registry: str
+    identifier: str
+    typeDef: McdocType
+
+    def title(self, ctx: McdocContext):
+        name = f"{self.registry.removeprefix("minecraft:")} [{self.identifier}]"
+        return self.typeDef.title(name, ctx)
+
+    def body(self, ctx: McdocContext):
+        return self.typeDef.render(ctx)
+
+
+@dataclass
+class DisambiguationResult:
+    query: str
+    identifiers: list[str]
+
+    def title(self, ctx: McdocContext):
+        return f"{len(self.identifiers)} results for {self.query}"
+
+    def body(self, ctx: McdocContext):
+        return "\n".join([f"* {ctx.symbols.compact_path(i)}" for i in self.identifiers])
+
+
+class McdocSymbols:
+    def __init__(self, data: dict):
+        self.symbols = {
+            str(key): deserialize_mcdoc(typeDef)
+            for key, typeDef in data.get("mcdoc", {}).items()
+        }
+        self.dispatchers = {
+            str(registry): {
+                str(key): deserialize_mcdoc(typeDef)
+                for key, typeDef in members.items()
+            }
+            for registry, members in data.get("mcdoc/dispatcher", {}).items()
+        }
+
+        self.names = defaultdict[str, list[str]](list)
+        for key in self.symbols:
+            parts = key.split("::")
+            for i in range(len(parts)):
+                name = "::".join(parts[i:])
+                self.names[name].append(key)
+        self.names = dict(self.names)
+
+        self.unique_suffixes = dict[str, str]()
+        for name, keys in self.names.items():
+            if len(keys) <= 1 or re.match(r"<anonymous \d+>", name):
+                continue
+            for key in keys:
+                parts = key.split("::")
+                for i in reversed(range(len(parts)-1)):
+                    suffix = "::".join(parts[i:])
+                    if not [k for k in keys if k is not key and k.endswith(f"::{suffix}")]:
+                        self.unique_suffixes[key] = suffix
+                        break
+
+    def search(self, query: str) -> Union[SymbolResult, DispatchResult, DisambiguationResult]:
+        if query in self.symbols:
+            return SymbolResult(query, self.symbols[query])
+
+        if query in self.names:
+            identifiers = self.names[query]
+            if len(identifiers) > 1:
+                return DisambiguationResult(query, identifiers)
+            elif len(identifiers) == 1:
+                return SymbolResult(identifiers[0], self.symbols[identifiers[0]])
+
+        parts = query.split(" ")
+        if len(parts) == 2:
+            registry, identifier = parts
+            if ":" not in registry:
+                registry = f"minecraft:{registry}"
+            identifier = identifier.removeprefix("minecraft:")
+            map = self.dispatchers.get(registry, None)
+            if map and identifier in map:
+                return DispatchResult(registry, identifier, map[identifier])
+            raise QueryReturnedNoResults(query)
+
+        identifier = query.removeprefix("minecraft:")
+        resources = self.dispatchers.get("minecraft:resource", {})
+        if query in resources:
+            return DispatchResult("resource", identifier, resources[identifier])
+
+        for registry, map in self.dispatchers.items():
+            if identifier in map:
+                return DispatchResult(registry, identifier, map[identifier])
+
+        raise QueryReturnedNoResults(query)
+
+    def compact_path(self, path: str) -> str:
+        if path in self.unique_suffixes:
+            return self.unique_suffixes[path]
+        return path.split("::")[-1]
+
+    def get(self, path: str) -> Optional[McdocType]:
+        return self.symbols.get(path, None)
+
+    def dispatch(self, registry: str, identifier: str) -> Optional[McdocType]:
+        return self.dispatchers.get(registry, {}).get(identifier, None)