aboutsummaryrefslogtreecommitdiffstats
path: root/bot/exts/evergreen
diff options
context:
space:
mode:
Diffstat (limited to 'bot/exts/evergreen')
-rw-r--r--bot/exts/evergreen/8bitify.py2
-rw-r--r--bot/exts/evergreen/battleship.py2
-rw-r--r--bot/exts/evergreen/game.py83
-rw-r--r--bot/exts/evergreen/xkcd.py89
4 files changed, 161 insertions, 15 deletions
diff --git a/bot/exts/evergreen/8bitify.py b/bot/exts/evergreen/8bitify.py
index c048d9bf..54e68f80 100644
--- a/bot/exts/evergreen/8bitify.py
+++ b/bot/exts/evergreen/8bitify.py
@@ -19,7 +19,7 @@ class EightBitify(commands.Cog):
@staticmethod
def quantize(image: Image) -> Image:
"""Reduces colour palette to 256 colours."""
- return image.quantize(colors=32)
+ return image.quantize()
@commands.command(name="8bitify")
async def eightbit_command(self, ctx: commands.Context) -> None:
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))
generated by cgit v1.2.3 (git 2.39.1) at 2025-10-24 07:39:01 +0000