diff --git a/backend/config/__init__.py b/backend/config/__init__.py index 4a89812e84..720c2485cb 100644 --- a/backend/config/__init__.py +++ b/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") ) diff --git a/backend/endpoints/sockets/scan.py b/backend/endpoints/sockets/scan.py index dcc4898522..bd0eed6db7 100644 --- a/backend/endpoints/sockets/scan.py +++ b/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) diff --git a/backend/handler/metadata/base_handler.py b/backend/handler/metadata/base_handler.py index e814d3d219..ec42ea486f 100644 --- a/backend/handler/metadata/base_handler.py +++ b/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" diff --git a/backend/tasks/scheduled/update_switch_titledb.py b/backend/tasks/scheduled/update_switch_titledb.py index bf48adb9ec..c7bd2a815e 100644 --- a/backend/tasks/scheduled/update_switch_titledb.py +++ b/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 diff --git a/backend/tests/endpoints/sockets/test_scan.py b/backend/tests/endpoints/sockets/test_scan.py index 6852860c4d..f1f60c9a61 100644 --- a/backend/tests/endpoints/sockets/test_scan.py +++ b/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 diff --git a/backend/tests/tasks/test_update_switch_titledb.py b/backend/tests/tasks/test_update_switch_titledb.py index 868571c3da..8d6b0f6f18 100644 --- a/backend/tests/tasks/test_update_switch_titledb.py +++ b/backend/tests/tasks/test_update_switch_titledb.py @@ -4,14 +4,62 @@ import pytest from tasks.scheduled.update_switch_titledb import ( + SWITCH_NAME_TO_ID_KEY, SWITCH_PRODUCT_ID_KEY, SWITCH_TITLEDB_INDEX_KEY, UpdateSwitchTitleDBTask, + normalize_switch_name, update_switch_titledb_task, ) from tasks.tasks import RemoteFilePullTask +def test_normalize_switch_name(): + assert normalize_switch_name("Super Mario Odyssey") == "super mario odyssey" + # Punctuation and casing collapse so file names and TitleDB names align. + assert normalize_switch_name( + "The Legend of Zelda: Breath of the Wild" + ) == normalize_switch_name("the legend of zelda breath of the wild") + assert normalize_switch_name(" ") == "" + + +class TestSwitchNameToIdIndex: + @pytest.fixture + def data(self): + return { + # Base game -> indexed. + "0100000000010000": {"id": "0100000000010000", "name": "Base Game"}, + # Update (0x800 set) -> skipped. + "0100000000010800": {"id": "0100000000010800", "name": "Base Game"}, + # DLC (0x1000 set) -> skipped. + "0100000000011000": {"id": "0100000000011000", "name": "DLC Pack"}, + # Two base titles sharing a name -> ambiguous, dropped from index. + "0100000000020000": {"id": "0100000000020000", "name": "Ambiguous"}, + "0100000000030000": {"id": "0100000000030000", "name": "Ambiguous"}, + } + + @patch.object(RemoteFilePullTask, "run") + @patch("tasks.scheduled.update_switch_titledb.async_cache.pipeline") + async def test_builds_unique_base_index( + self, mock_pipeline, mock_super_run, data + ): + mock_super_run.return_value = json.dumps(data).encode("utf-8") + + mock_pipe = AsyncMock() + mock_pipe.__aenter__ = AsyncMock(return_value=mock_pipe) + mock_pipe.__aexit__ = AsyncMock(return_value=None) + mock_pipeline.return_value = mock_pipe + + await UpdateSwitchTitleDBTask().run(force=True) + + name_index: dict[str, str] = {} + for call in mock_pipe.hset.call_args_list: + if call[0][0] == SWITCH_NAME_TO_ID_KEY: + name_index.update(call[1]["mapping"]) + + assert name_index == {"base game": "0100000000010000"} + + class TestUpdateSwitchTitleDBTask: @pytest.fixture def task(self): diff --git a/env.template b/env.template index c7c980ece7..10c5778726 100644 --- a/env.template +++ b/env.template @@ -86,6 +86,7 @@ ENABLE_SCHEDULED_RESCAN=false # Enable scheduled library re-scans SCHEDULED_RESCAN_CRON=0 3 * * * # Cron expression for scheduled re-scans ENABLE_SCHEDULED_UPDATE_SWITCH_TITLEDB=false # Enable scheduled Switch TitleDB index updates SCHEDULED_UPDATE_SWITCH_TITLEDB_CRON=0 4 * * * # Cron expression for scheduled Switch TitleDB updates +ENABLE_SWITCH_TITLE_ID_RENAME=false # During a scan, rename Switch files lacking a title ID to embed one (needs the Switch TitleDB index) ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA=false # Enable scheduled LaunchBox metadata updates SCHEDULED_UPDATE_LAUNCHBOX_METADATA_CRON=0 4 * * * # Cron expression for scheduled LaunchBox metadata updates ENABLE_SCHEDULED_CONVERT_IMAGES_TO_WEBP=false # Enable scheduled conversion of images to WebP