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/error_handler.py4
-rw-r--r--bot/exts/evergreen/game.py83
-rw-r--r--bot/exts/evergreen/snakes/_snakes_cog.py128
5 files changed, 132 insertions, 87 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/error_handler.py b/bot/exts/evergreen/error_handler.py
index 6e518435..99af1519 100644
--- a/bot/exts/evergreen/error_handler.py
+++ b/bot/exts/evergreen/error_handler.py
@@ -42,8 +42,8 @@ class CommandErrorHandler(commands.Cog):
@commands.Cog.listener()
async def on_command_error(self, ctx: commands.Context, error: commands.CommandError) -> None:
"""Activates when a command opens an error."""
- if hasattr(ctx.command, 'on_error'):
- logging.debug("A command error occured but the command had it's own error handler.")
+ if getattr(error, 'handled', False):
+ logging.debug(f"Command {ctx.command} had its error already handled locally; ignoring.")
return
error = getattr(error, 'original', error)
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/snakes/_snakes_cog.py b/bot/exts/evergreen/snakes/_snakes_cog.py
index 70bb0e73..d5e4f206 100644
--- a/bot/exts/evergreen/snakes/_snakes_cog.py
+++ b/bot/exts/evergreen/snakes/_snakes_cog.py
@@ -15,7 +15,8 @@ import aiohttp
import async_timeout
from PIL import Image, ImageDraw, ImageFont
from discord import Colour, Embed, File, Member, Message, Reaction
-from discord.ext.commands import BadArgument, Bot, Cog, CommandError, Context, bot_has_permissions, group
+from discord.errors import HTTPException
+from discord.ext.commands import Bot, Cog, CommandError, Context, bot_has_permissions, group
from bot.constants import ERROR_REPLIES, Tokens
from bot.exts.evergreen.snakes import _utils as utils
@@ -151,6 +152,7 @@ class Snakes(Cog):
self.snake_idioms = utils.get_resource("snake_idioms")
self.snake_quizzes = utils.get_resource("snake_quiz")
self.snake_facts = utils.get_resource("snake_facts")
+ self.num_movie_pages = None
# region: Helper methods
@staticmethod
@@ -739,71 +741,68 @@ class Snakes(Cog):
@snakes_group.command(name='movie')
async def movie_command(self, ctx: Context) -> None:
"""
- Gets a random snake-related movie from OMDB.
+ Gets a random snake-related movie from TMDB.
Written by Samuel.
Modified by gdude.
+ Modified by Will Da Silva.
"""
- url = "http://www.omdbapi.com/"
- page = random.randint(1, 27)
+ # Initially 8 pages are fetched. The actual number of pages is set after the first request.
+ page = random.randint(1, self.num_movie_pages or 8)
- response = await self.bot.http_session.get(
- url,
- params={
- "s": "snake",
- "page": page,
- "type": "movie",
- "apikey": Tokens.omdb
- }
- )
- data = await response.json()
- movie = random.choice(data["Search"])["imdbID"]
-
- response = await self.bot.http_session.get(
- url,
- params={
- "i": movie,
- "apikey": Tokens.omdb
- }
- )
- data = await response.json()
-
- embed = Embed(
- title=data["Title"],
- color=SNAKE_COLOR
- )
-
- del data["Response"], data["imdbID"], data["Title"]
-
- for key, value in data.items():
- if not value or value == "N/A" or key in ("Response", "imdbID", "Title", "Type"):
- continue
+ async with ctx.typing():
+ response = await self.bot.http_session.get(
+ "https://api.themoviedb.org/3/search/movie",
+ params={
+ "query": "snake",
+ "page": page,
+ "language": "en-US",
+ "api_key": Tokens.tmdb,
+ }
+ )
+ data = await response.json()
+ if self.num_movie_pages is None:
+ self.num_movie_pages = data["total_pages"]
+ movie = random.choice(data["results"])["id"]
+
+ response = await self.bot.http_session.get(
+ f"https://api.themoviedb.org/3/movie/{movie}",
+ params={
+ "language": "en-US",
+ "api_key": Tokens.tmdb,
+ }
+ )
+ data = await response.json()
- if key == "Ratings": # [{'Source': 'Internet Movie Database', 'Value': '7.6/10'}]
- rating = random.choice(value)
+ embed = Embed(title=data["title"], color=SNAKE_COLOR)
- if rating["Source"] != "Internet Movie Database":
- embed.add_field(name=f"Rating: {rating['Source']}", value=rating["Value"])
+ if data["poster_path"] is not None:
+ embed.set_image(url=f"https://images.tmdb.org/t/p/original{data['poster_path']}")
- continue
+ if data["overview"]:
+ embed.add_field(name="Overview", value=data["overview"])
- if key == "Poster":
- embed.set_image(url=value)
- continue
+ if data["release_date"]:
+ embed.add_field(name="Release Date", value=data["release_date"])
- elif key == "imdbRating":
- key = "IMDB Rating"
+ if data["genres"]:
+ embed.add_field(name="Genres", value=", ".join([x["name"] for x in data["genres"]]))
- elif key == "imdbVotes":
- key = "IMDB Votes"
+ if data["vote_count"]:
+ embed.add_field(name="Rating", value=f"{data['vote_average']}/10 ({data['vote_count']} votes)", inline=True)
- embed.add_field(name=key, value=value, inline=True)
+ if data["budget"] and data["revenue"]:
+ embed.add_field(name="Budget", value=data["budget"], inline=True)
+ embed.add_field(name="Revenue", value=data["revenue"], inline=True)
- embed.set_footer(text="Data provided by the OMDB API")
+ embed.set_footer(text="This product uses the TMDb API but is not endorsed or certified by TMDb.")
+ embed.set_thumbnail(url="https://i.imgur.com/LtFtC8H.png")
- await ctx.channel.send(
- embed=embed
- )
+ try:
+ await ctx.channel.send(embed=embed)
+ except HTTPException as err:
+ await ctx.channel.send("An error occurred while fetching a snake-related movie!")
+ raise err from None
@snakes_group.command(name='quiz')
@locked()
@@ -1126,26 +1125,15 @@ class Snakes(Cog):
# endregion
# region: Error handlers
- @get_command.error
@card_command.error
- @video_command.error
async def command_error(self, ctx: Context, error: CommandError) -> None:
"""Local error handler for the Snake Cog."""
- embed = Embed()
- embed.colour = Colour.red()
-
- if isinstance(error, BadArgument):
- embed.description = str(error)
- embed.title = random.choice(ERROR_REPLIES)
-
- elif isinstance(error, OSError):
- log.error(f"snake_card encountered an OSError: {error} ({error.original})")
+ original_error = getattr(error, "original", None)
+ if isinstance(original_error, OSError):
+ error.handled = True
+ embed = Embed()
+ embed.colour = Colour.red()
+ log.error(f"snake_card encountered an OSError: {error} ({original_error})")
embed.description = "Could not generate the snake card! Please try again."
embed.title = random.choice(ERROR_REPLIES)
-
- else:
- log.error(f"Unhandled tag command error: {error} ({error.original})")
- return
-
- await ctx.send(embed=embed)
- # endregion
+ await ctx.send(embed=embed)
generated by cgit v1.2.3 (git 2.39.1) at 2025-10-26 11:54:46 +0000