diff --git a/backend/adapters/services/steamgriddb.py b/backend/adapters/services/steamgriddb.py new file mode 100644 index 000000000..d4486b86a --- /dev/null +++ b/backend/adapters/services/steamgriddb.py @@ -0,0 +1,175 @@ +import http +import itertools +import json +from collections.abc import AsyncIterator, Collection +from typing import Literal, cast + +import aiohttp +import yarl +from adapters.services.steamgriddb_types import ( + SGDBDimension, + SGDBGame, + SGDBGrid, + SGDBGridList, + SGDBMime, + SGDBStyle, + SGDBTag, + SGDBType, +) +from aiohttp.client import ClientTimeout +from config import STEAMGRIDDB_API_KEY +from exceptions.endpoint_exceptions import SGDBInvalidAPIKeyException +from logger.logger import log +from utils.context import ctx_aiohttp_session + + +async def auth_middleware( + req: aiohttp.ClientRequest, handler: aiohttp.ClientHandlerType +) -> aiohttp.ClientResponse: + """SteamGridDB API authentication mechanism.""" + req.headers["Authorization"] = f"Bearer {STEAMGRIDDB_API_KEY}" + return await handler(req) + + +class SteamGridDBService: + """Service to interact with the SteamGridDB API. + + Reference: https://www.steamgriddb.com/api/v2 + """ + + def __init__( + self, + base_url: str | None = None, + ) -> None: + self.url = yarl.URL(base_url or "https://steamgriddb.com/api/v2") + + async def _request(self, url: str, request_timeout: int = 120) -> dict: + aiohttp_session = ctx_aiohttp_session.get() + log.debug( + "API request: URL=%s, Timeout=%s", + url, + request_timeout, + ) + try: + res = await aiohttp_session.get( + url, + middlewares=(auth_middleware,), + timeout=ClientTimeout(total=request_timeout), + ) + res.raise_for_status() + return await res.json() + except aiohttp.ClientResponseError as exc: + if exc.status == http.HTTPStatus.UNAUTHORIZED: + raise SGDBInvalidAPIKeyException from exc + # Log the error and return an empty dict if the request fails with a different code + log.error(exc) + return {} + except json.decoder.JSONDecodeError as exc: + log.error( + "Failed to decode JSON response from SteamGridDB: %s", + str(exc), + ) + return {} + + async def get_grids_for_game( + self, + game_id: int, + *, + styles: Collection[SGDBStyle] | None = None, + dimensions: Collection[SGDBDimension] | None = None, + mimes: Collection[SGDBMime] | None = None, + types: Collection[SGDBType] | None = None, + any_of_tags: Collection[SGDBTag] | None = None, + is_nsfw: bool | Literal["any"] | None = None, + is_humor: bool | Literal["any"] | None = None, + is_epilepsy: bool | Literal["any"] | None = None, + limit: int | None = None, + page_number: int | None = None, + ) -> SGDBGridList: + """Retrieve grids by game ID. + + Reference: https://www.steamgriddb.com/api/v2#tag/GRIDS/operation/getGridsByGameId + """ + params: dict[str, list[str]] = {} + if styles: + params["styles"] = [",".join(styles)] + if dimensions: + params["dimensions"] = [",".join(dimensions)] + if mimes: + params["mimes"] = [",".join(mimes)] + if types: + params["types"] = [",".join(types)] + if any_of_tags: + params["oneoftag"] = [",".join(any_of_tags)] + if is_nsfw is not None: + params["nsfw"] = [str(is_nsfw).lower()] + if is_humor is not None: + params["humor"] = [str(is_humor).lower()] + if is_epilepsy is not None: + params["epilepsy"] = [str(is_epilepsy).lower()] + if limit is not None: + params["limit"] = [str(limit)] + if page_number is not None: + params["page"] = [str(page_number)] + + url = self.url.joinpath("grids/game", str(game_id)).with_query(**params) + response = await self._request(str(url)) + if not response: + return SGDBGridList( + page=0, + total=0, + limit=limit or 50, + data=[], + ) + return cast(SGDBGridList, response) + + async def iter_grids_for_game( + self, + game_id: int, + *, + styles: Collection[SGDBStyle] | None = None, + dimensions: Collection[SGDBDimension] | None = None, + mimes: Collection[SGDBMime] | None = None, + types: Collection[SGDBType] | None = None, + any_of_tags: Collection[SGDBTag] | None = None, + is_nsfw: bool | Literal["any"] | None = None, + is_humor: bool | Literal["any"] | None = None, + is_epilepsy: bool | Literal["any"] | None = None, + ) -> AsyncIterator[SGDBGrid]: + """Iterate through grids by game ID. + + Reference: https://www.steamgriddb.com/api/v2#tag/GRIDS/operation/getGridsByGameId + """ + page_size = 50 # Maximum page size for this endpoint. + offset = 0 + + for page_number in itertools.count(start=0): + response = await self.get_grids_for_game( + game_id, + styles=styles, + dimensions=dimensions, + mimes=mimes, + types=types, + any_of_tags=any_of_tags, + is_nsfw=is_nsfw, + is_humor=is_humor, + is_epilepsy=is_epilepsy, + limit=page_size, + page_number=page_number, + ) + results = response["data"] + for result in results: + yield result + + offset += len(results) + if len(results) < page_size or offset >= response["total"]: + break + + async def search_games(self, term: str) -> list[SGDBGame]: + """Search for games by name. + + Reference: https://www.steamgriddb.com/api/v2#tag/SEARCH/operation/searchGrids + """ + url = self.url.joinpath("search/autocomplete", term) + response = await self._request(str(url)) + return cast(list[SGDBGame], response.get("data", [])) diff --git a/backend/adapters/services/steamgriddb_types.py b/backend/adapters/services/steamgriddb_types.py new file mode 100644 index 000000000..bade565f4 --- /dev/null +++ b/backend/adapters/services/steamgriddb_types.py @@ -0,0 +1,75 @@ +import enum +from typing import Mapping, TypedDict + + +@enum.unique +class SGDBStyle(enum.StrEnum): + ALTERNATE = "alternate" + BLURRED = "blurred" + WHITE_LOGO = "white_logo" + MATERIAL = "material" + NO_LOGO = "no_logo" + + +@enum.unique +class SGDBDimension(enum.StrEnum): + STEAM_HORIZONTAL = "460x215" + STEAM_HORIZONTAL_2X = "920x430" + STEAM_VERTICAL = "600x900" + GOG_GALAXY_TILE = "342x482" + GOG_GALAXY_COVER = "660x930" + SQUARE_512 = "512x512" + SQUARE_1024 = "1024x1024" + + +@enum.unique +class SGDBMime(enum.StrEnum): + PNG = "image/png" + JPEG = "image/jpeg" + WEBP = "image/webp" + + +@enum.unique +class SGDBType(enum.StrEnum): + STATIC = "static" + ANIMATED = "animated" + + +@enum.unique +class SGDBTag(enum.StrEnum): + HUMOR = "humor" + NSFW = "nsfw" + EPILEPSY = "epilepsy" + + +class PaginatedResponse[T: Mapping](TypedDict): + page: int + total: int + limit: int + data: list[T] + + +class SGDBGridAuthor(TypedDict): + name: str + steam64: str + avatar: str + + +class SGDBGrid(TypedDict): + id: int + score: int + style: SGDBStyle + url: str + thumb: str + tags: list[str] + author: SGDBGridAuthor + + +class SGDBGame(TypedDict): + id: int + name: str + types: list[str] + verified: bool + + +SGDBGridList = PaginatedResponse[SGDBGrid] diff --git a/backend/handler/metadata/sgdb_handler.py b/backend/handler/metadata/sgdb_handler.py index 1ea113c3e..77c0baf3f 100644 --- a/backend/handler/metadata/sgdb_handler.py +++ b/backend/handler/metadata/sgdb_handler.py @@ -1,79 +1,30 @@ import asyncio -import itertools -import json from typing import Any, Final +from adapters.services.steamgriddb import SteamGridDBService +from adapters.services.steamgriddb_types import SGDBDimension, SGDBType from config import STEAMGRIDDB_API_KEY -from exceptions.endpoint_exceptions import SGDBInvalidAPIKeyException from logger.logger import log -from utils.context import ctx_httpx_client from .base_hander import MetadataHandler # Used to display the Mobygames API status in the frontend STEAMGRIDDB_API_ENABLED: Final = bool(STEAMGRIDDB_API_KEY) -# SteamGridDB dimensions -STEAMVERTICAL: Final = "600x900" -GALAXY342: Final = "342x482" -GALAXY660: Final = "660x930" -SQUARE512: Final = "512x512" -SQUARE1024: Final = "1024x1024" - -# SteamGridDB types -STATIC: Final = "static" -ANIMATED: Final = "animated" - -SGDB_API_COVER_LIMIT: Final = 50 - class SGDBBaseHandler(MetadataHandler): def __init__(self) -> None: - self.BASE_URL = "https://www.steamgriddb.com/api/v2" - self.search_endpoint = f"{self.BASE_URL}/search/autocomplete" - self.grid_endpoint = f"{self.BASE_URL}/grids/game" - self.headers = { - "Authorization": f"Bearer {STEAMGRIDDB_API_KEY}", - "Accept": "*/*", - } - self.masked_headers = self._mask_sensitive_values(self.headers) - self.timeout = 120 + self.sgdb_service = SteamGridDBService() async def get_details(self, search_term: str) -> list[dict[str, Any]]: - httpx_client = ctx_httpx_client.get() - url = f"{self.search_endpoint}/{search_term}" - - log.debug( - "API request: Method=GET, URL=%s, Headers=%s, Timeout=%s", - url, - self.masked_headers, - self.timeout, - ) - - res = await httpx_client.get( - url, - headers=self.headers, - timeout=self.timeout, - ) - - try: - if res.status_code == 401: - raise SGDBInvalidAPIKeyException - search_response = res.json() - except json.decoder.JSONDecodeError as exc: - log.error( - "Failed to decode JSON response from SteamGridDB: %s", - str(exc), - ) - return [] - - if len(search_response["data"]) == 0: + games = await self.sgdb_service.search_games(term=search_term) + if not games: log.warning(f"Could not find '{search_term}' on SteamGridDB") return [] tasks = [ self._get_game_covers(game_id=game["id"], game_name=game["name"]) - for game in search_response["data"] + for game in games ] results = await asyncio.gather(*tasks) @@ -82,48 +33,23 @@ class SGDBBaseHandler(MetadataHandler): async def _get_game_covers( self, game_id: int, game_name: str ) -> dict[str, Any] | None: - httpx_client = ctx_httpx_client.get() - game_covers = [] - - for page in itertools.count(start=0): - url = f"{self.grid_endpoint}/{game_id}" - params = { - "dimensions": f"{STEAMVERTICAL},{GALAXY342},{GALAXY660},{SQUARE512},{SQUARE1024}", - "types": f"{STATIC},{ANIMATED}", - "limit": SGDB_API_COVER_LIMIT, - "page": page, - } - - log.debug( - "API request: Method=GET, URL=%s, Headers=%s, Params=%s, Timeout=%s", - url, - self.masked_headers, - params, - self.timeout, + game_covers = [ + cover + async for cover in self.sgdb_service.iter_grids_for_game( + game_id=game_id, + dimensions=( + SGDBDimension.STEAM_VERTICAL, + SGDBDimension.GOG_GALAXY_TILE, + SGDBDimension.GOG_GALAXY_COVER, + SGDBDimension.SQUARE_512, + SGDBDimension.SQUARE_1024, + ), + types=( + SGDBType.STATIC, + SGDBType.ANIMATED, + ), ) - - res = await httpx_client.get( - url, - headers=self.headers, - timeout=self.timeout, - params=params, - ) - - try: - covers_response = res.json() - except json.decoder.JSONDecodeError as exc: - log.error( - "Failed to decode JSON response from SteamGridDB: %s", - str(exc), - ) - return None - - page_covers = covers_response["data"] - game_covers.extend(page_covers) - - if len(page_covers) < SGDB_API_COVER_LIMIT: - break - + ] if not game_covers: return None