Merge pull request #3381 from DevYukine/feat/playmatch-multi-provider

feat(playmatch): add SteamGridDB, ScreenScraper, MobyGames & Launchbox hash support

Author: gantoine

backend/handler/metadata/playmatch_handler.py

@@ -1,6 +1,6 @@
 import json
 from enum import Enum
-from typing import NotRequired, TypedDict
+from typing import Final, NotRequired, TypedDict
 
 import httpx
 import yarl
@@ -16,31 +16,53 @@
 
 class PlaymatchProvider(str, Enum):
     IGDB = "IGDB"
+    STEAM_GRID_DB = "SteamGridDB"
+    SCREEN_SCRAPER = "ScreenScraper"
+    MOBY_GAMES = "MobyGames"
+    LAUNCH_BOX = "LaunchBox"
+    EMU_READY = "EmuReady"
+    OPEN_VGDB = "OpenVGDB"
+
+
+# Tag is the uppercased Playmatch MetadataProvider name.
+# Playmatch parses it case-insensitively but spacing must match.
+# Tags Playmatch doesn't yet know are kept so older RomM clients keep
+# submitting the right tag once Playmatch adds support.
+PLAYMATCH_TAG_TO_ATTR: Final[dict[str, str]] = {
+    "IGDB": "igdb_id",
+    "MOBYGAMES": "moby_id",
+    "SCREENSCRAPER": "ss_id",
+    "RETRO_ACHIEVEMENTS": "ra_id",
+    "LAUNCHBOX": "launchbox_id",
+    "HASHEOUS": "hasheous_id",
+    "TGDB": "tgdb_id",
+    "FLASHPOINT": "flashpoint_id",
+    "HOWLONGTOBEAT": "hltb_id",
+    "LIBRETRO": "libretro_id",
+    "STEAMGRIDDB": "sgdb_id",
+    "GAMELIST": "gamelist_id",
+}
+
+# Rom attrs the scan handler actually consumes from a Playmatch lookup.
+# Other tags exist only for outbound suggestions.
+PLAYMATCH_LOOKUP_ROM_ATTRS: frozenset[str] = frozenset(
+    {"igdb_id", "moby_id", "ss_id", "launchbox_id", "sgdb_id"}
+)
 
-
-# (rom attribute, provider tag). Playmatch drops unknown tags server-side.
-_PLAYMATCH_PROVIDER_TAGS: tuple[tuple[str, str], ...] = (
-    ("igdb_id", "IGDB"),
-    ("moby_id", "MOBY_GAMES"),
-    ("ss_id", "SCREENSCRAPER"),
-    ("ra_id", "RETRO_ACHIEVEMENTS"),
-    ("launchbox_id", "LAUNCHBOX"),
-    ("hasheous_id", "HASHEOUS"),
-    ("tgdb_id", "TGDB"),
-    ("flashpoint_id", "FLASHPOINT"),
-    ("hltb_id", "HOWLONGTOBEAT"),
-    ("libretro_id", "LIBRETRO"),
-    ("sgdb_id", "STEAMGRIDDB"),
-    ("gamelist_id", "GAMELIST"),
+# MetadataSource values (StrEnum) for which Playmatch can return ids. Typed as
+# strings so this module stays free of scan_handler imports. EmuReady and
+# OpenVGDB are in Playmatch's enum but have no RomM counterpart yet.
+PLAYMATCH_SUPPORTED_SOURCES: frozenset[str] = frozenset(
+    {"igdb", "moby", "ss", "launchbox", "sgdb"}
 )
 
 
 class GameMatchType(str, Enum):
     SHA256 = "SHA256"
     SHA1 = "SHA1"
     MD5 = "MD5"
-    FileNameAndSize = "FileNameAndSize"
-    NoMatch = "NoMatch"
+    FILE_NAME_AND_SIZE = "FileNameAndSize"
+    NO_MATCH = "NoMatch"
 
 
 class PlaymatchExternalMetadata(TypedDict):
@@ -55,6 +77,17 @@ class PlaymatchExternalMetadata(TypedDict):
 
 class PlaymatchRomMatch(TypedDict):
     igdb_id: int | None
+    moby_id: int | None
+    ss_id: int | None
+    launchbox_id: int | None
+    sgdb_id: int | None
+    ra_id: int | None
+    hasheous_id: int | None
+    tgdb_id: int | None
+    flashpoint_id: str | None
+    hltb_id: int | None
+    gamelist_id: str | None
+    libretro_id: str | None
 
 
 class PlaymatchHandler(MetadataHandler):
@@ -124,7 +157,7 @@ async def _request(self, url: str, query: dict) -> dict:
                 detail="Can't connect to Playmatch, check your internet connection",
             ) from exc
         except json.JSONDecodeError as exc:
-            log.error("Error decoding JSON response from ScreenScraper: %s", exc)
+            log.error("Error decoding JSON response from Playmatch: %s", exc)
             return {}
 
     async def lookup_rom(self, files: list[RomFile]) -> PlaymatchRomMatch:
@@ -135,15 +168,30 @@ async def lookup_rom(self, files: list[RomFile]) -> PlaymatchRomMatch:
         :return: A PlaymatchRomMatch objects containing the matched ROM information.
         :raises HTTPException: If the request fails or the service is unavailable.
         """
+        fallback_rom = PlaymatchRomMatch(
+            igdb_id=None,
+            moby_id=None,
+            ss_id=None,
+            launchbox_id=None,
+            sgdb_id=None,
+            ra_id=None,
+            hasheous_id=None,
+            tgdb_id=None,
+            flashpoint_id=None,
+            hltb_id=None,
+            gamelist_id=None,
+            libretro_id=None,
+        )
+
         if not self.is_enabled():
-            return PlaymatchRomMatch(igdb_id=None)
+            return fallback_rom
 
         first_file = next(
             (file for file in files if file.file_size_bytes > 0),
             None,
         )
         if first_file is None:
-            return PlaymatchRomMatch(igdb_id=None)
+            return fallback_rom
 
         try:
             response = await self._request(
@@ -157,45 +205,61 @@ async def lookup_rom(self, files: list[RomFile]) -> PlaymatchRomMatch:
             )
         except httpx.HTTPStatusError:
             # We silently fail if the service is unavailable as this should not block the rest of RomM.
-            return PlaymatchRomMatch(igdb_id=None)
+            return fallback_rom
 
         game_match_type = response.get("gameMatchType", None)
-        if game_match_type == GameMatchType.NoMatch:
+        if game_match_type == GameMatchType.NO_MATCH:
             log.debug("No match found for the provided ROM file.")
-            return PlaymatchRomMatch(igdb_id=None)
+            return fallback_rom
 
         externalMetadata = response.get("externalMetadata", [])
         if len(externalMetadata) == 0:
             log.debug("No external metadata found for the matched ROM file.")
-            return PlaymatchRomMatch(igdb_id=None)
-
-        igdb_id = None
+            return fallback_rom
 
+        result = fallback_rom
         for metadata in externalMetadata:
             provider_name = metadata.get("providerName", None)
             provider_game_id = metadata.get("providerId", None)
-            if provider_name == PlaymatchProvider.IGDB and provider_game_id is not None:
+            if not provider_name or provider_game_id is None:
+                continue
+
+            attr = PLAYMATCH_TAG_TO_ATTR.get(provider_name.upper())
+            if not attr or attr not in PLAYMATCH_LOOKUP_ROM_ATTRS:
+                continue
+
+            try:
+                parsed_id = int(provider_game_id)
+            except (TypeError, ValueError):
                 log.debug(
-                    "Playmatch found IGDB match with IGDB ID: %s", provider_game_id
+                    "Playmatch returned non-int ID for %s: %r",
+                    provider_name,
+                    provider_game_id,
                 )
-                igdb_id = int(provider_game_id)
+                continue
+
+            log.debug("Playmatch found %s match with id: %s", provider_name, parsed_id)
+            result[attr] = parsed_id  # trunk-ignore(mypy/literal-required)
 
-        return PlaymatchRomMatch(igdb_id=igdb_id)
+        return result
 
     @staticmethod
     def is_manual_match(form_fields_set: set[str]) -> bool:
         """True if the submitted form contains any Playmatch-tracked provider id field."""
-        return any(attr in form_fields_set for attr, _ in _PLAYMATCH_PROVIDER_TAGS)
+        return any(attr in form_fields_set for attr in PLAYMATCH_TAG_TO_ATTR.values())
 
     async def submit_manual_match_suggestion(self, rom: Rom) -> None:
-        """Fire-and-forget suggestion POST. No-ops if disabled or no provider IDs are set; never raises."""
+        """
+        Fire-and-forget suggestion POST.
+        No-ops if disabled or no provider IDs are set.
+        """
         try:
             if not self.is_enabled():
                 return
 
             mappings = [
                 {"provider": tag, "providerId": str(getattr(rom, attr))}
-                for attr, tag in _PLAYMATCH_PROVIDER_TAGS
+                for tag, attr in PLAYMATCH_TAG_TO_ATTR.items()
                 if getattr(rom, attr, None)
             ]
             if not mappings: