Merge pull request #3817 from rommapp/fix/igdb-sgdb-series-prefix-matching

fix(igdb): don't match series games to their base title

Author: gantoine

backend/handler/metadata/igdb_handler.py

@@ -64,6 +64,10 @@
 # Regex to detect IGDB ID tags in filenames like (igdb-12345)
 IGDB_TAG_REGEX = re.compile(r"\(igdb-(\d+)\)", re.IGNORECASE)
 
+# Jaro-Winkler score of an exact (post-normalization) title match. Only a first
+# pass hitting this may be trusted without widening the search.
+EXACT_MATCH_SCORE: Final = 1.0
+
 
 class IGDBPlatform(TypedDict):
     slug: str
@@ -558,6 +562,23 @@ def extract_igdb_id_from_filename(fs_name: str) -> int | None:
             return int(match.group(1))
         return None
 
+    def _is_prefix_superset_match(self, search_term: str, candidate_name: str) -> bool:
+        """Whether one title's words are a proper prefix of the other's.
+
+        Jaro-Winkler scores a base title and a longer variant that starts with
+        it (e.g. "Portable Ops" vs "Portable Ops Plus") well above the match
+        threshold, so a fuzzy pass can settle for the base when the variant is
+        simply absent from that pass's candidates. Detecting this prefix/superset
+        ambiguity lets the caller widen the search before committing. (#3805)
+        """
+        search_tokens = self.normalize_search_term(search_term).split()
+        candidate_tokens = self.normalize_search_term(candidate_name).split()
+        if not search_tokens or not candidate_tokens:
+            return False
+
+        shorter, longer = sorted((search_tokens, candidate_tokens), key=len)
+        return len(shorter) < len(longer) and longer[: len(shorter)] == shorter
+
     async def _search_rom(
         self, search_term: str, platform_igdb_id: int, with_game_type: bool = False
     ) -> Game | None:
@@ -578,18 +599,18 @@ async def _search_rom(
             game_type_filter = ""
 
         log.debug("Searching in games endpoint with game_type %s", game_type_filter)
-        where_filter = f"{_build_platforms_where(platform_igdb_id)} {game_type_filter}"
+        base_where = _build_platforms_where(platform_igdb_id)
 
         # Special case for ScummVM games
         # https://github.com/rommapp/romm/issues/2424
         scummvm_platform = self.get_platform(UPS.SCUMMVM)
         if scummvm_platform["igdb_id"] == platform_igdb_id:
-            where_filter = f"keywords=[{platform_igdb_id}] {game_type_filter}"
+            base_where = f"keywords=[{platform_igdb_id}]"
 
         roms = await self.igdb_service.list_games(
             search_term=search_term,
             fields=GAMES_FIELDS,
-            where=where_filter,
+            where=f"{base_where} {game_type_filter}",
             limit=self.pagination_limit,
         )
 
@@ -599,12 +620,37 @@ async def _search_rom(
             search_term,
             list(games_by_name.keys()),
         )
-        if best_match:
+
+        # Trust an exact first-pass hit outright. A non-exact hit that is only a
+        # prefix/superset of the search term (e.g. matching "Portable Ops" for a
+        # "Portable Ops Plus" search) may be a near-miss for a more specific
+        # variant this pass never saw, so widen the search and re-rank across
+        # every candidate before committing. (#3805)
+        if best_match is not None and (
+            best_score >= EXACT_MATCH_SCORE
+            or not self._is_prefix_superset_match(search_term, best_match)
+        ):
             log.debug(
                 f"Found match for '{search_term}' -> '{best_match}' (score: {best_score:.3f})"
             )
             return games_by_name[best_match]
 
+        extra_roms: list[Game] = []
+
+        # The game_type filter can hide a more specific variant that IGDB
+        # classifies as an excluded type (e.g. an expansion). Re-query without
+        # it so such variants become candidates.
+        if game_type_filter:
+            log.debug("Searching in games endpoint without game_type")
+            extra_roms.extend(
+                await self.igdb_service.list_games(
+                    search_term=search_term,
+                    fields=GAMES_FIELDS,
+                    where=base_where,
+                    limit=self.pagination_limit,
+                )
+            )
+
         log.debug("Searching expanded in search endpoint")
         roms_expanded = await self.igdb_service.search(
             fields=SEARCH_FIELDS,
@@ -629,25 +675,30 @@ async def _search_rom(
                 unique_game_ids,
             )
             id_filter = " | ".join(f"id={gid}" for gid in unique_game_ids)
-            extra_roms = await self.igdb_service.list_games(
-                fields=GAMES_FIELDS,
-                where=f"({id_filter})",
-                limit=self.pagination_limit,
+            extra_roms.extend(
+                await self.igdb_service.list_games(
+                    fields=GAMES_FIELDS,
+                    where=f"({id_filter})",
+                    limit=self.pagination_limit,
+                )
             )
 
-            extra_games_by_name = _index_games_by_searchable_name(extra_roms)
-
+        if extra_roms:
+            # Re-rank across the union of every pass so an exact variant surfaced
+            # only after widening can outrank the first-pass near-miss on the
+            # base title. The base stays in the pool, so it remains the fallback
+            # when no better match exists.
+            games_by_name = _index_games_by_searchable_name(roms + extra_roms)
             best_match, best_score = self.find_best_match(
                 search_term,
-                list(extra_games_by_name.keys()),
+                list(games_by_name.keys()),
             )
-            if best_match:
-                log.debug(
-                    f"Found match for '{search_term}' -> '{best_match}' (score: {best_score:.3f})"
-                )
-                return extra_games_by_name[best_match]
 
-            roms.extend(extra_roms)
+        if best_match:
+            log.debug(
+                f"Found match for '{search_term}' -> '{best_match}' (score: {best_score:.3f})"
+            )
+            return games_by_name[best_match]
 
         return None
 

backend/tests/handler/metadata/test_igdb_handler.py

@@ -705,3 +705,177 @@ async def test_serial_at_filename_start_resolves_title(self):
         mock_hget.assert_awaited_once_with(PS1_SERIAL_INDEX_KEY, "SCUS-94163")
         assert result.get("name") == "Gran Turismo"
         assert result["igdb_id"] is None
+
+
+class TestIsPrefixSupersetMatch:
+    """Unit tests for the prefix/superset title heuristic (issue #3805)."""
+
+    @pytest.mark.parametrize(
+        ("search_term", "candidate", "expected"),
+        [
+            # A more specific variant's search term extends the base title.
+            (
+                "metal gear solid portable ops plus",
+                "Metal Gear Solid: Portable Ops",
+                True,
+            ),
+            # Reversed: the base term is a prefix of the variant candidate.
+            (
+                "Metal Gear Solid: Portable Ops",
+                "Metal Gear Solid: Portable Ops Plus",
+                True,
+            ),
+            ("pokemon ranger shadows of almia", "Pokemon Ranger", True),
+            # Extra word is not a trailing suffix, so not a prefix relationship.
+            ("sonic hedgehog", "Sonic the Hedgehog", False),
+            # Identical titles are an exact match, not a prefix ambiguity.
+            ("Metal Gear Solid", "metal gear solid", False),
+            # Unrelated titles.
+            ("contra", "Probotector", False),
+        ],
+    )
+    def test_prefix_superset_detection(self, search_term, candidate, expected):
+        handler = IGDBHandler()
+        assert handler._is_prefix_superset_match(search_term, candidate) is expected
+
+
+class TestSearchRomPrefixSupersetVariant:
+    """A base title that is a prefix of the searched variant must not be
+    accepted when the more specific variant exists (issue #3805).
+
+    'Metal Gear Solid - Portable Ops Plus' and 'Portable Ops' (and the Pokemon
+    Ranger series) scored the same IGDB id because the first search pass only
+    returned the base game and its ~0.99 Jaro-Winkler score cleared the match
+    threshold.
+    """
+
+    BASE_ID = 1001
+    VARIANT_ID = 1002
+    BASE_NAME = "Metal Gear Solid: Portable Ops"
+    VARIANT_NAME = "Metal Gear Solid: Portable Ops Plus"
+
+    @pytest.mark.asyncio
+    async def test_variant_excluded_by_game_type_is_recovered(self):
+        """When the game_type filter hides the variant (IGDB classifies it as an
+        expansion), dropping the filter must surface it so the exact match wins
+        over the base near-miss."""
+        handler = IGDBHandler()
+
+        base = _make_game(self.BASE_ID, self.BASE_NAME)
+        variant = _make_game(self.VARIANT_ID, self.VARIANT_NAME)
+
+        async def mock_list_games(
+            search_term=None, fields=None, where=None, limit=None
+        ):
+            # game_type-filtered pass excludes the variant (an expansion type).
+            if where and "game_type" in where:
+                return [base]
+            # Re-query without the game_type filter surfaces both.
+            return [base, variant]
+
+        with (
+            patch(
+                "handler.metadata.igdb_handler.IGDBHandler.is_enabled",
+                return_value=True,
+            ),
+            patch.object(
+                handler.igdb_service,
+                "list_games",
+                side_effect=mock_list_games,
+            ),
+            patch.object(
+                handler.igdb_service,
+                "search",
+                new_callable=AsyncMock,
+                return_value=[],
+            ),
+        ):
+            result = await handler._search_rom(
+                "metal gear solid portable ops plus",
+                GENESIS_IGDB_ID,
+                with_game_type=True,
+            )
+
+        assert result is not None
+        assert result["id"] == self.VARIANT_ID, (
+            f"Expected the '{self.VARIANT_NAME}' variant (id={self.VARIANT_ID}), "
+            f"got {result.get('name')} (id={result.get('id')}). The base title is "
+            "only a prefix near-miss and must not win over the exact variant."
+        )
+
+    @pytest.mark.asyncio
+    async def test_base_title_still_matches_when_it_is_the_target(self):
+        """Scanning the base game itself must return the base on the first pass
+        without any widening (exact match short-circuits)."""
+        handler = IGDBHandler()
+
+        base = _make_game(self.BASE_ID, self.BASE_NAME)
+        variant = _make_game(self.VARIANT_ID, self.VARIANT_NAME)
+
+        search_mock = AsyncMock(return_value=[])
+
+        async def mock_list_games(
+            search_term=None, fields=None, where=None, limit=None
+        ):
+            # First pass includes both; the base is an exact match.
+            if where and "game_type" in where:
+                return [base, variant]
+            raise AssertionError("widening should not run for an exact match")
+
+        with (
+            patch(
+                "handler.metadata.igdb_handler.IGDBHandler.is_enabled",
+                return_value=True,
+            ),
+            patch.object(
+                handler.igdb_service,
+                "list_games",
+                side_effect=mock_list_games,
+            ),
+            patch.object(handler.igdb_service, "search", search_mock),
+        ):
+            result = await handler._search_rom(
+                "metal gear solid portable ops",
+                GENESIS_IGDB_ID,
+                with_game_type=True,
+            )
+
+        assert result is not None
+        assert result["id"] == self.BASE_ID
+        search_mock.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_non_prefix_fuzzy_match_does_not_widen(self):
+        """A benign non-exact match that is not a prefix/superset must be
+        returned from the first pass without extra queries."""
+        handler = IGDBHandler()
+
+        game = _make_game(42, "Sonic the Hedgehog")
+        search_mock = AsyncMock(return_value=[])
+
+        async def mock_list_games(
+            search_term=None, fields=None, where=None, limit=None
+        ):
+            if where and "game_type" in where:
+                return [game]
+            raise AssertionError("widening should not run for a non-prefix match")
+
+        with (
+            patch(
+                "handler.metadata.igdb_handler.IGDBHandler.is_enabled",
+                return_value=True,
+            ),
+            patch.object(
+                handler.igdb_service,
+                "list_games",
+                side_effect=mock_list_games,
+            ),
+            patch.object(handler.igdb_service, "search", search_mock),
+        ):
+            result = await handler._search_rom(
+                "sonic hedgehog", GENESIS_IGDB_ID, with_game_type=True
+            )
+
+        assert result is not None
+        assert result["id"] == 42
+        search_mock.assert_not_awaited()