Implement issues/prs tagging

.github/workflows/deploy.yaml

@@ -15,6 +15,7 @@ jobs:
         env:
           COMPOSE_PROFILES: production
           TOKEN: ${{ secrets.TOKEN }}
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Display Logs
         run: python3 -m exendlr bobux-admin "bot is ready"

docker-compose.yaml

@@ -7,3 +7,4 @@ services:
       - /home/exenifix/bobux-data/admin:/app/data
     environment:
       - TOKEN
+      - GH_TOKEN

ext/github.py

@@ -0,0 +1,48 @@
+import asyncio
+import re
+
+import disnake
+
+from utils.bot import Cog
+from utils.constants import DEVELOPER_ROLE_ID
+
+REPO_TAG_PATTERN = re.compile(r"([-A-Za-z0-9_]*)#(\d+)")
+
+
+class GithubCog(Cog):
+    @Cog.listener()
+    async def on_message(self, msg: disnake.Message):
+        if msg.author.get_role(DEVELOPER_ROLE_ID) is None:
+            return
+        content = msg.clean_content
+        tags: list[tuple[str, str]] = re.findall(REPO_TAG_PATTERN, content)  # type: ignore
+        if len(tags) == 0:
+            return
+        ordered: dict[str, set[int]] = {}
+        for tag in tags:
+            repo = tag[0] or "Bobux"
+            if repo not in self.bot.github.repositories:
+                continue
+            num = int(tag[1])
+            if repo not in ordered:
+                ordered[repo] = set()
+            ordered[repo].add(num)
+        print(ordered)
+        await msg.add_reaction("<a:loading:1082740409240928316>")
+
+        fetched_items = 0
+        embed = disnake.Embed(title="Referencing Issues and PRs", color=0x00FFFF)
+        for repo, numbers in ordered.items():
+            if fetched_items > 8:
+                break
+            coros = [self.bot.github.get_item(repo, num) for num in numbers]
+            if fetched_items + len(coros) > 8:
+                coros = coros[: 8 - fetched_items]
+            fetched_items += len(coros)
+            items = list(filter(lambda x: x is not None, await asyncio.gather(*coros)))
+            if len(items) == 0:
+                continue
+            embed.add_field(repo, "\n".join(i.get_txt() for i in items), inline=False)
+        await msg.clear_reactions()
+        if len(embed.fields) > 0:
+            await msg.reply(embed=embed, fail_if_not_exists=False)

utils/bot.py

@@ -10,6 +10,7 @@
 
 from utils import env, paths
 from utils.database import Database
+from utils.github import Github
 
 REQUIRED_DIRS = [paths.LOGS]
 for p in REQUIRED_DIRS:
@@ -26,8 +27,10 @@ def __init__(self) -> None:
             guild_typing=True,
             guilds=True,
             members=True,
+            message_content=True,
         )
         self.db = Database()
+        self.github = Github()
         self.log = FileLogger("BOT", folder=paths.LOGS)
 
         super().__init__(
@@ -39,6 +42,7 @@ async def start(self, *args: Any, **kwargs: Any) -> None:
         self.log.info("Starting bot...")
         await self.db.connect()
         await self.db.setup()
+        await self.github.setup()
 
         await super().start(*args, **kwargs)
 
@@ -57,6 +61,8 @@ async def on_error(self, event_method: str, *args: Any, **kwargs: Any) -> None:
     async def close(self) -> None:
         self.log.info("Shutting down...")
         await self.db.close()
+        await self.github.close()
+        await super().close()
         self.log.ok("Bot was shut down successfully")
 
     def auto_setup(self, module_name: str) -> None:

utils/constants.py

@@ -1,4 +1,6 @@
 GUILD_ID = 835880446041784350
+DEVELOPER_ROLE_ID = 842043338927636510
 POINTS_ASSIGNERS_ROLES_IDS = [842043338927636510, 842043787991842847]
 BUGPOINTS_ROLES = {5: 1057329593754857473, 15: 1057329717159678042, 30: 1057329737447510026}
 SUGGESTION_POINTS_ROLES = {3: 1061264336888279081, 10: 1061264486163546172, 20: 1061264331381149757}
+ORG_NAME = "BobuxBot"

utils/env.py

@@ -4,6 +4,7 @@
 
 class MainEnvironment(EnvironmentProfile):
     TOKEN: str
+    GH_TOKEN: str
 
 
 load_dotenv()

utils/github.py

@@ -0,0 +1,77 @@
+from dataclasses import dataclass
+from enum import Enum
+
+from aiohttp import ClientResponseError, ClientSession
+
+from utils import env
+from utils.constants import ORG_NAME
+
+
+class State(Enum):
+    OPEN = "open"
+    CLOSED = "closed"
+    MERGED = "merged"
+
+
+class ItemType(Enum):
+    ISSUE = 0
+    PR = 1
+
+
+@dataclass
+class Item:
+    number: int
+    title: str
+    url: str
+    state: State
+    type: ItemType
+
+    @property
+    def emoji(self) -> str:
+        if self.type == ItemType.PR:
+            return {
+                State.OPEN: "<:propen:1082736247673466911>",
+                State.CLOSED: "<:prclosed:1082736251167309875>",
+                State.MERGED: "<:merged:1082736253667127337>",
+            }[self.state]
+        return {State.OPEN: "<:isopened:1082737004225232947>", State.CLOSED: "<:isclosed:1082737006754406481>"}[
+            self.state
+        ]
+
+    def get_txt(self) -> str:
+        return f"{self.emoji} [{self.title[:64]}{'...' if len(self.title) > 64 else ''} (#{self.number})]({self.url})"
+
+
+class Github:
+    _session: ClientSession
+
+    def __init__(self) -> None:
+        self.repositories: list[str] = []  # presumed to be constant, never updates
+
+    async def setup(self) -> None:
+        self._session = ClientSession(
+            "https://api.github.com",
+            headers={
+                "Accept": "application/vnd.github+json",
+                "Authorization": f"Bearer {env.main.GH_TOKEN}",
+                "X-GitHub-Api-Version": "2022-11-28",
+            },
+            raise_for_status=True,
+        )
+        async with self._session.get(f"/orgs/{ORG_NAME}/repos") as resp:
+            self.repositories = [r["name"] for r in await resp.json()]
+
+    async def close(self) -> None:
+        await self._session.close()
+
+    async def get_item(self, repository: str, number: int) -> Item | None:
+        try:
+            r = await self._session.get(f"/repos/{ORG_NAME}/{repository}/issues/{number}")
+            data = await r.json()
+            item_type = ItemType.PR if "pull_request" in data else ItemType.ISSUE
+            state = State(data["state"])
+            if item_type == ItemType.PR and state == State.CLOSED and data["pull_request"]["merged_at"] is not None:
+                state = State.MERGED
+            return Item(number, data["title"], data["html_url"], state, item_type)
+        except ClientResponseError:
+            return None