Add Gitub Category

.env-example

@@ -1,4 +1,9 @@
 TOKEN="your bot token here"  # here is a tutorial on how to get setup a bot and get this
 PREFIX="!"  # the prefix the bot should use, will default to "!" if this is not present
+
 CHANNEL_DEVALERTS=""
 CHANNEL_DEVLOG=""
+CHANNEL_DEV_GURKBOT=""
+CHANNEL_DEV_REAGURK=""
+CHANNEL_DEV_GURKLANG=""
+CHANNEL_DEV_BRANDING=""

bot/bot.py

@@ -1,5 +1,6 @@
 import os
 
+from aiohttp import ClientSession
 from discord import Embed
 from discord.ext import commands
 from loguru import logger
@@ -11,6 +12,7 @@ class Bot(commands.Bot):
     """The core of the bot."""
 
     def __init__(self) -> None:
+        self.http_session = ClientSession()
         super().__init__(command_prefix=constants.PREFIX)
         self.load_extensions()
 
@@ -43,8 +45,13 @@ async def startup_greeting(self) -> None:
         """Announce presence to the devlog channel."""
         embed = Embed(description="Connected!")
         embed.set_author(
-            name="Gurkbot",
-            url=constants.BOT_REPO_URL,
-            icon_url=self.user.avatar_url,
+            name="Gurkbot", url=constants.BOT_REPO_URL, icon_url=self.user.avatar_url
         )
         await self.get_channel(constants.Channels.devlog).send(embed=embed)
+
+    async def close(self) -> None:
+        """Close Http session when bot is shutting down."""
+        await super().close()
+
+        if self.http_session:
+            await self.http_session.close()

bot/constants.py

@@ -14,6 +14,14 @@
 LOG_FILE = pathlib.Path("log/gurkbot.log")
 
 
+class Emojis(NamedTuple):
+    issue_emoji = "<:IssueOpen:794119024367632405>"
+    issue_closed_emoji = "<:IssueClosed:794118652219359253>"
+    pull_request_emoji = "<:PROpen:794118652014231562>"
+    pull_request_closed_emoji = "<:PRClosed:794120818908463134>"
+    merge_emoji = "<:PRMerged:794119023687761941>"
+
+
 class Colours:
     green = 0x1F8B4C
     yellow = 0xF1C502
@@ -24,6 +32,11 @@ class Channels(NamedTuple):
     devalerts = int(os.getenv("CHANNEL_DEVALERTS", 796695123177766982))
     devlog = int(os.getenv("CHANNEL_DEVLOG", 789431367167377448))
 
+    dev_gurkbot = int(os.getenv("CHANNEL_DEV_GURKBOT", 789431367167377448))
+    dev_reagurk = int(os.getenv("CHANNEL_DEV_REAGURK", 789431367167377448))
+    dev_gurklang = int(os.getenv("CHANNEL_DEV_GURKLANG", 789431367167377448))
+    dev_branding = int(os.getenv("CHANNEL_DEV_BRANDING", 789431367167377448))
+
 
 # Bot replies
 with pathlib.Path("bot/resources/bot_replies.yml").open(encoding="utf8") as file:

bot/exts/github/_issues.py

@@ -0,0 +1,112 @@
+from random import choice
+from typing import Optional
+
+import discord
+from aiohttp import ClientSession
+from bot.constants import Channels, ERROR_REPLIES, Emojis
+from discord import Embed
+from discord.ext import commands
+from loguru import logger
+
+BAD_RESPONSE = {
+    404: "Issue/pull request not located! Please enter a valid number!",
+    403: "Rate limit has been hit! Please try again later!",
+}
+MAX_REQUESTS = 5
+REPO_CHANNEL_MAP = {
+    Channels.dev_reagurk: "reagurk",
+    Channels.dev_gurkbot: "gurkbot",
+    Channels.dev_gurklang: "py-gurklang",
+    Channels.dev_branding: "branding",
+}
+
+
+class Issues:
+    """Cog that allows users to retrieve issues from GitHub."""
+
+    def __init__(self, http_session: ClientSession) -> None:
+        self.http_session = http_session
+
+    @staticmethod
+    def get_repo(channel: discord.TextChannel) -> Optional[str]:
+        """Get repository for the particular channel."""
+        return REPO_CHANNEL_MAP.get(channel.id)
+
+    @staticmethod
+    def error_embed(error_msg: str) -> Embed:
+        """Generate Error Embed for Issues command."""
+        embed = discord.Embed(
+            title=choice(ERROR_REPLIES),
+            color=discord.Color.red(),
+            description=error_msg,
+        )
+        return embed
+
+    async def issue(
+        self,
+        channel: discord.TextChannel,
+        numbers: commands.Greedy[int],
+        repository: Optional[str],
+        user: str,
+    ) -> Embed:
+        """Retrieve issue(s) from a GitHub repository."""
+        links = []
+        numbers = set(numbers)
+
+        repository = repository if repository else self.get_repo(channel)
+
+        if len(numbers) > MAX_REQUESTS:
+            embed = self.error_embed(
+                "You can specify a maximum of {MAX_REQUESTS} issues/PRs only."
+            )
+            return embed
+
+        for number in numbers:
+            url = f"https://api.github.com/repos/{user}/{repository}/issues/{number}"
+            merge_url = (
+                f"https://api.github.com/repos/{user}/{repository}/pulls/{number}/merge"
+            )
+
+            logger.trace(f"Querying GH issues API: {url}")
+            async with self.http_session.get(url) as r:
+                json_data = await r.json()
+
+            if r.status in BAD_RESPONSE:
+                logger.warning(f"Received response {r.status} from: {url}")
+                embed = self.error_embed(f"#{number} {BAD_RESPONSE.get(r.status)}")
+                return embed
+
+            if "issues" in json_data.get("html_url"):
+                icon_url = (
+                    Emojis.issue_emoji
+                    if json_data.get("state") == "open"
+                    else Emojis.issue_closed_emoji
+                )
+
+            else:
+                logger.info(
+                    f"PR provided, querying GH pulls API for additional information: {merge_url}"
+                )
+                async with self.http_session.get(merge_url) as m:
+                    if json_data.get("state") == "open":
+                        icon_url = Emojis.pull_request_emoji
+                    elif m.status == 204:
+                        icon_url = Emojis.merge_emoji
+                    else:
+                        icon_url = Emojis.pull_request_closed_emoji
+
+            issue_url = json_data.get("html_url")
+            links.append(
+                (
+                    icon_url,
+                    f"[{user}/{repository}] #{number} {json_data.get('title')}",
+                    issue_url,
+                )
+            )
+
+        resp = discord.Embed(
+            colour=discord.Color.green(),
+            description="\n".join("{0} [{1}]({2})".format(*link) for link in links),
+        )
+        resp.set_author(name="GitHub", url=f"https://github.com/{user}/{repository}")
+        return resp

bot/exts/github/_profile.py

@@ -0,0 +1,92 @@
+from datetime import datetime
+from random import choice
+from typing import Optional
+
+import discord
+from aiohttp import ClientSession
+from bot.constants import ERROR_REPLIES
+from discord import Embed
+
+
+class GithubInfo:
+    """Fetches info from GitHub."""
+
+    def __init__(self, http_session: ClientSession) -> None:
+        self.http_session = http_session
+
+    async def fetch_data(self, url: str) -> dict:
+        """Retrieve data as a dictionary."""
+        async with self.http_session.get(url) as r:
+            return await r.json()
+
+    @staticmethod
+    def get_data(username: Optional[str], user_data: dict, org_data: dict) -> Embed:
+        """Return embed containing filtered github data for user."""
+        orgs = [org["login"] for org in org_data]
+
+        if user_data.get("message") != "Not Found":
+            # Forming blog link
+            if user_data["blog"].startswith("http"):  # Blog link is complete
+                blog = user_data["blog"]
+            elif user_data["blog"]:  # Blog exists but the link is not complete
+                blog = f"https://{user_data['blog']}"
+            else:
+                blog = "No website link available"
+
+            embed = discord.Embed(
+                title=f"{user_data['login']}'s GitHub profile info",
+                description=f"```{user_data['bio']}```\n"
+                if user_data["bio"] is not None
+                else "",
+                colour=discord.Colour.green(),
+                timestamp=datetime.strptime(
+                    user_data["created_at"], "%Y-%m-%dT%H:%M:%SZ"
+                ),
+            )
+            embed.set_thumbnail(url=user_data["avatar_url"])
+            embed.set_footer(text="Account created at")
+
+            embed.add_field(
+                name="Followers",
+                value=f"[{user_data['followers']}]({user_data['html_url']}?tab=followers)",
+            )
+            embed.add_field(name="\u200b", value="\u200b")
+            embed.add_field(
+                name="Following",
+                value=f"[{user_data['following']}]({user_data['html_url']}?tab=following)",
+            )
+            embed.add_field(
+                name="Public repos",
+                value=f"[{user_data['public_repos']}]({user_data['html_url']}?tab=repositories)",
+            )
+            embed.add_field(name="\u200b", value="\u200b")
+            embed.add_field(
+                name="Gists",
+                value=f"[{user_data['public_gists']}](https://gist.github.com/{username})",
+            )
+            embed.add_field(
+                name="Organizations",
+                value=" | ".join(orgs) if orgs else "No Organizations",
+            )
+            embed.add_field(name="\u200b", value="\u200b")
+            embed.add_field(name="Website", value=blog)
+
+            return embed
+
+    async def get_github_info(self, username: str) -> Embed:
+        """Fetches a user's GitHub information."""
+        user_data = await self.fetch_data(f"https://api.github.com/users/{username}")
+
+        # User_data will not have a message key if the user exists
+        if user_data.get("message") is not None:
+            embed = discord.Embed(
+                title=choice(ERROR_REPLIES),
+                description=f"The profile for `{username}` was not found.",
+                colour=discord.Colour.red(),
+            )
+            return embed
+
+        org_data = await self.fetch_data(user_data["organizations_url"])
+        embed = self.get_data(username, user_data, org_data)
+
+        return embed

bot/exts/github/_source.py

@@ -0,0 +1,52 @@
+import re
+from inspect import getsourcelines
+from typing import Optional
+
+import discord
+from aiohttp import ClientSession
+from bot import constants
+from discord.ext.commands import Command
+
+doc_reg_class = r'("""|\'\'\')([\s\S]*?)(\1\s*)'
+
+
+class Source:
+    """Displays information about the bot's source code."""
+
+    def __init__(
+        self, http_session: ClientSession, bot_avatar: discord.asset.Asset
+    ) -> None:
+        self.http_session = http_session
+        self.MAX_FIELD_LENGTH = 500
+        self.bot_avatar = bot_avatar
+
+    async def inspect(self, cmd: Optional[Command]) -> Optional[discord.Embed]:
+        """Display information and a GitHub link to the source code of a command."""
+        if cmd is None:
+            return
+
+        module = cmd.module
+        code_lines, start_line = getsourcelines(cmd.callback)
+        url = (
+            f"<{constants.BOT_REPO_URL}/tree/main/"
+            f'{"/".join(module.split("."))}.py#L{start_line}>\n'
+        )
+
+        source_code = "".join(code_lines)
+        sanitized = source_code.replace("`", "\u200B`")
+        sanitized = re.sub(doc_reg_class, "", sanitized)
+        # The help argument of commands.command gets changed to `help=`
+        sanitized = sanitized.replace("help=", 'help=""')
+
+        if len(sanitized) > self.MAX_FIELD_LENGTH:
+            sanitized = (
+                sanitized[: self.MAX_FIELD_LENGTH]
+                + "\n... (truncated - too many lines)"
+            )
+
+        embed = discord.Embed(title="Gurkbot's Source Link", description=f"{url}")
+        embed.add_field(
+            name="Source Code Snippet", value=f"```python\n{sanitized}\n```"
+        )
+
+        return embed