diff options
| -rw-r--r-- | Pipfile | 1 | ||||
| -rw-r--r-- | Pipfile.lock | 10 | ||||
| -rw-r--r-- | bot/constants.py | 43 | ||||
| -rw-r--r-- | bot/decorators.py | 23 | ||||
| -rw-r--r-- | bot/errors.py | 6 | ||||
| -rw-r--r-- | bot/exts/backend/branding.py | 569 | ||||
| -rw-r--r-- | bot/exts/backend/error_handler.py | 8 | ||||
| -rw-r--r-- | bot/exts/utils/utils.py | 9 | ||||
| -rw-r--r-- | bot/seasons.py | 181 | ||||
| -rw-r--r-- | config-default.yml | 10 |
10 files changed, 853 insertions, 7 deletions
@@ -26,6 +26,7 @@ requests = "~=2.22" sentry-sdk = "~=0.14" sphinx = "~=2.2" statsd = "~=3.3" +arrow = "~=0.17" emoji = "~=0.6" [dev-packages] diff --git a/Pipfile.lock b/Pipfile.lock index 541db1627..215c64bed 100644 --- a/Pipfile.lock +++ b/Pipfile.lock @@ -1,7 +1,7 @@ { "_meta": { "hash": { - "sha256": "3ccb368599709d2970f839fc3721cfeebcd5a2700fed7231b2ce38a080828325" + "sha256": "d08ba836e630ae64a560ef879216d24801e8069dddf7d51b00e885efcf24c2ff" }, "pipfile-spec": 6, "requires": { @@ -82,6 +82,14 @@ ], "version": "==0.7.12" }, + "arrow": { + "hashes": [ + "sha256:e098abbd9af3665aea81bdd6c869e93af4feb078e98468dd351c383af187aac5", + "sha256:ff08d10cda1d36c68657d6ad20d74fbea493d980f8b2d45344e00d6ed2bf6ed4" + ], + "index": "pypi", + "version": "==0.17.0" + }, "async-rediscache": { "extras": [ "fakeredis" diff --git a/bot/constants.py b/bot/constants.py index 08ae0d52f..3475ee4e3 100644 --- a/bot/constants.py +++ b/bot/constants.py @@ -13,7 +13,7 @@ their default values from `config-default.yml`. import logging import os from collections.abc import Mapping -from enum import Enum +from enum import Enum, IntEnum from pathlib import Path from typing import Dict, List, Optional @@ -249,6 +249,9 @@ class Colours(metaclass=YAMLGetter): soft_green: int soft_orange: int bright_green: int + orange: int + pink: int + purple: int class DuckPond(metaclass=YAMLGetter): @@ -299,6 +302,8 @@ class Emojis(metaclass=YAMLGetter): comments: str user: str + ok_hand: str + class Icons(metaclass=YAMLGetter): section = "style" @@ -493,6 +498,7 @@ class Keys(metaclass=YAMLGetter): section = "keys" site_api: Optional[str] + github: Optional[str] class URLs(metaclass=YAMLGetter): @@ -613,6 +619,12 @@ class VoiceGate(metaclass=YAMLGetter): voice_ping_delete_delay: int +class Branding(metaclass=YAMLGetter): + section = "branding" + + cycle_frequency: int + + class Event(Enum): """ Event names. This does not include every event (for example, raw @@ -639,6 +651,35 @@ class Event(Enum): voice_state_update = "voice_state_update" +class Month(IntEnum): + JANUARY = 1 + FEBRUARY = 2 + MARCH = 3 + APRIL = 4 + MAY = 5 + JUNE = 6 + JULY = 7 + AUGUST = 8 + SEPTEMBER = 9 + OCTOBER = 10 + NOVEMBER = 11 + DECEMBER = 12 + + def __str__(self) -> str: + return self.name.title() + + +class AssetType(Enum): + """ + Discord media assets. + + The values match exactly the kwarg keys that can be passed to `Guild.edit`. + """ + + BANNER = "banner" + SERVER_ICON = "icon" + + # Debug mode DEBUG_MODE = 'local' in os.environ.get("SITE_URL", "local") diff --git a/bot/decorators.py b/bot/decorators.py index 063c8f878..0b50cc365 100644 --- a/bot/decorators.py +++ b/bot/decorators.py @@ -1,4 +1,5 @@ import asyncio +import functools import logging import typing as t from contextlib import suppress @@ -8,7 +9,7 @@ from discord import Member, NotFound from discord.ext import commands from discord.ext.commands import Cog, Context -from bot.constants import Channels, RedirectOutput +from bot.constants import Channels, DEBUG_MODE, RedirectOutput from bot.utils import function from bot.utils.checks import in_whitelist_check @@ -153,3 +154,23 @@ def respect_role_hierarchy(member_arg: function.Argument) -> t.Callable: await func(*args, **kwargs) return wrapper return decorator + + +def mock_in_debug(return_value: t.Any) -> t.Callable: + """ + Short-circuit function execution if in debug mode and return `return_value`. + + The original function name, and the incoming args and kwargs are DEBUG level logged + upon each call. This is useful for expensive operations, i.e. media asset uploads + that are prone to rate-limits but need to be tested extensively. + """ + def decorator(func: t.Callable) -> t.Callable: + @functools.wraps(func) + async def wrapped(*args, **kwargs) -> t.Any: + """Short-circuit and log if in debug mode.""" + if DEBUG_MODE: + log.debug(f"Function {func.__name__} called with args: {args}, kwargs: {kwargs}") + return return_value + return await func(*args, **kwargs) + return wrapped + return decorator diff --git a/bot/errors.py b/bot/errors.py index 65d715203..a3484830b 100644 --- a/bot/errors.py +++ b/bot/errors.py @@ -18,3 +18,9 @@ class LockedResourceError(RuntimeError): f"Cannot operate on {self.type.lower()} `{self.id}`; " "it is currently locked and in use by another operation." ) + + +class BrandingError(Exception): + """Exception raised by the BrandingManager cog.""" + + pass diff --git a/bot/exts/backend/branding.py b/bot/exts/backend/branding.py new file mode 100644 index 000000000..7ce85aab2 --- /dev/null +++ b/bot/exts/backend/branding.py @@ -0,0 +1,569 @@ +import asyncio +import itertools +import logging +import random +import typing as t +from datetime import datetime, time, timedelta + +import arrow +import async_timeout +import discord +from async_rediscache import RedisCache +from discord.ext import commands + +from bot.bot import Bot +from bot.constants import AssetType, Branding, Colours, Emojis, Guild, Keys, MODERATION_ROLES +from bot.decorators import in_whitelist, mock_in_debug +from bot.errors import BrandingError +from bot.seasons import SeasonBase, get_all_seasons, get_current_season, get_season + +log = logging.getLogger(__name__) + +STATUS_OK = 200 # HTTP status code + +FILE_BANNER = "banner.png" +FILE_AVATAR = "avatar.png" +SERVER_ICONS = "server_icons" + +BRANDING_URL = "https://api.github.com/repos/python-discord/branding/contents" + +PARAMS = {"ref": "master"} # Target branch +HEADERS = {"Accept": "application/vnd.github.v3+json"} # Ensure we use API v3 + +# A GitHub token is not necessary for the cog to operate, +# unauthorized requests are however limited to 60 per hour +if Keys.github: + HEADERS["Authorization"] = f"token {Keys.github}" + + +class GitHubFile(t.NamedTuple): + """ + Represents a remote file on GitHub. + + The `sha` hash is kept so that we can determine that a file has changed, + despite its filename remaining unchanged. + """ + + download_url: str + path: str + sha: str + + +def pretty_files(files: t.Iterable[GitHubFile]) -> str: + """Provide a human-friendly representation of `files`.""" + return "\n".join(file.path for file in files) + + +def time_until_midnight() -> timedelta: + """ + Determine amount of time until the next-up UTC midnight. + + The exact `midnight` moment is actually delayed to 5 seconds after, in order + to avoid potential problems due to imprecise sleep. + """ + now = datetime.utcnow() + tomorrow = now + timedelta(days=1) + midnight = datetime.combine(tomorrow, time(second=5)) + + return midnight - now + + +class BrandingManager(commands.Cog): + """ + Manages the guild's branding. + + The purpose of this cog is to help automate the synchronization of the branding + repository with the guild. It is capable of discovering assets in the repository + via GitHub's API, resolving download urls for them, and delegating + to the `bot` instance to upload them to the guild. + + BrandingManager is designed to be entirely autonomous. Its `daemon` background task awakens + once a day (see `time_until_midnight`) to detect new seasons, or to cycle icons within a single + season. The daemon can be turned on and off via the `daemon` cmd group. The value set via + its `start` and `stop` commands is persisted across sessions. If turned on, the daemon will + automatically start on the next bot start-up. Otherwise, it will wait to be started manually. + + All supported operations, e.g. setting seasons, applying the branding, or cycling icons, can + also be invoked manually, via the following API: + + branding list + - Show all available seasons + + branding set <season_name> + - Set the cog's internal state to represent `season_name`, if it exists + - If no `season_name` is given, set chronologically current season + - This will not automatically apply the season's branding to the guild, + the cog's state can be detached from the guild + - Seasons can therefore be 'previewed' using this command + + branding info + - View detailed information about resolved assets for current season + + branding refresh + - Refresh internal state, i.e. synchronize with branding repository + + branding apply + - Apply the current internal state to the guild, i.e. upload the assets + + branding cycle + - If there are multiple available icons for current season, randomly pick + and apply the next one + + The daemon calls these methods autonomously as appropriate. The use of this cog + is locked to moderation roles. As it performs media asset uploads, it is prone to + rate-limits - the `apply` command should be used with caution. The `set` command can, + however, be used freely to 'preview' seasonal branding and check whether paths have been + resolved as appropriate. + + While the bot is in debug mode, it will 'mock' asset uploads by logging the passed + download urls and pretending that the upload was successful. Make use of this + to test this cog's behaviour. + """ + + current_season: t.Type[SeasonBase] + + banner: t.Optional[GitHubFile] + + available_icons: t.List[GitHubFile] + remaining_icons: t.List[GitHubFile] + + days_since_cycle: t.Iterator + + daemon: t.Optional[asyncio.Task] + + # Branding configuration + branding_configuration = RedisCache() + + def __init__(self, bot: Bot) -> None: + """ + Assign safe default values on init. + + At this point, we don't have information about currently available branding. + Most of these attributes will be overwritten once the daemon connects, or once + the `refresh` command is used. + """ + self.bot = bot + self.current_season = get_current_season() + + self.banner = None + + self.available_icons = [] + self.remaining_icons = [] + + self.days_since_cycle = itertools.cycle([None]) + + should_run = self.bot.loop.run_until_complete(self.branding_configuration.get("daemon_active")) + + if should_run: + self.daemon = self.bot.loop.create_task(self._daemon_func()) + else: + self.daemon = None + + @property + def _daemon_running(self) -> bool: + """True if the daemon is currently active, False otherwise.""" + return self.daemon is not None and not self.daemon.done() + + async def _daemon_func(self) -> None: + """ + Manage all automated behaviour of the BrandingManager cog. + + Once a day, the daemon will perform the following tasks: + - Update `current_season` + - Poll GitHub API to see if the available branding for `current_season` has changed + - Update assets if changes are detected (banner, guild icon, bot avatar, bot nickname) + - Check whether it's time to cycle guild icons + + The internal loop runs once when activated, then periodically at the time + given by `time_until_midnight`. + + All method calls in the internal loop are considered safe, i.e. no errors propagate + to the daemon's loop. The daemon itself does not perform any error handling on its own. + """ + await self.bot.wait_until_guild_available() + + while True: + self.current_season = get_current_season() + branding_changed = await self.refresh() + + if branding_changed: + await self.apply() + + elif next(self.days_since_cycle) == Branding.cycle_frequency: + await self.cycle() + + until_midnight = time_until_midnight() + await asyncio.sleep(until_midnight.total_seconds()) + + async def _info_embed(self) -> discord.Embed: + """Make an informative embed representing current season.""" + info_embed = discord.Embed(description=self.current_season.description, colour=self.current_season.colour) + + # If we're in a non-evergreen season, also show active months + if self.current_season is not SeasonBase: + title = f"{self.current_season.season_name} ({', '.join(str(m) for m in self.current_season.months)})" + else: + title = self.current_season.season_name + + # Use the author field to show the season's name and avatar if available + info_embed.set_author(name=title) + + banner = self.banner.path if self.banner is not None else "Unavailable" + info_embed.add_field(name="Banner", value=banner, inline=False) + + icons = pretty_files(self.available_icons) or "Unavailable" + info_embed.add_field(name="Available icons", value=icons, inline=False) + + # Only display cycle frequency if we're actually cycling + if len(self.available_icons) > 1 and Branding.cycle_frequency: + info_embed.set_footer(text=f"Icon cycle frequency: {Branding.cycle_frequency}") + + return info_embed + + async def _reset_remaining_icons(self) -> None: + """Set `remaining_icons` to a shuffled copy of `available_icons`.""" + self.remaining_icons = random.sample(self.available_icons, k=len(self.available_icons)) + + async def _reset_days_since_cycle(self) -> None: + """ + Reset the `days_since_cycle` iterator based on configured frequency. + + If the current season only has 1 icon, or if `Branding.cycle_frequency` is falsey, + the iterator will always yield None. This signals that the icon shouldn't be cycled. + + Otherwise, it will yield ints in range [1, `Branding.cycle_frequency`] indefinitely. + When the iterator yields a value equal to `Branding.cycle_frequency`, it is time to cycle. + """ + if len(self.available_icons) > 1 and Branding.cycle_frequency: + sequence = range(1, Branding.cycle_frequency + 1) + else: + sequence = [None] + + self.days_since_cycle = itertools.cycle(sequence) + + async def _get_files(self, path: str, include_dirs: bool = False) -> t.Dict[str, GitHubFile]: + """ + Get files at `path` in the branding repository. + + If `include_dirs` is False (default), only returns files at `path`. + Otherwise, will return both files and directories. Never returns symlinks. + + Return dict mapping from filename to corresponding `GitHubFile` instance. + This may return an empty dict if the response status is non-200, + or if the target directory is empty. + """ + url = f"{BRANDING_URL}/{path}" + async with self.bot.http_session.get(url, headers=HEADERS, params=PARAMS) as resp: + # Short-circuit if we get non-200 response + if resp.status != STATUS_OK: + log.error(f"GitHub API returned non-200 response: {resp}") + return {} + directory = await resp.json() # Directory at `path` + + allowed_types = {"file", "dir"} if include_dirs else {"file"} + return { + file["name"]: GitHubFile(file["download_url"], file["path"], file["sha"]) + for file in directory + if file["type"] in allowed_types + } + + async def refresh(self) -> bool: + """ + Synchronize available assets with branding repository. + + If the current season is not the evergreen, and lacks at least one asset, + we use the evergreen seasonal dir as fallback for missing assets. + + Finally, if neither the seasonal nor fallback branding directories contain + an asset, it will simply be ignored. + + Return True if the branding has changed. This will be the case when we enter + a new season, or when something changes in the current seasons's directory + in the branding repository. + """ + old_branding = (self.banner, self.available_icons) + seasonal_dir = await self._get_files(self.current_season.branding_path, include_dirs=True) + + # Only make a call to the fallback directory if there is something to be gained + branding_incomplete = any( + asset not in seasonal_dir + for asset in (FILE_BANNER, FILE_AVATAR, SERVER_ICONS) + ) + if branding_incomplete and self.current_season is not SeasonBase: + fallback_dir = await self._get_files(SeasonBase.branding_path, include_dirs=True) + else: + fallback_dir = {} + + # Resolve assets in this directory, None is a safe value + self.banner = seasonal_dir.get(FILE_BANNER) or fallback_dir.get(FILE_BANNER) + + # Now resolve server icons by making a call to the proper sub-directory + if SERVER_ICONS in seasonal_dir: + icons_dir = await self._get_files(f"{self.current_season.branding_path}/{SERVER_ICONS}") + self.available_icons = list(icons_dir.values()) + + elif SERVER_ICONS in fallback_dir: + icons_dir = await self._get_files(f"{SeasonBase.branding_path}/{SERVER_ICONS}") + self.available_icons = list(icons_dir.values()) + + else: + self.available_icons = [] # This should never be the case, but an empty list is a safe value + + # GitHubFile instances carry a `sha` attr so this will pick up if a file changes + branding_changed = old_branding != (self.banner, self.available_icons) + + if branding_changed: + log.info(f"New branding detected (season: {self.current_season.season_name})") + await self._reset_remaining_icons() + await self._reset_days_since_cycle() + + return branding_changed + + async def cycle(self) -> bool: + """ + Apply the next-up server icon. + + Returns True if an icon is available and successfully gets applied, False otherwise. + """ + if not self.available_icons: + log.info("Cannot cycle: no icons for this season") + return False + + if not self.remaining_icons: + log.info("Reset & shuffle remaining icons") + await self._reset_remaining_icons() + + next_up = self.remaining_icons.pop(0) + success = await self.set_icon(next_up.download_url) + + return success + + async def apply(self) -> t.List[str]: + """ + Apply current branding to the guild and bot. + + This delegates to the bot instance to do all the work. We only provide download urls + for available assets. Assets unavailable in the branding repo will be ignored. + + Returns a list of names of all failed assets. An asset is considered failed + if it isn't found in the branding repo, or if something goes wrong while the + bot is trying to apply it. + + An empty list denotes that all assets have been applied successfully. + """ + report = {asset: False for asset in ("banner", "icon")} + + if self.banner is not None: + report["banner"] = await self.set_banner(self.banner.download_url) + + report["icon"] = await self.cycle() + + failed_assets = [asset for asset, succeeded in report.items() if not succeeded] + return failed_assets + + @in_whitelist(roles=MODERATION_ROLES) + @commands.group(name="branding") + async def branding_cmds(self, ctx: commands.Context) -> None: + """Manual branding control.""" + if not ctx.invoked_subcommand: + await ctx.send_help(ctx.command) + + @branding_cmds.command(name="list", aliases=["ls"]) + async def branding_list(self, ctx: commands.Context) -> None: + """List all available seasons and branding sources.""" + embed = discord.Embed(title="Available seasons", colour=Colours.soft_green) + + for season in get_all_seasons(): + if season is SeasonBase: + active_when = "always" + else: + active_when = f"in {', '.join(str(m) for m in season.months)}" + + description = ( + f"Active {active_when}\n" + f"Branding: {season.branding_path}" + ) + embed.add_field(name=season.season_name, value=description, inline=False) + + await ctx.send(embed=embed) + + @branding_cmds.command(name="set") + async def branding_set(self, ctx: commands.Context, *, season_name: t.Optional[str] = None) -> None: + """ + Manually set season, or reset to current if none given. + + Season search is a case-less comparison against both seasonal class name, + and its `season_name` attr. + + This only pre-loads the cog's internal state to the chosen season, but does not + automatically apply the branding. As that is an expensive operation, the `apply` + command must be called explicitly after this command finishes. + + This means that this command can be used to 'preview' a season gathering info + about its available assets, without applying them to the guild. + + If the daemon is running, it will automatically reset the season to current when + it wakes up. The season set via this command can therefore remain 'detached' from + what it should be - the daemon will make sure that it's set back properly. + """ + if season_name is None: + new_season = get_current_season() + else: + new_season = get_season(season_name) + if new_season is None: + raise BrandingError("No such season exists") + + if self.current_season is new_season: + raise BrandingError(f"Season {self.current_season.season_name} already active") + + self.current_season = new_season + await self.branding_refresh(ctx) + + @branding_cmds.command(name="info", aliases=["status"]) + async def branding_info(self, ctx: commands.Context) -> None: + """ + Show available assets for current season. + + This can be used to confirm that assets have been resolved properly. + When `apply` is used, it attempts to upload exactly the assets listed here. + """ + await ctx.send(embed=await self._info_embed()) + + @branding_cmds.command(name="refresh") + async def branding_refresh(self, ctx: commands.Context) -> None: + """Sync currently available assets with branding repository.""" + async with ctx.typing(): + await self.refresh() + await self.branding_info(ctx) + + @branding_cmds.command(name="apply") + async def branding_apply(self, ctx: commands.Context) -> None: + """ + Apply current season's branding to the guild. + + Use `info` to check which assets will be applied. Shows which assets have + failed to be applied, if any. + """ + async with ctx.typing(): + failed_assets = await self.apply() + if failed_assets: + raise BrandingError(f"Failed to apply following assets: {', '.join(failed_assets)}") + + response = discord.Embed(description=f"All assets applied {Emojis.ok_hand}", colour=Colours.soft_green) + await ctx.send(embed=response) + + @branding_cmds.command(name="cycle") + async def branding_cycle(self, ctx: commands.Context) -> None: + """ + Apply the next-up guild icon, if multiple are available. + + The order is random. + """ + async with ctx.typing(): + success = await self.cycle() + if not success: + raise BrandingError("Failed to cycle icon") + + response = discord.Embed(description=f"Success {Emojis.ok_hand}", colour=Colours.soft_green) + await ctx.send(embed=response) + + @branding_cmds.group(name="daemon", aliases=["d", "task"]) + async def daemon_group(self, ctx: commands.Context) -> None: + """Control the background daemon.""" + if not ctx.invoked_subcommand: + await ctx.send_help(ctx.command) + + @daemon_group.command(name="status") + async def daemon_status(self, ctx: commands.Context) -> None: + """Check whether daemon is currently active.""" + if self._daemon_running: + remaining_time = (arrow.utcnow() + time_until_midnight()).humanize() + response = discord.Embed(description=f"Daemon running {Emojis.ok_hand}", colour=Colours.soft_green) + response.set_footer(text=f"Next refresh {remaining_time}") + else: + response = discord.Embed(description="Daemon not running", colour=Colours.soft_red) + + await ctx.send(embed=response) + + @daemon_group.command(name="start") + async def daemon_start(self, ctx: commands.Context) -> None: + """If the daemon isn't running, start it.""" + if self._daemon_running: + raise BrandingError("Daemon already running!") + + self.daemon = self.bot.loop.create_task(self._daemon_func()) + await self.branding_configuration.set("daemon_active", True) + + response = discord.Embed(description=f"Daemon started {Emojis.ok_hand}", colour=Colours.soft_green) + await ctx.send(embed=response) + + @daemon_group.command(name="stop") + async def daemon_stop(self, ctx: commands.Context) -> None: + """If the daemon is running, stop it.""" + if not self._daemon_running: + raise BrandingError("Daemon not running!") + + self.daemon.cancel() + await self.branding_configuration.set("daemon_active", False) + + response = discord.Embed(description=f"Daemon stopped {Emojis.ok_hand}", colour=Colours.soft_green) + await ctx.send(embed=response) + + async def _fetch_image(self, url: str) -> bytes: + """Retrieve and read image from `url`.""" + log.debug(f"Getting image from: {url}") + async with self.bot.http_session.get(url) as resp: + return await resp.read() + + async def _apply_asset(self, target: discord.Guild, asset: AssetType, url: str) -> bool: + """ + Internal method for applying media assets to the guild. + + This shouldn't be called directly. The purpose of this method is mainly generic + error handling to reduce needless code repetition. + + Return True if upload was successful, False otherwise. + """ + log.info(f"Attempting to set {asset.name}: {url}") + + kwargs = {asset.value: await self._fetch_image(url)} + try: + async with async_timeout.timeout(5): + await target.edit(**kwargs) + + except asyncio.TimeoutError: + log.info("Asset upload timed out") + return False + + except discord.HTTPException as discord_error: + log.exception("Asset upload failed", exc_info=discord_error) + return False + + else: + log.info("Asset successfully applied") + return True + + @mock_in_debug(return_value=True) + async def set_banner(self, url: str) -> bool: + """Set the guild's banner to image at `url`.""" + guild = self.bot.get_guild(Guild.id) + if guild is None: + log.info("Failed to get guild instance, aborting asset upload") + return False + + return await self._apply_asset(guild, AssetType.BANNER, url) + + @mock_in_debug(return_value=True) + async def set_icon(self, url: str) -> bool: + """Sets the guild's icon to image at `url`.""" + guild = self.bot.get_guild(Guild.id) + if guild is None: + log.info("Failed to get guild instance, aborting asset upload") + return False + + return await self._apply_asset(guild, AssetType.SERVER_ICON, url) + + +def setup(bot: Bot) -> None: + """Load BrandingManager cog.""" + bot.add_cog(BrandingManager(bot)) diff --git a/bot/exts/backend/error_handler.py b/bot/exts/backend/error_handler.py index c643d346e..b6c19d504 100644 --- a/bot/exts/backend/error_handler.py +++ b/bot/exts/backend/error_handler.py @@ -1,5 +1,6 @@ import contextlib import logging +import random import typing as t from discord import Embed @@ -8,9 +9,9 @@ from sentry_sdk import push_scope from bot.api import ResponseCodeError from bot.bot import Bot -from bot.constants import Channels, Colours +from bot.constants import Channels, Colours, ERROR_REPLIES from bot.converters import TagNameConverter -from bot.errors import LockedResourceError +from bot.errors import BrandingError, LockedResourceError from bot.utils.checks import InWhitelistCheckFailure log = logging.getLogger(__name__) @@ -78,6 +79,9 @@ class ErrorHandler(Cog): await self.handle_api_error(ctx, e.original) elif isinstance(e.original, LockedResourceError): await ctx.send(f"{e.original} Please wait for it to finish and try again later.") + elif isinstance(e.original, BrandingError): + await ctx.send(embed=self._get_error_embed(random.choice(ERROR_REPLIES), str(e.original))) + return else: await self.handle_unexpected_error(ctx, e.original) return # Exit early to avoid logging. diff --git a/bot/exts/utils/utils.py b/bot/exts/utils/utils.py index 8e7e6ba36..df9d6661d 100644 --- a/bot/exts/utils/utils.py +++ b/bot/exts/utils/utils.py @@ -12,7 +12,7 @@ from discord.ext.commands import BadArgument, Cog, Context, clean_content, comma from discord.utils import snowflake_time from bot.bot import Bot -from bot.constants import Channels, MODERATION_ROLES, STAFF_ROLES +from bot.constants import Channels, Keys, MODERATION_ROLES, STAFF_ROLES from bot.converters import Snowflake from bot.decorators import in_whitelist from bot.pagination import LinePaginator @@ -48,6 +48,11 @@ ICON_URL = "https://www.python.org/static/opengraph-icon-200x200.png" pep_cache = AsyncCache() +# Add GitHub token when it's set to raise limit of requests per hour +GITHUB_HEADERS = {} +if Keys.github: + GITHUB_HEADERS["Authorization"] = f"token {Keys.github}" + class Utils(Cog): """A selection of utilities which don't have a clear category.""" @@ -215,7 +220,7 @@ class Utils(Cog): await self.bot.wait_until_ready() log.trace("Started refreshing PEP URLs.") - async with self.bot.http_session.get(self.PEPS_LISTING_API_URL) as resp: + async with self.bot.http_session.get(self.PEPS_LISTING_API_URL, headers=GITHUB_HEADERS) as resp: listing = await resp.json() log.trace("Got PEP URLs listing from GitHub API") diff --git a/bot/seasons.py b/bot/seasons.py new file mode 100644 index 000000000..3b39f03cd --- /dev/null +++ b/bot/seasons.py @@ -0,0 +1,181 @@ +import logging +import typing as t +from datetime import datetime + +from bot.constants import Colours, Month +from bot.errors import BrandingError + +log = logging.getLogger(__name__) + + +class SeasonBase: + """ + Base for Seasonal classes. + + This serves as the off-season fallback for when no specific + seasons are active. + + Seasons are 'registered' simply by inheriting from `SeasonBase`. + We discover them by calling `__subclasses__`. + """ + + season_name: str = "Evergreen" + bot_name: str = "SeasonalBot" + + colour: str = Colours.soft_green + description: str = "The default season!" + + branding_path: str = "seasonal/evergreen" + + months: t.Set[Month] = set(Month) + + +class Christmas(SeasonBase): + """Branding for December.""" + + season_name = "Festive season" + bot_name = "MerryBot" + + colour = Colours.soft_red + description = ( + "The time is here to get into the festive spirit! No matter who you are, where you are, " + "or what beliefs you may follow, we hope every one of you enjoy this festive season!" + ) + + branding_path = "seasonal/christmas" + + months = {Month.DECEMBER} + + +class Easter(SeasonBase): + """Branding for April.""" + + season_name = "Easter" + bot_name = "BunnyBot" + + colour = Colours.bright_green + description = ( + "Bunny here, bunny there, bunny everywhere! Here at Python Discord, we celebrate " + "our version of Easter during the entire month of April." + ) + + branding_path = "seasonal/easter" + + months = {Month.APRIL} + + +class Halloween(SeasonBase): + """Branding for October.""" + + season_name = "Halloween" + bot_name = "NeonBot" + + colour = Colours.orange + description = "Trick or treat?!" + + branding_path = "seasonal/halloween" + + months = {Month.OCTOBER} + + +class Pride(SeasonBase): + """Branding for June.""" + + season_name = "Pride" + bot_name = "ProudBot" + + colour = Colours.pink + description = ( + "The month of June is a special month for us at Python Discord. It is very important to us " + "that everyone feels welcome here, no matter their origin, identity or sexuality. During the " + "month of June, while some of you are participating in Pride festivals across the world, " + "we will be celebrating individuality and commemorating the history and challenges " + "of the LGBTQ+ community with a Pride event of our own!" + ) + + branding_path = "seasonal/pride" + + months = {Month.JUNE} + + +class Valentines(SeasonBase): + """Branding for February.""" + + season_name = "Valentines" + bot_name = "TenderBot" + + colour = Colours.pink + description = "Love is in the air!" + + branding_path = "seasonal/valentines" + + months = {Month.FEBRUARY} + + +class Wildcard(SeasonBase): + """Branding for August.""" + + season_name = "Wildcard" + bot_name = "RetroBot" + + colour = Colours.purple + description = "A season full of surprises!" + + months = {Month.AUGUST} + + +def get_all_seasons() -> t.List[t.Type[SeasonBase]]: + """Give all available season classes.""" + return [SeasonBase] + SeasonBase.__subclasses__() + + +def get_current_season() -> t.Type[SeasonBase]: + """Give active season, based on current UTC month.""" + current_month = Month(datetime.utcnow().month) + + active_seasons = tuple( + season + for season in SeasonBase.__subclasses__() + if current_month in season.months + ) + + if not active_seasons: + return SeasonBase + + return active_seasons[0] + + +def get_season(name: str) -> t.Optional[t.Type[SeasonBase]]: + """ + Give season such that its class name or its `season_name` attr match `name` (caseless). + + If no such season exists, return None. + """ + name = name.casefold() + + for season in get_all_seasons(): + matches = (season.__name__.casefold(), season.season_name.casefold()) + + if name in matches: + return season + + +def _validate_season_overlap() -> None: + """ + Raise BrandingError if there are any colliding seasons. + + This serves as a local test to ensure that seasons haven't been misconfigured. + """ + month_to_season = {} + + for season in SeasonBase.__subclasses__(): + for month in season.months: + colliding_season = month_to_season.get(month) + + if colliding_season: + raise BrandingError(f"Season {season} collides with {colliding_season} in {month.name}") + else: + month_to_season[month] = season + + +_validate_season_overlap() diff --git a/config-default.yml b/config-default.yml index 006743342..9e1c9c012 100644 --- a/config-default.yml +++ b/config-default.yml @@ -28,6 +28,9 @@ style: soft_green: 0x68c290 soft_orange: 0xf9cb54 bright_green: 0x01d277 + orange: 0xe67e22 + pink: 0xcf84e0 + purple: 0xb734eb emojis: defcon_disabled: "<:defcondisabled:470326273952972810>" @@ -68,6 +71,8 @@ style: comments: "<:reddit_comments:755845255001014384>" user: "<:reddit_users:755845303822974997>" + ok_hand: ":ok_hand:" + icons: crown_blurple: "https://cdn.discordapp.com/emojis/469964153289965568.png" crown_green: "https://cdn.discordapp.com/emojis/469964154719961088.png" @@ -323,6 +328,7 @@ filter: keys: site_api: !ENV "BOT_API_KEY" + github: !ENV "GITHUB_TOKEN" urls: @@ -533,5 +539,9 @@ voice_gate: voice_ping_delete_delay: 60 # Seconds before deleting the bot's ping to user in Voice Gate +branding: + cycle_frequency: 3 # How many days bot wait before refreshing server icon + + config: required_keys: ['bot.token'] |