aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorGravatar Sebastiaan Zeeff <[email protected]>2020-11-30 00:42:55 +0100
committerGravatar Sebastiaan Zeeff <[email protected]>2020-11-30 00:53:30 +0100
commit0c79e90c8f9637fd6136047fecdec5813d2ccca3 (patch)
tree05e77d2e549487bf3cce418c845318362a515aad
parentAdd constants parsing for multiple leaderboards (diff)
Remove adventofcode.py in favour of subpackage
As the Advent of Code file was getting massive, I've removed the old single-file based extension as I'm going to replace it with a partially rewritten subpackage-based extension.
-rw-r--r--bot/exts/christmas/adventofcode.py983
1 files changed, 0 insertions, 983 deletions
diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py
deleted file mode 100644
index b188059e..00000000
--- a/bot/exts/christmas/adventofcode.py
+++ /dev/null
@@ -1,983 +0,0 @@
-import asyncio
-import json
-import logging
-import math
-import re
-import typing
-from datetime import datetime, timedelta
-from pathlib import Path
-from typing import List, Tuple
-
-import aiohttp
-import discord
-from async_rediscache import RedisCache
-from bs4 import BeautifulSoup
-from discord.ext import commands
-from pytz import timezone
-
-from bot.bot import Bot
-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, seasonal_task
-
-log = logging.getLogger(__name__)
-
-AOC_REQUEST_HEADER = {"user-agent": "PythonDiscord AoC Event Bot"}
-
-EST = timezone("EST")
-COUNTDOWN_STEP = 60 * 5
-
-AOC_WHITELIST = WHITELISTED_CHANNELS + (Channels.advent_of_code, Channels.advent_of_code_staff)
-
-
-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!"""
-
- # Mapping for AoC PyDis community leaderboard IDs -> cached amount of members in leaderboard.
- public_leaderboard_members = RedisCache()
-
- # We don't want that users join to multiple leaderboards, so return only 1 code to user.
- # User ID -> AoC Leaderboard ID
- user_leaderboards = RedisCache()
-
- # We must keep track when user got (and what) stars, because we have multiple leaderboards.
- # Format: User ID -> AoCCachedMember (pickle)
- public_user_data = RedisCache()
-
- def __init__(self, bot: 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.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_staff_leaderboard = None
-
- self.countdown_task = None
- self.status_task = None
- self.leaderboard_member_update_task = self.bot.loop.create_task(self.leaderboard_members_updater())
-
- 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)
-
- self.leaderboard_join_codes = {
- aoc_id: join_code for aoc_id, join_code in zip(
- AocConfig.leaderboard_public_ids, AocConfig.leaderboard_public_join_codes
- )
- }
- self.leaderboard_cookies = {
- aoc_id: cookie for aoc_id, cookie in zip(
- AocConfig.leaderboard_public_ids, Tokens.aoc_public_session_cookies
- )
- }
-
- self.last_updated = None
- self.staff_last_updated = None
- self.refresh_lock = asyncio.Lock()
- self.staff_refresh_lock = asyncio.Lock()
-
- @seasonal_task(Month.DECEMBER, sleep_time=60 * 30)
- async def leaderboard_members_updater(self) -> None:
- """Updates public leaderboards cached member amounts in every 30 minutes."""
- # Check whether we are in the 25 days of advent
- if not is_in_advent():
- return
-
- # Update every leaderboard with our session cookies
- for aoc_id, cookie in self.leaderboard_cookies.items():
- leaderboard = await AocPrivateLeaderboard.from_url(aoc_id, cookie)
- await self.public_leaderboard_members.set(aoc_id, len(leaderboard.members))
-
- async def refresh_leaderboard(self) -> None:
- """Updates public PyDis leaderboard scores based on dates."""
- self.last_updated = datetime.utcnow()
- leaderboard_users = {}
- leaderboards = [
- await AocPrivateLeaderboard.json_from_url(aoc_id, cookie)
- for aoc_id, cookie in self.leaderboard_cookies.items()
- ]
-
- # Check if any requests failed
- if False in leaderboards:
- log.warning("Unable to get one or more of the public leaderboards. Not updating cache.")
- return
-
- for leaderboard in leaderboards:
- for member_id, data in leaderboard["members"].items():
- leaderboard_users[int(member_id)] = {
- "name": data.get("name", "Anonymous User"),
- "aoc_id": int(member_id),
- "days": {
- day: {
- "star_one": "1" in stars,
- "star_two": "2" in stars,
- "star_one_earned": int(stars["1"]["get_star_ts"]) if "1" in stars else None,
- "star_two_earned": int(stars["2"]["get_star_ts"]) if "2" in stars else None,
- } for day, stars in data.get("completion_day_level", {}).items()
- }
- }
-
- # Iterate over every advent day
- for day in range(1, 26):
- day = str(day)
- star_one_users = []
- star_two_users = []
-
- for user, user_data in leaderboard_users.items():
- if day in user_data["days"]:
- if user_data["days"][day]["star_one"]:
- star_one_users.append({
- "id": user,
- "earned": datetime.fromtimestamp(user_data["days"][day]["star_one_earned"]),
- })
-
- if user_data["days"][day]["star_two"]:
- star_two_users.append({
- "id": user,
- "earned": datetime.fromtimestamp(user_data["days"][day]["star_two_earned"]),
- })
-
- # Sort these lists based on the time a user earnt a star
- star_one_users = sorted(star_one_users, key=lambda k: k["earned"])[:100]
- star_two_users = sorted(star_two_users, key=lambda k: k["earned"])[:100]
-
- points = 100
- for star_user_one in star_one_users:
- if "score" in leaderboard_users[star_user_one["id"]]:
- leaderboard_users[star_user_one["id"]]["score"] += points
- else:
- leaderboard_users[star_user_one["id"]]["score"] = points
- points -= 1
-
- points = 100
- for star_user_two in star_two_users:
- if "score" in leaderboard_users[star_user_two["id"]]:
- leaderboard_users[star_user_two["id"]]["score"] += points
- else:
- leaderboard_users[star_user_two["id"]]["score"] = points
- points -= 1
-
- # Attach star completions for building the response later
- for user, user_data in leaderboard_users.items():
- completions_star_one = sum([1 for day in user_data["days"].values() if day["star_one"]])
- completions_star_two = sum([1 for day in user_data["days"].values() if day["star_two"]])
-
- leaderboard_users[user]["star_one_completions"] = completions_star_one
- leaderboard_users[user]["star_two_completions"] = completions_star_two
-
- # Finally, clear old cache and persist everything to Redis
- await self.public_user_data.clear()
- [await self.public_user_data.set(user, json.dumps(user_data)) for user, user_data in leaderboard_users.items()]
-
- async def check_leaderboard(self) -> None:
- """Checks should be public leaderboard refreshed and refresh when required."""
- async with self.refresh_lock:
- secs = AocConfig.leaderboard_cache_age_threshold_seconds
- if self.last_updated is None or self.last_updated < datetime.utcnow() - timedelta(seconds=secs):
- await self.refresh_leaderboard()
-
- async def check_staff_leaderboard(self) -> None:
- """Checks should be staff leaderboard refreshed and refresh when required."""
- async with self.staff_refresh_lock:
- secs = AocConfig.leaderboard_cache_age_threshold_seconds
- if self.staff_last_updated is None or self.staff_last_updated < datetime.utcnow() - timedelta(seconds=secs):
- leaderboard = await AocPrivateLeaderboard.from_url(
- AocConfig.leaderboard_staff_id,
- Tokens.aoc_staff_session_cookie
- )
-
- if leaderboard is not None:
- self.staff_last_updated = datetime.utcnow()
- self.cached_staff_leaderboard = leaderboard
- else:
- log.warning("Can't update staff leaderboard. Got unexpected response.")
-
- async def get_leaderboard(self, members_amount: int, context: commands.Context) -> typing.Union[str, bool]:
- """Generates leaderboard based on Redis data."""
- # When we don't have users in cache, warn and return False.
- if await self.public_user_data.length() == 0:
- log.warning("Don't have cache for displaying AoC public leaderboard.")
- return False
-
- leaderboard_members = sorted(
- [json.loads(data) for user, data in await self.public_user_data.items()],
- key=lambda k: k.get("score", 0),
- reverse=True
- )[:members_amount]
-
- stargroup = f"{Emojis.star}, {Emojis.star * 2}"
- header = f"{' ' * 3}{'Score'} {'Name':^25} {stargroup:^7}\n{'-' * 44}"
- table = ""
- for i, member in enumerate(leaderboard_members):
- if member["name"] == "Anonymous User":
- name = f"{member['name']} #{member['aoc_id']}"
- else:
- name = member["name"]
-
- table += (
- f"{i + 1:2}) {member['score']:4} {name:25.25} "
- f"({member['star_one_completions']:2}, {member['star_two_completions']:2})\n"
- )
- else:
- table = f"```{header}\n{table}```"
-
- return table
-
- @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")
-
- if ctx.channel.id == Channels.advent_of_code_staff:
- join_code = AocConfig.leaderboard_staff_join_code
- log.info(f"{author.name} ({author.id}) ran command in staff AoC channel. Returning staff code.")
- else:
- # Ensure we use the same leaderboard code for the same user
- if await self.user_leaderboards.contains(ctx.author.id):
- join_code = self.leaderboard_join_codes[await self.user_leaderboards.get(ctx.author.id)]
- log.info(f"{author.name} ({author.id}) has a cached AoC join code, returning it.")
- else:
- # Find the leaderboard that has the least members inside from cache
- least_id, least = 0, 200
- for aoc_id, amount in await self.public_leaderboard_members.items():
- log.info(amount, least)
- if amount < least:
- least, least_id = amount, aoc_id
-
- join_code = self.leaderboard_join_codes[least_id]
- # Persist this code to Redis, so we can get it later again.
- await self.user_leaderboards.set(ctx.author.id, least_id)
- log.info(f"{author.name} ({author.id}) got new join code. Persisted it to cache.")
-
- info_str = (
- "Head over to https://adventofcode.com/leaderboard/private "
- f"with code `{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():
- staff = ctx.channel.id == Channels.advent_of_code_staff
- number_of_people_to_display = await self._check_n_entries(ctx, number_of_people_to_display)
-
- if staff:
- await self.check_staff_leaderboard()
- if self.cached_staff_leaderboard is not None:
- members_to_print = self.cached_staff_leaderboard.top_n(number_of_people_to_display)
- table = AocPrivateLeaderboard.build_leaderboard_embed(members_to_print)
- else:
- log.warning("Missing AoC staff leaderboard cache.")
- table = False
- else:
- await self.check_leaderboard()
- table = await self.get_leaderboard(number_of_people_to_display, ctx)
-
- # When we don't have cache, show it to user.
- if table is False:
- await ctx.send(
- ":x: Sorry, we can't get our leaderboard cache. Please let staff know about it."
- )
- return
-
- # Build embed
- aoc_embed = discord.Embed(
- description=(
- "Total members: "
- f"{len(self.cached_staff_leaderboard.members) if staff else await self.public_user_data.length()}"
- ),
- colour=Colours.soft_green,
- timestamp=self.staff_last_updated if staff else self.last_updated
- )
- if ctx.channel.id == Channels.advent_of_code_staff:
- aoc_embed.set_author(
- name="Advent of Code",
- url=f"{self._base_url}/leaderboard/private/view/{AocConfig.leaderboard_staff_id}"
- )
- elif await self.user_leaderboards.contains(ctx.author.id):
- aoc_embed.set_author(
- name="Advent of Code",
- url=f"{self._base_url}/leaderboard/private/view/{await self.user_leaderboards.get(ctx.author.id)}"
- )
- else:
- aoc_embed.set_author(name="Advent of Code")
- 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():
- is_staff = ctx.channel.id == Channels.advent_of_code_staff
- if is_staff:
- await self.check_staff_leaderboard()
- if self.cached_staff_leaderboard is None:
- await ctx.send(":x: Missing leaderboard cache.")
- return
- else:
- await self.check_leaderboard()
-
- # Build ASCII table
- if is_staff:
- total_members = len(self.cached_staff_leaderboard.members)
- else:
- total_members = await self.public_user_data.length()
- if total_members == 0:
- await ctx.send(":x: Missing leaderboard cache. Please notify staff about it.")
- return
-
- _star = Emojis.star
- header = f"{'Day':4}{_star:^8}{_star*2:^4}{'% ' + _star:^8}{'% ' + _star*2:^4}\n{'='*35}"
- table = ""
- if is_staff:
- for day, completions in enumerate(self.cached_staff_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"
- else:
- completions = {}
- # Build data for completion rates
- for _, user_data in await self.public_user_data.items():
- user_data = json.loads(user_data)
- for day, stars in user_data["days"].items():
- day = int(day)
- if day not in completions:
- completions[day] = [0, 0]
-
- if stars["star_one"]:
- completions[day][0] += 1
- if stars["star_two"]:
- completions[day][1] += 1
-
- for day, completion in completions.items():
- per_one_star = f"{(completion[0]/total_members)*100:.2f}"
- per_two_star = f"{(completion[1] / total_members) * 100:.2f}"
-
- table += f"{day:3}){completion[0]:^8}{completion[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.staff_last_updated if is_staff else self.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)
-
- 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) -> 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
- """
- leaderboard = self.cached_global_leaderboard
- if not leaderboard:
- log.debug("No cached global leaderboard found")
- self.cached_global_leaderboard = await AocGlobalLeaderboard.from_url()
- 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 global leaderboard age less than threshold ({age_seconds} seconds old)")
- else:
- log.debug(f"Cached global leaderboard age greater than threshold ({age_seconds} seconds old)")
- self.cached_global_leaderboard = await AocGlobalLeaderboard.from_url()
-
- leaderboard = self.cached_global_leaderboard
- if not leaderboard:
- await ctx.send(
- "",
- embed=_error_embed_helper(
- title="Something's gone wrong and there's no cached global 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}"
- )
- 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
-
- 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, cookie: str, year: int = AocConfig.year
- ) -> typing.Union[dict, bool]:
- """
- 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(headers=AOC_REQUEST_HEADER, cookies={"session": cookie}) as session:
- async with session.get(api_url) as resp:
- if resp.status == 200:
- try:
- raw_dict = await resp.json()
- except aiohttp.ContentTypeError:
- log.warning(f"Got invalid response type from AoC API. Leaderboard ID {leaderboard_id}")
- return False
- else:
- log.warning(f"Bad response received from AoC ({resp.status}), check session cookie")
- return False
-
- 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, leaderboard_id: int, cookie: str) -> typing.Union["AocPrivateLeaderboard", None]:
- """Helper wrapping of AocPrivateLeaderboard.json_from_url and AocPrivateLeaderboard.from_json."""
- api_json = await cls.json_from_url(leaderboard_id, cookie)
- if api_json is not False:
- return cls.from_json(api_json)
- else:
- return None
-
- @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: Bot) -> None:
- """Advent of Code Cog load."""
- bot.add_cog(AdventOfCode(bot))