aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--bot/constants.py67
-rw-r--r--bot/exts/events/advent_of_code/__init__.py10
-rw-r--r--bot/exts/events/advent_of_code/_caches.py5
-rw-r--r--bot/exts/events/advent_of_code/_cog.py488
-rw-r--r--bot/exts/events/advent_of_code/_helpers.py642
-rw-r--r--bot/exts/events/advent_of_code/views/dayandstarview.py82
-rw-r--r--bot/resources/events/advent_of_code/about.json27
-rw-r--r--bot/utils/members.py47
8 files changed, 0 insertions, 1368 deletions
diff --git a/bot/constants.py b/bot/constants.py
index eb9ee4b3..a0b8bd60 100644
--- a/bot/constants.py
+++ b/bot/constants.py
@@ -1,12 +1,9 @@
-import dataclasses
import enum
import logging
-from datetime import datetime
from os import environ
from typing import NamedTuple
__all__ = (
- "AdventOfCode",
"Branding",
"Cats",
"Channels",
@@ -39,65 +36,6 @@ log = logging.getLogger(__name__)
PYTHON_PREFIX = "!"
[email protected]
-class AdventOfCodeLeaderboard:
- id: str
- _session: str
- join_code: str
-
- # If we notice that the session for this board expired, we set
- # this attribute to `True`. We will emit a Sentry error so we
- # can handle it, but, in the meantime, we'll try using the
- # fallback session to make sure the commands still work.
- use_fallback_session: bool = False
-
- @property
- def session(self) -> str:
- """Return either the actual `session` cookie or the fallback cookie."""
- if self.use_fallback_session:
- log.trace(f"Returning fallback cookie for board `{self.id}`.")
- return AdventOfCode.fallback_session
-
- return self._session
-
-
-def _parse_aoc_leaderboard_env() -> dict[str, AdventOfCodeLeaderboard]:
- """
- Parse the environment variable containing leaderboard information.
-
- A leaderboard should be specified in the format `id,session,join_code`,
- without the backticks. If more than one leaderboard needs to be added to
- the constant, separate the individual leaderboards with `::`.
-
- Example ENV: `id1,session1,join_code1::id2,session2,join_code2`
- """
- raw_leaderboards = environ.get("AOC_LEADERBOARDS", "")
- if not raw_leaderboards:
- return {}
-
- leaderboards = {}
- for leaderboard in raw_leaderboards.split("::"):
- leaderboard_id, session, join_code = leaderboard.split(",")
- leaderboards[leaderboard_id] = AdventOfCodeLeaderboard(leaderboard_id, session, join_code)
-
- return leaderboards
-
-
-class AdventOfCode:
- # Information for the several leaderboards we have
- leaderboards = _parse_aoc_leaderboard_env()
- staff_leaderboard_id = environ.get("AOC_STAFF_LEADERBOARD_ID", "")
- fallback_session = environ.get("AOC_FALLBACK_SESSION", "")
-
- # Other Advent of Code constants
- ignored_days = environ.get("AOC_IGNORED_DAYS", "").split(",")
- leaderboard_displayed_members = 10
- leaderboard_cache_expiry_seconds = 1800
- max_day_and_star_results = 15
- year = int(environ.get("AOC_YEAR", datetime.utcnow().year))
- role_id = int(environ.get("AOC_ROLE_ID", 518565788744024082))
-
-
class Branding:
cycle_frequency = int(environ.get("CYCLE_FREQUENCY", 3)) # 0: never, 1: every day, 2: every other day, ...
@@ -107,8 +45,6 @@ class Cats:
class Channels(NamedTuple):
- advent_of_code = int(environ.get("AOC_CHANNEL_ID", 897932085766004786))
- advent_of_code_commands = int(environ.get("AOC_COMMANDS_CHANNEL_ID", 897932607545823342))
algos_and_data_structs = 650401909852864553
bot_commands = 267659945086812160
community_meta = 267659945086812160
@@ -193,8 +129,6 @@ class Colours:
class Emojis:
cross_mark = "\u274C"
- star = "\u2B50"
- christmas_tree = "\U0001F384"
check = "\u2611"
envelope = "\U0001F4E8"
trashcan = environ.get("TRASHCAN_EMOJI", "<:trashcan:637136429717389331>")
@@ -303,7 +237,6 @@ class Roles(NamedTuple):
helpers = int(environ.get("ROLE_HELPERS", 267630620367257601))
core_developers = 587606783669829632
everyone = int(environ.get("BOT_GUILD", 267624335836053506))
- aoc_completionist = int(environ.get("AOC_COMPLETIONIST_ROLE_ID", 916691790181056532))
class Tokens(NamedTuple):
diff --git a/bot/exts/events/advent_of_code/__init__.py b/bot/exts/events/advent_of_code/__init__.py
deleted file mode 100644
index 33c3971a..00000000
--- a/bot/exts/events/advent_of_code/__init__.py
+++ /dev/null
@@ -1,10 +0,0 @@
-from bot.bot import Bot
-
-
-async def setup(bot: Bot) -> None:
- """Set up the Advent of Code extension."""
- # Import the Cog at runtime to prevent side effects like defining
- # RedisCache instances too early.
- from ._cog import AdventOfCode
-
- await bot.add_cog(AdventOfCode(bot))
diff --git a/bot/exts/events/advent_of_code/_caches.py b/bot/exts/events/advent_of_code/_caches.py
deleted file mode 100644
index 32d5394f..00000000
--- a/bot/exts/events/advent_of_code/_caches.py
+++ /dev/null
@@ -1,5 +0,0 @@
-import async_rediscache
-
-leaderboard_counts = async_rediscache.RedisCache(namespace="AOC_leaderboard_counts")
-leaderboard_cache = async_rediscache.RedisCache(namespace="AOC_leaderboard_cache")
-assigned_leaderboard = async_rediscache.RedisCache(namespace="AOC_assigned_leaderboard")
diff --git a/bot/exts/events/advent_of_code/_cog.py b/bot/exts/events/advent_of_code/_cog.py
deleted file mode 100644
index 49140a3f..00000000
--- a/bot/exts/events/advent_of_code/_cog.py
+++ /dev/null
@@ -1,488 +0,0 @@
-import json
-import logging
-from datetime import datetime, timedelta
-from pathlib import Path
-from typing import Optional
-
-import arrow
-import discord
-from async_rediscache import RedisCache
-from discord.ext import commands, tasks
-
-from bot.bot import Bot
-from bot.constants import (
- AdventOfCode as AocConfig, Channels, Client, Colours, Emojis, Month, PYTHON_PREFIX, Roles, WHITELISTED_CHANNELS
-)
-from bot.exts.events.advent_of_code import _helpers
-from bot.exts.events.advent_of_code.views.dayandstarview import AoCDropdownView
-from bot.utils import members
-from bot.utils.decorators import InChannelCheckFailure, in_month, whitelist_override, with_role
-from bot.utils.exceptions import MovedCommandError
-
-log = logging.getLogger(__name__)
-
-AOC_REQUEST_HEADER = {"user-agent": "PythonDiscord AoC Event Bot"}
-
-AOC_WHITELIST_RESTRICTED = WHITELISTED_CHANNELS + (Channels.advent_of_code_commands,)
-
-# Some commands can be run in the regular advent of code channel
-# They aren't spammy and foster discussion
-AOC_WHITELIST = AOC_WHITELIST_RESTRICTED + (Channels.advent_of_code,)
-
-
-class AdventOfCode(commands.Cog):
- """Advent of Code festivities! Ho Ho Ho!"""
-
- # Redis Cache for linking Discord IDs to Advent of Code usernames
- # RedisCache[member_id: aoc_username_string]
- account_links = RedisCache()
-
- # A dict with keys of member_ids to block from getting the role
- # RedisCache[member_id: None]
- completionist_block_list = 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/events/advent_of_code/about.json")
- self.cached_about_aoc = self._build_about_embed()
-
- notification_coro = _helpers.new_puzzle_notification(self.bot)
- self.notification_task = self.bot.loop.create_task(notification_coro)
- self.notification_task.set_name("Daily AoC Notification")
- self.notification_task.add_done_callback(_helpers.background_task_callback)
-
- status_coro = _helpers.countdown_status(self.bot)
- self.status_task = self.bot.loop.create_task(status_coro)
- self.status_task.set_name("AoC Status Countdown")
- self.status_task.add_done_callback(_helpers.background_task_callback)
-
- # Don't start task while event isn't running
- # self.completionist_task.start()
-
- @tasks.loop(minutes=10.0)
- async def completionist_task(self) -> None:
- """
- Give members who have completed all 50 AoC stars the completionist role.
-
- Runs on a schedule, as defined in the task.loop decorator.
- """
- guild = self.bot.get_guild(Client.guild)
- completionist_role = guild.get_role(Roles.aoc_completionist)
- if completionist_role is None:
- log.warning("Could not find the AoC completionist role; cancelling completionist task.")
- self.completionist_task.cancel()
- return
-
- aoc_name_to_member_id = {
- aoc_name: member_id
- for member_id, aoc_name in await self.account_links.items()
- }
-
- try:
- leaderboard = await _helpers.fetch_leaderboard()
- except _helpers.FetchingLeaderboardFailedError:
- await self.bot.log_to_dev_log("Unable to fetch AoC leaderboard during role sync.")
- return
-
- placement_leaderboard = json.loads(leaderboard["placement_leaderboard"])
-
- for member_aoc_info in placement_leaderboard.values():
- if not member_aoc_info["stars"] == 50:
- # Only give the role to people who have completed all 50 stars
- continue
-
- aoc_name = member_aoc_info["name"] or f"Anonymous #{member_aoc_info['id']}"
-
- member_id = aoc_name_to_member_id.get(aoc_name)
- if not member_id:
- log.debug(f"Could not find member_id for {member_aoc_info['name']}, not giving role.")
- continue
-
- member = await members.get_or_fetch_member(guild, member_id)
- if member is None:
- log.debug(f"Could not find {member_id}, not giving role.")
- continue
-
- if completionist_role in member.roles:
- log.debug(f"{member.name} ({member.mention}) already has the completionist role.")
- continue
-
- if not await self.completionist_block_list.contains(member_id):
- log.debug(f"Giving completionist role to {member.name} ({member.mention}).")
- await members.handle_role_change(member, member.add_roles, completionist_role)
-
- @commands.group(name="adventofcode", aliases=("aoc",))
- @whitelist_override(channels=AOC_WHITELIST)
- async def adventofcode_group(self, ctx: commands.Context) -> None:
- """All of the Advent of Code commands."""
- if not ctx.invoked_subcommand:
- await self.bot.invoke_help_command(ctx)
-
- @with_role(Roles.admins)
- @adventofcode_group.command(
- name="block",
- brief="Block a user from getting the completionist role.",
- )
- async def block_from_role(self, ctx: commands.Context, member: discord.Member) -> None:
- """Block the given member from receiving the AoC completionist role, removing it from them if needed."""
- completionist_role = ctx.guild.get_role(Roles.aoc_completionist)
- if completionist_role in member.roles:
- await member.remove_roles(completionist_role)
-
- await self.completionist_block_list.set(member.id, "sentinel")
- await ctx.send(f":+1: Blocked {member.mention} from getting the AoC completionist role.")
-
- @commands.guild_only()
- @adventofcode_group.command(
- name="subscribe",
- aliases=("sub", "notifications", "notify", "notifs", "unsubscribe", "unsub"),
- help=f"NOTE: This command has been moved to {PYTHON_PREFIX}subscribe",
- )
- @whitelist_override(channels=AOC_WHITELIST)
- async def aoc_subscribe(self, ctx: commands.Context) -> None:
- """
- Deprecated role command.
-
- This command has been moved to bot, and will be removed in the future.
- """
- raise MovedCommandError(f"{PYTHON_PREFIX}subscribe")
-
- @adventofcode_group.command(name="countdown", aliases=("count", "c"), brief="Return time left until next day")
- @whitelist_override(channels=AOC_WHITELIST)
- async def aoc_countdown(self, ctx: commands.Context) -> None:
- """Return time left until next day."""
- if _helpers.is_in_advent():
- tomorrow, _ = _helpers.time_left_to_est_midnight()
- next_day_timestamp = int(tomorrow.timestamp())
-
- await ctx.send(f"Day {tomorrow.day} starts <t:{next_day_timestamp}:R>.")
- return
-
- datetime_now = arrow.now(_helpers.EST)
- # Calculate the delta to this & next year's December 1st to see which one is closest and not in the past
- this_year = arrow.get(datetime(datetime_now.year, 12, 1), _helpers.EST)
- next_year = arrow.get(datetime(datetime_now.year + 1, 12, 1), _helpers.EST)
- deltas = (dec_first - datetime_now for dec_first in (this_year, next_year))
- delta = min(delta for delta in deltas if delta >= timedelta()) # timedelta() gives 0 duration delta
-
- next_aoc_timestamp = int((datetime_now + delta).timestamp())
-
- await ctx.send(
- "The Advent of Code event is not currently running. "
- f"The next event will start <t:{next_aoc_timestamp}:R>."
- )
-
- @adventofcode_group.command(name="about", aliases=("ab", "info"), brief="Learn about Advent of Code")
- @whitelist_override(channels=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)
-
- @commands.guild_only()
- @adventofcode_group.command(name="join", aliases=("j",), brief="Learn how to join the leaderboard (via DM)")
- @whitelist_override(channels=AOC_WHITELIST)
- async def join_leaderboard(self, ctx: commands.Context) -> None:
- """DM the user the information for joining the Python Discord leaderboard."""
- current_date = datetime.now()
- allowed_months = (Month.NOVEMBER.value, Month.DECEMBER.value)
- if not (
- current_date.month in allowed_months and current_date.year == AocConfig.year or
- current_date.month == Month.JANUARY.value and current_date.year == AocConfig.year + 1
- ):
- # Only allow joining the leaderboard in the run up to AOC and the January following.
- await ctx.send(f"The Python Discord leaderboard for {current_date.year} is not yet available!")
- return
-
- author = ctx.author
- log.info(f"{author.name} ({author.id}) has requested a PyDis AoC leaderboard code")
-
- if AocConfig.staff_leaderboard_id and any(r.id == Roles.helpers for r in author.roles):
- join_code = AocConfig.leaderboards[AocConfig.staff_leaderboard_id].join_code
- else:
- try:
- join_code = await _helpers.get_public_join_code(author)
- except _helpers.FetchingLeaderboardFailedError:
- await ctx.send(":x: Failed to get join code! Notified maintainers.")
- return
-
- if not join_code:
- log.error(f"Failed to get a join code for user {author} ({author.id})")
- error_embed = discord.Embed(
- title="Unable to get join code",
- description="Failed to get a join code to one of our boards. Please notify staff.",
- colour=discord.Colour.red(),
- )
- await ctx.send(embed=error_embed)
- return
-
- info_str = [
- "To join our leaderboard, follow these steps:",
- "• Log in on https://adventofcode.com",
- "• Head over to https://adventofcode.com/leaderboard/private",
- f"• Use this code `{join_code}` to join the Python Discord leaderboard!",
- ]
- try:
- await author.send("\n".join(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)
-
- @in_month(Month.NOVEMBER, Month.DECEMBER, Month.JANUARY)
- @adventofcode_group.command(
- name="link",
- aliases=("connect",),
- brief="Tie your Discord account with your Advent of Code name."
- )
- @whitelist_override(channels=AOC_WHITELIST)
- async def aoc_link_account(self, ctx: commands.Context, *, aoc_name: str = None) -> None:
- """
- Link your Discord Account to your Advent of Code name.
-
- Stored in a Redis Cache with the format of `Discord ID: Advent of Code Name`
- """
- cache_items = await self.account_links.items()
- cache_aoc_names = [value for _, value in cache_items]
-
- if aoc_name:
- # Let's check the current values in the cache to make sure it isn't already tied to a different account
- if aoc_name == await self.account_links.get(ctx.author.id):
- await ctx.reply(f"{aoc_name} is already tied to your account.")
- return
- elif aoc_name in cache_aoc_names:
- log.info(
- f"{ctx.author} ({ctx.author.id}) tried to connect their account to {aoc_name},"
- " but it's already connected to another user."
- )
- await ctx.reply(
- f"{aoc_name} is already tied to another account."
- " Please contact an admin if you believe this is an error."
- )
- return
-
- # Update an existing link
- if old_aoc_name := await self.account_links.get(ctx.author.id):
- log.info(f"Changing link for {ctx.author} ({ctx.author.id}) from {old_aoc_name} to {aoc_name}.")
- await self.account_links.set(ctx.author.id, aoc_name)
- await ctx.reply(f"Your linked account has been changed to {aoc_name}.")
- else:
- # Create a new link
- log.info(f"Linking {ctx.author} ({ctx.author.id}) to account {aoc_name}.")
- await self.account_links.set(ctx.author.id, aoc_name)
- await ctx.reply(f"You have linked your Discord ID to {aoc_name}.")
- else:
- # User has not supplied a name, let's check if they're in the cache or not
- if cache_name := await self.account_links.get(ctx.author.id):
- await ctx.reply(f"You have already linked an Advent of Code account: {cache_name}.")
- else:
- await ctx.reply(
- "You have not linked an Advent of Code account."
- " Please re-run the command with one specified."
- )
-
- @in_month(Month.NOVEMBER, Month.DECEMBER, Month.JANUARY)
- @adventofcode_group.command(
- name="unlink",
- aliases=("disconnect",),
- brief="Tie your Discord account with your Advent of Code name."
- )
- @whitelist_override(channels=AOC_WHITELIST)
- async def aoc_unlink_account(self, ctx: commands.Context) -> None:
- """
- Unlink your Discord ID with your Advent of Code leaderboard name.
-
- Deletes the entry that was Stored in the Redis cache.
- """
- if aoc_cache_name := await self.account_links.get(ctx.author.id):
- log.info(f"Unlinking {ctx.author} ({ctx.author.id}) from Advent of Code account {aoc_cache_name}")
- await self.account_links.delete(ctx.author.id)
- await ctx.reply(f"We have removed the link between your Discord ID and {aoc_cache_name}.")
- else:
- log.info(f"Attempted to unlink {ctx.author} ({ctx.author.id}), but no link was found.")
- await ctx.reply("You don't have an Advent of Code account linked.")
-
- @in_month(Month.DECEMBER, Month.JANUARY)
- @adventofcode_group.command(
- name="dayandstar",
- aliases=("daynstar", "daystar"),
- brief="Get a view that lets you filter the leaderboard by day and star",
- )
- @whitelist_override(channels=AOC_WHITELIST_RESTRICTED)
- async def aoc_day_and_star_leaderboard(
- self,
- ctx: commands.Context,
- maximum_scorers_day_and_star: Optional[int] = 10
- ) -> None:
- """Have the bot send a View that will let you filter the leaderboard by day and star."""
- if maximum_scorers_day_and_star > AocConfig.max_day_and_star_results or maximum_scorers_day_and_star <= 0:
- raise commands.BadArgument(
- f"The maximum number of results you can query is {AocConfig.max_day_and_star_results}"
- )
- async with ctx.typing():
- try:
- leaderboard = await _helpers.fetch_leaderboard()
- except _helpers.FetchingLeaderboardFailedError:
- await ctx.send(":x: Unable to fetch leaderboard!")
- return
- # This is a dictionary that contains solvers in respect of day, and star.
- # e.g. 1-1 means the solvers of the first star of the first day and their completion time
- per_day_and_star = json.loads(leaderboard['leaderboard_per_day_and_star'])
- view = AoCDropdownView(
- day_and_star_data=per_day_and_star,
- maximum_scorers=maximum_scorers_day_and_star,
- original_author=ctx.author
- )
- message = await ctx.send(
- content="Please select a day and a star to filter by!",
- view=view
- )
- await view.wait()
- await message.edit(view=None)
-
- @in_month(Month.DECEMBER, Month.JANUARY)
- @adventofcode_group.command(
- name="leaderboard",
- aliases=("board", "lb"),
- brief="Get a snapshot of the PyDis private AoC leaderboard",
- )
- @whitelist_override(channels=AOC_WHITELIST_RESTRICTED)
- async def aoc_leaderboard(self, ctx: commands.Context, *, aoc_name: Optional[str] = None) -> None:
- """
- Get the current top scorers of the Python Discord Leaderboard.
-
- Additionally you can specify an `aoc_name` that will append the
- specified profile's personal stats to the top of the leaderboard
- """
- # Strip quotes from the AoC username if needed (e.g. "My Name" -> My Name)
- # This is to keep compatibility with those already used to wrapping the AoC name in quotes
- # Note: only strips one layer of quotes to allow names with quotes at the start and end
- # e.g. ""My Name"" -> "My Name"
- if aoc_name and aoc_name.startswith('"') and aoc_name.endswith('"'):
- aoc_name = aoc_name[1:-1]
-
- # Check if an advent of code account is linked in the Redis Cache if aoc_name is not given
- if (aoc_cache_name := await self.account_links.get(ctx.author.id)) and aoc_name is None:
- aoc_name = aoc_cache_name
-
- async with ctx.typing():
- try:
- leaderboard = await _helpers.fetch_leaderboard(self_placement_name=aoc_name)
- except _helpers.FetchingLeaderboardFailedError:
- await ctx.send(":x: Unable to fetch leaderboard!")
- return
-
- number_of_participants = leaderboard["number_of_participants"]
-
- top_count = min(AocConfig.leaderboard_displayed_members, number_of_participants)
- self_placement_header = " (and your personal stats compared to the top 10)" if aoc_name else ""
- header = f"Here's our current top {top_count}{self_placement_header}! {Emojis.christmas_tree * 3}"
- table = "```\n" \
- f"{leaderboard['placement_leaderboard'] if aoc_name else leaderboard['top_leaderboard']}" \
- "\n```"
- info_embed = _helpers.get_summary_embed(leaderboard)
-
- await ctx.send(content=f"{header}\n\n{table}", embed=info_embed)
- return
-
- @in_month(Month.DECEMBER, Month.JANUARY)
- @adventofcode_group.command(
- name="global",
- aliases=("globalboard", "gb"),
- brief="Get a link to the global leaderboard",
- )
- @whitelist_override(channels=AOC_WHITELIST_RESTRICTED)
- async def aoc_global_leaderboard(self, ctx: commands.Context) -> None:
- """Get a link to the global Advent of Code leaderboard."""
- url = self.global_leaderboard_url
- global_leaderboard = discord.Embed(
- title="Advent of Code — Global Leaderboard",
- description=f"You can find the global leaderboard [here]({url})."
- )
- global_leaderboard.set_thumbnail(url=_helpers.AOC_EMBED_THUMBNAIL)
- await ctx.send(embed=global_leaderboard)
-
- @adventofcode_group.command(
- name="stats",
- aliases=("dailystats", "ds"),
- brief="Get daily statistics for the Python Discord leaderboard"
- )
- @whitelist_override(channels=AOC_WHITELIST_RESTRICTED)
- async def private_leaderboard_daily_stats(self, ctx: commands.Context) -> None:
- """Send an embed with daily completion statistics for the Python Discord leaderboard."""
- try:
- leaderboard = await _helpers.fetch_leaderboard()
- except _helpers.FetchingLeaderboardFailedError:
- await ctx.send(":x: Can't fetch leaderboard for stats right now!")
- return
-
- # The daily stats are serialized as JSON as they have to be cached in Redis
- daily_stats = json.loads(leaderboard["daily_stats"])
- async with ctx.typing():
- lines = ["Day ⭐ ⭐⭐ | %⭐ %⭐⭐\n================================"]
- for day, stars in daily_stats.items():
- star_one = stars["star_one"]
- star_two = stars["star_two"]
- p_star_one = star_one / leaderboard["number_of_participants"]
- p_star_two = star_two / leaderboard["number_of_participants"]
- lines.append(
- f"{day:>2}) {star_one:>4} {star_two:>4} | {p_star_one:>7.2%} {p_star_two:>7.2%}"
- )
- table = "\n".join(lines)
- info_embed = _helpers.get_summary_embed(leaderboard)
- await ctx.send(f"```\n{table}\n```", embed=info_embed)
-
- @with_role(Roles.admins)
- @adventofcode_group.command(
- name="refresh",
- aliases=("fetch",),
- brief="Force a refresh of the leaderboard cache.",
- )
- async def refresh_leaderboard(self, ctx: commands.Context) -> None:
- """
- Force a refresh of the leaderboard cache.
-
- Note: This should be used sparingly, as we want to prevent sending too
- many requests to the Advent of Code server.
- """
- async with ctx.typing():
- try:
- await _helpers.fetch_leaderboard(invalidate_cache=True)
- except _helpers.FetchingLeaderboardFailedError:
- await ctx.send(":x: Something went wrong while trying to refresh the cache!")
- else:
- await ctx.send("\N{OK Hand Sign} Refreshed leaderboard cache!")
-
- def cog_unload(self) -> None:
- """Cancel season-related tasks on cog unload."""
- log.debug("Unloading the cog and canceling the background task.")
- self.notification_task.cancel()
- self.status_task.cancel()
- self.completionist_task.cancel()
-
- def _build_about_embed(self) -> discord.Embed:
- """Build and return the informational "About AoC" embed from the resources file."""
- embed_fields = json.loads(self.about_aoc_filepath.read_text("utf8"))
-
- about_embed = discord.Embed(
- title=self._base_url,
- colour=Colours.soft_green,
- url=self._base_url,
- timestamp=datetime.utcnow()
- )
- about_embed.set_author(name="Advent of Code", url=self._base_url)
- for field in embed_fields:
- about_embed.add_field(**field)
-
- about_embed.set_footer(text="Last Updated")
- return about_embed
-
- async def cog_command_error(self, ctx: commands.Context, error: Exception) -> None:
- """Custom error handler if an advent of code command was posted in the wrong channel."""
- if isinstance(error, InChannelCheckFailure):
- await ctx.send(f":x: Please use <#{Channels.advent_of_code_commands}> for aoc commands instead.")
- error.handled = True
diff --git a/bot/exts/events/advent_of_code/_helpers.py b/bot/exts/events/advent_of_code/_helpers.py
deleted file mode 100644
index abd80b77..00000000
--- a/bot/exts/events/advent_of_code/_helpers.py
+++ /dev/null
@@ -1,642 +0,0 @@
-import asyncio
-import collections
-import datetime
-import json
-import logging
-import math
-import operator
-from typing import Any, Optional
-
-import aiohttp
-import arrow
-import discord
-from discord.ext import commands
-
-from bot.bot import Bot
-from bot.constants import AdventOfCode, Channels, Colours
-from bot.exts.events.advent_of_code import _caches
-
-log = logging.getLogger(__name__)
-
-PASTE_URL = "https://paste.pythondiscord.com/documents"
-RAW_PASTE_URL_TEMPLATE = "https://paste.pythondiscord.com/raw/{key}"
-
-# Base API URL for Advent of Code Private Leaderboards
-AOC_API_URL = "https://adventofcode.com/{year}/leaderboard/private/view/{leaderboard_id}.json"
-AOC_REQUEST_HEADER = {"user-agent": "PythonDiscord AoC Event Bot"}
-
-# Leaderboard Line Template
-AOC_TABLE_TEMPLATE = "{rank: >4} | {name:25.25} | {score: >5} | {stars}"
-HEADER = AOC_TABLE_TEMPLATE.format(rank="", name="Name", score="Score", stars="⭐, ⭐⭐")
-HEADER = f"{HEADER}\n{'-' * (len(HEADER) + 2)}"
-HEADER_LINES = len(HEADER.splitlines())
-TOP_LEADERBOARD_LINES = HEADER_LINES + AdventOfCode.leaderboard_displayed_members
-
-# Keys that need to be set for a cached leaderboard
-REQUIRED_CACHE_KEYS = (
- "full_leaderboard",
- "top_leaderboard",
- "full_leaderboard_url",
- "leaderboard_fetched_at",
- "number_of_participants",
- "daily_stats",
-)
-
-AOC_EMBED_THUMBNAIL = (
- "https://raw.githubusercontent.com/python-discord"
- "/branding/main/seasonal/christmas/server_icons/festive_256.gif"
-)
-
-# Create an easy constant for the EST timezone
-EST = "America/New_York"
-
-# Step size for the challenge countdown status
-COUNTDOWN_STEP = 60 * 5
-
-# Create namedtuple that combines a participant's name and their completion
-# time for a specific star. We're going to use this later to order the results
-# for each star to compute the rank score.
-StarResult = collections.namedtuple("StarResult", "member_id completion_time")
-
-
-class UnexpectedRedirect(aiohttp.ClientError):
- """Raised when an unexpected redirect was detected."""
-
-
-class UnexpectedResponseStatus(aiohttp.ClientError):
- """Raised when an unexpected redirect was detected."""
-
-
-class FetchingLeaderboardFailedError(Exception):
- """Raised when one or more leaderboards could not be fetched at all."""
-
-
-def _format_leaderboard_line(rank: int, data: dict[str, Any], *, is_author: bool) -> str:
- """
- Build a string representing a line of the leaderboard.
-
- Parameters:
- rank:
- Rank in the leaderboard of this entry.
-
- data:
- Mapping with entry information.
-
- Keyword arguments:
- is_author:
- Whether to address the name displayed in the returned line
- personally.
-
- Returns:
- A formatted line for the leaderboard.
- """
- return AOC_TABLE_TEMPLATE.format(
- rank=rank,
- name=data['name'] if not is_author else f"(You) {data['name']}",
- score=str(data['score']),
- stars=f"({data['star_1']}, {data['star_2']})"
- )
-
-
-def leaderboard_sorting_function(entry: tuple[str, dict]) -> tuple[int, int]:
- """
- Provide a sorting value for our leaderboard.
-
- The leaderboard is sorted primarily on the score someone has received and
- secondary on the number of stars someone has completed.
- """
- result = entry[1]
- return result["score"], result["star_2"] + result["star_1"]
-
-
-def _parse_raw_leaderboard_data(raw_leaderboard_data: dict) -> dict:
- """
- Parse the leaderboard data received from the AoC website.
-
- The data we receive from AoC is structured by member, not by day/star. This
- means that we need to "transpose" the data to a per star structure in order
- to calculate the rank scores each individual should get.
-
- As we need our data both "per participant" as well as "per day", we return
- the parsed and analyzed data in both formats.
- """
- # We need to get an aggregate of completion times for each star of each day,
- # instead of per participant to compute the rank scores. This dictionary will
- # provide such a transposed dataset.
- star_results = collections.defaultdict(list)
-
- # As we're already iterating over the participants, we can record the number of
- # first stars and second stars they've achieved right here and now. This means
- # we won't have to iterate over the participants again later.
- leaderboard = {}
-
- # The data we get from the AoC website is structured by member, not by day/star,
- # which means we need to iterate over the members to transpose the data to a per
- # star view. We need that per star view to compute rank scores per star.
- per_day_star_stats = collections.defaultdict(list)
- for member in raw_leaderboard_data.values():
- name = member["name"] if member["name"] else f"Anonymous #{member['id']}"
- member_id = member["id"]
- leaderboard[member_id] = {"name": name, "score": 0, "star_1": 0, "star_2": 0}
-
- # Iterate over all days for this participant
- for day, stars in member["completion_day_level"].items():
- # Iterate over the complete stars for this day for this participant
- for star, data in stars.items():
- # Record completion of this star for this individual
- leaderboard[member_id][f"star_{star}"] += 1
-
- # Record completion datetime for this participant for this day/star
- completion_time = datetime.datetime.fromtimestamp(int(data["get_star_ts"]))
- star_results[(day, star)].append(
- StarResult(member_id=member_id, completion_time=completion_time)
- )
- per_day_star_stats[f"{day}-{star}"].append(
- {'completion_time': int(data["get_star_ts"]), 'member_name': name}
- )
- for key in per_day_star_stats:
- per_day_star_stats[key] = sorted(per_day_star_stats[key], key=operator.itemgetter('completion_time'))
-
- # Now that we have a transposed dataset that holds the completion time of all
- # participants per star, we can compute the rank-based scores each participant
- # should get for that star.
- max_score = len(leaderboard)
- for (day, _star), results in star_results.items():
- # If this day should not count in the ranking, skip it.
- if day in AdventOfCode.ignored_days:
- continue
-
- sorted_result = sorted(results, key=operator.attrgetter("completion_time"))
- for rank, star_result in enumerate(sorted_result):
- leaderboard[star_result.member_id]["score"] += max_score - rank
-
- # Since dictionaries now retain insertion order, let's use that
- sorted_leaderboard = dict(
- sorted(leaderboard.items(), key=leaderboard_sorting_function, reverse=True)
- )
-
- # Create summary stats for the stars completed for each day of the event.
- daily_stats = {}
- for day in range(1, 26):
- day = str(day)
- star_one = len(star_results.get((day, "1"), []))
- star_two = len(star_results.get((day, "2"), []))
- # By using a dictionary instead of namedtuple here, we can serialize
- # this data to JSON in order to cache it in Redis.
- daily_stats[day] = {"star_one": star_one, "star_two": star_two}
-
- return {"daily_stats": daily_stats, "leaderboard": sorted_leaderboard, 'per_day_and_star': per_day_star_stats}
-
-
-def _format_leaderboard(leaderboard: dict[str, dict], self_placement_name: str = None) -> str:
- """Format the leaderboard using the AOC_TABLE_TEMPLATE."""
- leaderboard_lines = [HEADER]
- self_placement_exists = False
- for rank, data in enumerate(leaderboard.values(), start=1):
- if self_placement_name and data["name"].lower() == self_placement_name.lower():
- leaderboard_lines.insert(
- 1,
- AOC_TABLE_TEMPLATE.format(
- rank=rank,
- name=f"(You) {data['name']}",
- score=str(data["score"]),
- stars=f"({data['star_1']}, {data['star_2']})"
- )
- )
- self_placement_exists = True
- continue
- leaderboard_lines.append(
- AOC_TABLE_TEMPLATE.format(
- rank=rank,
- name=data["name"],
- score=str(data["score"]),
- stars=f"({data['star_1']}, {data['star_2']})"
- )
- )
- if self_placement_name and not self_placement_exists:
- raise commands.BadArgument(
- "Sorry, your profile does not exist in this leaderboard."
- "\n\n"
- "To join our leaderboard, run the command `.aoc join`."
- " If you've joined recently, please wait up to 30 minutes for our leaderboard to refresh."
- )
- return "\n".join(leaderboard_lines)
-
-
-async def _leaderboard_request(url: str, board: str, cookies: dict) -> dict[str, Any]:
- """Make a leaderboard request using the specified session cookie."""
- async with aiohttp.request("GET", url, headers=AOC_REQUEST_HEADER, cookies=cookies) as resp:
- # The Advent of Code website redirects silently with a 200 response if a
- # session cookie has expired, is invalid, or was not provided.
- if str(resp.url) != url:
- log.error(f"Fetching leaderboard `{board}` failed! Check the session cookie.")
- raise UnexpectedRedirect(f"redirected unexpectedly to {resp.url} for board `{board}`")
-
- # Every status other than `200` is unexpected, not only 400+
- if not resp.status == 200:
- log.error(f"Unexpected response `{resp.status}` while fetching leaderboard `{board}`")
- raise UnexpectedResponseStatus(f"status `{resp.status}`")
-
- return await resp.json()
-
-
-async def _fetch_leaderboard_data() -> dict[str, Any]:
- """Fetch data for all leaderboards and return a pooled result."""
- year = AdventOfCode.year
-
- # We'll make our requests one at a time to not flood the AoC website with
- # up to six simultaneous requests. This may take a little longer, but it
- # does avoid putting unnecessary stress on the Advent of Code website.
-
- # Container to store the raw data of each leaderboard
- participants = {}
- for leaderboard in AdventOfCode.leaderboards.values():
- leaderboard_url = AOC_API_URL.format(year=year, leaderboard_id=leaderboard.id)
-
- # Two attempts, one with the original session cookie and one with the fallback session
- for attempt in range(1, 3):
- log.debug(f"Attempting to fetch leaderboard `{leaderboard.id}` ({attempt}/2)")
- cookies = {"session": leaderboard.session}
- try:
- raw_data = await _leaderboard_request(leaderboard_url, leaderboard.id, cookies)
- except UnexpectedRedirect:
- if cookies["session"] == AdventOfCode.fallback_session:
- log.error("It seems like the fallback cookie has expired!")
- raise FetchingLeaderboardFailedError from None
-
- # If we're here, it means that the original session did not
- # work. Let's fall back to the fallback session.
- leaderboard.use_fallback_session = True
- continue
- except aiohttp.ClientError:
- # Don't retry, something unexpected is wrong and it may not be the session.
- raise FetchingLeaderboardFailedError from None
- else:
- # Get the participants and store their current count.
- board_participants = raw_data["members"]
- await _caches.leaderboard_counts.set(leaderboard.id, len(board_participants))
- participants.update(board_participants)
- break
- else:
- log.error(f"reached 'unreachable' state while fetching board `{leaderboard.id}`.")
- raise FetchingLeaderboardFailedError
-
- log.info(f"Fetched leaderboard information for {len(participants)} participants")
- return participants
-
-
-async def _upload_leaderboard(leaderboard: str) -> str:
- """Upload the full leaderboard to our paste service and return the URL."""
- async with aiohttp.request("POST", PASTE_URL, data=leaderboard) as resp:
- try:
- resp_json = await resp.json()
- except Exception:
- log.exception("Failed to upload full leaderboard to paste service")
- return ""
-
- if "key" in resp_json:
- return RAW_PASTE_URL_TEMPLATE.format(key=resp_json["key"])
-
- log.error(f"Unexpected response from paste service while uploading leaderboard {resp_json}")
- return ""
-
-
-def _get_top_leaderboard(full_leaderboard: str) -> str:
- """Get the leaderboard up to the maximum specified entries."""
- return "\n".join(full_leaderboard.splitlines()[:TOP_LEADERBOARD_LINES])
-
-
-@_caches.leaderboard_cache.atomic_transaction
-async def fetch_leaderboard(invalidate_cache: bool = False, self_placement_name: str = None) -> dict:
- """
- Get the current Python Discord combined leaderboard.
-
- The leaderboard is cached and only fetched from the API if the current data
- is older than the lifetime set in the constants. To prevent multiple calls
- to this function fetching new leaderboard information in case of a cache
- miss, this function is locked to one call at a time using a decorator.
- """
- cached_leaderboard = await _caches.leaderboard_cache.to_dict()
- # Check if the cached leaderboard contains everything we expect it to. If it
- # does not, this probably means the cache has not been created yet or has
- # expired in Redis. This check also accounts for a malformed cache.
- if invalidate_cache or any(key not in cached_leaderboard for key in REQUIRED_CACHE_KEYS):
- log.info("No leaderboard cache available, fetching leaderboards...")
- # Fetch the raw data
- raw_leaderboard_data = await _fetch_leaderboard_data()
-
- # Parse it to extract "per star, per day" data and participant scores
- parsed_leaderboard_data = _parse_raw_leaderboard_data(raw_leaderboard_data)
-
- leaderboard = parsed_leaderboard_data["leaderboard"]
- number_of_participants = len(leaderboard)
- formatted_leaderboard = _format_leaderboard(leaderboard)
- full_leaderboard_url = await _upload_leaderboard(formatted_leaderboard)
- leaderboard_fetched_at = datetime.datetime.now(datetime.timezone.utc).isoformat()
-
- cached_leaderboard = {
- "placement_leaderboard": json.dumps(raw_leaderboard_data),
- "full_leaderboard": formatted_leaderboard,
- "top_leaderboard": _get_top_leaderboard(formatted_leaderboard),
- "full_leaderboard_url": full_leaderboard_url,
- "leaderboard_fetched_at": leaderboard_fetched_at,
- "number_of_participants": number_of_participants,
- "daily_stats": json.dumps(parsed_leaderboard_data["daily_stats"]),
- "leaderboard_per_day_and_star": json.dumps(parsed_leaderboard_data["per_day_and_star"])
- }
-
- # Store the new values in Redis
- await _caches.leaderboard_cache.update(cached_leaderboard)
-
- # Set an expiry on the leaderboard RedisCache
- with await _caches.leaderboard_cache._get_pool_connection() as connection:
- await connection.expire(
- _caches.leaderboard_cache.namespace,
- AdventOfCode.leaderboard_cache_expiry_seconds
- )
- if self_placement_name:
- formatted_placement_leaderboard = _parse_raw_leaderboard_data(
- json.loads(cached_leaderboard["placement_leaderboard"])
- )["leaderboard"]
- cached_leaderboard["placement_leaderboard"] = _get_top_leaderboard(
- _format_leaderboard(formatted_placement_leaderboard, self_placement_name=self_placement_name)
- )
- return cached_leaderboard
-
-
-def get_summary_embed(leaderboard: dict) -> discord.Embed:
- """Get an embed with the current summary stats of the leaderboard."""
- leaderboard_url = leaderboard["full_leaderboard_url"]
- refresh_minutes = AdventOfCode.leaderboard_cache_expiry_seconds // 60
- refreshed_unix = int(datetime.datetime.fromisoformat(leaderboard["leaderboard_fetched_at"]).timestamp())
-
- aoc_embed = discord.Embed(colour=Colours.soft_green)
-
- aoc_embed.description = (
- f"The leaderboard is refreshed every {refresh_minutes} minutes.\n"
- f"Last Updated: <t:{refreshed_unix}:t>"
- )
- aoc_embed.add_field(
- name="Number of Participants",
- value=leaderboard["number_of_participants"],
- inline=True,
- )
- if leaderboard_url:
- aoc_embed.add_field(
- name="Full Leaderboard",
- value=f"[Python Discord Leaderboard]({leaderboard_url})",
- inline=True,
- )
- aoc_embed.set_author(name="Advent of Code", url=leaderboard_url)
- aoc_embed.set_thumbnail(url=AOC_EMBED_THUMBNAIL)
-
- return aoc_embed
-
-
-async def get_public_join_code(author: discord.Member) -> Optional[str]:
- """
- Get the join code for one of the non-staff leaderboards.
-
- If a user has previously requested a join code and their assigned board
- hasn't filled up yet, we'll return the same join code to prevent them from
- getting join codes for multiple boards.
- """
- # Make sure to fetch new leaderboard information if the cache is older than
- # 30 minutes. While this still means that there could be a discrepancy
- # between the current leaderboard state and the numbers we have here, this
- # should work fairly well given the buffer of slots that we have.
- await fetch_leaderboard()
- previously_assigned_board = await _caches.assigned_leaderboard.get(author.id)
- current_board_counts = await _caches.leaderboard_counts.to_dict()
-
- # Remove the staff board from the current board counts as it should be ignored.
- current_board_counts.pop(AdventOfCode.staff_leaderboard_id, None)
-
- # If this user has already received a join code, we'll give them the
- # exact same one to prevent them from joining multiple boards and taking
- # up multiple slots.
- if previously_assigned_board:
- # Check if their previously assigned board still has room for them
- if current_board_counts.get(previously_assigned_board, 0) < 200:
- log.info(f"{author} ({author.id}) was already assigned to a board with open slots.")
- return AdventOfCode.leaderboards[previously_assigned_board].join_code
-
- log.info(
- f"User {author} ({author.id}) previously received the join code for "
- f"board `{previously_assigned_board}`, but that board's now full. "
- "Assigning another board to this user."
- )
-
- # If we don't have the current board counts cached, let's force fetching a new cache
- if not current_board_counts:
- log.warning("Leaderboard counts were missing from the cache unexpectedly!")
- await fetch_leaderboard(invalidate_cache=True)
- current_board_counts = await _caches.leaderboard_counts.to_dict()
-
- # Find the board with the current lowest participant count. As we can't
- best_board, _count = min(current_board_counts.items(), key=operator.itemgetter(1))
-
- if current_board_counts.get(best_board, 0) >= 200:
- log.warning(f"User {author} `{author.id}` requested a join code, but all boards are full!")
- return
-
- log.info(f"Assigning user {author} ({author.id}) to board `{best_board}`")
- await _caches.assigned_leaderboard.set(author.id, best_board)
-
- # Return the join code for this board
- return AdventOfCode.leaderboards[best_board].join_code
-
-
-def is_in_advent() -> bool:
- """
- Check if we're currently on an Advent of Code day, excluding 25 December.
-
- This helper function is used to check whether or not a feature that prepares
- something for the next Advent of Code challenge should run. As the puzzle
- published on the 25th is the last puzzle, this check excludes that date.
- """
- return arrow.now(EST).day in range(1, 25) and arrow.now(EST).month == 12
-
-
-def time_left_to_est_midnight() -> tuple[datetime.datetime, datetime.timedelta]:
- """Calculate the amount of time left until midnight EST/UTC-5."""
- # Change all time properties back to 00:00
- todays_midnight = arrow.now(EST).replace(
- microsecond=0,
- second=0,
- minute=0,
- hour=0
- )
-
- # We want tomorrow so add a day on
- tomorrow = todays_midnight + datetime.timedelta(days=1)
-
- # Calculate the timedelta between the current time and midnight
- return tomorrow, tomorrow - arrow.now(EST)
-
-
-async def wait_for_advent_of_code(*, hours_before: int = 1) -> None:
- """
- Wait for the Advent of Code event to start.
-
- This function returns `hours_before` (default: 1) the Advent of Code
- actually starts. This allows functions to schedule and execute code that
- needs to run before the event starts.
-
- If the event has already started, this function returns immediately.
-
- Note: The "next Advent of Code" is determined based on the current value
- of the `AOC_YEAR` environment variable. This allows callers to exit early
- if we're already past the Advent of Code edition the bot is currently
- configured for.
- """
- start = arrow.get(datetime.datetime(AdventOfCode.year, 12, 1), EST)
- target = start - datetime.timedelta(hours=hours_before)
- now = arrow.now(EST)
-
- # If we've already reached or passed to target, we
- # simply return immediately.
- if now >= target:
- return
-
- delta = target - now
- await asyncio.sleep(delta.total_seconds())
-
-
-async def countdown_status(bot: Bot) -> None:
- """
- Add the time until the next challenge is published to the bot's status.
-
- This function sleeps until 2 hours before the event and exists one hour
- after the last challenge has been published. It will not start up again
- automatically for next year's event, as it will wait for the environment
- variable AOC_YEAR to be updated.
-
- This ensures that the task will only start sleeping again once the next
- event approaches and we're making preparations for that event.
- """
- log.debug("Initializing status countdown task.")
- # We wait until 2 hours before the event starts. Then we
- # set our first countdown status.
- await wait_for_advent_of_code(hours_before=2)
-
- # Log that we're going to start with the countdown status.
- log.info("The Advent of Code has started or will start soon, starting countdown status.")
-
- # Calculate when the task needs to stop running. To prevent the task from
- # sleeping for the entire year, it will only wait in the currently
- # configured year. This means that the task will only start hibernating once
- # we start preparing the next event by changing environment variables.
- last_challenge = arrow.get(datetime.datetime(AdventOfCode.year, 12, 25), EST)
- end = last_challenge + datetime.timedelta(hours=1)
-
- while arrow.now(EST) < end:
- _, time_left = time_left_to_est_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"
-
- log.trace(f"Changing presence to {playing!r}")
- # 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
- log.trace(f"The countdown status task will sleep for {delay} seconds.")
- await asyncio.sleep(delay)
-
-
-async def new_puzzle_notification(bot: Bot) -> None:
- """
- Announce the release of a new Advent of Code puzzle.
-
- This background task hibernates until just before the Advent of Code starts
- and will then start announcing puzzles as they are published. After the
- event has finished, this task will terminate.
- """
- # We wake up one hour before the event starts to prepare the announcement
- # of the release of the first puzzle.
- await wait_for_advent_of_code(hours_before=1)
-
- log.info("The Advent of Code has started or will start soon, waking up notification task.")
-
- aoc_channel = bot.get_channel(Channels.advent_of_code)
- aoc_role = aoc_channel.guild.get_role(AdventOfCode.role_id)
-
- if not aoc_channel:
- log.error("Could not find the AoC channel to send notification in")
- return
-
- if not aoc_role:
- log.error("Could not find the AoC role to announce the daily puzzle")
- return
-
- # The last event day is 25 December, so we only have to schedule
- # a reminder if the current day is before 25 December.
- end = arrow.get(datetime.datetime(AdventOfCode.year, 12, 25), EST)
- while arrow.now(EST) < end:
- log.trace("Started puzzle notification loop.")
- tomorrow, time_left = time_left_to_est_midnight()
-
- # Use `total_seconds` to get the time left in fractional seconds This
- # should wake us up very close to the target. As a safe guard, the sleep
- # duration is padded with 0.1 second to make sure we wake up after
- # midnight.
- sleep_seconds = time_left.total_seconds() + 0.1
- log.trace(f"The puzzle notification task will sleep for {sleep_seconds} seconds")
- await asyncio.sleep(sleep_seconds)
-
- puzzle_url = f"https://adventofcode.com/{AdventOfCode.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 aoc_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=[aoc_role],
- )
- )
-
- # Ensure that we don't send duplicate announcements by sleeping to well
- # over midnight. This means we're certain to calculate the time to the
- # next midnight at the top of the loop.
- await asyncio.sleep(120)
-
-
-def background_task_callback(task: asyncio.Task) -> None:
- """Check if the finished background task failed to make sure we log errors."""
- if task.cancelled():
- log.info(f"Background task `{task.get_name()}` was cancelled.")
- elif exception := task.exception():
- log.error(f"Background task `{task.get_name()}` failed:", exc_info=exception)
- else:
- log.info(f"Background task `{task.get_name()}` exited normally.")
diff --git a/bot/exts/events/advent_of_code/views/dayandstarview.py b/bot/exts/events/advent_of_code/views/dayandstarview.py
deleted file mode 100644
index f0ebc803..00000000
--- a/bot/exts/events/advent_of_code/views/dayandstarview.py
+++ /dev/null
@@ -1,82 +0,0 @@
-from datetime import datetime
-
-import discord
-
-AOC_DAY_AND_STAR_TEMPLATE = "{rank: >4} | {name:25.25} | {completion_time: >10}"
-
-
-class AoCDropdownView(discord.ui.View):
- """Interactive view to filter AoC stats by Day and Star."""
-
- def __init__(self, original_author: discord.Member, day_and_star_data: dict[str: dict], maximum_scorers: int):
- super().__init__()
- self.day = 0
- self.star = 0
- self.data = day_and_star_data
- self.maximum_scorers = maximum_scorers
- self.original_author = original_author
-
- def generate_output(self) -> str:
- """
- Generates a formatted codeblock with AoC statistics based on the currently selected day and star.
-
- Optionally, when the requested day and star data does not exist yet it returns an error message.
- """
- header = AOC_DAY_AND_STAR_TEMPLATE.format(
- rank="Rank",
- name="Name", completion_time="Completion time (UTC)"
- )
- lines = [f"{header}\n{'-' * (len(header) + 2)}"]
- if not (day_and_star_data := self.data.get(f"{self.day}-{self.star}")):
- return ":x: The requested data for the specified day and star does not exist yet."
- for rank, scorer in enumerate(day_and_star_data[:self.maximum_scorers]):
- time_data = datetime.fromtimestamp(scorer['completion_time']).strftime("%I:%M:%S %p")
- lines.append(AOC_DAY_AND_STAR_TEMPLATE.format(
- datastamp="",
- rank=rank + 1,
- name=scorer['member_name'],
- completion_time=time_data)
- )
- joined_lines = "\n".join(lines)
- return f"Statistics for Day: {self.day}, Star: {self.star}.\n ```\n{joined_lines}\n```"
-
- async def interaction_check(self, interaction: discord.Interaction) -> bool:
- """Global check to ensure that the interacting user is the user who invoked the command originally."""
- if interaction.user != self.original_author:
- await interaction.response.send_message(
- ":x: You can't interact with someone else's response. Please run the command yourself!",
- ephemeral=True
- )
- return False
- return True
-
- @discord.ui.select(
- placeholder="Day",
- options=[discord.SelectOption(label=str(i)) for i in range(1, 26)],
- custom_id="day_select"
- )
- async def day_select(self, _: discord.Interaction, select: discord.ui.Select) -> None:
- """Dropdown to choose a Day of the AoC."""
- self.day = select.values[0]
-
- @discord.ui.select(
- placeholder="Star",
- options=[discord.SelectOption(label=str(i)) for i in range(1, 3)],
- custom_id="star_select"
- )
- async def star_select(self, _: discord.Interaction, select: discord.ui.Select) -> None:
- """Dropdown to choose either the first or the second star."""
- self.star = select.values[0]
-
- @discord.ui.button(label="Fetch", style=discord.ButtonStyle.blurple)
- async def fetch(self, interaction: discord.Interaction, _: discord.ui.Button) -> None:
- """Button that fetches the statistics based on the dropdown values."""
- if self.day == 0 or self.star == 0:
- await interaction.response.send_message(
- "You have to select a value from both of the dropdowns!",
- ephemeral=True
- )
- else:
- await interaction.response.edit_message(content=self.generate_output())
- self.day = 0
- self.star = 0
diff --git a/bot/resources/events/advent_of_code/about.json b/bot/resources/events/advent_of_code/about.json
deleted file mode 100644
index dd0fe59a..00000000
--- a/bot/resources/events/advent_of_code/about.json
+++ /dev/null
@@ -1,27 +0,0 @@
-[
- {
- "name": "What is Advent of Code?",
- "value": "Advent of Code (AoC) is a series of small programming puzzles for a variety of skill levels, run every year during the month of December.\n\nThey are self-contained and are just as appropriate for an expert who wants to stay sharp as they are for a beginner who is just learning to code. Each puzzle calls upon different skills and has two parts that build on a theme.",
- "inline": false
- },
- {
- "name": "How do I sign up?",
- "value": "Sign up with one of these services:",
- "inline": true
- },
- {
- "name": "Auth Services",
- "value": "GitHub\nGoogle\nTwitter\nReddit",
- "inline": true
- },
- {
- "name": "How does scoring work?",
- "value": "For the [global leaderboard](https://adventofcode.com/leaderboard), the first person to get a star first gets 100 points, the second person gets 99 points, and so on down to 1 point at 100th place.\n\nFor private leaderboards, the first person to get a star gets N points, where N is the number of people on the leaderboard. The second person to get the star gets N-1 points and so on and so forth.",
- "inline": false
- },
- {
- "name": "Join our private leaderboard!",
- "value": "Come join the Python Discord private leaderboard and compete against other people in the community! Get the join code using `.aoc join` and visit the [private leaderboard page](https://adventofcode.com/leaderboard/private) to join our leaderboard.",
- "inline": false
- }
-]
diff --git a/bot/utils/members.py b/bot/utils/members.py
deleted file mode 100644
index de5850ca..00000000
--- a/bot/utils/members.py
+++ /dev/null
@@ -1,47 +0,0 @@
-import logging
-import typing as t
-
-import discord
-
-log = logging.getLogger(__name__)
-
-
-async def get_or_fetch_member(guild: discord.Guild, member_id: int) -> t.Optional[discord.Member]:
- """
- Attempt to get a member from cache; on failure fetch from the API.
-
- Return `None` to indicate the member could not be found.
- """
- if member := guild.get_member(member_id):
- log.trace("%s retrieved from cache.", member)
- else:
- try:
- member = await guild.fetch_member(member_id)
- except discord.errors.NotFound:
- log.trace("Failed to fetch %d from API.", member_id)
- return None
- log.trace("%s fetched from API.", member)
- return member
-
-
-async def handle_role_change(
- member: discord.Member,
- coro: t.Callable[..., t.Coroutine],
- role: discord.Role
-) -> None:
- """
- Change `member`'s cooldown role via awaiting `coro` and handle errors.
-
- `coro` is intended to be `discord.Member.add_roles` or `discord.Member.remove_roles`.
- """
- try:
- await coro(role)
- except discord.NotFound:
- log.debug(f"Failed to change role for {member} ({member.id}): member not found")
- except discord.Forbidden:
- log.error(
- f"Forbidden to change role for {member} ({member.id}); "
- f"possibly due to role hierarchy"
- )
- except discord.HTTPException as e:
- log.error(f"Failed to change role for {member} ({member.id}): {e.status} {e.code}")
generated by cgit v1.2.3 (git 2.39.1) at 2025-10-31 01:06:52 +0000