feat(scan): embed Switch title IDs into file names during scan

backend/config/__init__.py

@@ -213,6 +213,9 @@ def _get_env(var: str, fallback: str | None = None) -> str | None:
     "SCHEDULED_UPDATE_SWITCH_TITLEDB_CRON",
     "0 4 * * *",  # At 4:00 AM every day
 )
+ENABLE_SWITCH_TITLE_ID_RENAME: Final[bool] = safe_str_to_bool(
+    _get_env("ENABLE_SWITCH_TITLE_ID_RENAME")
+)
 ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA: Final[bool] = safe_str_to_bool(
     _get_env("ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA")
 )

backend/endpoints/sockets/scan.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import os
 from dataclasses import dataclass
 from itertools import batched
 from typing import Any, Final
@@ -12,7 +13,14 @@
 from sqlalchemy.exc import IntegrityError
 
 from adapters.services.screenscraper import reset_daily_quota as reset_ss_daily_quota
-from config import DEV_MODE, REDIS_URL, SCAN_TIMEOUT, SCAN_WORKERS, TASK_RESULT_TTL
+from config import (
+    DEV_MODE,
+    ENABLE_SWITCH_TITLE_ID_RENAME,
+    REDIS_URL,
+    SCAN_TIMEOUT,
+    SCAN_WORKERS,
+    TASK_RESULT_TTL,
+)
 from config.config_manager import MetadataMediaType
 from config.config_manager import config_manager as cm
 from endpoints.responses import TaskType
@@ -23,6 +31,7 @@
     FOLDER_STRUCT_MSG,
     FirmwareNotFoundException,
     FolderStructureNotMatchException,
+    RomAlreadyExistsException,
     RomsNotFoundException,
 )
 from exceptions.socket_exceptions import ScanStoppedException
@@ -36,6 +45,12 @@
 )
 from handler.filesystem.roms_handler import FSRom
 from handler.metadata import meta_gamelist_handler, meta_hltb_handler
+from handler.metadata.base_handler import (
+    SWITCH_PRODUCT_ID_REGEX,
+    SWITCH_TITLEDB_REGEX,
+    UniversalPlatformSlug as UPS,
+    switch_name_to_product_id,
+)
 from handler.metadata.ss_handler import add_ss_auth_to_url, get_preferred_media_types
 from handler.redis_handler import get_job_func_name, high_prio_queue, redis_client
 from handler.scan_handler import (
@@ -61,6 +76,9 @@
 
 STOP_SCAN_FLAG: Final = "scan:stop"
 
+SWITCH_PLATFORM_SLUGS: Final = frozenset((UPS.SWITCH, UPS.SWITCH_2))
+SWITCH_SERVED_EXTENSIONS: Final = frozenset((".nsp", ".xci", ".nsz", ".xcz", ".nro"))
+
 
 def _clone_track_meta(src: TrackMeta | None, rom_id: int) -> TrackMeta | None:
     """Build a fresh TrackMeta from a scanned (transient) one for a new RomFile."""
@@ -246,6 +264,68 @@ def _should_get_rom_files(
     )
 
 
+async def _maybe_add_switch_title_id(
+    platform: Platform, fs_rom: FSRom, rom: Rom | None
+) -> None:
+    """Rename a flat Switch ROM lacking a title ID to embed one from the TitleDB.
+
+    Opt-in via ENABLE_SWITCH_TITLE_ID_RENAME. Only base games whose name maps to
+    a single title ID are renamed, so tools that parse title IDs out of the file
+    name (e.g. CyberFoil) can index them.
+    """
+    if (
+        not ENABLE_SWITCH_TITLE_ID_RENAME
+        or not fs_rom["flat"]
+        or platform.slug not in SWITCH_PLATFORM_SLUGS
+    ):
+        return
+
+    fs_name = fs_rom["fs_name"]
+    stem, ext = os.path.splitext(fs_name)
+    if ext.lower() not in SWITCH_SERVED_EXTENSIONS:
+        return
+
+    # Skip files that already carry a title ID.
+    if SWITCH_PRODUCT_ID_REGEX.search(fs_name) or SWITCH_TITLEDB_REGEX.search(fs_name):
+        return
+
+    clean_name = fs_rom_handler.get_file_name_with_no_tags(fs_name)
+    title_id = await switch_name_to_product_id(clean_name)
+    if not title_id:
+        return
+
+    new_fs_name = f"{stem} [{title_id}][v0]{ext}"
+    roms_path = fs_rom_handler.get_roms_fs_structure(platform.fs_slug)
+    try:
+        await fs_rom_handler.rename_fs_rom(fs_name, new_fs_name, roms_path)
+    except RomAlreadyExistsException:
+        log.warning(
+            f"Skipping Switch title ID rename for {hl(fs_name)}: "
+            f"{hl(new_fs_name)} already exists"
+        )
+        return
+
+    fs_rom["fs_name"] = new_fs_name
+    log.info(
+        f"Renamed {hl(fs_name)} to {hl(new_fs_name, color=BLUE)} (Switch title ID)"
+    )
+
+    # Move an already-tracked entry and its files onto the new name, mirroring
+    # the manual rename endpoint, so nothing is orphaned under the old name.
+    if rom is not None:
+        db_rom_handler.update_rom(rom.id, {"fs_name": new_fs_name})
+        for file in rom.files:
+            new_file_name = file.file_name.replace(fs_name, new_fs_name)
+            new_file_path = file.file_path.replace(fs_name, new_fs_name)
+            db_rom_handler.update_rom_file(
+                file.id,
+                {"file_name": new_file_name, "file_path": new_file_path},
+            )
+            file.file_name = new_file_name
+            file.file_path = new_file_path
+        rom.fs_name = new_fs_name
+
+
 # There's an order of operations here that is important:
 # 1. Read the list of roms from the filesystem
 # 2. Check if ROM should be scanned based on the scan type
@@ -268,6 +348,8 @@ async def _identify_rom(
     if redis_client.get(STOP_SCAN_FLAG):
         return
 
+    await _maybe_add_switch_title_id(platform, fs_rom, rom)
+
     # Update properties that don't require metadata
     parsed_tags = fs_rom_handler.parse_tags(fs_rom["fs_name"])
     roms_path = fs_rom_handler.get_roms_fs_structure(platform.fs_slug)

backend/handler/metadata/base_handler.py

@@ -13,8 +13,10 @@
 from handler.redis_handler import async_cache
 from logger.logger import log
 from tasks.scheduled.update_switch_titledb import (
+    SWITCH_NAME_TO_ID_KEY,
     SWITCH_PRODUCT_ID_KEY,
     SWITCH_TITLEDB_INDEX_KEY,
+    normalize_switch_name,
 )
 
 jarowinkler = JaroWinkler()
@@ -27,6 +29,18 @@
 SWITCH_PRODUCT_ID_REGEX: Final = re.compile(r"(0100[0-9A-F]{12})")
 
 
+async def switch_name_to_product_id(name: str) -> str | None:
+    """Resolve a game name to a unique base-game Switch title ID, or None.
+
+    Backed by the reverse index built in update_switch_titledb_task; returns
+    None when the index is absent or the name is unknown/ambiguous.
+    """
+    key = normalize_switch_name(name)
+    if not key or not (await async_cache.exists(SWITCH_NAME_TO_ID_KEY)):
+        return None
+    return await async_cache.hget(SWITCH_NAME_TO_ID_KEY, key)
+
+
 # No regex needed for MAME
 MAME_XML_KEY: Final = "romm:mame_xml"
 

backend/tasks/scheduled/update_switch_titledb.py

@@ -1,4 +1,5 @@
 import json
+import re
 from itertools import batched
 from typing import Any, Final
 
@@ -15,6 +16,34 @@
 
 SWITCH_TITLEDB_INDEX_KEY: Final = "romm:switch_titledb"
 SWITCH_PRODUCT_ID_KEY: Final = "romm:switch_product_id"
+# Maps a normalized game name to a unique base-game title ID, for injecting the
+# ID into a Switch file name that lacks one. Built in the same update task.
+SWITCH_NAME_TO_ID_KEY: Final = "romm:switch_name_to_id"
+
+_NON_ALNUM_PATTERN: Final = re.compile(r"[^a-z0-9]+")
+
+
+def normalize_switch_name(name: str) -> str:
+    """Normalize a game name for name-based title ID lookups."""
+    return _NON_ALNUM_PATTERN.sub(" ", name.lower()).strip()
+
+
+def _base_title_id(entry: dict) -> str | None:
+    """Return the entry's title ID if it is a base game, else None.
+
+    Nintendo title IDs are 16 hex digits. Base games clear the low 13 bits;
+    updates set 0x800 and DLC sets the 0x1000 bit. See
+    https://switchbrew.org/wiki/Title_list.
+    """
+    title_id = entry.get("id")
+    if not title_id:
+        return None
+    try:
+        if int(title_id, 16) & 0x1FFF == 0:
+            return title_id
+    except ValueError:
+        return None
+    return None
 
 
 class UpdateSwitchTitleDBTask(RemoteFilePullTask):
@@ -61,6 +90,30 @@ async def run(self, force: bool = False) -> dict[str, Any]:
                 }
                 if product_map:
                     await pipe.hset(SWITCH_PRODUCT_ID_KEY, mapping=product_map)
+
+            # Reverse index: normalized base-game name -> title ID. Only keep
+            # names that resolve to a single base title so ambiguous names are
+            # never auto-renamed to the wrong game.
+            name_to_id: dict[str, str] = {}
+            ambiguous: set[str] = set()
+            for entry in relevant_data.values():
+                base_id = _base_title_id(entry)
+                if not base_id or not entry.get("name"):
+                    continue
+                key = normalize_switch_name(entry["name"])
+                if not key:
+                    continue
+                existing = name_to_id.get(key)
+                if existing is None:
+                    name_to_id[key] = base_id
+                elif existing != base_id:
+                    ambiguous.add(key)
+            for key in ambiguous:
+                name_to_id.pop(key, None)
+
+            for name_batch in batched(name_to_id.items(), 2000, strict=False):
+                await pipe.hset(SWITCH_NAME_TO_ID_KEY, mapping=dict(name_batch))
+
             await pipe.execute()
 
         # Final progress update

backend/tests/endpoints/sockets/test_scan.py

@@ -7,12 +7,14 @@
 from endpoints.sockets.scan import (
     ScanStats,
     _identify_rom,
+    _maybe_add_switch_title_id,
     reject_unauthorized_scan,
     scan_handler,
     scan_platforms,
     should_scan_rom,
     stop_scan_handler,
 )
+from exceptions.fs_exceptions import RomAlreadyExistsException
 from handler.auth.constants import Scope
 from handler.filesystem.roms_handler import (
     FSRom,
@@ -696,3 +698,139 @@ def test_url_contains_fs_path_and_name(self, handler: FSRomsHandler):
         assert url is not None
         assert fs_path in url
         assert fs_name in url
+
+
+class TestMaybeAddSwitchTitleId:
+    """`_maybe_add_switch_title_id` embeds a title ID into a Switch file name."""
+
+    TITLE_ID = "0100000000010000"
+
+    def _switch_platform(self, slug=UPS.SWITCH):
+        platform = Platform(name="Switch", slug=slug, fs_slug="switch")
+        platform.id = 1
+        return platform
+
+    def _fs_rom(self, fs_name="Super Mario Odyssey.nsp", flat=True) -> FSRom:
+        return {
+            "fs_name": fs_name,
+            "flat": flat,
+            "nested": not flat,
+            "files": [],
+            "crc_hash": "",
+            "md5_hash": "",
+            "sha1_hash": "",
+            "ra_hash": "",
+        }
+
+    def _enable(self, mocker, enabled=True, title_id=TITLE_ID):
+        mocker.patch.object(
+            scan_module, "ENABLE_SWITCH_TITLE_ID_RENAME", enabled
+        )
+        lookup = mocker.patch.object(
+            scan_module,
+            "switch_name_to_product_id",
+            AsyncMock(return_value=title_id),
+        )
+        rename = mocker.patch.object(
+            scan_module.fs_rom_handler, "rename_fs_rom", AsyncMock()
+        )
+        return lookup, rename
+
+    async def test_renames_flat_switch_file(self, mocker):
+        lookup, rename = self._enable(mocker)
+        fs_rom = self._fs_rom()
+
+        await _maybe_add_switch_title_id(self._switch_platform(), fs_rom, None)
+
+        expected = f"Super Mario Odyssey [{self.TITLE_ID}][v0].nsp"
+        assert fs_rom["fs_name"] == expected
+        rename.assert_awaited_once()
+        old, new, _path = rename.await_args.args
+        assert old == "Super Mario Odyssey.nsp"
+        assert new == expected
+        lookup.assert_awaited_once_with("Super Mario Odyssey")
+
+    async def test_skips_when_flag_disabled(self, mocker):
+        _lookup, rename = self._enable(mocker, enabled=False)
+        fs_rom = self._fs_rom()
+
+        await _maybe_add_switch_title_id(self._switch_platform(), fs_rom, None)
+
+        assert fs_rom["fs_name"] == "Super Mario Odyssey.nsp"
+        rename.assert_not_awaited()
+
+    async def test_skips_when_title_id_already_present(self, mocker):
+        lookup, rename = self._enable(mocker)
+        fs_rom = self._fs_rom(f"Super Mario Odyssey [{self.TITLE_ID}][v0].nsp")
+
+        await _maybe_add_switch_title_id(self._switch_platform(), fs_rom, None)
+
+        lookup.assert_not_awaited()
+        rename.assert_not_awaited()
+
+    async def test_skips_non_switch_platform(self, mocker):
+        _lookup, rename = self._enable(mocker)
+        fs_rom = self._fs_rom()
+
+        platform = Platform(name="Test", slug="test", fs_slug="test")
+        platform.id = 1
+        await _maybe_add_switch_title_id(platform, fs_rom, None)
+
+        rename.assert_not_awaited()
+
+    async def test_skips_non_served_extension(self, mocker):
+        _lookup, rename = self._enable(mocker)
+        fs_rom = self._fs_rom("Super Mario Odyssey.zip")
+
+        await _maybe_add_switch_title_id(self._switch_platform(), fs_rom, None)
+
+        rename.assert_not_awaited()
+
+    async def test_skips_nested_rom(self, mocker):
+        _lookup, rename = self._enable(mocker)
+        fs_rom = self._fs_rom(flat=False)
+
+        await _maybe_add_switch_title_id(self._switch_platform(), fs_rom, None)
+
+        rename.assert_not_awaited()
+
+    async def test_skips_when_name_unresolved(self, mocker):
+        _lookup, rename = self._enable(mocker, title_id=None)
+        fs_rom = self._fs_rom()
+
+        await _maybe_add_switch_title_id(self._switch_platform(), fs_rom, None)
+
+        assert fs_rom["fs_name"] == "Super Mario Odyssey.nsp"
+        rename.assert_not_awaited()
+
+    async def test_leaves_name_when_target_exists(self, mocker):
+        _lookup, rename = self._enable(mocker)
+        rename.side_effect = RomAlreadyExistsException("dup")
+        db = mocker.patch.object(scan_module, "db_rom_handler")
+        fs_rom = self._fs_rom()
+
+        await _maybe_add_switch_title_id(self._switch_platform(), fs_rom, None)
+
+        assert fs_rom["fs_name"] == "Super Mario Odyssey.nsp"
+        db.update_rom.assert_not_called()
+
+    async def test_moves_existing_db_entry(self, mocker):
+        self._enable(mocker)
+        db = mocker.patch.object(scan_module, "db_rom_handler")
+        fs_rom = self._fs_rom()
+
+        rom_file = MagicMock(
+            file_name="Super Mario Odyssey.nsp",
+            file_path="switch/roms/Super Mario Odyssey.nsp",
+        )
+        rom = MagicMock(id=7, files=[rom_file])
+
+        await _maybe_add_switch_title_id(self._switch_platform(), fs_rom, rom)
+
+        expected = f"Super Mario Odyssey [{self.TITLE_ID}][v0].nsp"
+        db.update_rom.assert_called_once_with(7, {"fs_name": expected})
+        db.update_rom_file.assert_called_once()
+        _file_id, data = db.update_rom_file.call_args.args
+        assert data["file_name"] == expected
+        assert rom.fs_name == expected
+        assert rom_file.file_name == expected