implement archive of nethys support

src/bot/discord/bot.py

@@ -4,6 +4,7 @@ import disnake
 from disnake.ext.commands import bot
 
 from src.bot.discord.commands.dice import DiceCog
+from src.bot.discord.commands.pathfinder_aon import PathfinderArchiveOfNethysCog
 from src.bot.discord.commands.pathfinder_wiki import PathfinderWikiCog
 
 
@@ -25,6 +26,7 @@ class NucuBot(bot.Bot):
         )
         discord_bot.add_cog(DiceCog(discord_bot))
         discord_bot.add_cog(PathfinderWikiCog(discord_bot))
+        discord_bot.add_cog(PathfinderArchiveOfNethysCog(discord_bot))
         return discord_bot
 
     async def on_ready(self):

src/bot/discord/commands/pathfinder_aon.py

@@ -0,0 +1,88 @@
+import datetime
+
+import disnake
+import re
+from disnake.ext import commands
+
+from src.knowledge.pathfinder_archive_of_nethys import PathfinderArchiveOfNethysClient
+
+
+async def pathfinder_aon_lookup_autocomplete(
+    inter: disnake.ApplicationCommandInteraction, user_input: str
+) -> list[str]:
+    """
+    Autocompletes pathfinder archive of nethys queries.
+    """
+    async with PathfinderArchiveOfNethysClient() as aon_client:
+        if user_input == "":
+            return [
+                "(archetype-182) Zombie",
+                "(archetype-26) Recall Knowledge",
+                "(rules-458) Flanking",
+                "(rules-733) Runes",
+                "(rules-1115) Diseases",
+                "(class-23) Kineticist",
+                "(rules-1133) Magic",
+                "(rules-28) Golarion",
+                "(rules-387) Actions",
+                "(rules-1026) Action Economy",
+            ]
+        data = await aon_client.search_pages(user_input)
+        return list(map(lambda x: f"({x.id}) - {x.name}", data))
+
+
+class PathfinderArchiveOfNethysButton(disnake.ui.View):
+    """
+    PathfinderArchiveOfNethysButton adds a view button for visiting an archive of nethys link.
+    """
+
+    def __init__(self, url: str):
+        super().__init__()
+        self.add_item(disnake.ui.Button(label="View Page", url=url))
+
+
+class PathfinderArchiveOfNethysCog(commands.Cog):
+    """
+    PathfinderArchiveOfNethysCog is the Cog that implements Pathfinder Archive of Nethys related commands.
+    """
+
+    def __init__(self, bot):
+        self.bot = bot
+        self._id_regex = r"\((.*)\) .*"
+
+    @commands.slash_command(name="aon", description="Lookup a Archive of Nethys page.")
+    async def lookup(
+        self,
+        inter: disnake.ApplicationCommandInteraction,
+        query: str = commands.Param(autocomplete=pathfinder_aon_lookup_autocomplete),
+    ):
+        """
+        Looks up a page on Archive of Nethys.
+        """
+        async with PathfinderArchiveOfNethysClient() as aon_client:
+            match = re.match(self._id_regex, query)
+            if match:
+                data = await aon_client.search_page_by_id(document_id=match.groups()[0])
+            else:
+                data = await aon_client.search_pages(query)
+                if len(data) == 0:
+                    data = None
+                else:
+                    data = data[0]
+            if data is None:
+                await inter.send(
+                    f"@{inter.author} nothing was found for the given query: {query}"
+                )
+            else:
+                embed = disnake.Embed(
+                    title=data.name,
+                    description=data.text,
+                    timestamp=datetime.datetime.now(),
+                )
+                await inter.send(
+                    f"@{inter.author} requested {query}",
+                    embed=embed,
+                    view=PathfinderArchiveOfNethysButton(
+                        url=aon_client.format_url(data.url)
+                    ),
+                )

src/bot/discord/commands/pathfinder_wiki.py

@@ -7,12 +7,19 @@ from src.knowledge.pathfinder_wiki import PathfinderWikiClient
 async def pathfinder_wiki_lookup_autocomplete(
     inter: disnake.ApplicationCommandInteraction, user_input: str
 ) -> list[str]:
+    """
+    Autocompletes pathfinder wiki queries.
+    """
     async with PathfinderWikiClient() as wiki_client:
         data = await wiki_client.search_pages(user_input)
         return list(map(lambda x: x.key, data))
 
 
 class PathfinderWikiCog(commands.Cog):
+    """
+    PathfinderWikiCog implements commands related to the pathfinder wiki.
+    """
+
     def __init__(self, bot):
         self.bot = bot
 
@@ -24,5 +31,8 @@ class PathfinderWikiCog(commands.Cog):
         inter: disnake.ApplicationCommandInteraction,
         query: str = commands.Param(autocomplete=pathfinder_wiki_lookup_autocomplete),
     ):
+        """
+        Looks up a page on Pathfinder wiki.
+        """
         wiki_link = PathfinderWikiClient.get_page_link(query)
         await inter.send(f"@{inter.author} here's the wiki link: {wiki_link}")

src/knowledge/pathfinder_archive_of_nethys.py

@@ -1,5 +1,6 @@
 import asyncio
 import dataclasses
+import typing
 
 import aiohttp
 
@@ -35,19 +36,63 @@ class PathfinderArchiveOfNethysClient:
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         await self.session.close()
 
-    async def search_pages(self, query: str) -> list[PathfinderArchiveOfNethysDocument]:
+    async def search_page_by_id(
+        self, document_id: str
+    ) -> typing.Optional[PathfinderArchiveOfNethysDocument]:
+        """
+            Searches the Archive of Nethys by an id.
+        :param document_id: - The document id.
+        :return: - The Archive of Nethys document.
+        """
+        if self.session is None:
+            self.session = aiohttp.ClientSession()
+
+        if document_id == "":
+            return None
+
+        async with self.session.post(
+            self._aon_elasticsearch_base_url,
+            json={
+                "size": 1,
+                "query": {
+                    "query_string": {
+                        "query": f"{document_id}",
+                        "default_field": "id",
+                    }
+                },
+                "fields": ["name", "text", "url"],
+                "_source": False,
+            },
+        ) as response:
+            result = await response.json()
+            for item in result.get("hits", {}).get("hits", []):
+                return PathfinderArchiveOfNethysDocument(
+                    id=item.get("_id"),
+                    name=item.get("fields", {}).get("name", ["Unknown"])[0],
+                    text=item.get("fields", {}).get("text", ["Unknown"])[0],
+                    url=item.get("fields", {}).get("url", ["Unknown"])[0],
+                )
+
+    async def search_pages(
+        self, query: str, size: int = 10
+    ) -> list[PathfinderArchiveOfNethysDocument]:
         """
             Searches the Archive of Nethys pages
         :param query: The search query.
+        :param size: The maximum number of items to return in a single query.
         :return: A list of: func:`PathfinderWikiPage`.
         """
         if self.session is None:
             self.session = aiohttp.ClientSession()
 
+        if query == "":
+            return []
+
         found_items = []
         async with self.session.post(
             self._aon_elasticsearch_base_url,
             json={
+                "size": size,
                 "query": {
                     "query_string": {"query": f"{query}*", "default_field": "name"}
                 },
@@ -68,7 +113,10 @@ class PathfinderArchiveOfNethysClient:
             return found_items
 
     @staticmethod
-    def get_page_link(page_url) -> str:
+    def format_url(page_url) -> str:
+        """
+        Formats an archive of nethys url into a full url.
+        """
         return f"https://2e.aonprd.com/{page_url}"
 
     async def close(self):
@@ -80,6 +128,6 @@ if __name__ == "__main__":
     result = asyncio.run(pf.search_pages("zom"))
     print(result)
     for item in result:
-        print(pf.get_page_link(item.url))
+        print(pf.format_url(item.url))
 
     asyncio.run(pf.close())

src/knowledge/pathfinder_wiki.py

@@ -60,6 +60,9 @@ class PathfinderWikiClient:
         return f"https://pathfinderwiki.com/wiki/{page_key}"
 
     async def close(self):
+        """
+        Closes the client.
+        """
         await self.session.close()