Remake advent of code cog.

- Remove tree.sync on setup
+ Add tree.sync on connection to gateway
+ Ability to load cogs in module format
+ Slashcommand error are logged as expected

Author: GDWR

Pipfile

@@ -17,6 +17,7 @@ aiodns = "~=3.0.0"
 PyJWT = "~=2.3.0"
 matplotlib = "*"
 "discord.py" = "*"
+pydantic = "*"
 
 [dev-packages]
 flake8 = "~=4.0.1"

Pipfile.lock

@@ -1,4 +1,5 @@
 import asyncio
+import importlib
 import inspect
 import pkgutil
 import socket
@@ -23,9 +24,12 @@ def on_error(name: str) -> NoReturn:
         raise ImportError(name=name)
 
     for module in pkgutil.walk_packages(cogs.__path__, f"{cogs.__name__}.", onerror=on_error):
+        if any(name.startswith("_") for name in module.name.split(".")):
+            continue  # Ignore modules/packages with a name starting with _
+
         if module.ispkg:
-            _import = __import__(module.name)
-            if not inspect.isfunction(getattr(_import, "setup", None)):
+            imported = importlib.import_module(module.name)
+            if not inspect.isfunction(getattr(imported, "setup", None)):
                 continue
 
         yield module.name

bot/__main__.py

@@ -1,7 +1,8 @@
 import logging
+from typing import Union
 
 import aiohttp
-
+import discord
 from discord.ext.commands import Bot, CommandError, Context
 
 from bot.disable import DisableApi
@@ -34,16 +35,20 @@ def __init__(
         self.graphql = GraphQLClient(session=session)
 
     async def setup_hook(self) -> None:
-        """Sync the application command tree before bot connects for commands."""
-        await self.tree.sync()
+        """Assign a error handler for Interaction commands."""
+        self.tree.on_error = self.on_command_error
 
-    @staticmethod
-    async def on_ready() -> None:
-        """Runs when the bot is connected."""
+    async def on_ready(self) -> None:
+        """Runs when the bot is connected. Sync Interaction/app_commands when connected to the gateway."""
         log.info('Awaiting...')
         log.info("Bot Is Ready For Commands")
+        await self.tree.sync()
 
-    async def on_command_error(self, ctx: Context, exception: CommandError) -> None:
+    async def on_command_error(
+            self,
+            ctx: Union[Context, discord.Interaction],
+            exception: Union[CommandError, discord.app_commands.AppCommandError]
+    ) -> None:
         """Fired when exception happens."""
         log.error(
             "Exception happened while executing command",

bot/bot.py

@@ -0,0 +1,15 @@
+import logging
+
+from bot.bot import Friendo
+from bot.settings import AOC_JOIN_CODE
+from ._cog import AdventOfCode
+
+logger = logging.getLogger("advent_of_code")
+
+
+async def setup(bot: Friendo) -> None:
+    """Sets up the AdventOfCode cog."""
+    if AOC_JOIN_CODE is not None:
+        await bot.add_cog(AdventOfCode(bot))
+    else:
+        logger.warning("Skipping setup for advent of code as a `AOC_JOIN_CODE` wasn't provided")

bot/cogs/advent_of_code.py

@@ -0,0 +1,103 @@
+import logging
+from datetime import datetime
+from itertools import cycle
+from operator import attrgetter
+
+import discord
+from discord import app_commands, Embed, Color
+from discord.ext import commands
+
+from bot.bot import Friendo
+from bot.settings import AOC_JOIN_CODE, AOC_SESSION_COOKIE, AOC_LEADERBOARD_ID
+from ._types import Leaderboard, LeaderboardMember
+
+logger = logging.getLogger("advent_of_code")
+
+
+class AdventOfCode(commands.GroupCog):
+    """Commands for Advent of Code."""
+
+    def __init__(self, bot: Friendo):
+        self.bot = bot
+        super().__init__()
+
+    async def fetch_leaderboard(self, year: int) -> Leaderboard:
+        """Get the leaderboard's state for a specified year."""
+        url = f"https://adventofcode.com/{year}/leaderboard/private/view/{AOC_LEADERBOARD_ID}.json"
+        cookies = {'session': AOC_SESSION_COOKIE}
+
+        # AoC Author has requested applications to provide a url to the tool in the User-Agent
+        headers = {'User-Agent': 'github.com/fisher60/friendo-bot'}
+
+        async with self.bot.session.get(url, cookies=cookies, headers=headers) as response:
+            data = await response.json()
+            return Leaderboard(**data)
+
+    @staticmethod
+    def _create_leaderboard_message(members: list[LeaderboardMember], amount: int) -> str:
+        reset = ""
+        red = ""
+        green = ""
+        yellow = ""
+
+        get_line_color = cycle([red, green])
+
+        formatted_message = f"Here is our top {amount}\n```ansi\n"
+        for rank, member in enumerate(members[:amount], start=1):
+            formatted_message += (
+                f"{next(get_line_color)}{rank:0>2} {yellow}★{reset} {member.name} ({member.local_score})\n"
+            )
+
+        formatted_message += "```"
+
+        return formatted_message
+
+    @app_commands.command()
+    @app_commands.describe(
+        year="View the leaderboard from a specific year, defaults to the current year",
+        amount="How many members to view, defaults to 10"
+    )
+    async def leaderboard(
+            self,
+            interaction: discord.Interaction,
+            year: app_commands.Range[int, 2015, None] = None,
+            amount: app_commands.Range[int, 0, None] = 10
+    ) -> None:
+        """ Get the current Advent of Code leaderboard """
+        if year is None:
+            year = datetime.now().year
+        elif year > datetime.now().year:  # Don't allow years that haven't happened yet
+            await interaction.response.send_message(
+                f"> Please select a valid year 2015 - {datetime.now().year}",
+                ephemeral=True
+            )
+            return
+
+        leaderboard = await self.fetch_leaderboard(year)
+
+        all_member = list(filter(attrgetter("local_score"), leaderboard.members.values()))
+        all_member.sort(key=attrgetter("local_score"), reverse=True)
+
+        embed = Embed(
+            title=f"Advent of Code {leaderboard.year}",
+            description=self._create_leaderboard_message(all_member, amount),
+            url="https://adventofcode.com",
+            colour=Color.gold(),
+        )
+        embed.add_field(
+            name="\u200b",
+            value="Join the fun with `/advent-of-code join`"
+        )
+
+        await interaction.response.send_message(embed=embed)
+
+    @app_commands.command()
+    async def join(self, interaction: discord.Interaction) -> None:
+        """Find out how to join the Advent of Code leaderboard!"""
+        await interaction.response.send_message(
+            ">>> To join the leaderboard, follow these steps:\n"
+            "\t1. Log in on https://adventofcode.com\n"
+            "\t2. Head over to https://adventofcode.com/leaderboard/private\n"
+            f"\t3. Use this code `{AOC_JOIN_CODE}`",
+            ephemeral=True
+        )

bot/cogs/advent_of_code/__init__.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class PartCompletion(BaseModel):
+    star_index: int
+    get_star_ts: int
+
+
+class LeaderboardMember(BaseModel):
+    id: int
+    name: str
+    stars: int
+    local_score: int
+    global_score: int
+    last_star_ts: int
+    completion_day_level: dict[int, dict[int, PartCompletion]]
+
+
+class Leaderboard(BaseModel):
+    owner_id: int
+    event: int
+    members: dict[int, LeaderboardMember]
+
+    @property
+    def year(self) -> int:
+        return self.event

bot/cogs/advent_of_code/_cog.py

@@ -21,6 +21,10 @@
 
 AOC_SESSION_COOKIE = environ.get("AOC_SESSION_COOKIE")
 AOC_JOIN_CODE = environ.get("AOC_JOIN_CODE")
+if AOC_JOIN_CODE:
+    AOC_LEADERBOARD_ID = AOC_JOIN_CODE.split("-")[0]
+else:
+    AOC_LEADERBOARD_ID = None
 
 FRIENDO_API_USER = environ.get("FRIENDO_API_USER")
 FRIENDO_API_PASS = environ.get("FRIENDO_API_PASS")
@@ -34,6 +38,4 @@
 
 GITHUB_REPO = "https://github.com/fisher60/friendo-bot"
 
-AOC_LEADERBOARD_LINK = "https://adventofcode.com/2021/leaderboard/private/view/991705.json"
-
 API_COGS = ["events", "memes"]