From 0712a27181842a0af52f8051b5601a75874feeca Mon Sep 17 00:00:00 2001 From: FirePlank <44502537+FirePlank@users.noreply.github.com> Date: Mon, 6 Nov 2023 23:57:04 +0200 Subject: [PATCH] Migrated advent of code cog to new extension format --- bot/cogs/adventofcode.py | 311 ------------------------ bot/extensions/adventofcode/__init__.py | 9 + bot/extensions/adventofcode/commands.py | 216 ++++++++++++++++ bot/extensions/adventofcode/tasks.py | 53 ++++ cli.py | 1 + 5 files changed, 279 insertions(+), 311 deletions(-) delete mode 100644 bot/cogs/adventofcode.py create mode 100644 bot/extensions/adventofcode/__init__.py create mode 100644 bot/extensions/adventofcode/commands.py create mode 100644 bot/extensions/adventofcode/tasks.py diff --git a/bot/cogs/adventofcode.py b/bot/cogs/adventofcode.py deleted file mode 100644 index e2afb7be..00000000 --- a/bot/cogs/adventofcode.py +++ /dev/null @@ -1,311 +0,0 @@ -import asyncio -import logging -import re -from datetime import datetime, timedelta - -import aiohttp -import discord -import inflect -import pytz -from bs4 import BeautifulSoup -from discord.ext import commands - -from bot.config import settings - -log = logging.getLogger(__name__) -loop = asyncio.get_event_loop() - -previous_year = 0 if datetime.now(tz=pytz.timezone("EST")).strftime("%m") == "12" else 1 -YEAR = int(datetime.now(tz=pytz.timezone("EST")).strftime("%Y")) - previous_year - -API_URL = f"https://adventofcode.com/{YEAR}/leaderboard/private/view/975452.json" -INTERVAL = 120 -AOC_REQUEST_HEADER = {"user-agent": "TWT AoC Event Bot"} -AOC_SESSION_COOKIE = {"session": settings.aoc.session_cookie} -ENGINE = inflect.engine() - - -def time_left_to_aoc_midnight(): - """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(tz=pytz.timezone("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(tz=pytz.timezone("UTC")) - - -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 ( - int(datetime.now(tz=pytz.timezone("EST")).day) in range(1, 25) - and int(datetime.now(tz=pytz.timezone("EST")).month) == 12 - ): - tomorrow, time_left = time_left_to_aoc_midnight() - - # Prevent bot from being slightly too early in trying to announce today's puzzle - await asyncio.sleep(time_left.seconds + 1) - - channel = bot.get_channel(settings.aoc.channel_id) - - if not channel: - break - - aoc_role = channel.guild.get_role(settings.aoc.role_id) - if not aoc_role: - break - - puzzle_url = f"https://adventofcode.com/{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): - async with bot.session.head(puzzle_url, raise_for_status=False) as resp: - if resp.status == 200: - break - await asyncio.sleep(10) - else: - break - - await channel.send( - f"{aoc_role.mention} Good morning! Day {tomorrow.day} is ready to be attempted. " - f"View it online now at {puzzle_url}. Good luck!", - allowed_mentions=discord.AllowedMentions( - everyone=False, - users=False, - roles=[discord.Object(settings.aoc.role_id)], - ), - ) - - -class Member: - def __init__(self, results): - self.global_score = results["global_score"] - self.name = results["name"] - self.stars = results["stars"] - self.last_star_ts = results["last_star_ts"] - self.completion_day_level = results["completion_day_level"] - self.id = results["id"] - self.local_score = results["local_score"] - - -class AdventOfCode(commands.Cog, name="Advent of Code"): - def __init__(self, bot): - self.bot = bot - - self._base_url = f"https://adventofcode.com/{YEAR}" - self.global_leaderboard_url = f"https://adventofcode.com/{YEAR}/leaderboard" - self.private_leaderboard_url = f"{self._base_url}/leaderboard/private/view/975452" - - countdown_coro = day_countdown(self.bot) - self.countdown_task = loop.create_task(countdown_coro) - - @commands.group(name="adventofcode", aliases=("aoc",)) - 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", - ) - async def aoc_subscribe(self, ctx: commands.Context) -> None: - """Assign the role for notifications about new days being ready.""" - if ctx.channel.id != settings.aoc.channel_id: - await ctx.send(f"Please use the <#{settings.aoc.channel_id}> channel") - return - - role = ctx.guild.get_role(settings.aoc.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") - async def aoc_unsubscribe(self, ctx: commands.Context) -> None: - """Remove the role for notifications about new days being ready.""" - if ctx.channel.id != settings.aoc.channel_id: - await ctx.send(f"Please use the <#{settings.aoc.channel_id}> channel") - return - - role = ctx.guild.get_role(settings.aoc.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 Aoc Finishes", - ) - async def aoc_countdown(self, ctx: commands.Context) -> None: - """Return time left until Aoc Finishes.""" - if ctx.channel.id != settings.aoc.channel_id: - await ctx.send(f"Please use the <#{settings.aoc.channel_id}> channel") - return - - if ( - int(datetime.now(tz=pytz.timezone("EST")).day) in range(1, 25) - and int(datetime.now(tz=pytz.timezone("EST")).month) == 12 - ): - days = 24 - int(datetime.now().strftime("%d")) - hours = 23 - int(datetime.now().strftime("%H")) - minutes = 59 - int(datetime.now().strftime("%M")) - - embed = discord.Embed( - title="Advent of Code", - description=f"There are {str(days)} days {str(hours)} hours " - f"and {str(minutes)} minutes left until AOC gets over.", - ) - embed.set_footer(text=ctx.author.display_name, icon_url=ctx.author.display_avatar.url) - await ctx.send(embed=embed) - - else: - await ctx.send("Aoc Hasn't Started Yet!") - - @adventofcode_group.command(name="join", aliases=("j",), brief="Learn how to join the leaderboard (via DM)") - async def join_leaderboard(self, ctx: commands.Context) -> None: - """DM the user the information for joining the TWT AoC private leaderboard.""" - if ctx.channel.id != settings.aoc.channel_id: - await ctx.send(f"Please use the <#{settings.aoc.channel_id}> channel") - return - - author = ctx.message.author - - info_str = ( - "Head over to https://adventofcode.com/leaderboard/private " - "with code `975452-d90a48b0` to join the TWT private leaderboard!" - ) - try: - await author.send(info_str) - except discord.errors.Forbidden: - await ctx.send(f":x: {author.mention}, please (temporarily) enable DMs to receive the join code") - else: - await ctx.message.add_reaction("\U0001F4E8") - - @adventofcode_group.command( - name="leaderboard", - aliases=("board", "lb"), - brief="Get a snapshot of the TWT private AoC leaderboard", - ) - async def aoc_leaderboard(self, ctx: commands.Context): - if ctx.channel.id != settings.aoc.channel_id: - return await ctx.send(f"Please use the <#{settings.aoc.channel_id}> channel") - - api_url = API_URL - - async with aiohttp.ClientSession(cookies=AOC_SESSION_COOKIE, headers=AOC_REQUEST_HEADER) as session: - async with session.get(api_url) as resp: - if resp.status == 200: - leaderboard = await resp.json() - else: - resp.raise_for_status() - - members = [Member(leaderboard["members"][id]) for id in leaderboard["members"]] - - embed = discord.Embed( - title=f"{ctx.guild.name} Advent of Code Leaderboard", - colour=discord.Colour(0x68C290), - url=f"https://adventofcode.com/{YEAR}/leaderboard/private/view/975452", - ) - - leaderboard = { - "owner_id": leaderboard["owner_id"], - "event": leaderboard["event"], - "members": members, - } - members = leaderboard["members"] - - for i, member in enumerate(sorted(members, key=lambda x: x.local_score, reverse=True)[:10], 1): - embed.add_field( - name=f"{ENGINE.ordinal(i)} Place: {member.name} ({member.stars} :star:)", - value=f"Local Score: {member.local_score} | Global Score: {member.global_score}", - inline=False, - ) - - tomorrow, _ = time_left_to_aoc_midnight() - embed.set_footer(text=f"Current Day: {tomorrow.day - 1}/25") - - await ctx.send(embed=embed) - - @adventofcode_group.command( - name="global", - aliases=("globalboard", "gb"), - brief="Get a snapshot of the global AoC leaderboard", - ) - async def global_leaderboard(self, ctx: commands.Context, number_of_people_to_display: int = 10): - if ctx.channel.id != settings.aoc.channel_id: - await ctx.send(f"Please use the <#{settings.aoc.channel_id}>") - return - - aoc_url = f"https://adventofcode.com/{YEAR}/leaderboard" - number_of_people_to_display = min(25, number_of_people_to_display) - - 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: - 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() - - # 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)) - - s_desc = "\n".join( - f"`{index}` {lb_list[index-1][2]} - {lb_list[index-1][1]} " - for index, title in enumerate(lb_list[:number_of_people_to_display], start=1) - ) - - embed = discord.Embed( - title="Advent of Code Global Leaderboard", - colour=discord.Colour(0x68C290), - url="https://adventofcode.com", - description=s_desc, - ) - await ctx.send(embed=embed) - - -async def setup(bot): - await bot.add_cog(AdventOfCode(bot)) diff --git a/bot/extensions/adventofcode/__init__.py b/bot/extensions/adventofcode/__init__.py new file mode 100644 index 00000000..df59c28a --- /dev/null +++ b/bot/extensions/adventofcode/__init__.py @@ -0,0 +1,9 @@ +from bot.core import DiscordBot + +from .commands import AdventOfCode +from .tasks import AdventOfCodeTasks + + +async def setup(bot: DiscordBot) -> None: + await bot.add_cog(AdventOfCode(bot=bot)) + await bot.add_cog(AdventOfCodeTasks(bot=bot)) diff --git a/bot/extensions/adventofcode/commands.py b/bot/extensions/adventofcode/commands.py new file mode 100644 index 00000000..932bb117 --- /dev/null +++ b/bot/extensions/adventofcode/commands.py @@ -0,0 +1,216 @@ +import logging +import re +from datetime import datetime + +import aiohttp +import discord +import pytz +from bs4 import BeautifulSoup +from discord import app_commands +from discord.ext import commands + +from bot import core +from bot.config import settings + +log = logging.getLogger(__name__) + +YEAR = datetime.now(tz=pytz.timezone("EST")).year +API_URL = f"https://adventofcode.com/{YEAR}/leaderboard/private/view/975452.json" +INTERVAL = 120 +AOC_REQUEST_HEADER = {"user-agent": "TWT AoC Event Bot"} +AOC_SESSION_COOKIE = {"session": settings.aoc.session_cookie} + + +def ordinal(n: int): + if 11 <= (n % 100) <= 13: + suffix = "th" + else: + suffix = ["th", "st", "nd", "rd", "th"][min(n % 10, 4)] + return str(n) + suffix + + +class Member: + def __init__(self, results): + self.global_score = results["global_score"] + self.name = results["name"] + self.stars = results["stars"] + self.last_star_ts = results["last_star_ts"] + self.completion_day_level = results["completion_day_level"] + self.id = results["id"] + self.local_score = results["local_score"] + + +class AdventOfCode(commands.GroupCog, group_name="aoc"): + def __init__(self, bot): + self.bot = bot + + @property + def role(self) -> discord.Role: + return self.bot.get_role(settings.aoc.role_id) + + @app_commands.command() + async def subscribe(self, interaction: core.InteractionType) -> None: + """Subscribe to receive notifications for new puzzles""" + + if self.role not in interaction.user.roles: + await interaction.user.add_roles(self.role) + await interaction.response.send_message( + "Okay! You have been __subscribed__ to notifications about new Advent of Code tasks." + "You can run `/aoc unsubscribe` to disable them again for you.", + ephemeral=True, + ) + else: + await interaction.response.send_message( + "Hey, you already are receiving notifications about new Advent of Code tasks." + "If you don't want them any more, run `/aoc unsubscribe` instead.", + ephemeral=True, + ) + + @app_commands.command() + async def unsubscribe(self, interaction: core.InteractionType) -> None: + """Unsubscribe from receiving notifications for new puzzles""" + + if self.role in interaction.user.roles: + await interaction.user.remove_roles(self.role) + await interaction.response.send_message( + "Okay! You have been __unsubscribed__ from notifications about new Advent of Code tasks.", + ephemeral=True, + ) + else: + await interaction.response.send_message( + "Hey, you don't even get any notifications about new Advent of Code tasks currently anyway.", + ephemeral=True, + ) + + @app_commands.command() + async def countdown(self, interaction: core.InteractionType) -> None: + """Get the time left until the next puzzle is released""" + + if ( + int(datetime.now(tz=pytz.timezone("EST")).day) in range(1, 25) + and int(datetime.now(tz=pytz.timezone("EST")).month) == 12 + ): + days = 24 - int(datetime.now().strftime("%d")) + hours = 23 - int(datetime.now().strftime("%H")) + minutes = 59 - int(datetime.now().strftime("%M")) + + embed = discord.Embed( + title="Advent of Code", + description=f"There are {str(days)} days {str(hours)} hours " + f"and {str(minutes)} minutes left until AOC gets over.", + ) + await interaction.response.send_message(embed=embed, ephemeral=True) + + else: + await interaction.response.send_message("Advent of Code is not currently running.", ephemeral=True) + + @app_commands.command(name="join") + async def join_leaderboard(self, interaction: core.InteractionType) -> None: + """Learn how to join the leaderboard""" + + await interaction.response.send_message( + "Head over to https://adventofcode.com/leaderboard/private" + "with the code `975452-d90a48b0` to join the TWT private leaderboard!", + ephemeral=True, + ) + + @app_commands.command() + async def leaderboard(self, interaction: core.InteractionType) -> None: + """Get a snapshot of the TWT private AoC leaderboard""" + + if interaction.channel_id != settings.aoc.channel_id: + return await interaction.response.send_message( + f"Please use the <#{settings.aoc.channel_id}> channel", ephemeral=True + ) + + async with aiohttp.ClientSession(cookies=AOC_SESSION_COOKIE, headers=AOC_REQUEST_HEADER) as session: + async with session.get(API_URL) as resp: + if resp.status == 200: + leaderboard = await resp.json() + else: + resp.raise_for_status() + + members = [Member(leaderboard["members"][id]) for id in leaderboard["members"]] + + embed = discord.Embed( + title=f"{interaction.guild.name} Advent of Code Leaderboard", + colour=discord.Colour(0x68C290), + url=f"https://adventofcode.com/{YEAR}/leaderboard/private/view/975452", + ) + + leaderboard = { + "owner_id": leaderboard["owner_id"], + "event": leaderboard["event"], + "members": members, + } + members = leaderboard["members"] + + for i, member in enumerate(sorted(members, key=lambda x: x.local_score, reverse=True)[:10], 1): + embed.add_field( + name=f"{ordinal(i)} Place: {member.name} ({member.stars} ⭐)", + value=f"Local Score: {member.local_score} | Global Score: {member.global_score}", + inline=False, + ) + embed.set_footer(text=f"Current Day: {datetime.now(tz=pytz.timezone('EST')).day}/25") + + await interaction.response.send_message(embed=embed) + + @app_commands.command(name="global") + async def global_leaderboard(self, interaction: core.InteractionType, number_of_people_to_display: int = 10): + """Get a snapshot of the global AoC leaderboard""" + if interaction.channel_id != settings.aoc.channel_id: + return await interaction.response.send_message( + f"Please use the <#{settings.aoc.channel_id}> channel", ephemeral=True + ) + + aoc_url = f"https://adventofcode.com/{YEAR}/leaderboard" + number_of_people_to_display = min(25, number_of_people_to_display) + + 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: + 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() + + # 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)) + + s_desc = "\n".join( + f"`{index}` {lb_list[index-1][2]} - {lb_list[index-1][1]} " + for index, title in enumerate(lb_list[:number_of_people_to_display], start=1) + ) + + embed = discord.Embed( + title="Advent of Code Global Leaderboard", + colour=discord.Colour(0x68C290), + url="https://adventofcode.com", + description=s_desc, + ) + await interaction.response.send_message(embed=embed) + + +async def setup(bot): + await bot.add_cog(AdventOfCode(bot)) diff --git a/bot/extensions/adventofcode/tasks.py b/bot/extensions/adventofcode/tasks.py new file mode 100644 index 00000000..ab0e394b --- /dev/null +++ b/bot/extensions/adventofcode/tasks.py @@ -0,0 +1,53 @@ +import asyncio +import datetime as dt +from datetime import datetime + +import aiohttp +import discord +import pytz +from discord.ext import commands, tasks + +from bot import core +from bot.config import settings + +aoc_time = dt.time(hour=0, minute=0, second=1, tzinfo=pytz.timezone("EST")) +YEAR = datetime.now(tz=pytz.timezone("EST")).year + + +class AdventOfCodeTasks(commands.Cog): + """Tasks for the Advent of Code cog.""" + + def __init__(self, bot: core.DiscordBot): + self.bot = bot + self.daily_puzzle.start() + + def cog_unload(self) -> None: + self.daily_puzzle.cancel() + + @property + def channel(self) -> discord.TextChannel: + return self.bot.get_channel(settings.aoc.channel_id) + + @tasks.loop(time=aoc_time) + async def daily_puzzle(self) -> None: + """Post the daily Advent of Code puzzle""" + + day = datetime.now(tz=pytz.timezone("EST")).day + month = datetime.now(tz=pytz.timezone("EST")).month + if day > 25 or month != 12: + return + + puzzle_url = f"https://adventofcode.com/{YEAR}/day/{day}" + # Check if the puzzle is already available + for retry in range(4): + async with aiohttp.ClientSession(raise_for_status=False) as session: + async with session.get(puzzle_url) as resp: + if resp.status == 200: + break + await asyncio.sleep(10) + + await self.channel.send( + f"<@&{settings.aoc.role_id}> Good morning! Day {day} is ready to be attempted." + f"View it online now at {puzzle_url}. Good luck!", + allowed_mentions=discord.AllowedMentions(roles=True), + ) diff --git a/cli.py b/cli.py index 7908687f..0677de46 100644 --- a/cli.py +++ b/cli.py @@ -138,6 +138,7 @@ async def main(ctx): "bot.extensions.levelling", "bot.extensions.persistent_roles", "bot.extensions.polls", + "bot.extensions.adventofcode", "bot.cogs._help", "bot.cogs.clashofcode", "bot.cogs.roles",