From e956170ab1be2ff2af803ce6ea46c6bcb2f91837 Mon Sep 17 00:00:00 2001 From: ks129 <45097959+ks129@users.noreply.github.com> Date: Tue, 24 Nov 2020 17:39:42 +0200 Subject: Add staff AoC channel to whitelist --- bot/exts/christmas/adventofcode.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'bot/exts/christmas/adventofcode.py') diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py index b3fe0623..fdd84d5d 100644 --- a/bot/exts/christmas/adventofcode.py +++ b/bot/exts/christmas/adventofcode.py @@ -25,7 +25,7 @@ AOC_SESSION_COOKIE = {"session": Tokens.aoc_session_cookie} EST = timezone("EST") COUNTDOWN_STEP = 60 * 5 -AOC_WHITELIST = WHITELISTED_CHANNELS + (Channels.advent_of_code,) +AOC_WHITELIST = WHITELISTED_CHANNELS + (Channels.advent_of_code, Channels.advent_of_code_staff) def is_in_advent() -> bool: -- cgit v1.2.3 From 85ddb6d8c06722f09842339a43e91d0a88fd9946 Mon Sep 17 00:00:00 2001 From: ks129 <45097959+ks129@users.noreply.github.com> Date: Wed, 25 Nov 2020 18:21:30 +0200 Subject: Implement different invitation codes for staff and public leaderboards --- bot/exts/christmas/adventofcode.py | 77 ++++++++++++++++++++++++++++++++------ 1 file changed, 66 insertions(+), 11 deletions(-) (limited to 'bot/exts/christmas/adventofcode.py') diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py index fdd84d5d..be1c733a 100644 --- a/bot/exts/christmas/adventofcode.py +++ b/bot/exts/christmas/adventofcode.py @@ -9,18 +9,19 @@ 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 +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"} -AOC_SESSION_COOKIE = {"session": Tokens.aoc_session_cookie} EST = timezone("EST") COUNTDOWN_STEP = 60 * 5 @@ -131,12 +132,19 @@ async def day_countdown(bot: commands.Bot) -> None: class AdventOfCode(commands.Cog): """Advent of Code festivities! Ho Ho Ho!""" - def __init__(self, bot: commands.Bot): + # 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_join_codes = 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.private_leaderboard_url = f"{self._base_url}/leaderboard/private/view/{AocConfig.leaderboard_id}" + self.private_leaderboard_url = f"{self._base_url}/leaderboard/private/view/{AocConfig.leaderboard_staff_id}" self.about_aoc_filepath = Path("./bot/resources/advent_of_code/about.json") self.cached_about_aoc = self._build_about_embed() @@ -146,6 +154,7 @@ class AdventOfCode(commands.Cog): 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) @@ -153,6 +162,32 @@ class AdventOfCode(commands.Cog): 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 + ) + } + + @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.""" + # Whole December isn't advent + if not is_in_advent(): + return + + # Update every leaderboard for what we have session cookie + for aoc_id, cookie in self.leaderboard_cookies.items(): + leaderboard = await AocPrivateLeaderboard.from_url(aoc_id, cookie) + log.info(leaderboard.members) + # Update only when API return any members + if len(leaderboard.members) > 0: + await self.public_leaderboard_members.set(aoc_id, len(leaderboard.members)) + @in_month(Month.DECEMBER) @commands.group(name="adventofcode", aliases=("aoc",)) @override_in_channel(AOC_WHITELIST) @@ -234,9 +269,29 @@ class AdventOfCode(commands.Cog): 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: + # We want that user get only 1 code + if await self.user_join_codes.contains(ctx.author.id): + join_code = await self.user_join_codes.get(ctx.author.id) + log.info(f"{author.name} ({author.id}) have already cached AoC join code. Returning it.") + else: + 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_join_codes.set(ctx.author.id, join_code) + 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 `{AocConfig.leaderboard_join_code}` to join the PyDis private leaderboard!" + f"with code `{join_code}` to join the PyDis private leaderboard!" ) try: await author.send(info_str) @@ -580,8 +635,8 @@ class AocPrivateLeaderboard: @staticmethod async def json_from_url( - leaderboard_id: int = AocConfig.leaderboard_id, year: int = AocConfig.year - ) -> "AocPrivateLeaderboard": + leaderboard_id: int, cookie: str, year: int = AocConfig.year + ) -> dict: """ Request the API JSON from Advent of Code for leaderboard_id for the specified year's event. @@ -590,7 +645,7 @@ class AocPrivateLeaderboard: 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 aiohttp.ClientSession(headers=AOC_REQUEST_HEADER, cookies={"session": cookie}) as session: async with session.get(api_url) as resp: if resp.status == 200: raw_dict = await resp.json() @@ -608,9 +663,9 @@ class AocPrivateLeaderboard: ) @classmethod - async def from_url(cls) -> "AocPrivateLeaderboard": + async def from_url(cls, leaderboard_id: int, cookie: str) -> "AocPrivateLeaderboard": """Helper wrapping of AocPrivateLeaderboard.json_from_url and AocPrivateLeaderboard.from_json.""" - api_json = await cls.json_from_url() + api_json = await cls.json_from_url(leaderboard_id, cookie) return cls.from_json(api_json) @staticmethod @@ -738,6 +793,6 @@ def _error_embed_helper(title: str, description: str) -> discord.Embed: return discord.Embed(title=title, description=description, colour=discord.Colour.red()) -def setup(bot: commands.Bot) -> None: +def setup(bot: Bot) -> None: """Advent of Code Cog load.""" bot.add_cog(AdventOfCode(bot)) -- cgit v1.2.3 From c3bbbbd114489d12b24a93ab3ab835fa85eb15e6 Mon Sep 17 00:00:00 2001 From: ks129 <45097959+ks129@users.noreply.github.com> Date: Thu, 26 Nov 2020 20:01:59 +0200 Subject: Implement staff and public leaderboards --- bot/exts/christmas/adventofcode.py | 255 ++++++++++++++++++++++++++++--------- 1 file changed, 197 insertions(+), 58 deletions(-) (limited to 'bot/exts/christmas/adventofcode.py') diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py index be1c733a..9b7780ae 100644 --- a/bot/exts/christmas/adventofcode.py +++ b/bot/exts/christmas/adventofcode.py @@ -136,21 +136,24 @@ class AdventOfCode(commands.Cog): 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 ID -> Join code user_join_codes = 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.private_leaderboard_url = f"{self._base_url}/leaderboard/private/view/{AocConfig.leaderboard_staff_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.cached_staff_leaderboard = None self.countdown_task = None self.status_task = None @@ -173,6 +176,11 @@ class AdventOfCode(commands.Cog): ) } + 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.""" @@ -183,11 +191,129 @@ class AdventOfCode(commands.Cog): # Update every leaderboard for what we have session cookie for aoc_id, cookie in self.leaderboard_cookies.items(): leaderboard = await AocPrivateLeaderboard.from_url(aoc_id, cookie) - log.info(leaderboard.members) # Update only when API return any members if len(leaderboard.members) > 0: 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() + ] + + 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 user star earning time + 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 + + # Put completions also in to make building easier 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): + self.staff_last_updated = datetime.utcnow() + self.cached_staff_leaderboard = await AocPrivateLeaderboard.from_url( + AocConfig.leaderboard_staff_id, + Tokens.aoc_staff_session_cookie + ) + + async def get_leaderboard(self, members_amount: int) -> str: + """Generates leaderboard based on Redis data.""" + await self.check_leaderboard() + leaderboard_members = sorted( + [json.loads(data) for user, data in await self.public_user_data.items()], key=lambda k: k["score"] + )[: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) @@ -316,26 +442,32 @@ class AdventOfCode(commands.Cog): 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 - + 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) - # 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) + if staff: + await self.check_staff_leaderboard() + members_to_print = self.cached_staff_leaderboard.top_n(number_of_people_to_display) + table = AocPrivateLeaderboard.build_leaderboard_embed(members_to_print) + else: + table = await self.get_leaderboard(number_of_people_to_display) # Build embed aoc_embed = discord.Embed( - description=f"Total members: {len(self.cached_private_leaderboard.members)}", + 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.cached_private_leaderboard.last_updated + timestamp=self.staff_last_updated if staff else self.last_updated ) - aoc_embed.set_author(name="Advent of Code", url=self.private_leaderboard_url) + 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}" + ) + else: + aoc_embed.set_author(name="Advent of Code") aoc_embed.set_footer(text="Last Updated") await ctx.send( @@ -356,29 +488,54 @@ class AdventOfCode(commands.Cog): 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 + is_staff = ctx.channel.id == Channels.advent_of_code_staff + if is_staff: + await self.check_staff_leaderboard() + else: + await self.check_leaderboard() # Build ASCII table - total_members = len(self.cached_private_leaderboard.members) + if is_staff: + total_members = len(self.cached_staff_leaderboard.members) + else: + total_members = await self.public_user_data.length() + _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}" + 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" + 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.cached_private_leaderboard.last_updated + 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") @@ -402,7 +559,7 @@ class AdventOfCode(commands.Cog): 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) + await self._check_leaderboard_cache(ctx) if not self.cached_global_leaderboard: # Feedback on issues with leaderboard caching are sent by _check_leaderboard_cache() @@ -425,41 +582,31 @@ class AdventOfCode(commands.Cog): embed=aoc_embed, ) - async def _check_leaderboard_cache(self, ctx: commands.Context, global_board: bool = False) -> None: + 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 """ - # 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) + leaderboard = self.cached_global_leaderboard if not leaderboard: - log.debug(f"No cached {_shortstr} leaderboard found") - await self._boardgetter(global_board) + 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 {_shortstr} leaderboard age less than threshold ({age_seconds} seconds old)") + log.debug(f"Cached global 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) + log.debug(f"Cached global leaderboard age greater than threshold ({age_seconds} seconds old)") + self.cached_global_leaderboard = await AocGlobalLeaderboard.from_url() - leaderboard = getattr(self, leaderboard_str) + leaderboard = self.cached_global_leaderboard if not leaderboard: await ctx.send( "", embed=_error_embed_helper( - title=f"Something's gone wrong and there's no cached {_shortstr} leaderboard!", + title="Something's gone wrong and there's no cached global leaderboard!", description="Please check in with a staff member.", ), ) @@ -475,8 +622,7 @@ class AdventOfCode(commands.Cog): ) 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" + f"integer less than or equal to {max_entries}" ) number_of_people_to_display = max_entries @@ -496,13 +642,6 @@ class AdventOfCode(commands.Cog): 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.") -- cgit v1.2.3 From eddb38e9c091e8e0f3cb4529ec501091f71d901b Mon Sep 17 00:00:00 2001 From: ks129 <45097959+ks129@users.noreply.github.com> Date: Thu, 26 Nov 2020 20:05:29 +0200 Subject: Store AoC leaderboard IDs instead join codes for users mapping --- bot/exts/christmas/adventofcode.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) (limited to 'bot/exts/christmas/adventofcode.py') diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py index 9b7780ae..70bdc50a 100644 --- a/bot/exts/christmas/adventofcode.py +++ b/bot/exts/christmas/adventofcode.py @@ -136,8 +136,8 @@ class AdventOfCode(commands.Cog): public_leaderboard_members = RedisCache() # We don't want that users join to multiple leaderboards, so return only 1 code to user. - # User ID -> Join code - user_join_codes = RedisCache() + # 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) @@ -400,8 +400,8 @@ class AdventOfCode(commands.Cog): log.info(f"{author.name} ({author.id}) ran command in staff AoC channel. Returning staff code.") else: # We want that user get only 1 code - if await self.user_join_codes.contains(ctx.author.id): - join_code = await self.user_join_codes.get(ctx.author.id) + 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}) have already cached AoC join code. Returning it.") else: least_id, least = 0, 200 @@ -412,7 +412,7 @@ class AdventOfCode(commands.Cog): join_code = self.leaderboard_join_codes[least_id] # Persist this code to Redis, so we can get it later again. - await self.user_join_codes.set(ctx.author.id, join_code) + 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 = ( @@ -466,6 +466,11 @@ class AdventOfCode(commands.Cog): 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") -- cgit v1.2.3 From 40461802eca56b9d92e627e275b402bdcd3e6824 Mon Sep 17 00:00:00 2001 From: ks129 <45097959+ks129@users.noreply.github.com> Date: Fri, 27 Nov 2020 17:31:05 +0200 Subject: Add comment about choosing leaderboard for user --- bot/exts/christmas/adventofcode.py | 1 + 1 file changed, 1 insertion(+) (limited to 'bot/exts/christmas/adventofcode.py') diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py index 70bdc50a..2b8ac7bf 100644 --- a/bot/exts/christmas/adventofcode.py +++ b/bot/exts/christmas/adventofcode.py @@ -404,6 +404,7 @@ class AdventOfCode(commands.Cog): join_code = self.leaderboard_join_codes[await self.user_leaderboards.get(ctx.author.id)] log.info(f"{author.name} ({author.id}) have already cached AoC join code. Returning it.") else: + # Find leaderboard that have least members inside (based on cache) least_id, least = 0, 200 for aoc_id, amount in await self.public_leaderboard_members.items(): log.info(amount, least) -- cgit v1.2.3 From ab28341bfdcf91d0ebf6ce1e4bc492b748067330 Mon Sep 17 00:00:00 2001 From: ks129 <45097959+ks129@users.noreply.github.com> Date: Fri, 27 Nov 2020 17:32:03 +0200 Subject: Remove unnecessary check for members in leaderboard updater task --- bot/exts/christmas/adventofcode.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) (limited to 'bot/exts/christmas/adventofcode.py') diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py index 2b8ac7bf..0d73eb3f 100644 --- a/bot/exts/christmas/adventofcode.py +++ b/bot/exts/christmas/adventofcode.py @@ -191,9 +191,7 @@ class AdventOfCode(commands.Cog): # Update every leaderboard for what we have session cookie for aoc_id, cookie in self.leaderboard_cookies.items(): leaderboard = await AocPrivateLeaderboard.from_url(aoc_id, cookie) - # Update only when API return any members - if len(leaderboard.members) > 0: - await self.public_leaderboard_members.set(aoc_id, len(leaderboard.members)) + 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.""" -- cgit v1.2.3 From bede07a6983a1d15461cd65914f41df624244dc3 Mon Sep 17 00:00:00 2001 From: ks129 <45097959+ks129@users.noreply.github.com> Date: Fri, 27 Nov 2020 18:00:35 +0200 Subject: Handle leaderboard cache create/update fail --- bot/exts/christmas/adventofcode.py | 64 +++++++++++++++++++++++++++++++------- 1 file changed, 52 insertions(+), 12 deletions(-) (limited to 'bot/exts/christmas/adventofcode.py') diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py index 0d73eb3f..9d7289c2 100644 --- a/bot/exts/christmas/adventofcode.py +++ b/bot/exts/christmas/adventofcode.py @@ -3,6 +3,7 @@ import json import logging import math import re +import typing from datetime import datetime, timedelta from pathlib import Path from typing import List, Tuple @@ -202,6 +203,11 @@ class AdventOfCode(commands.Cog): for aoc_id, cookie in self.leaderboard_cookies.items() ] + # Check does this have any failed requests + 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)] = { @@ -281,15 +287,24 @@ class AdventOfCode(commands.Cog): 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): - self.staff_last_updated = datetime.utcnow() - self.cached_staff_leaderboard = await AocPrivateLeaderboard.from_url( + leaderboard = await AocPrivateLeaderboard.from_url( AocConfig.leaderboard_staff_id, Tokens.aoc_staff_session_cookie ) - async def get_leaderboard(self, members_amount: int) -> str: + 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.""" - await self.check_leaderboard() + # When we don't have users in cache, log warning 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["score"] )[:members_amount] @@ -446,10 +461,22 @@ class AdventOfCode(commands.Cog): if staff: await self.check_staff_leaderboard() - members_to_print = self.cached_staff_leaderboard.top_n(number_of_people_to_display) - table = AocPrivateLeaderboard.build_leaderboard_embed(members_to_print) + 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: - table = await self.get_leaderboard(number_of_people_to_display) + 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( @@ -495,6 +522,9 @@ class AdventOfCode(commands.Cog): 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() @@ -503,6 +533,9 @@ class AdventOfCode(commands.Cog): 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}" @@ -779,7 +812,7 @@ class AocPrivateLeaderboard: @staticmethod async def json_from_url( leaderboard_id: int, cookie: str, year: int = AocConfig.year - ) -> dict: + ) -> typing.Union[dict, bool]: """ Request the API JSON from Advent of Code for leaderboard_id for the specified year's event. @@ -791,10 +824,14 @@ class AocPrivateLeaderboard: 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: - raw_dict = await resp.json() + 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") - resp.raise_for_status() + return False return raw_dict @@ -806,10 +843,13 @@ class AocPrivateLeaderboard: ) @classmethod - async def from_url(cls, leaderboard_id: int, cookie: str) -> "AocPrivateLeaderboard": + 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) - return cls.from_json(api_json) + if api_json is not False: + return cls.from_json(api_json) + else: + return None @staticmethod def _sorted_members(injson: dict) -> list: -- cgit v1.2.3 From 6be2a47e32e5e07ab23d2796b5e551be1ec7b0b6 Mon Sep 17 00:00:00 2001 From: ks129 <45097959+ks129@users.noreply.github.com> Date: Fri, 27 Nov 2020 18:02:09 +0200 Subject: Use default 0 for score and reverse leaderboard members --- bot/exts/christmas/adventofcode.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) (limited to 'bot/exts/christmas/adventofcode.py') diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py index 9d7289c2..feff09aa 100644 --- a/bot/exts/christmas/adventofcode.py +++ b/bot/exts/christmas/adventofcode.py @@ -306,7 +306,9 @@ class AdventOfCode(commands.Cog): return False leaderboard_members = sorted( - [json.loads(data) for user, data in await self.public_user_data.items()], key=lambda k: k["score"] + [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}" -- cgit v1.2.3 From ffa3619a19515c147013669339f51d36f83a5f29 Mon Sep 17 00:00:00 2001 From: ks129 <45097959+ks129@users.noreply.github.com> Date: Fri, 27 Nov 2020 19:14:27 +0200 Subject: Fix adventofcode extension (and constants) grammar Co-authored-by: Joe Banks --- bot/constants.py | 4 ++-- bot/exts/christmas/adventofcode.py | 20 ++++++++++---------- 2 files changed, 12 insertions(+), 12 deletions(-) (limited to 'bot/exts/christmas/adventofcode.py') diff --git a/bot/constants.py b/bot/constants.py index bbfe7c3f..d24551f6 100644 --- a/bot/constants.py +++ b/bot/constants.py @@ -33,7 +33,7 @@ class AdventOfCode: leaderboard_cache_age_threshold_seconds = 3600 leaderboard_public_ids = [645282] leaderboard_staff_id = 957532 - # Public join codes in environment must be in same order than in AdventOfCode.leaderboard_public_ids + # Public join codes in environment must be in the same order as AdventOfCode.leaderboard_public_ids leaderboard_public_join_codes = environ.get("AOC_PUBLIC_JOIN_CODES", "").split(",") leaderboard_staff_join_code = environ.get("AOC_STAFF_JOIN_CODE", "") leaderboard_max_displayed_members = 10 @@ -204,7 +204,7 @@ class Roles(NamedTuple): class Tokens(NamedTuple): giphy = environ.get("GIPHY_TOKEN") - # Public AoC cookies in environment must be in same order than in AdventOfCode.leaderboard_public_ids + # Public AoC cookies in environment must be in the same order as AdventOfCode.leaderboard_public_ids aoc_public_session_cookies = environ.get("AOC_PUBLIC_SESSION_COOKIES", "").split(",") aoc_staff_session_cookie = environ.get("AOC_STAFF_SESSION_COOKIE") omdb = environ.get("OMDB_API_KEY") diff --git a/bot/exts/christmas/adventofcode.py b/bot/exts/christmas/adventofcode.py index feff09aa..b188059e 100644 --- a/bot/exts/christmas/adventofcode.py +++ b/bot/exts/christmas/adventofcode.py @@ -185,11 +185,11 @@ class AdventOfCode(commands.Cog): @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.""" - # Whole December isn't advent + # Check whether we are in the 25 days of advent if not is_in_advent(): return - # Update every leaderboard for what we have session cookie + # 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)) @@ -203,7 +203,7 @@ class AdventOfCode(commands.Cog): for aoc_id, cookie in self.leaderboard_cookies.items() ] - # Check does this have any failed requests + # 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 @@ -243,7 +243,7 @@ class AdventOfCode(commands.Cog): "earned": datetime.fromtimestamp(user_data["days"][day]["star_two_earned"]), }) - # Sort these lists based on user star earning time + # 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] @@ -263,7 +263,7 @@ class AdventOfCode(commands.Cog): leaderboard_users[star_user_two["id"]]["score"] = points points -= 1 - # Put completions also in to make building easier later. + # 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"]]) @@ -271,7 +271,7 @@ class AdventOfCode(commands.Cog): 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 + # 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()] @@ -300,7 +300,7 @@ class AdventOfCode(commands.Cog): 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, log warning and return False. + # 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 @@ -414,12 +414,12 @@ class AdventOfCode(commands.Cog): join_code = AocConfig.leaderboard_staff_join_code log.info(f"{author.name} ({author.id}) ran command in staff AoC channel. Returning staff code.") else: - # We want that user get only 1 code + # 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}) have already cached AoC join code. Returning it.") + log.info(f"{author.name} ({author.id}) has a cached AoC join code, returning it.") else: - # Find leaderboard that have least members inside (based on cache) + # 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) -- cgit v1.2.3 From 0c79e90c8f9637fd6136047fecdec5813d2ccca3 Mon Sep 17 00:00:00 2001 From: Sebastiaan Zeeff Date: Mon, 30 Nov 2020 00:42:55 +0100 Subject: 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. --- bot/exts/christmas/adventofcode.py | 983 ------------------------------------- 1 file changed, 983 deletions(-) delete mode 100644 bot/exts/christmas/adventofcode.py (limited to 'bot/exts/christmas/adventofcode.py') 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'][: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'][: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)) -- cgit v1.2.3