First pass of pobierz command

wulkabot/bot.py

@@ -29,6 +29,9 @@ class Wulkabot(commands.Bot):
     async def on_connect(self) -> None:
         print(f"Connected as {self.user}")
 
+    async def on_command_error(self, context: commands.Context, exception: commands.errors.CommandError, /) -> None:
+        await context.send(f"Error! {exception}")
+
     async def close(self) -> None:
         await super().close()
         await self.http_client.close()

wulkabot/cogs/wulkanowy.py

@@ -0,0 +1,38 @@
+"""
+Wulkabot
+Copyright (C) 2022-present Stanisław Jelnicki
+"""
+
+
+import asyncio
+
+import discord
+from discord import app_commands
+from discord.ext import commands
+
+from .. import bot
+from ..utils import github, wulkanowy_manager
+
+
+class Wulkanowy(commands.Cog):
+    def __init__(self, bot: bot.Wulkabot) -> None:
+        super().__init__()
+        self.bot = bot
+        self.github = github.GitHub()
+        self.wulkanowy_manager = wulkanowy_manager.WulkanowyManager()
+
+    async def cog_unload(self) -> None:
+        await self.github.close()
+        await self.wulkanowy_manager.close()
+
+    @app_commands.command()
+    async def pobierz(self, interaction: discord.Interaction):
+        branches = await self.github.fetch_branches("wulkanowy", "wulkanowy")
+        builds = await asyncio.gather(
+            *(map(self.wulkanowy_manager.fetch_branch_build, branches)), return_exceptions=True
+        )
+        await interaction.response.send_message("\n".join(map(str, builds)), suppress_embeds=True)
+
+
+async def setup(bot: bot.Wulkabot):
+    await bot.add_cog(Wulkanowy(bot))

wulkabot/utils/github.py

@@ -3,8 +3,6 @@ Wulkabot
 Copyright (C) 2022-present Stanisław Jelnicki
 """
 
-from typing import Any
-
 import aiohttp
 
 
@@ -12,10 +10,16 @@ class GitHub:
     def __init__(self) -> None:
         self._http = aiohttp.ClientSession(base_url="https://api.github.com")
 
-    async def fetch_repo(self, owner: str, repo: str) -> dict[str, Any] | None:
+    async def fetch_repo(self, owner: str, repo: str) -> dict[str, str | int | None]:
         response = await self._http.get(f"/repos/{owner}/{repo}")
-        if response.ok:
+        response.raise_for_status()
         return await response.json()
 
+    async def fetch_branches(self, owner: str, repo: str) -> list[str]:
+        response = await self._http.get(f"/repos/{owner}/{repo}/branches")
+        response.raise_for_status()
+        branches = await response.json()
+        return [branch["name"] for branch in branches]
+
     async def close(self):
         await self._http.close()

wulkabot/utils/wulkanowy_manager.py

@@ -0,0 +1,45 @@
+"""
+Wulkabot
+Copyright (C) 2022-present Stanisław Jelnicki
+"""
+
+from typing import Any
+
+import aiohttp
+
+BASE_URL = "https://manager.wulkanowy.net.pl"
+WULKANOWY_HASH = "daeff1893f3c8128"
+
+
+class WulkanowyBuild:
+    def __init__(self, data: dict[str, Any]) -> None:
+        self.build_number: int = data["build_number"]
+        self.build_slug: str = data["build_slug"]
+        self.artifact_slug: str = data["artifact_slug"]
+
+    @property
+    def download_url(self) -> str:
+        return f"{BASE_URL}/v1/download/app/{WULKANOWY_HASH}/build/{self.build_slug}/artifact/{self.artifact_slug}"
+
+    def __str__(self) -> str:
+        return self.download_url
+
+
+class WulkanowyManagerException(Exception):
+    pass
+
+
+class WulkanowyManager:
+    def __init__(self) -> None:
+        self._http = aiohttp.ClientSession(base_url=BASE_URL)
+
+    async def fetch_branch_build(self, branch: str) -> WulkanowyBuild:
+        response = await self._http.get(f"/v1/build/app/{WULKANOWY_HASH}/branch/{branch}")
+        response.raise_for_status()
+        json = await response.json()
+        if not json["success"]:
+            raise WulkanowyManagerException(json["error"])
+        return WulkanowyBuild(json["data"])
+
+    async def close(self):
+        await self._http.close()