diff --git a/backend/config/__init__.py b/backend/config/__init__.py index 500ae9571..c713bf73c 100644 --- a/backend/config/__init__.py +++ b/backend/config/__init__.py @@ -133,6 +133,9 @@ def _get_env(var: str, fallback: str | None = None) -> str | None: _get_env("FLASHPOINT_API_ENABLED") ) +# RAWG +RAWG_API_KEY: Final[str | None] = _get_env("RAWG_API_KEY") + # HOWLONGTOBEAT HLTB_API_ENABLED: Final[bool] = safe_str_to_bool(_get_env("HLTB_API_ENABLED")) diff --git a/backend/handler/metadata/__init__.py b/backend/handler/metadata/__init__.py index 7a64e6e0c..bdd7eeea4 100644 --- a/backend/handler/metadata/__init__.py +++ b/backend/handler/metadata/__init__.py @@ -8,6 +8,7 @@ from .moby_handler import MobyGamesHandler from .playmatch_handler import PlaymatchHandler from .ra_handler import RAHandler +from .rawg_handler import RAWGHandler from .sgdb_handler import SGDBBaseHandler from .ss_handler import SSHandler from .tgdb_handler import TGDBHandler @@ -25,3 +26,4 @@ meta_flashpoint_handler = FlashpointHandler() meta_gamelist_handler = GamelistHandler() meta_hltb_handler = HLTBHandler() +meta_rawg_handler = RAWGHandler() diff --git a/backend/handler/metadata/rawg_handler.py b/backend/handler/metadata/rawg_handler.py new file mode 100644 index 000000000..3a5193c79 --- /dev/null +++ b/backend/handler/metadata/rawg_handler.py @@ -0,0 +1,279 @@ +import re +from typing import Final, NotRequired, TypedDict + +import httpx +import pydash +from config import RAWG_API_KEY +from logger.logger import log +from utils.context import ctx_httpx_client + +from .base_handler import BaseRom, MetadataHandler +from .base_handler import UniversalPlatformSlug as UPS + +# Filename tag like (rawg-12345), matching the convention the other handlers +# use so an operator can pin a match by hand. +RAWG_TAG_REGEX = re.compile(r"\(rawg-(\d+)\)", re.IGNORECASE) + +RAWG_API_URL: Final = "https://api.rawg.io/api" + + +class RAWGPlatform(TypedDict): + slug: str + rawg_slug: str | None + name: NotRequired[str] + + +class RAWGMetadata(TypedDict): + rawg_score: str + genres: list[str] + esrb_rating: str + developers: list[str] + publishers: list[str] + first_release_date: int | None + + +class RAWGRom(BaseRom): + rawg_id: int | None + rawg_metadata: NotRequired[RAWGMetadata] + + +def extract_metadata_from_rawg_rom(rom: dict) -> RAWGMetadata: + return RAWGMetadata( + { + "rawg_score": str(rom.get("rating", "")), + "genres": [g["name"] for g in rom.get("genres", []) if g.get("name")], + "esrb_rating": pydash.get(rom, "esrb_rating.name", "") or "", + "developers": [ + d["name"] for d in rom.get("developers", []) or [] if d.get("name") + ], + "publishers": [ + p["name"] for p in rom.get("publishers", []) or [] if p.get("name") + ], + "first_release_date": rom.get("released_timestamp"), + } + ) + + +# RomM's universal slug -> RAWG's platform slug. +# +# Only mapped where RAWG actually has the platform. RAWG is a modern and PC +# focused database, so a good deal of the retro catalogue is simply absent -- +# mapping those to a near neighbour would return confidently wrong matches, +# which is worse than returning nothing. +SLUG_TO_RAWG_SLUG: dict[UPS, str] = { + UPS._3DO: "3do", + UPS.AMIGA: "commodore-amiga", + UPS.ANDROID: "android", + UPS.ATARI2600: "atari-2600", + UPS.ATARI5200: "atari-5200", + UPS.ATARI7800: "atari-7800", + UPS.ATARI_ST: "atari-st", + UPS.NEW_NINTENDON3DS: "nintendo-3ds", + UPS.DC: "dreamcast", + UPS.DOS: "pc", + UPS.GB: "game-boy", + UPS.GBA: "game-boy-advance", + UPS.GBC: "game-boy-color", + UPS.GENESIS: "genesis", + UPS.IOS: "ios", + UPS.JAGUAR: "jaguar", + UPS.LINUX: "linux", + UPS.LYNX: "atari-lynx", + UPS.MAC: "macos", + UPS.N64: "nintendo-64", + UPS.NDS: "nintendo-ds", + UPS.NEO_GEO_CD: "neogeo", + UPS.NES: "nes", + UPS.NGC: "gamecube", + UPS.PS2: "playstation2", + UPS.PS3: "playstation3", + UPS.PS4: "playstation4", + UPS.PS5: "playstation5", + UPS.PSP: "psp", + UPS.PSVITA: "ps-vita", + UPS.PSX: "playstation1", + UPS.SATURN: "sega-saturn", + UPS.SEGA32: "sega-32x", + UPS.SEGACD: "sega-cd", + UPS.SMS: "sega-master-system", + UPS.SNES: "snes", + UPS.SWITCH: "nintendo-switch", + UPS.WII: "wii", + UPS.WIIU: "wii-u", + UPS.WIN: "pc", + UPS.XBOX: "xbox-old", + UPS.XBOX360: "xbox360", + UPS.XBOXONE: "xone", + UPS.N3DS: "nintendo-3ds", +} + + +class RAWGHandler(MetadataHandler): + """ + Handler for RAWG.io, a large general-purpose video game database. + + RAWG is strongest on modern and PC titles and thinner on the retro + catalogue than IGDB, MobyGames or ScreenScraper, so it is offered as an + additional source rather than a replacement for any of them. Its free tier + is generous, which makes it a reasonable first source for anybody who has + not obtained keys elsewhere. + """ + + def __init__(self) -> None: + self.search_endpoint = f"{RAWG_API_URL}/games" + self.min_similarity_score: Final = 0.7 + + @classmethod + def is_enabled(cls) -> bool: + return bool(RAWG_API_KEY) + + async def heartbeat(self) -> bool: + if not self.is_enabled(): + return False + + try: + response = await self._request(self.search_endpoint, {"page_size": 1}) + except Exception as e: + log.error("Error checking RAWG API: %s", e) + return False + + return bool(response) + + @staticmethod + def extract_rawg_id_from_filename(fs_name: str) -> int | None: + """Extract a RAWG id from a filename tag like (rawg-12345).""" + match = RAWG_TAG_REGEX.search(fs_name) + return int(match.group(1)) if match else None + + async def _request(self, url: str, params: dict) -> dict: + httpx_client = ctx_httpx_client.get() + try: + res = await httpx_client.get( + url, + params={**params, "key": RAWG_API_KEY}, + timeout=120, + ) + res.raise_for_status() + return res.json() + except httpx.NetworkError as exc: + log.critical("Connection error: can't connect to RAWG", exc_info=True) + raise exc + except httpx.HTTPStatusError as err: + # RAWG answers 401 for a bad key and 404 for an unknown id. Neither + # is a reason to fail a whole scan -- an empty result lets the next + # handler in the chain try. + if err.response.status_code == 401: + log.error("RAWG rejected the API key") + return {} + except httpx.TimeoutException: + log.debug("Request to RAWG timed out") + return {} + + def get_platform(self, slug: str) -> RAWGPlatform: + platform = SLUG_TO_RAWG_SLUG.get(slug.lower(), None) # type: ignore[arg-type] + if not platform: + return RAWGPlatform(rawg_slug=None, slug=slug) + + return RAWGPlatform(rawg_slug=platform, slug=slug) + + async def get_rom(self, fs_name: str, platform_rawg_slug: str) -> RAWGRom: + fallback_rom = RAWGRom(rawg_id=None) + + if not self.is_enabled(): + return fallback_rom + + # A pinned id in the filename is an explicit instruction and beats any + # amount of searching. + rawg_id = self.extract_rawg_id_from_filename(fs_name) + if rawg_id: + return await self.get_rom_by_id(rawg_id) + + if not platform_rawg_slug: + return fallback_rom + + search_term = self.normalize_search_term(fs_name) + if not search_term: + return fallback_rom + + res = await self._request( + self.search_endpoint, + { + "search": search_term, + "platforms_slug": platform_rawg_slug, + "page_size": 10, + }, + ) + results = res.get("results", []) or [] + if not results: + return fallback_rom + + games_by_name = {game["name"]: game for game in results if game.get("name")} + best_match, best_score = self.find_best_match( + search_term, + list(games_by_name.keys()), + self.min_similarity_score, + ) + if not best_match: + return fallback_rom + + log.debug( + f"Found RAWG match for '{search_term}' -> '{best_match}' " + f"(score: {best_score:.3f})" + ) + return await self.get_rom_by_id(games_by_name[best_match]["id"]) + + async def get_rom_by_id(self, rawg_id: int) -> RAWGRom: + if not self.is_enabled(): + return RAWGRom(rawg_id=None) + + res = await self._request(f"{self.search_endpoint}/{rawg_id}", {}) + if not res or not res.get("name"): + return RAWGRom(rawg_id=None) + + return RAWGRom( + rawg_id=rawg_id, + name=res["name"], + summary=res.get("description_raw", "") or "", + url_cover=self.normalize_cover_url(res.get("background_image", "") or ""), + url_screenshots=[ + self.normalize_cover_url(shot["image"]) + for shot in (res.get("short_screenshots") or []) + if shot.get("image") + ], + rawg_metadata=extract_metadata_from_rawg_rom(res), + ) + + async def get_matched_roms_by_name( + self, search_term: str, platform_rawg_slug: str | None + ) -> list[RAWGRom]: + """Candidates for a manual match, unfiltered by similarity. + + The operator is choosing, so a low-scoring result is theirs to reject + rather than something to hide from them. + """ + if not self.is_enabled(): + return [] + + params: dict = {"search": search_term, "page_size": 20} + if platform_rawg_slug: + params["platforms_slug"] = platform_rawg_slug + + res = await self._request(self.search_endpoint, params) + return [ + RAWGRom( + rawg_id=game["id"], + name=game["name"], + summary="", + url_cover=self.normalize_cover_url( + game.get("background_image", "") or "" + ), + url_screenshots=[ + self.normalize_cover_url(shot["image"]) + for shot in (game.get("short_screenshots") or []) + if shot.get("image") + ], + rawg_metadata=extract_metadata_from_rawg_rom(game), + ) + for game in (res.get("results") or []) + if game.get("id") and game.get("name") + ] diff --git a/backend/tests/handler/metadata/test_rawg_handler.py b/backend/tests/handler/metadata/test_rawg_handler.py new file mode 100644 index 000000000..aedd347cd --- /dev/null +++ b/backend/tests/handler/metadata/test_rawg_handler.py @@ -0,0 +1,93 @@ +from unittest.mock import patch + +import pytest +from handler.metadata.base_handler import UniversalPlatformSlug as UPS +from handler.metadata.rawg_handler import ( + SLUG_TO_RAWG_SLUG, + RAWGHandler, + extract_metadata_from_rawg_rom, +) + + +@pytest.fixture +def rawg_handler(): + return RAWGHandler() + + +def test_is_disabled_without_a_key(rawg_handler): + with patch("handler.metadata.rawg_handler.RAWG_API_KEY", None): + assert rawg_handler.is_enabled() is False + + with patch("handler.metadata.rawg_handler.RAWG_API_KEY", "some-key"): + assert rawg_handler.is_enabled() is True + + +@pytest.mark.asyncio +async def test_get_rom_returns_a_fallback_when_disabled(rawg_handler): + with patch("handler.metadata.rawg_handler.RAWG_API_KEY", None): + rom = await rawg_handler.get_rom("Chrono Trigger (USA).sfc", "snes") + + assert rom["rawg_id"] is None + + +@pytest.mark.asyncio +async def test_get_rom_returns_a_fallback_without_a_platform(rawg_handler): + """An unmapped platform must not become an unfiltered search across every + platform RAWG knows -- that returns a confident match for the wrong + system.""" + with patch("handler.metadata.rawg_handler.RAWG_API_KEY", "some-key"): + rom = await rawg_handler.get_rom("Chrono Trigger (USA).sfc", "") + + assert rom["rawg_id"] is None + + +def test_get_platform_maps_known_slugs(rawg_handler): + assert rawg_handler.get_platform("snes")["rawg_slug"] == "snes" + assert rawg_handler.get_platform("psx")["rawg_slug"] == "playstation1" + assert rawg_handler.get_platform("genesis")["rawg_slug"] == "genesis" + + +def test_get_platform_returns_none_for_an_unmapped_slug(rawg_handler): + """RAWG is thin on retro, so plenty of RomM platforms have no counterpart. + Returning None keeps the handler out of the way instead of guessing.""" + platform = rawg_handler.get_platform("neo-geo-pocket-color") + + assert platform["rawg_slug"] is None + assert platform["slug"] == "neo-geo-pocket-color" + + +def test_get_platform_is_case_insensitive(rawg_handler): + assert rawg_handler.get_platform("SNES")["rawg_slug"] == "snes" + + +def test_every_mapped_platform_is_a_real_universal_slug(): + """A typo here maps a platform nobody can reach, and it fails silently.""" + for slug in SLUG_TO_RAWG_SLUG: + assert isinstance(slug, UPS) + + +def test_extract_rawg_id_from_filename(rawg_handler): + assert rawg_handler.extract_rawg_id_from_filename("Game (rawg-12345).sfc") == 12345 + assert rawg_handler.extract_rawg_id_from_filename("Game (RAWG-99).sfc") == 99 + assert rawg_handler.extract_rawg_id_from_filename("Game (USA).sfc") is None + + +def test_extract_metadata_handles_a_sparse_response(): + """RAWG omits fields freely -- `esrb_rating` is null for most retro + titles, and `developers` is absent from search results entirely.""" + metadata = extract_metadata_from_rawg_rom( + {"rating": 4.5, "genres": [{"name": "RPG"}], "esrb_rating": None} + ) + + assert metadata["rawg_score"] == "4.5" + assert metadata["genres"] == ["RPG"] + assert metadata["esrb_rating"] == "" + assert metadata["developers"] == [] + assert metadata["publishers"] == [] + + +def test_extract_metadata_of_an_empty_response(): + metadata = extract_metadata_from_rawg_rom({}) + + assert metadata["genres"] == [] + assert metadata["first_release_date"] is None