diff options
Diffstat (limited to 'bot/exts')
| -rw-r--r-- | bot/exts/evergreen/battleship.py | 2 | ||||
| -rw-r--r-- | bot/exts/evergreen/game.py | 83 | ||||
| -rw-r--r-- | bot/exts/evergreen/xkcd.py | 89 | ||||
| -rw-r--r-- | bot/exts/halloween/spookynamerate.py | 401 |
4 files changed, 561 insertions, 14 deletions
diff --git a/bot/exts/evergreen/battleship.py b/bot/exts/evergreen/battleship.py index 9bc374e6..fa3fb35c 100644 --- a/bot/exts/evergreen/battleship.py +++ b/bot/exts/evergreen/battleship.py @@ -140,7 +140,7 @@ class Game: @staticmethod def get_square(grid: Grid, square: str) -> Square: """Grabs a square from a grid with an inputted key.""" - index = ord(square[0]) - ord("A") + index = ord(square[0].upper()) - ord("A") number = int(square[1:]) return grid[number-1][index] # -1 since lists are indexed from 0 diff --git a/bot/exts/evergreen/game.py b/bot/exts/evergreen/game.py index d0fd7a40..d37be0e2 100644 --- a/bot/exts/evergreen/game.py +++ b/bot/exts/evergreen/game.py @@ -2,7 +2,8 @@ import difflib import logging import random import re -from datetime import datetime as dt +from asyncio import sleep +from datetime import datetime as dt, timedelta from enum import IntEnum from typing import Any, Dict, List, Optional, Tuple @@ -17,10 +18,25 @@ from bot.utils.decorators import with_role from bot.utils.pagination import ImagePaginator, LinePaginator # Base URL of IGDB API -BASE_URL = "https://api-v3.igdb.com" +BASE_URL = "https://api.igdb.com/v4" -HEADERS = { - "user-key": Tokens.igdb, +CLIENT_ID = Tokens.igdb_client_id +CLIENT_SECRET = Tokens.igdb_client_secret + +# The number of seconds before expiry that we attempt to re-fetch a new access token +ACCESS_TOKEN_RENEWAL_WINDOW = 60*60*24*2 + +# URL to request API access token +OAUTH_URL = "https://id.twitch.tv/oauth2/token" + +OAUTH_PARAMS = { + "client_id": CLIENT_ID, + "client_secret": CLIENT_SECRET, + "grant_type": "client_credentials" +} + +BASE_HEADERS = { + "Client-ID": CLIENT_ID, "Accept": "application/json" } @@ -135,8 +151,47 @@ class Games(Cog): self.http_session: ClientSession = bot.http_session self.genres: Dict[str, int] = {} - - self.refresh_genres_task.start() + self.headers = BASE_HEADERS + + self.bot.loop.create_task(self.renew_access_token()) + + async def renew_access_token(self) -> None: + """Refeshes V4 access token a number of seconds before expiry. See `ACCESS_TOKEN_RENEWAL_WINDOW`.""" + while True: + async with self.http_session.post(OAUTH_URL, params=OAUTH_PARAMS) as resp: + result = await resp.json() + if resp.status != 200: + # If there is a valid access token continue to use that, + # otherwise unload cog. + if "Authorization" in self.headers: + time_delta = timedelta(seconds=ACCESS_TOKEN_RENEWAL_WINDOW) + logger.error( + "Failed to renew IGDB access token. " + f"Current token will last for {time_delta} " + f"OAuth response message: {result['message']}" + ) + else: + logger.warning( + "Invalid OAuth credentials. Unloading Games cog. " + f"OAuth response message: {result['message']}" + ) + self.bot.remove_cog('Games') + + return + + self.headers["Authorization"] = f"Bearer {result['access_token']}" + + # Attempt to renew before the token expires + next_renewal = result["expires_in"] - ACCESS_TOKEN_RENEWAL_WINDOW + + time_delta = timedelta(seconds=next_renewal) + logger.info(f"Successfully renewed access token. Refreshing again in {time_delta}") + + # This will be true the first time this loop runs. + # Since we now have an access token, its safe to start this task. + if self.genres == {}: + self.refresh_genres_task.start() + await sleep(next_renewal) @tasks.loop(hours=24.0) async def refresh_genres_task(self) -> None: @@ -156,9 +211,8 @@ class Games(Cog): async def _get_genres(self) -> None: """Create genres variable for games command.""" body = "fields name; limit 100;" - async with self.http_session.get(f"{BASE_URL}/genres", data=body, headers=HEADERS) as resp: + async with self.http_session.post(f"{BASE_URL}/genres", data=body, headers=self.headers) as resp: result = await resp.json() - genres = {genre["name"].capitalize(): genre["id"] for genre in result} # Replace complex names with names from ALIASES @@ -306,7 +360,7 @@ class Games(Cog): body = GAMES_LIST_BODY.format(**params) # Do request to IGDB API, create headers, URL, define body, return result - async with self.http_session.get(url=f"{BASE_URL}/games", data=body, headers=HEADERS) as resp: + async with self.http_session.post(url=f"{BASE_URL}/games", data=body, headers=self.headers) as resp: return await resp.json() async def create_page(self, data: Dict[str, Any]) -> Tuple[str, str]: @@ -348,7 +402,7 @@ class Games(Cog): # Define request body of IGDB API request and do request body = SEARCH_BODY.format(**{"term": search_term}) - async with self.http_session.get(url=f"{BASE_URL}/games", data=body, headers=HEADERS) as resp: + async with self.http_session.post(url=f"{BASE_URL}/games", data=body, headers=self.headers) as resp: data = await resp.json() # Loop over games, format them to good format, make line and append this to total lines @@ -377,7 +431,7 @@ class Games(Cog): "offset": offset }) - async with self.http_session.get(url=f"{BASE_URL}/companies", data=body, headers=HEADERS) as resp: + async with self.http_session.post(url=f"{BASE_URL}/companies", data=body, headers=self.headers) as resp: return await resp.json() async def create_company_page(self, data: Dict[str, Any]) -> Tuple[str, str]: @@ -418,7 +472,10 @@ class Games(Cog): def setup(bot: Bot) -> None: """Add/Load Games cog.""" # Check does IGDB API key exist, if not, log warning and don't load cog - if not Tokens.igdb: - logger.warning("No IGDB API key. Not loading Games cog.") + if not Tokens.igdb_client_id: + logger.warning("No IGDB client ID. Not loading Games cog.") + return + if not Tokens.igdb_client_secret: + logger.warning("No IGDB client secret. Not loading Games cog.") return bot.add_cog(Games(bot)) diff --git a/bot/exts/evergreen/xkcd.py b/bot/exts/evergreen/xkcd.py new file mode 100644 index 00000000..d3224bfe --- /dev/null +++ b/bot/exts/evergreen/xkcd.py @@ -0,0 +1,89 @@ +import logging +import re +from random import randint +from typing import Dict, Optional, Union + +from discord import Embed +from discord.ext import tasks +from discord.ext.commands import Cog, Context, command + +from bot.bot import Bot +from bot.constants import Colours + +log = logging.getLogger(__name__) + +COMIC_FORMAT = re.compile(r"latest|[0-9]+") +BASE_URL = "https://xkcd.com" + + +class XKCD(Cog): + """Retrieving XKCD comics.""" + + def __init__(self, bot: Bot) -> None: + self.bot = bot + self.latest_comic_info: Dict[str, Union[str, int]] = {} + self.get_latest_comic_info.start() + + def cog_unload(self) -> None: + """Cancels refreshing of the task for refreshing the most recent comic info.""" + self.get_latest_comic_info.cancel() + + @tasks.loop(minutes=30) + async def get_latest_comic_info(self) -> None: + """Refreshes latest comic's information ever 30 minutes. Also used for finding a random comic.""" + async with self.bot.http_session.get(f"{BASE_URL}/info.0.json") as resp: + if resp.status == 200: + self.latest_comic_info = await resp.json() + else: + log.debug(f"Failed to get latest XKCD comic information. Status code {resp.status}") + + @command(name="xkcd") + async def fetch_xkcd_comics(self, ctx: Context, comic: Optional[str]) -> None: + """ + Getting an xkcd comic's information along with the image. + + To get a random comic, don't type any number as an argument. To get the latest, type 'latest'. + """ + embed = Embed(title=f"XKCD comic '{comic}'") + + embed.colour = Colours.soft_red + + if comic and (comic := re.match(COMIC_FORMAT, comic)) is None: + embed.description = "Comic parameter should either be an integer or 'latest'." + await ctx.send(embed=embed) + return + + comic = randint(1, self.latest_comic_info['num']) if comic is None else comic.group(0) + + if comic == "latest": + info = self.latest_comic_info + else: + async with self.bot.http_session.get(f"{BASE_URL}/{comic}/info.0.json") as resp: + if resp.status == 200: + info = await resp.json() + else: + embed.title = f"XKCD comic #{comic}" + embed.description = f"{resp.status}: Could not retrieve xkcd comic #{comic}." + log.debug(f"Retrieving xkcd comic #{comic} failed with status code {resp.status}.") + await ctx.send(embed=embed) + return + + embed.title = f"XKCD comic #{info['num']}" + + if info["img"][-3:] in ("jpg", "png", "gif"): + embed.set_image(url=info["img"]) + date = f"{info['year']}/{info['month']}/{info['day']}" + embed.set_footer(text=f"{date} - #{info['num']}, \'{info['safe_title']}\'") + embed.colour = Colours.soft_green + else: + embed.description = ( + "The selected comic is interactive, and cannot be displayed within an embed.\n" + f"Comic can be viewed [here](https://xkcd.com/{info['num']})." + ) + + await ctx.send(embed=embed) + + +def setup(bot: Bot) -> None: + """Loading the XKCD cog.""" + bot.add_cog(XKCD(bot)) diff --git a/bot/exts/halloween/spookynamerate.py b/bot/exts/halloween/spookynamerate.py new file mode 100644 index 00000000..e2950343 --- /dev/null +++ b/bot/exts/halloween/spookynamerate.py @@ -0,0 +1,401 @@ +import asyncio +import json +import random +from collections import defaultdict +from datetime import datetime, timedelta +from logging import getLogger +from os import getenv +from pathlib import Path +from typing import Dict, Union + +from async_rediscache import RedisCache +from discord import Embed, Reaction, TextChannel, User +from discord.colour import Colour +from discord.ext import tasks +from discord.ext.commands import Bot, Cog, Context, group + +from bot.constants import Channels, Client, Colours, Month +from bot.utils.decorators import InMonthCheckFailure + +logger = getLogger(__name__) + +EMOJIS_VAL = { + "\N{Jack-O-Lantern}": 1, + "\N{Ghost}": 2, + "\N{Skull and Crossbones}": 3, + "\N{Zombie}": 4, + "\N{Face Screaming In Fear}": 5, +} +ADDED_MESSAGES = [ + "Let's see if you win?", + ":jack_o_lantern: SPOOKY :jack_o_lantern:", + "If you got it, haunt it.", + "TIME TO GET YOUR SPOOKY ON! :skull:", +] +PING = "<@{id}>" + +EMOJI_MESSAGE = "\n".join([f"- {emoji} {val}" for emoji, val in EMOJIS_VAL.items()]) +HELP_MESSAGE_DICT = { + "title": "Spooky Name Rate", + "description": f"Help for the `{Client.prefix}spookynamerate` command", + "color": Colours.soft_orange, + "fields": [ + { + "name": "How to play", + "value": ( + "Everyday, the bot will post a random name, which you will need to spookify using your creativity.\n" + "You can rate each message according to how scary it is.\n" + "At the end of the day, the author of the message with most reactions will be the winner of the day.\n" + f"On a scale of 1 to {len(EMOJIS_VAL)}, the reactions order:\n" + f"{EMOJI_MESSAGE}" + ), + "inline": False, + }, + { + "name": "How do I add my spookified name?", + "value": f"Simply type `{Client.prefix}spookynamerate add my name`", + "inline": False, + }, + { + "name": "How do I *delete* my spookified name?", + "value": f"Simply type `{Client.prefix}spookynamerate delete`", + "inline": False, + }, + ], +} + + +class SpookyNameRate(Cog): + """ + A game that asks the user to spookify or halloweenify a name that is given everyday. + + It sends a random name everyday. The user needs to try and spookify it to his best ability and + send that name back using the `spookynamerate add entry` command + """ + + # This cache stores the message id of each added word along with a dictionary which contains the name the author + # added, the author's id, and the author's score (which is 0 by default) + messages = RedisCache() + + # The data cache stores small information such as the current name that is going on and whether it is the first time + # the bot is running + data = RedisCache() + debug = getenv('SPOOKYNAMERATE_DEBUG', False) # Enable if you do not want to limit the commands to October or if + # you do not want to wait till 12 UTC. Note: if debug is enabled and you run `.cogs reload spookynamerate`, it + # will automatically start the scoring and announcing the result (without waiting for 12, so do not expect it to.). + # Also, it won't wait for the two hours (when the poll closes). + + def __init__(self, bot: Bot) -> None: + self.bot = bot + + names_data = self.load_json( + Path("bot", "resources", "halloween", "spookynamerate_names.json") + ) + self.first_names = names_data["first_names"] + self.last_names = names_data["last_names"] + # the names are from https://www.mockaroo.com/ + + self.name = None + + self.bot.loop.create_task(self.load_vars()) + + self.first_time = None + self.poll = False + self.announce_name.start() + self.checking_messages = asyncio.Lock() + # Define an asyncio.Lock() to make sure the dictionary isn't changed + # when checking the messages for duplicate emojis' + + async def load_vars(self) -> None: + """Loads the variables that couldn't be loaded in __init__.""" + self.first_time = await self.data.get("first_time", True) + self.name = await self.data.get("name") + + @group(name="spookynamerate", invoke_without_command=True) + async def spooky_name_rate(self, ctx: Context) -> None: + """Get help on the Spooky Name Rate game.""" + await ctx.send(embed=Embed.from_dict(HELP_MESSAGE_DICT)) + + @spooky_name_rate.command(name="list", aliases=["all", "entries"]) + async def list_entries(self, ctx: Context) -> None: + """Send all the entries up till now in a single embed.""" + await ctx.send(embed=await self.get_responses_list(final=False)) + + @spooky_name_rate.command(name="name") + async def tell_name(self, ctx: Context) -> None: + """Tell the current random name.""" + if not self.poll: + await ctx.send(f"The name is **{self.name}**") + return + + await ctx.send( + f"The name ~~is~~ was **{self.name}**. The poll has already started, so you cannot " + "add an entry." + ) + + @spooky_name_rate.command(name="add", aliases=["register"]) + async def add_name(self, ctx: Context, *, name: str) -> None: + """Use this command to add/register your spookified name.""" + if self.poll: + logger.info(f"{ctx.message.author} tried to add a name, but the poll had already started.") + await ctx.send("Sorry, the poll has started! You can try and participate in the next round though!") + return + + message = ctx.message + + for data in (json.loads(user_data) for _, user_data in await self.messages.items()): + if data["author"] == message.author.id: + await ctx.send( + "But you have already added an entry! Type " + f"`{self.bot.command_prefix}spookynamerate " + "delete` to delete it, and then you can add it again" + ) + return + + elif data["name"] == name: + await ctx.send("TOO LATE. Someone has already added this name.") + return + + msg = await (await self.get_channel()).send(f"{message.author.mention} added the name {name!r}!") + + await self.messages.set( + msg.id, + json.dumps( + { + "name": name, + "author": message.author.id, + "score": 0, + } + ), + ) + + for emoji in EMOJIS_VAL: + await msg.add_reaction(emoji) + + logger.info(f"{message.author} added the name {name!r}") + + @spooky_name_rate.command(name="delete") + async def delete_name(self, ctx: Context) -> None: + """Delete the user's name.""" + if self.poll: + await ctx.send("You can't delete your name since the poll has already started!") + return + for message_id, data in await self.messages.items(): + data = json.loads(data) + + if ctx.author.id == data["author"]: + await self.messages.delete(message_id) + await ctx.send(f'Name deleted successfully ({data["name"]!r})!') + return + + await ctx.send( + f"But you don't have an entry... :eyes: Type `{self.bot.command_prefix}spookynamerate add your entry`" + ) + + @Cog.listener() + async def on_reaction_add(self, reaction: Reaction, user: User) -> None: + """Ensures that each user adds maximum one reaction.""" + if user.bot or not await self.messages.contains(reaction.message.id): + return + + async with self.checking_messages: # Acquire the lock so that the dictionary isn't reset while iterating. + if reaction.emoji in EMOJIS_VAL: + # create a custom counter + reaction_counter = defaultdict(int) + for msg_reaction in reaction.message.reactions: + async for reaction_user in msg_reaction.users(): + if reaction_user == self.bot.user: + continue + reaction_counter[reaction_user] += 1 + + if reaction_counter[user] > 1: + await user.send( + "Sorry, you have already added a reaction, " + "please remove your reaction and try again." + ) + await reaction.remove(user) + return + + @tasks.loop(hours=24.0) + async def announce_name(self) -> None: + """Announces the name needed to spookify every 24 hours and the winner of the previous game.""" + if not self.in_allowed_month(): + return + + channel = await self.get_channel() + + if self.first_time: + await channel.send( + "Okkey... Welcome to the **Spooky Name Rate Game**! It's a relatively simple game.\n" + f"Everyday, a random name will be sent in <#{Channels.community_bot_commands}> " + "and you need to try and spookify it!\nRegister your name using " + f"`{self.bot.command_prefix}spookynamerate add spookified name`" + ) + + await self.data.set("first_time", False) + self.first_time = False + + else: + if await self.messages.items(): + await channel.send(embed=await self.get_responses_list(final=True)) + self.poll = True + if not SpookyNameRate.debug: + await asyncio.sleep(2 * 60 * 60) # sleep for two hours + + logger.info("Calculating score") + for message_id, data in await self.messages.items(): + data = json.loads(data) + + msg = await channel.fetch_message(message_id) + score = 0 + for reaction in msg.reactions: + reaction_value = EMOJIS_VAL.get(reaction.emoji, 0) # get the value of the emoji else 0 + score += reaction_value * (reaction.count - 1) # multiply by the num of reactions + # subtract one, since one reaction was done by the bot + + logger.debug(f"{self.bot.get_user(data['author'])} got a score of {score}") + data["score"] = score + await self.messages.set(message_id, json.dumps(data)) + + # Sort the winner messages + winner_messages = sorted( + ((msg_id, json.loads(usr_data)) for msg_id, usr_data in await self.messages.items()), + key=lambda x: x[1]["score"], + reverse=True, + ) + + winners = [] + for i, winner in enumerate(winner_messages): + winners.append(winner) + if len(winner_messages) > i + 1: + if winner_messages[i + 1][1]["score"] != winner[1]["score"]: + break + elif len(winner_messages) == (i + 1) + 1: # The next element is the last element + if winner_messages[i + 1][1]["score"] != winner[1]["score"]: + break + + # one iteration is complete + await channel.send("Today's Spooky Name Rate Game ends now, and the winner(s) is(are)...") + + async with channel.typing(): + await asyncio.sleep(1) # give the drum roll feel + + if not winners: # There are no winners (no participants) + await channel.send("Hmm... Looks like no one participated! :cry:") + return + + score = winners[0][1]["score"] + congratulations = "to all" if len(winners) > 1 else PING.format(id=winners[0][1]["author"]) + names = ", ".join(f'{win[1]["name"]} ({PING.format(id=win[1]["author"])})' for win in winners) + + # display winners, their names and scores + await channel.send( + f"Congratulations {congratulations}!\n" + f"You have a score of {score}!\n" + f"Your name{ 's were' if len(winners) > 1 else 'was'}:\n{names}" + ) + + # Send random party emojis + party = (random.choice([":partying_face:", ":tada:"]) for _ in range(random.randint(1, 10))) + await channel.send(" ".join(party)) + + async with self.checking_messages: # Acquire the lock to delete the messages + await self.messages.clear() # reset the messages + + # send the next name + self.name = f"{random.choice(self.first_names)} {random.choice(self.last_names)}" + await self.data.set("name", self.name) + + await channel.send( + "Let's move on to the next name!\nAnd the next name is...\n" + f"**{self.name}**!\nTry to spookify that... :smirk:" + ) + + self.poll = False # accepting responses + + @announce_name.before_loop + async def wait_till_scheduled_time(self) -> None: + """Waits till the next day's 12PM if crossed it, otherwise waits till the same day's 12PM.""" + if SpookyNameRate.debug: + return + + now = datetime.utcnow() + if now.hour < 12: + twelve_pm = now.replace(hour=12, minute=0, second=0, microsecond=0) + time_left = twelve_pm - now + await asyncio.sleep(time_left.seconds) + return + + tomorrow_12pm = now + timedelta(days=1) + tomorrow_12pm = tomorrow_12pm.replace(hour=12, minute=0, second=0, microsecond=0) + await asyncio.sleep((tomorrow_12pm - now).seconds) + + async def get_responses_list(self, final: bool = False) -> Embed: + """Returns an embed containing the responses of the people.""" + channel = await self.get_channel() + + embed = Embed(color=Colour.red()) + + if await self.messages.items(): + if final: + embed.title = "Spooky Name Rate is about to end!" + embed.description = ( + "This Spooky Name Rate round is about to end in 2 hours! You can review " + "the entries below! Have you rated other's names?" + ) + else: + embed.title = "All the spookified names!" + embed.description = "See a list of all the entries entered by everyone!" + else: + embed.title = "No one has added an entry yet..." + + for message_id, data in await self.messages.items(): + data = json.loads(data) + + embed.add_field( + name=(self.bot.get_user(data["author"]) or await self.bot.fetch_user(data["author"])).name, + value=f"[{(data)['name']}](https://discord.com/channels/{Client.guild}/{channel.id}/{message_id})", + ) + + return embed + + async def get_channel(self) -> Union[TextChannel, None]: + """Gets the sir-lancebot-channel after waiting until ready.""" + await self.bot.wait_until_ready() + channel = self.bot.get_channel( + Channels.community_bot_commands + ) or await self.bot.fetch_channel(Channels.community_bot_commands) + if not channel: + logger.warning("Bot is unable to get the #seasonalbot-commands channel. Please check the channel ID.") + return channel + + @staticmethod + def load_json(file: Path) -> Dict[str, str]: + """Loads a JSON file and returns its contents.""" + with file.open("r", encoding="utf-8") as f: + return json.load(f) + + @staticmethod + def in_allowed_month() -> bool: + """Returns whether running in the limited month.""" + if SpookyNameRate.debug: + return True + + if not Client.month_override: + return datetime.utcnow().month == Month.OCTOBER + return Client.month_override == Month.OCTOBER + + def cog_check(self, ctx: Context) -> bool: + """A command to check whether the command is being called in October.""" + if not self.in_allowed_month(): + raise InMonthCheckFailure("You can only use these commands in October!") + return True + + def cog_unload(self) -> None: + """Stops the announce_name task.""" + self.announce_name.cancel() + + +def setup(bot: Bot) -> None: + """Loads the SpookyNameRate Cog.""" + bot.add_cog(SpookyNameRate(bot)) |