feat: library-based similar games from shared metadata

backend/endpoints/roms/__init__.py

@@ -54,6 +54,10 @@
 )
 from handler.database import db_rom_handler, db_save_handler
 from handler.database.base_handler import sync_session
+from handler.database.roms_handler import (
+    MAX_SIMILAR_ROMS_LIMIT,
+    SIMILAR_ROMS_LIMIT,
+)
 from handler.filesystem import fs_resource_handler, fs_rom_handler
 from handler.filesystem.assets_handler import validate_image_upload
 from handler.metadata import (
@@ -1113,6 +1117,53 @@ def get_rom_simple(
     return SimpleRomSchema.from_orm_with_request(rom, request)
 
 
+@protected_route(
+    router.get,
+    "/{id}/similar",
+    [] if DISABLE_DOWNLOAD_ENDPOINT_AUTH else [Scope.ROMS_READ],
+    responses={status.HTTP_404_NOT_FOUND: {}},
+)
+def get_similar_roms(
+    request: Request,
+    id: Annotated[int, PathVar(description="Rom internal id.", ge=1)],
+    limit: Annotated[
+        int,
+        Query(
+            description="Max number of similar ROMs to return.",
+            ge=1,
+            le=MAX_SIMILAR_ROMS_LIMIT,
+        ),
+    ] = SIMILAR_ROMS_LIMIT,
+) -> list[SimpleRomSchema]:
+    """Return library ROMs similar to the given ROM.
+
+    Similarity is computed from the normalized `RomMetadata` signals
+    (franchises, collections, genres, companies, age ratings), so only ROMs
+    already in the library are returned. ROMs hidden from the caller are
+    excluded."""
+
+    rom = db_rom_handler.get_rom_simple(id)
+
+    if not rom:
+        raise RomNotFoundInDatabaseException(id)
+
+    assert_rom_visible(request, rom)
+
+    perms = get_permissions(request)
+    similar_ids = db_rom_handler.get_similar_rom_ids(
+        rom,
+        limit=limit,
+        hidden_platform_ids=list(perms.hidden_platform_ids),
+        hidden_rom_ids=list(perms.hidden_rom_ids),
+    )
+    if not similar_ids:
+        return []
+
+    roms_by_id = {r.id: r for r in db_rom_handler.get_roms_by_ids(similar_ids)}
+    ordered = [roms_by_id[i] for i in similar_ids if i in roms_by_id]
+    return [SimpleRomSchema.from_orm_with_request(r, request) for r in ordered]
+
+
 @protected_route(
     router.get,
     "/{id}",

backend/handler/database/roms_handler.py

@@ -66,6 +66,27 @@
 
 from .base_handler import DBBaseHandler
 
+# Default and hard cap for the "similar games in your library" list.
+SIMILAR_ROMS_LIMIT = 6
+MAX_SIMILAR_ROMS_LIMIT = 24
+# Per-signal weights for the metadata-overlap similarity score. Series
+# membership (franchise/collection) dominates, then genre, then the studio,
+# with age rating as a light tiebreaker.
+SIMILARITY_WEIGHTS = {
+    "franchises": 5.0,
+    "collections": 4.0,
+    "genres": 3.0,
+    "companies": 2.0,
+    "age_ratings": 1.0,
+}
+# Multiplier applied to a candidate's score when it shares the target's
+# platform, so same-console matches float to the top without eclipsing a
+# strong cross-platform series match.
+SAME_PLATFORM_BOOST = 1.25
+# Upper bound on weak-signal (genre/company/age) candidates pulled for scoring;
+# strong-signal (franchise/collection) candidates are always scored in full.
+MAX_SIMILAR_CANDIDATES = 400
+
 EJS_SUPPORTED_PLATFORMS = [
     UPS._3DO,
     UPS.AMIGA,
@@ -338,6 +359,125 @@ def get_roms_by_ids(
             return []
         return session.scalars(query.filter(Rom.id.in_(ids))).all()
 
+    @begin_session
+    def get_similar_rom_ids(
+        self,
+        rom: Rom,
+        *,
+        limit: int = SIMILAR_ROMS_LIMIT,
+        hidden_platform_ids: Sequence[int] | None = None,
+        hidden_rom_ids: Sequence[int] | None = None,
+        session: Session = None,  # type: ignore
+    ) -> list[int]:
+        """Rank other library ROMs by shared normalized metadata.
+
+        Similarity is a weighted overlap of the target ROM's `RomMetadata`
+        signals (franchises, collections, genres, companies, age ratings)
+        against every other identified ROM, with a multiplicative boost for
+        same-platform matches. Returns rom ids ordered by descending score,
+        capped at `limit`. Purely metadata-driven, so unidentified ROMs (no
+        `RomMetadata` row, or empty signals) yield an empty list.
+        """
+        target = session.get(RomMetadata, rom.id)
+        if target is None:
+            return []
+
+        signals = {
+            field: [v for v in (getattr(target, field) or []) if v]
+            for field in SIMILARITY_WEIGHTS
+        }
+        if not any(signals.values()):
+            return []
+
+        # Sibling ROMs are the same game (different dump/region); never
+        # surface them as "similar".
+        sibling_ids = set(
+            session.scalars(
+                select(SiblingRom.sibling_rom_id).where(SiblingRom.rom_id == rom.id)
+            ).all()
+        )
+        exclude_ids = {rom.id, *sibling_ids}
+        if hidden_rom_ids:
+            exclude_ids.update(hidden_rom_ids)
+
+        columns = (
+            RomMetadata.rom_id,
+            RomMetadata.franchises,
+            RomMetadata.collections,
+            RomMetadata.genres,
+            RomMetadata.companies,
+            RomMetadata.age_ratings,
+            RomMetadata.average_rating,
+            Rom.platform_id,
+        )
+
+        def _candidate_query(where):
+            query = (
+                select(*columns)
+                .join(Rom, Rom.id == RomMetadata.rom_id)
+                .where(where)
+                .where(RomMetadata.rom_id.not_in(exclude_ids))
+                .where(Rom.missing_from_fs.is_(False))
+            )
+            if hidden_platform_ids:
+                query = query.where(Rom.platform_id.not_in(hidden_platform_ids))
+            return query
+
+        # Strong signals (franchise / collection) are selective, so every
+        # match is scored. Weak signals (genre / company / age rating) can
+        # match a large slice of the library, so they are capped and ordered
+        # by rating to keep the candidate set bounded.
+        strong_conditions = [
+            json_array_contains_any(
+                getattr(RomMetadata, field), values, session=session
+            )
+            for field in ("franchises", "collections")
+            if (values := signals[field])
+        ]
+        weak_conditions = [
+            json_array_contains_any(
+                getattr(RomMetadata, field), values, session=session
+            )
+            for field in ("genres", "companies", "age_ratings")
+            if (values := signals[field])
+        ]
+
+        rows: dict[int, Any] = {}
+        if strong_conditions:
+            for row in session.execute(_candidate_query(or_(*strong_conditions))).all():
+                rows[row.rom_id] = row
+        if weak_conditions:
+            weak_query = (
+                _candidate_query(or_(*weak_conditions))
+                .order_by(
+                    # Portable NULLs-last: rating-less ROMs sort after rated ones.
+                    RomMetadata.average_rating.is_(None),
+                    RomMetadata.average_rating.desc(),
+                )
+                .limit(MAX_SIMILAR_CANDIDATES)
+            )
+            for row in session.execute(weak_query).all():
+                rows.setdefault(row.rom_id, row)
+
+        target_sets = {field: set(values) for field, values in signals.items()}
+
+        def _score(row) -> float:
+            score = 0.0
+            for field, weight in SIMILARITY_WEIGHTS.items():
+                shared = target_sets[field] & set(getattr(row, field) or [])
+                score += weight * len(shared)
+            if score and row.platform_id == rom.platform_id:
+                score *= SAME_PLATFORM_BOOST
+            return score
+
+        scored = [
+            (score, row.average_rating or 0.0, row.rom_id)
+            for row in rows.values()
+            if (score := _score(row)) > 0
+        ]
+        scored.sort(key=lambda t: (t[0], t[1], t[2]), reverse=True)
+        return [rom_id for _, _, rom_id in scored[:limit]]
+
     def get_files_for_roms(
         self,
         rom_ids: list[int],

backend/tests/endpoints/roms/test_rom.py

@@ -70,6 +70,46 @@ def test_get_rom_simple_missing_returns_404(client: TestClient, access_token: st
     assert response.status_code == status.HTTP_404_NOT_FOUND
 
 
+def test_get_similar_roms(
+    client: TestClient, access_token: str, platform: Platform, admin_user: User
+):
+    def _add(name: str, igdb_id: int, metadata: dict) -> Rom:
+        return db_rom_handler.add_rom(
+            Rom(
+                platform_id=platform.id,
+                name=name,
+                slug=name.lower().replace(" ", "-"),
+                fs_name=f"{name}.zip",
+                fs_path=f"{platform.slug}/roms",
+                igdb_id=igdb_id,
+                igdb_metadata=metadata,
+            )
+        )
+
+    target = _add("Similar Target", 611111, {"franchises": ["Metroid"]})
+    match = _add("Similar Match", 611112, {"franchises": ["Metroid"]})
+    _add("Similar Unrelated", 611113, {"genres": ["Racing"]})
+
+    response = client.get(
+        f"/api/roms/{target.id}/similar",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert response.status_code == status.HTTP_200_OK
+
+    body = response.json()
+    ids = [r["id"] for r in body]
+    assert match.id in ids
+    assert target.id not in ids
+
+
+def test_get_similar_roms_missing_returns_404(client: TestClient, access_token: str):
+    response = client.get(
+        "/api/roms/999999/similar",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+
+
 def test_download_multi_file_rom_content(
     client: TestClient, access_token: str, multi_file_rom: Rom
 ):