diff options
| -rw-r--r-- | bot/__init__.py | 5 | ||||
| -rw-r--r-- | bot/constants.py | 47 | ||||
| -rw-r--r-- | bot/exts/christmas/advent_of_code/__init__.py | 10 | ||||
| -rw-r--r-- | bot/exts/christmas/advent_of_code/_caches.py | 5 | ||||
| -rw-r--r-- | bot/exts/christmas/advent_of_code/_cog.py | 339 | ||||
| -rw-r--r-- | bot/exts/christmas/advent_of_code/_helpers.py | 348 | ||||
| -rw-r--r-- | bot/exts/christmas/adventofcode.py | 743 | ||||
| -rw-r--r-- | bot/resources/advent_of_code/about.json | 8 |
8 files changed, 748 insertions, 757 deletions
diff --git a/bot/__init__.py b/bot/__init__.py index a9a0865e..bdb18666 100644 --- a/bot/__init__.py +++ b/bot/__init__.py @@ -37,7 +37,8 @@ os.makedirs(log_dir, exist_ok=True) # File handler rotates logs every 5 MB file_handler = logging.handlers.RotatingFileHandler( - log_file, maxBytes=5 * (2**20), backupCount=10) + log_file, maxBytes=5 * (2**20), backupCount=10, encoding="utf-8", +) file_handler.setLevel(logging.TRACE if Client.debug else logging.DEBUG) # Console handler prints to terminal @@ -61,7 +62,7 @@ logging.basicConfig( format='%(asctime)s - %(name)s %(levelname)s: %(message)s', datefmt="%D %H:%M:%S", level=logging.TRACE if Client.debug else logging.DEBUG, - handlers=[console_handler, file_handler] + handlers=[console_handler, file_handler], ) logging.getLogger().info('Logging initialization complete') diff --git a/bot/constants.py b/bot/constants.py index 6999f321..292a242a 100644 --- a/bot/constants.py +++ b/bot/constants.py @@ -2,7 +2,7 @@ import enum import logging from datetime import datetime from os import environ -from typing import NamedTuple +from typing import Dict, NamedTuple __all__ = ( "AdventOfCode", @@ -29,11 +29,42 @@ __all__ = ( log = logging.getLogger(__name__) +class AdventOfCodeLeaderboard(NamedTuple): + id: str + session: str + join_code: str + + +def _parse_aoc_leaderboard_env() -> Dict[str, AdventOfCodeLeaderboard]: + """ + Parse the environment variable containing leaderboard information. + + A leaderboard should be specified in the format `id,session,join_code`, + without the backticks. If more than one leaderboard needs to be added to + the constant, separate the individual leaderboards with `::`. + + Example ENV: `id1,session1,join_code1::id2,session2,join_code2` + """ + raw_leaderboards = environ.get("AOC_LEADERBOARDS", "") + if not raw_leaderboards: + return {} + + leaderboards = {} + for leaderboard in raw_leaderboards.split("::"): + leaderboard_id, session, join_code = leaderboard.split(",") + leaderboards[leaderboard_id] = AdventOfCodeLeaderboard(leaderboard_id, session, join_code) + + return leaderboards + + class AdventOfCode: - leaderboard_cache_age_threshold_seconds = 3600 - leaderboard_id = 631135 - leaderboard_join_code = str(environ.get("AOC_JOIN_CODE", None)) - leaderboard_max_displayed_members = 10 + # Information for the several leaderboards we have + leaderboards = _parse_aoc_leaderboard_env() + staff_leaderboard_id = environ.get("AOC_STAFF_LEADERBOARD_ID", "") + + # Other Advent of Code constants + leaderboard_displayed_members = 10 + leaderboard_cache_expiry_seconds = 1800 year = int(environ.get("AOC_YEAR", datetime.utcnow().year)) role_id = int(environ.get("AOC_ROLE_ID", 518565788744024082)) @@ -44,7 +75,7 @@ class Branding: class Channels(NamedTuple): admins = 365960823622991872 - advent_of_code = int(environ.get("AOC_CHANNEL_ID", 517745814039166986)) + advent_of_code = int(environ.get("AOC_CHANNEL_ID", 782715290437943306)) announcements = int(environ.get("CHANNEL_ANNOUNCEMENTS", 354619224620138496)) big_brother_logs = 468507907357409333 bot = 267659945086812160 @@ -193,14 +224,14 @@ class Roles(NamedTuple): muted = 277914926603829249 owner = 267627879762755584 verified = 352427296948486144 - helpers = 267630620367257601 + helpers = int(environ.get("ROLE_HELPERS", 267630620367257601)) rockstars = 458226413825294336 core_developers = 587606783669829632 + events_lead = 778361735739998228 class Tokens(NamedTuple): giphy = environ.get("GIPHY_TOKEN") - aoc_session_cookie = environ.get("AOC_SESSION_COOKIE") omdb = environ.get("OMDB_API_KEY") youtube = environ.get("YOUTUBE_API_KEY") tmdb = environ.get("TMDB_API_KEY") diff --git a/bot/exts/christmas/advent_of_code/__init__.py b/bot/exts/christmas/advent_of_code/__init__.py new file mode 100644 index 00000000..3c521168 --- /dev/null +++ b/bot/exts/christmas/advent_of_code/__init__.py @@ -0,0 +1,10 @@ +from bot.bot import Bot + + +def setup(bot: Bot) -> None: + """Set up the Advent of Code extension.""" + # Import the Cog at runtime to prevent side effects like defining + # RedisCache instances too early. + from ._cog import AdventOfCode + + bot.add_cog(AdventOfCode(bot)) diff --git a/bot/exts/christmas/advent_of_code/_caches.py b/bot/exts/christmas/advent_of_code/_caches.py new file mode 100644 index 00000000..32d5394f --- /dev/null +++ b/bot/exts/christmas/advent_of_code/_caches.py @@ -0,0 +1,5 @@ +import async_rediscache + +leaderboard_counts = async_rediscache.RedisCache(namespace="AOC_leaderboard_counts") +leaderboard_cache = async_rediscache.RedisCache(namespace="AOC_leaderboard_cache") +assigned_leaderboard = async_rediscache.RedisCache(namespace="AOC_assigned_leaderboard") diff --git a/bot/exts/christmas/advent_of_code/_cog.py b/bot/exts/christmas/advent_of_code/_cog.py new file mode 100644 index 00000000..bc2a4724 --- /dev/null +++ b/bot/exts/christmas/advent_of_code/_cog.py @@ -0,0 +1,339 @@ +import asyncio +import json +import logging +import math +from datetime import datetime, timedelta +from pathlib import Path + +import discord +from discord.ext import commands + +from bot.bot import Bot +from bot.constants import ( + AdventOfCode as AocConfig, Channels, Colours, Emojis, Month, Roles, WHITELISTED_CHANNELS, +) +from bot.exts.christmas.advent_of_code import _helpers +from bot.utils.decorators import in_month, override_in_channel, with_role + +log = logging.getLogger(__name__) + +AOC_REQUEST_HEADER = {"user-agent": "PythonDiscord AoC Event Bot"} + +COUNTDOWN_STEP = 60 * 5 + +AOC_WHITELIST = WHITELISTED_CHANNELS + (Channels.advent_of_code,) + + +async def countdown_status(bot: commands.Bot) -> None: + """Set the playing status of the bot to the minutes & hours left until the next day's challenge.""" + while _helpers.is_in_advent(): + _, time_left = _helpers.time_left_to_aoc_midnight() + + aligned_seconds = int(math.ceil(time_left.seconds / COUNTDOWN_STEP)) * COUNTDOWN_STEP + hours, minutes = aligned_seconds // 3600, aligned_seconds // 60 % 60 + + if aligned_seconds == 0: + playing = "right now!" + elif aligned_seconds == COUNTDOWN_STEP: + playing = f"in less than {minutes} minutes" + elif hours == 0: + playing = f"in {minutes} minutes" + elif hours == 23: + playing = f"since {60 - minutes} minutes ago" + else: + playing = f"in {hours} hours and {minutes} minutes" + + # Status will look like "Playing in 5 hours and 30 minutes" + await bot.change_presence(activity=discord.Game(playing)) + + # Sleep until next aligned time or a full step if already aligned + delay = time_left.seconds % COUNTDOWN_STEP or COUNTDOWN_STEP + await asyncio.sleep(delay) + + +async def day_countdown(bot: commands.Bot) -> None: + """ + Calculate the number of seconds left until the next day of Advent. + + Once we have calculated this we should then sleep that number and when the time is reached, ping + the Advent of Code role notifying them that the new challenge is ready. + """ + while _helpers.is_in_advent(): + tomorrow, time_left = _helpers.time_left_to_aoc_midnight() + + # Prevent bot from being slightly too early in trying to announce today's puzzle + await asyncio.sleep(time_left.seconds + 1) + + channel = bot.get_channel(Channels.advent_of_code) + + if not channel: + log.error("Could not find the AoC channel to send notification in") + break + + aoc_role = channel.guild.get_role(AocConfig.role_id) + if not aoc_role: + log.error("Could not find the AoC role to announce the daily puzzle") + break + + puzzle_url = f"https://adventofcode.com/{AocConfig.year}/day/{tomorrow.day}" + + # Check if the puzzle is already available to prevent our members from spamming + # the puzzle page before it's available by making a small HEAD request. + for retry in range(1, 5): + log.debug(f"Checking if the puzzle is already available (attempt {retry}/4)") + async with bot.http_session.head(puzzle_url, raise_for_status=False) as resp: + if resp.status == 200: + log.debug("Puzzle is available; let's send an announcement message.") + break + log.debug(f"The puzzle is not yet available (status={resp.status})") + await asyncio.sleep(10) + else: + log.error("The puzzle does does not appear to be available at this time, canceling announcement") + break + + await channel.send( + f"{aoc_role.mention} Good morning! Day {tomorrow.day} is ready to be attempted. " + f"View it online now at {puzzle_url}. Good luck!", + allowed_mentions=discord.AllowedMentions( + everyone=False, + users=False, + roles=[discord.Object(AocConfig.role_id)], + ) + ) + + # Wait a couple minutes so that if our sleep didn't sleep enough + # time we don't end up announcing twice. + await asyncio.sleep(120) + + +class AdventOfCode(commands.Cog): + """Advent of Code festivities! Ho Ho Ho!""" + + def __init__(self, bot: Bot) -> None: + self.bot = bot + + self._base_url = f"https://adventofcode.com/{AocConfig.year}" + self.global_leaderboard_url = f"https://adventofcode.com/{AocConfig.year}/leaderboard" + + self.about_aoc_filepath = Path("./bot/resources/advent_of_code/about.json") + self.cached_about_aoc = self._build_about_embed() + + self.countdown_task = None + self.status_task = None + + countdown_coro = day_countdown(self.bot) + self.countdown_task = self.bot.loop.create_task(countdown_coro) + + status_coro = countdown_status(self.bot) + self.status_task = self.bot.loop.create_task(status_coro) + + @in_month(Month.DECEMBER) + @commands.group(name="adventofcode", aliases=("aoc",)) + @override_in_channel(AOC_WHITELIST) + async def adventofcode_group(self, ctx: commands.Context) -> None: + """All of the Advent of Code commands.""" + if not ctx.invoked_subcommand: + await ctx.send_help(ctx.command) + + @adventofcode_group.command( + name="subscribe", + aliases=("sub", "notifications", "notify", "notifs"), + brief="Notifications for new days" + ) + @override_in_channel(AOC_WHITELIST) + async def aoc_subscribe(self, ctx: commands.Context) -> None: + """Assign the role for notifications about new days being ready.""" + role = ctx.guild.get_role(AocConfig.role_id) + unsubscribe_command = f"{ctx.prefix}{ctx.command.root_parent} unsubscribe" + + if role not in ctx.author.roles: + await ctx.author.add_roles(role) + await ctx.send("Okay! You have been __subscribed__ to notifications about new Advent of Code tasks. " + f"You can run `{unsubscribe_command}` to disable them again for you.") + else: + await ctx.send("Hey, you already are receiving notifications about new Advent of Code tasks. " + f"If you don't want them any more, run `{unsubscribe_command}` instead.") + + @adventofcode_group.command(name="unsubscribe", aliases=("unsub",), brief="Notifications for new days") + @override_in_channel(AOC_WHITELIST) + async def aoc_unsubscribe(self, ctx: commands.Context) -> None: + """Remove the role for notifications about new days being ready.""" + role = ctx.guild.get_role(AocConfig.role_id) + + if role in ctx.author.roles: + await ctx.author.remove_roles(role) + await ctx.send("Okay! You have been __unsubscribed__ from notifications about new Advent of Code tasks.") + else: + await ctx.send("Hey, you don't even get any notifications about new Advent of Code tasks currently anyway.") + + @adventofcode_group.command(name="countdown", aliases=("count", "c"), brief="Return time left until next day") + @override_in_channel(AOC_WHITELIST) + async def aoc_countdown(self, ctx: commands.Context) -> None: + """Return time left until next day.""" + if not _helpers.is_in_advent(): + datetime_now = datetime.now(_helpers.EST) + + # Calculate the delta to this & next year's December 1st to see which one is closest and not in the past + this_year = datetime(datetime_now.year, 12, 1, tzinfo=_helpers.EST) + next_year = datetime(datetime_now.year + 1, 12, 1, tzinfo=_helpers.EST) + deltas = (dec_first - datetime_now for dec_first in (this_year, next_year)) + delta = min(delta for delta in deltas if delta >= timedelta()) # timedelta() gives 0 duration delta + + # Add a finer timedelta if there's less than a day left + if delta.days == 0: + delta_str = f"approximately {delta.seconds // 3600} hours" + else: + delta_str = f"{delta.days} days" + + await ctx.send(f"The Advent of Code event is not currently running. " + f"The next event will start in {delta_str}.") + return + + tomorrow, time_left = _helpers.time_left_to_aoc_midnight() + + hours, minutes = time_left.seconds // 3600, time_left.seconds // 60 % 60 + + await ctx.send(f"There are {hours} hours and {minutes} minutes left until day {tomorrow.day}.") + + @adventofcode_group.command(name="about", aliases=("ab", "info"), brief="Learn about Advent of Code") + @override_in_channel(AOC_WHITELIST) + async def about_aoc(self, ctx: commands.Context) -> None: + """Respond with an explanation of all things Advent of Code.""" + await ctx.send("", embed=self.cached_about_aoc) + + @adventofcode_group.command(name="join", aliases=("j",), brief="Learn how to join the leaderboard (via DM)") + @override_in_channel(AOC_WHITELIST) + async def join_leaderboard(self, ctx: commands.Context) -> None: + """DM the user the information for joining the Python Discord leaderboard.""" + author = ctx.message.author + log.info(f"{author.name} ({author.id}) has requested a PyDis AoC leaderboard code") + + if AocConfig.staff_leaderboard_id and any(r.id == Roles.helpers for r in author.roles): + join_code = AocConfig.leaderboards[AocConfig.staff_leaderboard_id].join_code + else: + join_code = await _helpers.get_public_join_code(author) + + if not join_code: + log.error(f"Failed to get a join code for user {author} ({author.id})") + error_embed = discord.Embed( + title="Unable to get join code", + description="Failed to get a join code to one of our boards. Please notify staff.", + colour=discord.Colour.red(), + ) + await ctx.send(embed=error_embed) + return + + info_str = ( + "Head over to https://adventofcode.com/leaderboard/private " + f"with code `{join_code}` to join the Python Discord leaderboard!" + ) + try: + await author.send(info_str) + except discord.errors.Forbidden: + log.debug(f"{author.name} ({author.id}) has disabled DMs from server members") + await ctx.send(f":x: {author.mention}, please (temporarily) enable DMs to receive the join code") + else: + await ctx.message.add_reaction(Emojis.envelope) + + @adventofcode_group.command( + name="leaderboard", + aliases=("board", "lb"), + brief="Get a snapshot of the PyDis private AoC leaderboard", + ) + @override_in_channel(AOC_WHITELIST) + async def aoc_leaderboard(self, ctx: commands.Context) -> None: + """Get the current top scorers of the Python Discord Leaderboard.""" + async with ctx.typing(): + leaderboard = await _helpers.fetch_leaderboard() + number_of_participants = leaderboard["number_of_participants"] + + top_count = min(AocConfig.leaderboard_displayed_members, number_of_participants) + header = f"Here's our current top {top_count}! {Emojis.christmas_tree * 3}" + + table = f"```\n{leaderboard['top_leaderboard']}\n```" + info_embed = _helpers.get_summary_embed(leaderboard) + + await ctx.send(content=f"{header}\n\n{table}", embed=info_embed) + + @adventofcode_group.command( + name="global", + aliases=("globalboard", "gb"), + brief="Get a link to the global leaderboard", + ) + @override_in_channel(AOC_WHITELIST) + async def aoc_global_leaderboard(self, ctx: commands.Context) -> None: + """Get a link to the global Advent of Code leaderboard.""" + url = self.global_leaderboard_url + global_leaderboard = discord.Embed( + title="Advent of Code — Global Leaderboard", + description=f"You can find the global leaderboard [here]({url})." + ) + global_leaderboard.set_thumbnail(url=_helpers.AOC_EMBED_THUMBNAIL) + await ctx.send(embed=global_leaderboard) + + @adventofcode_group.command( + name="stats", + aliases=("dailystats", "ds"), + brief="Get daily statistics for the Python Discord leaderboard" + ) + @override_in_channel(AOC_WHITELIST) + async def private_leaderboard_daily_stats(self, ctx: commands.Context) -> None: + """Send an embed with daily completion statistics for the Python Discord leaderboard.""" + leaderboard = await _helpers.fetch_leaderboard() + + # The daily stats are serialized as JSON as they have to be cached in Redis + daily_stats = json.loads(leaderboard["daily_stats"]) + async with ctx.typing(): + lines = ["Day ⭐ ⭐⭐ | %⭐ %⭐⭐\n================================"] + for day, stars in daily_stats.items(): + star_one = stars["star_one"] + star_two = stars["star_two"] + p_star_one = star_one / leaderboard["number_of_participants"] + p_star_two = star_two / leaderboard["number_of_participants"] + lines.append( + f"{day:>2}) {star_one:>4} {star_two:>4} | {p_star_one:>7.2%} {p_star_two:>7.2%}" + ) + table = "\n".join(lines) + info_embed = _helpers.get_summary_embed(leaderboard) + await ctx.send(f"```\n{table}\n```", embed=info_embed) + + @with_role(Roles.admin, Roles.events_lead) + @adventofcode_group.command( + name="refresh", + aliases=("fetch",), + brief="Force a refresh of the leaderboard cache.", + ) + async def refresh_leaderboard(self, ctx: commands.Context) -> None: + """ + Force a refresh of the leaderboard cache. + + Note: This should be used sparingly, as we want to prevent sending too + many requests to the Advent of Code server. + """ + async with ctx.typing(): + await _helpers.fetch_leaderboard(invalidate_cache=True) + await ctx.send("\N{OK Hand Sign} Refreshed leaderboard cache!") + + def cog_unload(self) -> None: + """Cancel season-related tasks on cog unload.""" + log.debug("Unloading the cog and canceling the background task.") + self.countdown_task.cancel() + self.status_task.cancel() + + def _build_about_embed(self) -> discord.Embed: + """Build and return the informational "About AoC" embed from the resources file.""" + with self.about_aoc_filepath.open("r", encoding="utf8") as f: + embed_fields = json.load(f) + + about_embed = discord.Embed( + title=self._base_url, + colour=Colours.soft_green, + url=self._base_url, + timestamp=datetime.utcnow() + ) + about_embed.set_author(name="Advent of Code", url=self._base_url) + for field in embed_fields: + about_embed.add_field(**field) + + about_embed.set_footer(text="Last Updated") + return about_embed diff --git a/bot/exts/christmas/advent_of_code/_helpers.py b/bot/exts/christmas/advent_of_code/_helpers.py new file mode 100644 index 00000000..7ac54322 --- /dev/null +++ b/bot/exts/christmas/advent_of_code/_helpers.py @@ -0,0 +1,348 @@ +import collections +import datetime +import json +import logging +import operator +import typing +from typing import Tuple + +import aiohttp +import discord +import pytz + +from bot.constants import AdventOfCode, Colours +from bot.exts.christmas.advent_of_code import _caches + +log = logging.getLogger(__name__) + +PASTE_URL = "https://paste.pythondiscord.com/documents" +RAW_PASTE_URL_TEMPLATE = "https://paste.pythondiscord.com/raw/{key}" + +# Base API URL for Advent of Code Private Leaderboards +AOC_API_URL = "https://adventofcode.com/{year}/leaderboard/private/view/{leaderboard_id}.json" +AOC_REQUEST_HEADER = {"user-agent": "PythonDiscord AoC Event Bot"} + +# Leaderboard Line Template +AOC_TABLE_TEMPLATE = "{rank: >4} | {name:25.25} | {score: >5} | {stars}" +HEADER = AOC_TABLE_TEMPLATE.format(rank="", name="Name", score="Score", stars="⭐, ⭐⭐") +HEADER = f"{HEADER}\n{'-' * (len(HEADER) + 2)}" +HEADER_LINES = len(HEADER.splitlines()) +TOP_LEADERBOARD_LINES = HEADER_LINES + AdventOfCode.leaderboard_displayed_members + +# Keys that need to be set for a cached leaderboard +REQUIRED_CACHE_KEYS = ( + "full_leaderboard", + "top_leaderboard", + "full_leaderboard_url", + "leaderboard_fetched_at", + "number_of_participants", + "daily_stats", +) + +AOC_EMBED_THUMBNAIL = ( + "https://raw.githubusercontent.com/python-discord" + "/branding/master/seasonal/christmas/server_icons/festive_256.gif" +) + +# Create an easy constant for the EST timezone +EST = pytz.timezone("EST") + +# Create namedtuple that combines a participant's name and their completion +# time for a specific star. We're going to use this later to order the results +# for each star to compute the rank score. +_StarResult = collections.namedtuple("StarResult", "name completion_time") + + +def _parse_raw_leaderboard_data(raw_leaderboard_data: dict) -> dict: + """ + Parse the leaderboard data received from the AoC website. + + The data we receive from AoC is structured by member, not by day/star. This + means that we need to "transpose" the data to a per star structure in order + to calculate the rank scores each individual should get. + + As we need our data both "per participant" as well as "per day", we return + the parsed and analyzed data in both formats. + """ + # We need to get an aggregate of completion times for each star of each day, + # instead of per participant to compute the rank scores. This dictionary will + # provide such a transposed dataset. + star_results = collections.defaultdict(list) + + # As we're already iterating over the participants, we can record the number of + # first stars and second stars they've achieved right here and now. This means + # we won't have to iterate over the participants again later. + leaderboard = {} + + # The data we get from the AoC website is structured by member, not by day/star, + # which means we need to iterate over the members to transpose the data to a per + # star view. We need that per star view to compute rank scores per star. + for member in raw_leaderboard_data.values(): + name = member["name"] if member["name"] else f"Anonymous #{member['id']}" + leaderboard[name] = {"score": 0, "star_1_count": 0, "star_2_count": 0} + + # Iterate over all days for this participant + for day, stars in member["completion_day_level"].items(): + # Iterate over the complete stars for this day for this participant + for star, data in stars.items(): + # Record completion of this star for this individual + leaderboard[name][f"star_{star}_count"] += 1 + + # Record completion datetime for this participant for this day/star + completion_time = datetime.datetime.fromtimestamp(int(data['get_star_ts'])) + star_results[(day, star)].append( + _StarResult(name=name, completion_time=completion_time) + ) + + # Now that we have a transposed dataset that holds the completion time of all + # participants per star, we can compute the rank-based scores each participant + # should get for that star. + max_score = len(leaderboard) + for star in star_results.values(): + for rank, star_result in enumerate(sorted(star, key=operator.itemgetter(1))): + leaderboard[star_result.name]["score"] += max_score - rank + + # Since dictionaries now retain insertion order, let's use that + sorted_leaderboard = dict( + sorted(leaderboard.items(), key=lambda t: t[1]["score"], reverse=True) + ) + + # Create summary stats for the stars completed for each day of the event. + daily_stats = {} + for day in range(1, 26): + star_one = len(star_results.get((day, 1), [])) + star_two = len(star_results.get((day, 1), [])) + # By using a dictionary instead of namedtuple here, we can serialize + # this data to JSON in order to cache it in Redis. + daily_stats[day] = {"star_one": star_one, "star_two": star_two} + + return {"daily_stats": daily_stats, "leaderboard": sorted_leaderboard} + + +def _format_leaderboard(leaderboard: typing.Dict[str, int]) -> str: + """Format the leaderboard using the AOC_TABLE_TEMPLATE.""" + leaderboard_lines = [HEADER] + for rank, (name, results) in enumerate(leaderboard.items(), start=1): + leaderboard_lines.append( + AOC_TABLE_TEMPLATE.format( + rank=rank, + name=name, + score=str(results["score"]), + stars=f"({results['star_1_count']}, {results['star_2_count']})" + ) + ) + + return "\n".join(leaderboard_lines) + + +async def _fetch_leaderboard_data() -> typing.Dict[str, typing.Any]: + """Fetch data for all leaderboards and return a pooled result.""" + year = AdventOfCode.year + + # We'll make our requests one at a time to not flood the AoC website with + # up to six simultaneous requests. This may take a little longer, but it + # does avoid putting unnecessary stress on the Advent of Code website. + + # Container to store the raw data of each leaderboard + participants = {} + for leaderboard in AdventOfCode.leaderboards.values(): + leaderboard_url = AOC_API_URL.format(year=year, leaderboard_id=leaderboard.id) + cookies = {"session": leaderboard.session} + + # We don't need to create a session if we're going to throw it away after each request + async with aiohttp.request( + "GET", leaderboard_url, headers=AOC_REQUEST_HEADER, cookies=cookies + ) as resp: + if resp.status == 200: + raw_data = await resp.json() + + # Get the participants and store their current count + board_participants = raw_data["members"] + await _caches.leaderboard_counts.set(leaderboard.id, len(board_participants)) + participants.update(board_participants) + else: + log.warning(f"Fetching data failed for leaderboard `{leaderboard.id}`") + resp.raise_for_status() + + log.info(f"Fetched leaderboard information for {len(participants)} participants") + return participants + + +async def _upload_leaderboard(leaderboard: str) -> str: + """Upload the full leaderboard to our paste service and return the URL.""" + async with aiohttp.request("POST", PASTE_URL, data=leaderboard) as resp: + try: + resp_json = await resp.json() + except Exception: + log.exception("Failed to upload full leaderboard to paste service") + return "" + + if "key" in resp_json: + return RAW_PASTE_URL_TEMPLATE.format(key=resp_json["key"]) + + log.error(f"Unexpected response from paste service while uploading leaderboard {resp_json}") + return "" + + +def _get_top_leaderboard(full_leaderboard: str) -> str: + """Get the leaderboard up to the maximum specified entries.""" + return "\n".join(full_leaderboard.splitlines()[:TOP_LEADERBOARD_LINES]) + + +@_caches.leaderboard_cache.atomic_transaction +async def fetch_leaderboard(invalidate_cache: bool = False) -> dict: + """ + Get the current Python Discord combined leaderboard. + + The leaderboard is cached and only fetched from the API if the current data + is older than the lifetime set in the constants. To prevent multiple calls + to this function fetching new leaderboard information in case of a cache + miss, this function is locked to one call at a time using a decorator. + """ + cached_leaderboard = await _caches.leaderboard_cache.to_dict() + + # Check if the cached leaderboard contains everything we expect it to. If it + # does not, this probably means the cache has not been created yet or has + # expired in Redis. This check also accounts for a malformed cache. + if invalidate_cache or any(key not in cached_leaderboard for key in REQUIRED_CACHE_KEYS): + log.info("No leaderboard cache available, fetching leaderboards...") + # Fetch the raw data + raw_leaderboard_data = await _fetch_leaderboard_data() + + # Parse it to extract "per star, per day" data and participant scores + parsed_leaderboard_data = _parse_raw_leaderboard_data(raw_leaderboard_data) + + leaderboard = parsed_leaderboard_data["leaderboard"] + number_of_participants = len(leaderboard) + formatted_leaderboard = _format_leaderboard(leaderboard) + full_leaderboard_url = await _upload_leaderboard(formatted_leaderboard) + leaderboard_fetched_at = datetime.datetime.utcnow().isoformat() + + cached_leaderboard = { + "full_leaderboard": formatted_leaderboard, + "top_leaderboard": _get_top_leaderboard(formatted_leaderboard), + "full_leaderboard_url": full_leaderboard_url, + "leaderboard_fetched_at": leaderboard_fetched_at, + "number_of_participants": number_of_participants, + "daily_stats": json.dumps(parsed_leaderboard_data["daily_stats"]), + } + + # Store the new values in Redis + await _caches.leaderboard_cache.update(cached_leaderboard) + + # Set an expiry on the leaderboard RedisCache + with await _caches.leaderboard_cache._get_pool_connection() as connection: + await connection.expire( + _caches.leaderboard_cache.namespace, + AdventOfCode.leaderboard_cache_expiry_seconds + ) + + return cached_leaderboard + + +def get_summary_embed(leaderboard: dict) -> discord.Embed: + """Get an embed with the current summary stats of the leaderboard.""" + leaderboard_url = leaderboard['full_leaderboard_url'] + + aoc_embed = discord.Embed( + colour=Colours.soft_green, + timestamp=datetime.datetime.fromisoformat(leaderboard["leaderboard_fetched_at"]), + ) + aoc_embed.add_field( + name="Number of Participants", + value=leaderboard["number_of_participants"], + inline=True, + ) + if leaderboard_url: + aoc_embed.add_field( + name="Full Leaderboard", + value=f"[Python Discord Leaderboard]({leaderboard_url})", + inline=True, + ) + aoc_embed.set_author(name="Advent of Code", url=leaderboard_url) + aoc_embed.set_footer(text="Last Updated") + aoc_embed.set_thumbnail(url=AOC_EMBED_THUMBNAIL) + + return aoc_embed + + +async def get_public_join_code(author: discord.Member) -> typing.Optional[str]: + """ + Get the join code for one of the non-staff leaderboards. + + If a user has previously requested a join code and their assigned board + hasn't filled up yet, we'll return the same join code to prevent them from + getting join codes for multiple boards. + """ + # Make sure to fetch new leaderboard information if the cache is older than + # 30 minutes. While this still means that there could be a discrepancy + # between the current leaderboard state and the numbers we have here, this + # should work fairly well given the buffer of slots that we have. + await fetch_leaderboard() + previously_assigned_board = await _caches.assigned_leaderboard.get(author.id) + current_board_counts = await _caches.leaderboard_counts.to_dict() + + # Remove the staff board from the current board counts as it should be ignored. + current_board_counts.pop(AdventOfCode.staff_leaderboard_id, None) + + # If this user has already received a join code, we'll give them the + # exact same one to prevent them from joining multiple boards and taking + # up multiple slots. + if previously_assigned_board: + # Check if their previously assigned board still has room for them + if current_board_counts.get(previously_assigned_board, 0) < 200: + log.info(f"{author} ({author.id}) was already assigned to a board with open slots.") + return AdventOfCode.leaderboards[previously_assigned_board].join_code + + log.info( + f"User {author} ({author.id}) previously received the join code for " + f"board `{previously_assigned_board}`, but that board's now full. " + "Assigning another board to this user." + ) + + # If we don't have the current board counts cached, let's force fetching a new cache + if not current_board_counts: + log.warning("Leaderboard counts were missing from the cache unexpectedly!") + await fetch_leaderboard(invalidate_cache=True) + current_board_counts = await _caches.leaderboard_counts.to_dict() + + # Find the board with the current lowest participant count. As we can't + best_board, _count = min(current_board_counts.items(), key=operator.itemgetter(1)) + + if current_board_counts.get(best_board, 0) >= 200: + log.warning(f"User {author} `{author.id}` requested a join code, but all boards are full!") + return + + log.info(f"Assigning user {author} ({author.id}) to board `{best_board}`") + await _caches.assigned_leaderboard.set(author.id, best_board) + + # Return the join code for this board + return AdventOfCode.leaderboards[best_board].join_code + + +def is_in_advent() -> bool: + """ + Check if we're currently on an Advent of Code day, excluding 25 December. + + This helper function is used to check whether or not a feature that prepares + something for the next Advent of Code challenge should run. As the puzzle + published on the 25th is the last puzzle, this check excludes that date. + """ + return datetime.datetime.now(EST).day in range(1, 25) and datetime.datetime.now(EST).month == 12 + + +def time_left_to_aoc_midnight() -> Tuple[datetime.datetime, datetime.timedelta]: + """Calculates the amount of time left until midnight in UTC-5 (Advent of Code maintainer timezone).""" + # Change all time properties back to 00:00 + todays_midnight = datetime.datetime.now(EST).replace( + microsecond=0, + second=0, + minute=0, + hour=0 + ) + + # We want tomorrow so add a day on + tomorrow = todays_midnight + datetime.timedelta(days=1) + + # Calculate the timedelta between the current time and midnight + return tomorrow, tomorrow - datetime.datetime.now(EST) diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py deleted file mode 100644 index b3fe0623..00000000 --- a/bot/exts/christmas/adventofcode.py +++ /dev/null @@ -1,743 +0,0 @@ -import asyncio -import json -import logging -import math -import re -from datetime import datetime, timedelta -from pathlib import Path -from typing import List, Tuple - -import aiohttp -import discord -from bs4 import BeautifulSoup -from discord.ext import commands -from pytz import timezone - -from bot.constants import AdventOfCode as AocConfig, Channels, Colours, Emojis, Month, Tokens, WHITELISTED_CHANNELS -from bot.utils import unlocked_role -from bot.utils.decorators import in_month, override_in_channel - -log = logging.getLogger(__name__) - -AOC_REQUEST_HEADER = {"user-agent": "PythonDiscord AoC Event Bot"} -AOC_SESSION_COOKIE = {"session": Tokens.aoc_session_cookie} - -EST = timezone("EST") -COUNTDOWN_STEP = 60 * 5 - -AOC_WHITELIST = WHITELISTED_CHANNELS + (Channels.advent_of_code,) - - -def is_in_advent() -> bool: - """Utility function to check if we are between December 1st and December 25th.""" - # Run the code from the 1st to the 24th - return datetime.now(EST).day in range(1, 25) and datetime.now(EST).month == 12 - - -def time_left_to_aoc_midnight() -> Tuple[datetime, timedelta]: - """Calculates the amount of time left until midnight in UTC-5 (Advent of Code maintainer timezone).""" - # Change all time properties back to 00:00 - todays_midnight = datetime.now(EST).replace(microsecond=0, - second=0, - minute=0, - hour=0) - - # We want tomorrow so add a day on - tomorrow = todays_midnight + timedelta(days=1) - - # Calculate the timedelta between the current time and midnight - return tomorrow, tomorrow - datetime.now(EST) - - -async def countdown_status(bot: commands.Bot) -> None: - """Set the playing status of the bot to the minutes & hours left until the next day's challenge.""" - while is_in_advent(): - _, time_left = time_left_to_aoc_midnight() - - aligned_seconds = int(math.ceil(time_left.seconds / COUNTDOWN_STEP)) * COUNTDOWN_STEP - hours, minutes = aligned_seconds // 3600, aligned_seconds // 60 % 60 - - if aligned_seconds == 0: - playing = "right now!" - elif aligned_seconds == COUNTDOWN_STEP: - playing = f"in less than {minutes} minutes" - elif hours == 0: - playing = f"in {minutes} minutes" - elif hours == 23: - playing = f"since {60 - minutes} minutes ago" - else: - playing = f"in {hours} hours and {minutes} minutes" - - # Status will look like "Playing in 5 hours and 30 minutes" - await bot.change_presence(activity=discord.Game(playing)) - - # Sleep until next aligned time or a full step if already aligned - delay = time_left.seconds % COUNTDOWN_STEP or COUNTDOWN_STEP - await asyncio.sleep(delay) - - -async def day_countdown(bot: commands.Bot) -> None: - """ - Calculate the number of seconds left until the next day of Advent. - - Once we have calculated this we should then sleep that number and when the time is reached, ping - the Advent of Code role notifying them that the new challenge is ready. - """ - while is_in_advent(): - tomorrow, time_left = time_left_to_aoc_midnight() - - # Correct `time_left.seconds` for the sleep we have after unlocking the role (-5) and adding - # a second (+1) as the bot is consistently ~0.5 seconds early in announcing the puzzles. - await asyncio.sleep(time_left.seconds - 4) - - channel = bot.get_channel(Channels.advent_of_code) - - if not channel: - log.error("Could not find the AoC channel to send notification in") - break - - aoc_role = channel.guild.get_role(AocConfig.role_id) - if not aoc_role: - log.error("Could not find the AoC role to announce the daily puzzle") - break - - async with unlocked_role(aoc_role, delay=5): - puzzle_url = f"https://adventofcode.com/{AocConfig.year}/day/{tomorrow.day}" - - # Check if the puzzle is already available to prevent our members from spamming - # the puzzle page before it's available by making a small HEAD request. - for retry in range(1, 5): - log.debug(f"Checking if the puzzle is already available (attempt {retry}/4)") - async with bot.http_session.head(puzzle_url, raise_for_status=False) as resp: - if resp.status == 200: - log.debug("Puzzle is available; let's send an announcement message.") - break - log.debug(f"The puzzle is not yet available (status={resp.status})") - await asyncio.sleep(10) - else: - log.error("The puzzle does does not appear to be available at this time, canceling announcement") - break - - await channel.send( - f"{aoc_role.mention} Good morning! Day {tomorrow.day} is ready to be attempted. " - f"View it online now at {puzzle_url}. Good luck!" - ) - - # Wait a couple minutes so that if our sleep didn't sleep enough - # time we don't end up announcing twice. - await asyncio.sleep(120) - - -class AdventOfCode(commands.Cog): - """Advent of Code festivities! Ho Ho Ho!""" - - def __init__(self, bot: commands.Bot): - self.bot = bot - - self._base_url = f"https://adventofcode.com/{AocConfig.year}" - self.global_leaderboard_url = f"https://adventofcode.com/{AocConfig.year}/leaderboard" - self.private_leaderboard_url = f"{self._base_url}/leaderboard/private/view/{AocConfig.leaderboard_id}" - - self.about_aoc_filepath = Path("./bot/resources/advent_of_code/about.json") - self.cached_about_aoc = self._build_about_embed() - - self.cached_global_leaderboard = None - self.cached_private_leaderboard = None - - self.countdown_task = None - self.status_task = None - - countdown_coro = day_countdown(self.bot) - self.countdown_task = self.bot.loop.create_task(countdown_coro) - - status_coro = countdown_status(self.bot) - self.status_task = self.bot.loop.create_task(status_coro) - - @in_month(Month.DECEMBER) - @commands.group(name="adventofcode", aliases=("aoc",)) - @override_in_channel(AOC_WHITELIST) - async def adventofcode_group(self, ctx: commands.Context) -> None: - """All of the Advent of Code commands.""" - if not ctx.invoked_subcommand: - await ctx.send_help(ctx.command) - - @adventofcode_group.command( - name="subscribe", - aliases=("sub", "notifications", "notify", "notifs"), - brief="Notifications for new days" - ) - @override_in_channel(AOC_WHITELIST) - async def aoc_subscribe(self, ctx: commands.Context) -> None: - """Assign the role for notifications about new days being ready.""" - role = ctx.guild.get_role(AocConfig.role_id) - unsubscribe_command = f"{ctx.prefix}{ctx.command.root_parent} unsubscribe" - - if role not in ctx.author.roles: - await ctx.author.add_roles(role) - await ctx.send("Okay! You have been __subscribed__ to notifications about new Advent of Code tasks. " - f"You can run `{unsubscribe_command}` to disable them again for you.") - else: - await ctx.send("Hey, you already are receiving notifications about new Advent of Code tasks. " - f"If you don't want them any more, run `{unsubscribe_command}` instead.") - - @adventofcode_group.command(name="unsubscribe", aliases=("unsub",), brief="Notifications for new days") - @override_in_channel(AOC_WHITELIST) - async def aoc_unsubscribe(self, ctx: commands.Context) -> None: - """Remove the role for notifications about new days being ready.""" - role = ctx.guild.get_role(AocConfig.role_id) - - if role in ctx.author.roles: - await ctx.author.remove_roles(role) - await ctx.send("Okay! You have been __unsubscribed__ from notifications about new Advent of Code tasks.") - else: - await ctx.send("Hey, you don't even get any notifications about new Advent of Code tasks currently anyway.") - - @adventofcode_group.command(name="countdown", aliases=("count", "c"), brief="Return time left until next day") - @override_in_channel(AOC_WHITELIST) - async def aoc_countdown(self, ctx: commands.Context) -> None: - """Return time left until next day.""" - if not is_in_advent(): - datetime_now = datetime.now(EST) - - # Calculate the delta to this & next year's December 1st to see which one is closest and not in the past - this_year = datetime(datetime_now.year, 12, 1, tzinfo=EST) - next_year = datetime(datetime_now.year + 1, 12, 1, tzinfo=EST) - deltas = (dec_first - datetime_now for dec_first in (this_year, next_year)) - delta = min(delta for delta in deltas if delta >= timedelta()) # timedelta() gives 0 duration delta - - # Add a finer timedelta if there's less than a day left - if delta.days == 0: - delta_str = f"approximately {delta.seconds // 3600} hours" - else: - delta_str = f"{delta.days} days" - - await ctx.send(f"The Advent of Code event is not currently running. " - f"The next event will start in {delta_str}.") - return - - tomorrow, time_left = time_left_to_aoc_midnight() - - hours, minutes = time_left.seconds // 3600, time_left.seconds // 60 % 60 - - await ctx.send(f"There are {hours} hours and {minutes} minutes left until day {tomorrow.day}.") - - @adventofcode_group.command(name="about", aliases=("ab", "info"), brief="Learn about Advent of Code") - @override_in_channel(AOC_WHITELIST) - async def about_aoc(self, ctx: commands.Context) -> None: - """Respond with an explanation of all things Advent of Code.""" - await ctx.send("", embed=self.cached_about_aoc) - - @adventofcode_group.command(name="join", aliases=("j",), brief="Learn how to join the leaderboard (via DM)") - @override_in_channel(AOC_WHITELIST) - async def join_leaderboard(self, ctx: commands.Context) -> None: - """DM the user the information for joining the PyDis AoC private leaderboard.""" - author = ctx.message.author - log.info(f"{author.name} ({author.id}) has requested the PyDis AoC leaderboard code") - - info_str = ( - "Head over to https://adventofcode.com/leaderboard/private " - f"with code `{AocConfig.leaderboard_join_code}` to join the PyDis private leaderboard!" - ) - try: - await author.send(info_str) - except discord.errors.Forbidden: - log.debug(f"{author.name} ({author.id}) has disabled DMs from server members") - await ctx.send(f":x: {author.mention}, please (temporarily) enable DMs to receive the join code") - else: - await ctx.message.add_reaction(Emojis.envelope) - - @adventofcode_group.command( - name="leaderboard", - aliases=("board", "lb"), - brief="Get a snapshot of the PyDis private AoC leaderboard", - ) - @override_in_channel(AOC_WHITELIST) - async def aoc_leaderboard(self, ctx: commands.Context, number_of_people_to_display: int = 10) -> None: - """ - Pull the top number_of_people_to_display members from the PyDis leaderboard and post an embed. - - For readability, number_of_people_to_display defaults to 10. A maximum value is configured in the - Advent of Code section of the bot constants. number_of_people_to_display values greater than this - limit will default to this maximum and provide feedback to the user. - """ - async with ctx.typing(): - await self._check_leaderboard_cache(ctx) - - if not self.cached_private_leaderboard: - # Feedback on issues with leaderboard caching are sent by _check_leaderboard_cache() - # Short circuit here if there's an issue - return - - number_of_people_to_display = await self._check_n_entries(ctx, number_of_people_to_display) - - # Generate leaderboard table for embed - members_to_print = self.cached_private_leaderboard.top_n(number_of_people_to_display) - table = AocPrivateLeaderboard.build_leaderboard_embed(members_to_print) - - # Build embed - aoc_embed = discord.Embed( - description=f"Total members: {len(self.cached_private_leaderboard.members)}", - colour=Colours.soft_green, - timestamp=self.cached_private_leaderboard.last_updated - ) - aoc_embed.set_author(name="Advent of Code", url=self.private_leaderboard_url) - aoc_embed.set_footer(text="Last Updated") - - await ctx.send( - content=f"Here's the current Top {number_of_people_to_display}! {Emojis.christmas_tree*3}\n\n{table}", - embed=aoc_embed, - ) - - @adventofcode_group.command( - name="stats", - aliases=("dailystats", "ds"), - brief="Get daily statistics for the PyDis private leaderboard" - ) - @override_in_channel(AOC_WHITELIST) - async def private_leaderboard_daily_stats(self, ctx: commands.Context) -> None: - """ - Respond with a table of the daily completion statistics for the PyDis private leaderboard. - - Embed will display the total members and the number of users who have completed each day's puzzle - """ - async with ctx.typing(): - await self._check_leaderboard_cache(ctx) - - if not self.cached_private_leaderboard: - # Feedback on issues with leaderboard caching are sent by _check_leaderboard_cache() - # Short circuit here if there's an issue - return - - # Build ASCII table - total_members = len(self.cached_private_leaderboard.members) - _star = Emojis.star - header = f"{'Day':4}{_star:^8}{_star*2:^4}{'% ' + _star:^8}{'% ' + _star*2:^4}\n{'='*35}" - table = "" - for day, completions in enumerate(self.cached_private_leaderboard.daily_completion_summary): - per_one_star = f"{(completions[0]/total_members)*100:.2f}" - per_two_star = f"{(completions[1]/total_members)*100:.2f}" - - table += f"{day+1:3}){completions[0]:^8}{completions[1]:^6}{per_one_star:^10}{per_two_star:^6}\n" - - table = f"```\n{header}\n{table}```" - - # Build embed - daily_stats_embed = discord.Embed( - colour=Colours.soft_green, timestamp=self.cached_private_leaderboard.last_updated - ) - daily_stats_embed.set_author(name="Advent of Code", url=self._base_url) - daily_stats_embed.set_footer(text="Last Updated") - - await ctx.send( - content=f"Here's the current daily statistics!\n\n{table}", embed=daily_stats_embed - ) - - @adventofcode_group.command( - name="global", - aliases=("globalboard", "gb"), - brief="Get a snapshot of the global AoC leaderboard", - ) - @override_in_channel(AOC_WHITELIST) - async def global_leaderboard(self, ctx: commands.Context, number_of_people_to_display: int = 10) -> None: - """ - Pull the top number_of_people_to_display members from the global AoC leaderboard and post an embed. - - For readability, number_of_people_to_display defaults to 10. A maximum value is configured in the - Advent of Code section of the bot constants. number_of_people_to_display values greater than this - limit will default to this maximum and provide feedback to the user. - """ - async with ctx.typing(): - await self._check_leaderboard_cache(ctx, global_board=True) - - if not self.cached_global_leaderboard: - # Feedback on issues with leaderboard caching are sent by _check_leaderboard_cache() - # Short circuit here if there's an issue - return - - number_of_people_to_display = await self._check_n_entries(ctx, number_of_people_to_display) - - # Generate leaderboard table for embed - members_to_print = self.cached_global_leaderboard.top_n(number_of_people_to_display) - table = AocGlobalLeaderboard.build_leaderboard_embed(members_to_print) - - # Build embed - aoc_embed = discord.Embed(colour=Colours.soft_green, timestamp=self.cached_global_leaderboard.last_updated) - aoc_embed.set_author(name="Advent of Code", url=self._base_url) - aoc_embed.set_footer(text="Last Updated") - - await ctx.send( - f"Here's the current global Top {number_of_people_to_display}! {Emojis.christmas_tree*3}\n\n{table}", - embed=aoc_embed, - ) - - async def _check_leaderboard_cache(self, ctx: commands.Context, global_board: bool = False) -> None: - """ - Check age of current leaderboard & pull a new one if the board is too old. - - global_board is a boolean to toggle between the global board and the Pydis private board - """ - # Toggle between global & private leaderboards - if global_board: - log.debug("Checking global leaderboard cache") - leaderboard_str = "cached_global_leaderboard" - _shortstr = "global" - else: - log.debug("Checking private leaderboard cache") - leaderboard_str = "cached_private_leaderboard" - _shortstr = "private" - - leaderboard = getattr(self, leaderboard_str) - if not leaderboard: - log.debug(f"No cached {_shortstr} leaderboard found") - await self._boardgetter(global_board) - else: - leaderboard_age = datetime.utcnow() - leaderboard.last_updated - age_seconds = leaderboard_age.total_seconds() - if age_seconds < AocConfig.leaderboard_cache_age_threshold_seconds: - log.debug(f"Cached {_shortstr} leaderboard age less than threshold ({age_seconds} seconds old)") - else: - log.debug(f"Cached {_shortstr} leaderboard age greater than threshold ({age_seconds} seconds old)") - await self._boardgetter(global_board) - - leaderboard = getattr(self, leaderboard_str) - if not leaderboard: - await ctx.send( - "", - embed=_error_embed_helper( - title=f"Something's gone wrong and there's no cached {_shortstr} leaderboard!", - description="Please check in with a staff member.", - ), - ) - - async def _check_n_entries(self, ctx: commands.Context, number_of_people_to_display: int) -> int: - """Check for n > max_entries and n <= 0.""" - max_entries = AocConfig.leaderboard_max_displayed_members - author = ctx.message.author - if not 0 <= number_of_people_to_display <= max_entries: - log.debug( - f"{author.name} ({author.id}) attempted to fetch an invalid number " - f" of entries from the AoC leaderboard ({number_of_people_to_display})" - ) - await ctx.send( - f":x: {author.mention}, number of entries to display must be a positive " - f"integer less than or equal to {max_entries}\n\n" - f"Head to {self.private_leaderboard_url} to view the entire leaderboard" - ) - number_of_people_to_display = max_entries - - return number_of_people_to_display - - def _build_about_embed(self) -> discord.Embed: - """Build and return the informational "About AoC" embed from the resources file.""" - with self.about_aoc_filepath.open("r", encoding="utf8") as f: - embed_fields = json.load(f) - - about_embed = discord.Embed(title=self._base_url, colour=Colours.soft_green, url=self._base_url) - about_embed.set_author(name="Advent of Code", url=self._base_url) - for field in embed_fields: - about_embed.add_field(**field) - - about_embed.set_footer(text=f"Last Updated (UTC): {datetime.utcnow()}") - - return about_embed - - async def _boardgetter(self, global_board: bool) -> None: - """Invoke the proper leaderboard getter based on the global_board boolean.""" - if global_board: - self.cached_global_leaderboard = await AocGlobalLeaderboard.from_url() - else: - self.cached_private_leaderboard = await AocPrivateLeaderboard.from_url() - - def cog_unload(self) -> None: - """Cancel season-related tasks on cog unload.""" - log.debug("Unloading the cog and canceling the background task.") - self.countdown_task.cancel() - self.status_task.cancel() - - -class AocMember: - """Object representing the Advent of Code user.""" - - def __init__(self, name: str, aoc_id: int, stars: int, starboard: list, local_score: int, global_score: int): - self.name = name - self.aoc_id = aoc_id - self.stars = stars - self.starboard = starboard - self.local_score = local_score - self.global_score = global_score - self.completions = self._completions_from_starboard(self.starboard) - - def __repr__(self): - """Generate a user-friendly representation of the AocMember & their score.""" - return f"<{self.name} ({self.aoc_id}): {self.local_score}>" - - @classmethod - def member_from_json(cls, injson: dict) -> "AocMember": - """ - Generate an AocMember from AoC's private leaderboard API JSON. - - injson is expected to be the dict contained in: - - AoC_APIjson['members'][<member id>:str] - - Returns an AocMember object - """ - return cls( - name=injson["name"] if injson["name"] else "Anonymous User", - aoc_id=int(injson["id"]), - stars=injson["stars"], - starboard=cls._starboard_from_json(injson["completion_day_level"]), - local_score=injson["local_score"], - global_score=injson["global_score"], - ) - - @staticmethod - def _starboard_from_json(injson: dict) -> list: - """ - Generate starboard from AoC's private leaderboard API JSON. - - injson is expected to be the dict contained in: - - AoC_APIjson['members'][<member id>:str]['completion_day_level'] - - Returns a list of 25 lists, where each nested list contains a pair of booleans representing - the code challenge completion status for that day - """ - # Basic input validation - if not isinstance(injson, dict): - raise ValueError - - # Initialize starboard - starboard = [] - for _i in range(25): - starboard.append([False, False]) - - # Iterate over days, which are the keys of injson (as str) - for day in injson: - idx = int(day) - 1 - # If there is a second star, the first star must be completed - if "2" in injson[day].keys(): - starboard[idx] = [True, True] - # If the day exists in injson, then at least the first star is completed - else: - starboard[idx] = [True, False] - - return starboard - - @staticmethod - def _completions_from_starboard(starboard: list) -> tuple: - """Return days completed, as a (1 star, 2 star) tuple, from starboard.""" - completions = [0, 0] - for day in starboard: - if day[0]: - completions[0] += 1 - if day[1]: - completions[1] += 1 - - return tuple(completions) - - -class AocPrivateLeaderboard: - """Object representing the Advent of Code private leaderboard.""" - - def __init__(self, members: list, owner_id: int, event_year: int): - self.members = members - self._owner_id = owner_id - self._event_year = event_year - self.last_updated = datetime.utcnow() - - self.daily_completion_summary = self.calculate_daily_completion() - - def top_n(self, n: int = 10) -> dict: - """ - Return the top n participants on the leaderboard. - - If n is not specified, default to the top 10 - """ - return self.members[:n] - - def calculate_daily_completion(self) -> List[tuple]: - """ - Calculate member completion rates by day. - - Return a list of tuples for each day containing the number of users who completed each part - of the challenge - """ - daily_member_completions = [] - for day in range(25): - one_star_count = 0 - two_star_count = 0 - for member in self.members: - if member.starboard[day][1]: - one_star_count += 1 - two_star_count += 1 - elif member.starboard[day][0]: - one_star_count += 1 - else: - daily_member_completions.append((one_star_count, two_star_count)) - - return(daily_member_completions) - - @staticmethod - async def json_from_url( - leaderboard_id: int = AocConfig.leaderboard_id, year: int = AocConfig.year - ) -> "AocPrivateLeaderboard": - """ - Request the API JSON from Advent of Code for leaderboard_id for the specified year's event. - - If no year is input, year defaults to the current year - """ - api_url = f"https://adventofcode.com/{year}/leaderboard/private/view/{leaderboard_id}.json" - - log.debug("Querying Advent of Code Private Leaderboard API") - async with aiohttp.ClientSession(cookies=AOC_SESSION_COOKIE, headers=AOC_REQUEST_HEADER) as session: - async with session.get(api_url) as resp: - if resp.status == 200: - raw_dict = await resp.json() - else: - log.warning(f"Bad response received from AoC ({resp.status}), check session cookie") - resp.raise_for_status() - - return raw_dict - - @classmethod - def from_json(cls, injson: dict) -> "AocPrivateLeaderboard": - """Generate an AocPrivateLeaderboard object from AoC's private leaderboard API JSON.""" - return cls( - members=cls._sorted_members(injson["members"]), owner_id=injson["owner_id"], event_year=injson["event"] - ) - - @classmethod - async def from_url(cls) -> "AocPrivateLeaderboard": - """Helper wrapping of AocPrivateLeaderboard.json_from_url and AocPrivateLeaderboard.from_json.""" - api_json = await cls.json_from_url() - return cls.from_json(api_json) - - @staticmethod - def _sorted_members(injson: dict) -> list: - """ - Generate a sorted list of AocMember objects from AoC's private leaderboard API JSON. - - Output list is sorted based on the AocMember.local_score - """ - members = [AocMember.member_from_json(injson[member]) for member in injson] - members.sort(key=lambda x: x.local_score, reverse=True) - - return members - - @staticmethod - def build_leaderboard_embed(members_to_print: List[AocMember]) -> str: - """ - Build a text table from members_to_print, a list of AocMember objects. - - Returns a string to be used as the content of the bot's leaderboard response - """ - stargroup = f"{Emojis.star}, {Emojis.star*2}" - header = f"{' '*3}{'Score'} {'Name':^25} {stargroup:^7}\n{'-'*44}" - table = "" - for i, member in enumerate(members_to_print): - if member.name == "Anonymous User": - name = f"{member.name} #{member.aoc_id}" - else: - name = member.name - - table += ( - f"{i+1:2}) {member.local_score:4} {name:25.25} " - f"({member.completions[0]:2}, {member.completions[1]:2})\n" - ) - else: - table = f"```{header}\n{table}```" - - return table - - -class AocGlobalLeaderboard: - """Object representing the Advent of Code global leaderboard.""" - - def __init__(self, members: List[tuple]): - self.members = members - self.last_updated = datetime.utcnow() - - def top_n(self, n: int = 10) -> dict: - """ - Return the top n participants on the leaderboard. - - If n is not specified, default to the top 10 - """ - return self.members[:n] - - @classmethod - async def from_url(cls) -> "AocGlobalLeaderboard": - """ - Generate an list of tuples for the entries on AoC's global leaderboard. - - Because there is no API for this, web scraping needs to be used - """ - aoc_url = f"https://adventofcode.com/{AocConfig.year}/leaderboard" - - async with aiohttp.ClientSession(headers=AOC_REQUEST_HEADER) as session: - async with session.get(aoc_url) as resp: - if resp.status == 200: - raw_html = await resp.text() - else: - log.warning(f"Bad response received from AoC ({resp.status}), check session cookie") - resp.raise_for_status() - - soup = BeautifulSoup(raw_html, "html.parser") - ele = soup.find_all("div", class_="leaderboard-entry") - - exp = r"(?:[ ]{,2}(\d+)\))?[ ]+(\d+)\s+([\w\(\)\#\@\-\d ]+)" - - lb_list = [] - for entry in ele: - # Strip off the AoC++ decorator - raw_str = entry.text.replace("(AoC++)", "").rstrip() - - # Use a regex to extract the info from the string to unify formatting - # Group 1: Rank - # Group 2: Global Score - # Group 3: Member string - r = re.match(exp, raw_str) - - rank = int(r.group(1)) if r.group(1) else None - global_score = int(r.group(2)) - - member = r.group(3) - if member.lower().startswith("(anonymous"): - # Normalize anonymous user string by stripping () and title casing - member = re.sub(r"[\(\)]", "", member).title() - - lb_list.append((rank, global_score, member)) - - return cls(lb_list) - - @staticmethod - def build_leaderboard_embed(members_to_print: List[tuple]) -> str: - """ - Build a text table from members_to_print, a list of tuples. - - Returns a string to be used as the content of the bot's leaderboard response - """ - header = f"{' '*4}{'Score'} {'Name':^25}\n{'-'*36}" - table = "" - for member in members_to_print: - # In the event of a tie, rank is None - if member[0]: - rank = f"{member[0]:3})" - else: - rank = f"{' ':4}" - table += f"{rank} {member[1]:4} {member[2]:25.25}\n" - else: - table = f"```{header}\n{table}```" - - return table - - -def _error_embed_helper(title: str, description: str) -> discord.Embed: - """Return a red-colored Embed with the given title and description.""" - return discord.Embed(title=title, description=description, colour=discord.Colour.red()) - - -def setup(bot: commands.Bot) -> None: - """Advent of Code Cog load.""" - bot.add_cog(AdventOfCode(bot)) diff --git a/bot/resources/advent_of_code/about.json b/bot/resources/advent_of_code/about.json index 91ae6813..dd0fe59a 100644 --- a/bot/resources/advent_of_code/about.json +++ b/bot/resources/advent_of_code/about.json @@ -6,22 +6,22 @@ }, { "name": "How do I sign up?", - "value": "AoC utilizes the following services' OAuth:", + "value": "Sign up with one of these services:", "inline": true }, { - "name": "Service", + "name": "Auth Services", "value": "GitHub\nGoogle\nTwitter\nReddit", "inline": true }, { "name": "How does scoring work?", - "value": "Getting a star first is worth 100 points, second is 99, and so on down to 1 point at 100th place.\n\nCheck out AoC's [global leaderboard](https://adventofcode.com/leaderboard) to see who's leading this year's event!", + "value": "For the [global leaderboard](https://adventofcode.com/leaderboard), the first person to get a star first gets 100 points, the second person gets 99 points, and so on down to 1 point at 100th place.\n\nFor private leaderboards, the first person to get a star gets N points, where N is the number of people on the leaderboard. The second person to get the star gets N-1 points and so on and so forth.", "inline": false }, { "name": "Join our private leaderboard!", - "value": "In addition to the global leaderboard, AoC also offers private leaderboards, where you can compete against a smaller group of friends!\n\nGet the join code using `.aoc join` and head over to AoC's [private leaderboard page](https://adventofcode.com/leaderboard/private) to join the PyDis private leaderboard!", + "value": "Come join the Python Discord private leaderboard and compete against other people in the community! Get the join code using `.aoc join` and visit the [private leaderboard page](https://adventofcode.com/leaderboard/private) to join our leaderboard.", "inline": false } ] |