diff --git a/backend/endpoints/sockets/scan.py b/backend/endpoints/sockets/scan.py index d45acc341..7e07c16cd 100644 --- a/backend/endpoints/sockets/scan.py +++ b/backend/endpoints/sockets/scan.py @@ -414,6 +414,16 @@ async def _identify_rom( _added_rom.gamelist_metadata[f"{media_type.value}_path"], ) + # Handle special media files from LaunchBox + if _added_rom.launchbox_metadata and MetadataSource.LAUNCHBOX in metadata_sources: + preferred_media_types = get_preferred_media_types() + for media_type in preferred_media_types: + if _added_rom.launchbox_metadata.get(f"{media_type.value}_path"): + await fs_resource_handler.store_media_file( + _added_rom.launchbox_metadata[f"{media_type.value}_url"], + _added_rom.launchbox_metadata[f"{media_type.value}_path"], + ) + # Store normal and locked badges if _added_rom.ra_metadata and MetadataSource.RA in metadata_sources: for ach in _added_rom.ra_metadata.get("achievements", []): diff --git a/backend/handler/filesystem/__init__.py b/backend/handler/filesystem/__init__.py index d11d2df22..7430823aa 100644 --- a/backend/handler/filesystem/__init__.py +++ b/backend/handler/filesystem/__init__.py @@ -1,5 +1,6 @@ from .assets_handler import FSAssetsHandler from .firmware_handler import FSFirmwareHandler +from .launchbox_handler import FSLaunchboxHandler from .platforms_handler import FSPlatformsHandler from .resources_handler import FSResourcesHandler from .roms_handler import FSRomsHandler @@ -11,3 +12,4 @@ fs_platform_handler = FSPlatformsHandler() fs_rom_handler = FSRomsHandler() fs_resource_handler = FSResourcesHandler() fs_sync_handler = FSSyncHandler() +fs_launchbox_handler = FSLaunchboxHandler() diff --git a/backend/handler/filesystem/launchbox_handler.py b/backend/handler/filesystem/launchbox_handler.py new file mode 100644 index 000000000..9c34f065b --- /dev/null +++ b/backend/handler/filesystem/launchbox_handler.py @@ -0,0 +1,9 @@ +from pathlib import Path + +from config import ROMM_BASE_PATH +from handler.filesystem.base_handler import FSHandler + + +class FSLaunchboxHandler(FSHandler): + def __init__(self) -> None: + super().__init__(base_path=str(Path(ROMM_BASE_PATH) / "launchbox")) diff --git a/backend/handler/filesystem/resources_handler.py b/backend/handler/filesystem/resources_handler.py index d92f2b003..680b6c279 100644 --- a/backend/handler/filesystem/resources_handler.py +++ b/backend/handler/filesystem/resources_handler.py @@ -19,6 +19,32 @@ from utils.validation import validate_url_for_http_request from .base_handler import CoverSize, FSHandler +LOCAL_FILE_SCHEMES = ("file://", "launchbox-file://") + + +def _resolve_local_file_uri(uri: str) -> Path | None: + """Resolve a local-file URI to an absolute Path, or None if unsafe/unknown. + + `file://` resolves under the ROM library root. `launchbox-file://` resolves + under the LaunchBox data root — LaunchBox metadata produces paths relative + to `/romm/launchbox`, which is not the same as the library root. + """ + from handler.filesystem import fs_launchbox_handler, fs_rom_handler + + if uri.startswith("launchbox-file://"): + try: + return fs_launchbox_handler.validate_path(uri[len("launchbox-file://") :]) + except ValueError: + return None + + if uri.startswith("file://"): + try: + return fs_rom_handler.validate_path(uri[len("file://") :]) + except ValueError: + return None + + return None + def _content_type_essence(header_value: str) -> str: """Return the MIME type token (before parameters), lowercased.""" @@ -95,27 +121,21 @@ class FSResourcesHandler(FSHandler): cover_file = f"{entity.fs_resources_path}/cover" await self.make_directory(cover_file) - # Handle file:// URLs for gamelist.xml - if url_cover.startswith("file://"): + # Handle local-file URIs from metadata handlers (gamelist, LaunchBox) + if url_cover.startswith(LOCAL_FILE_SCHEMES): try: - from handler.filesystem import fs_rom_handler - - validated = fs_rom_handler.validate_path( - url_cover[7:] # Remove "file://" prefix - ) - if await AnyioPath(validated).exists(): - # Copy the file to the resources directory - dest_path = f"{cover_file}/{size.value}.png" - await self.copy_file(validated, dest_path) - - if ENABLE_SCHEDULED_CONVERT_IMAGES_TO_WEBP: - self.image_converter.convert_to_webp( - self.validate_path(f"{cover_file}/{size.value}.png"), - force=True, - ) - else: - log.warning(f"Cover file not found: {str(validated)}") + resolved = _resolve_local_file_uri(url_cover) + if resolved is None or not await AnyioPath(resolved).exists(): + log.warning(f"Cover file not found: {url_cover}") return None + dest_path = f"{cover_file}/{size.value}.png" + await self.copy_file(resolved, dest_path) + + if ENABLE_SCHEDULED_CONVERT_IMAGES_TO_WEBP: + self.image_converter.convert_to_webp( + self.validate_path(f"{cover_file}/{size.value}.png"), + force=True, + ) except Exception as exc: log.error(f"Unable to copy cover file {url_cover}: {str(exc)}") return None @@ -275,20 +295,14 @@ class FSResourcesHandler(FSHandler): screenshot_path = f"{rom.fs_resources_path}/screenshots" await self.make_directory(screenshot_path) - # Handle file:// URLs for gamelist.xml - if url_screenhot.startswith("file://"): + # Handle local-file URIs from metadata handlers (gamelist, LaunchBox) + if url_screenhot.startswith(LOCAL_FILE_SCHEMES): try: - from handler.filesystem import fs_rom_handler - - validated = fs_rom_handler.validate_path( - url_screenhot[7:] # Remove "file://" prefix - ) - if await AnyioPath(validated).exists(): - # Copy the file to the resources directory - await self.copy_file(validated, f"{screenshot_path}/{idx}.jpg") - else: - log.warning(f"Screenshot file not found: {str(validated)}") + resolved = _resolve_local_file_uri(url_screenhot) + if resolved is None or not await AnyioPath(resolved).exists(): + log.warning(f"Screenshot file not found: {url_screenhot}") return None + await self.copy_file(resolved, f"{screenshot_path}/{idx}.jpg") except Exception as exc: log.error(f"Unable to copy screenshot file {url_screenhot}: {str(exc)}") return None @@ -395,20 +409,14 @@ class FSResourcesHandler(FSHandler): manual_path = f"{rom.fs_resources_path}/manual" await self.make_directory(manual_path) - # Handle file:// URLs for gamelist.xml - if url_manual.startswith("file://"): + # Handle local-file URIs from metadata handlers (gamelist, LaunchBox) + if url_manual.startswith(LOCAL_FILE_SCHEMES): try: - from handler.filesystem import fs_rom_handler - - validated = fs_rom_handler.validate_path( - url_manual[7:] # Remove "file://" prefix - ) - if await AnyioPath(validated).exists(): - # Copy the file to the resources directory - await self.copy_file(validated, f"{manual_path}/{rom.id}.pdf") - else: - log.warning(f"Manual file not found: {str(validated)}") + resolved = _resolve_local_file_uri(url_manual) + if resolved is None or not await AnyioPath(resolved).exists(): + log.warning(f"Manual file not found: {url_manual}") return None + await self.copy_file(resolved, f"{manual_path}/{rom.id}.pdf") except Exception as exc: log.error(f"Unable to copy manual file {url_manual}: {str(exc)}") return None @@ -542,17 +550,12 @@ class FSResourcesHandler(FSHandler): # Ensure destination directory exists await self.make_directory(directory) - # Handle file:// URLs for gamelist.xml - if url_media.startswith("file://"): + # Handle local-file URIs from metadata handlers (gamelist, LaunchBox) + if url_media.startswith(LOCAL_FILE_SCHEMES): try: - from handler.filesystem import fs_rom_handler - - validated = fs_rom_handler.validate_path( - url_media[7:] # Remove "file://" prefix - ) - file_path = AnyioPath(validated) - if await file_path.exists(): - await self.copy_file(Path(str(file_path)), dest_path) + resolved = _resolve_local_file_uri(url_media) + if resolved is not None and await AnyioPath(resolved).exists(): + await self.copy_file(resolved, dest_path) except Exception as exc: log.error(f"Unable to copy media file {url_media}: {str(exc)}") return None diff --git a/backend/handler/metadata/launchbox_handler/handler.py b/backend/handler/metadata/launchbox_handler/handler.py index 0cd9d5f6c..c164ddd98 100644 --- a/backend/handler/metadata/launchbox_handler/handler.py +++ b/backend/handler/metadata/launchbox_handler/handler.py @@ -128,6 +128,16 @@ class LaunchboxHandler(MetadataHandler): else: search_term = fs_name + # Resolve MAME arcade filename (e.g. wrlok_l3.zip) to its full title + # via LaunchBox's Mame.xml before name-based lookup. + if platform_slug == UPS.ARCADE: + mame_entry = await self._remote.get_mame_entry(fs_name) + if mame_entry: + name = (mame_entry.get("Name") or "").strip() + if name: + search_term = name + fallback_rom = LaunchboxRom(launchbox_id=None, name=name) + # We replace " - "/"- " with ": " to match Launchbox's naming convention search_term = re.sub(DASH_COLON_REGEX, ": ", search_term).lower() @@ -152,6 +162,8 @@ class LaunchboxHandler(MetadataHandler): remote=index_entry, remote_images=remote_images, remote_enabled=remote_available, + platform_name=get_platform(platform_slug).get("name"), + fs_name=fs_name, ) return build_rom( @@ -162,7 +174,12 @@ class LaunchboxHandler(MetadataHandler): ) async def get_rom_by_id( - self, database_id: int, *, remote_enabled: bool = True + self, + database_id: int, + *, + remote_enabled: bool = True, + fs_name: str | None = None, + platform_slug: str | None = None, ) -> LaunchboxRom: if not self.is_enabled(): return LaunchboxRom(launchbox_id=None) @@ -174,17 +191,42 @@ class LaunchboxHandler(MetadataHandler): if not remote: return LaunchboxRom(launchbox_id=None) + # Merge local-only fields when a local LaunchBox install has the same game + local: dict[str, str] | None = None + if fs_name and platform_slug: + candidate = await self._local.get_rom(fs_name, platform_slug) + if ( + candidate is not None + and safe_int(candidate.get("DatabaseID")) == database_id + ): + local = candidate + + platform_name = ( + get_platform(platform_slug).get("name") if platform_slug else None + ) remote_images = await self._remote.fetch_images( remote=remote, remote_enabled=remote_enabled ) - media_req = remote_media_req( - remote=remote, - remote_images=remote_images, - remote_enabled=remote_enabled, - ) + if local is not None: + media_req = local_media_req( + platform_name=platform_name, + fs_name=fs_name or "", + local=local, + remote=remote, + remote_images=remote_images, + remote_enabled=remote_enabled, + ) + else: + media_req = remote_media_req( + remote=remote, + remote_images=remote_images, + remote_enabled=remote_enabled, + platform_name=platform_name, + fs_name=fs_name or "", + ) return build_rom( - local=None, + local=local, remote=remote, launchbox_id=database_id, media_req=media_req, diff --git a/backend/handler/metadata/launchbox_handler/media.py b/backend/handler/metadata/launchbox_handler/media.py index 889690b8f..d4759d268 100644 --- a/backend/handler/metadata/launchbox_handler/media.py +++ b/backend/handler/metadata/launchbox_handler/media.py @@ -1,10 +1,18 @@ from pathlib import Path +from typing import TYPE_CHECKING +from config.config_manager import MetadataMediaType +from handler.filesystem import fs_resource_handler +from handler.metadata.ss_handler import get_preferred_media_types from utils.database import safe_str_to_bool +if TYPE_CHECKING: + from models.rom import Rom + from .types import ( LAUNCHBOX_IMAGES_DIR, LAUNCHBOX_MANUALS_DIR, + LAUNCHBOX_VIDEOS_DIR, LaunchboxImage, LaunchboxMetadata, LaunchboxRom, @@ -22,6 +30,8 @@ from .utils import ( sanitize_filename, ) +VIDEO_EXTS: tuple[str, ...] = (".mp4", ".webm", ".avi", ".mkv", ".mov", ".wmv") + def local_media_req( *, @@ -49,11 +59,19 @@ def remote_media_req( remote: dict | None, remote_images: list[dict] | None, remote_enabled: bool, + platform_name: str | None = None, + fs_name: str = "", ) -> MediaRequest: title = ((remote or {}).get("Name") or "").strip() + # Without a platform_name, _build_local_media_context bails and local + # Images/Manuals/Videos never get searched. Fall back to the platform + # recorded on the remote entry so remote-matched ROMs can still surface + # on-disk media. + if not platform_name and remote: + platform_name = (remote.get("Platform") or "").strip() or None return MediaRequest( - None, - "", + platform_name, + fs_name, title, None, remote_images, @@ -324,6 +342,28 @@ def _get_manuals(req: MediaRequest) -> str | None: return manual +def _get_video(req: MediaRequest) -> str | None: + """Resolve a local LaunchBox video for the given ROM. + + LaunchBox stores videos flat under `Videos//.` + (no region or category subdirectories). Return a `launchbox-file://` URL + to the first match, or None. + """ + ctx = _build_local_media_context( + req, LAUNCHBOX_VIDEOS_DIR, include_region_hints=False + ) + if ctx is None: + return None + + for stem in ctx["stems"]: + for ext in VIDEO_EXTS: + candidate = ctx["base"] / f"{stem}{ext}" + if candidate.is_file(): + return file_uri_for_local_path(candidate) + + return None + + def _get_images(req: MediaRequest) -> list[LaunchboxImage]: images: list[LaunchboxImage] = [] @@ -495,6 +535,30 @@ def build_launchbox_metadata( ) +def populate_rom_specific_paths( + metadata: LaunchboxMetadata, rom: "Rom" +) -> LaunchboxMetadata: + """Populate rom-specific media paths on a LaunchBox metadata dict. + + Called after the Rom is known (in the scan pipeline) to compute the + destination path for local media that the handler surfaced a URL for. + Currently just covers video. + """ + if ( + MetadataMediaType.VIDEO in get_preferred_media_types() + and "video_url" in metadata + and metadata.get("video_url") + ): + base = fs_resource_handler.get_media_resources_path( + rom.platform_id, rom.id, MetadataMediaType.VIDEO + ) + ext = Path(metadata["video_url"]).suffix.lower() + if ext not in VIDEO_EXTS: + ext = ".mp4" + metadata["video_path"] = f"{base}/video{ext}" + return metadata + + def build_rom( *, local: dict[str, str] | None, @@ -509,10 +573,12 @@ def build_rom( url_cover: str | None = None url_screenshots: list[str] = [] url_manual: str | None = None + video_url: str | None = None if media_req is not None: url_cover = _get_cover(media_req) url_screenshots = _get_screenshots(media_req) url_manual = _get_manuals(media_req) + video_url = _get_video(media_req) url_screenshots = url_screenshots or [] name = ( @@ -532,6 +598,13 @@ def build_rom( ).strip() launchbox_id = int(launchbox_id) if launchbox_id is not None else None + metadata = build_launchbox_metadata( + local=local, + remote=remote, + images=images, + ) + if video_url: + metadata["video_url"] = video_url return LaunchboxRom( launchbox_id=launchbox_id, name=name, @@ -539,9 +612,5 @@ def build_rom( url_cover=url_cover or "", url_screenshots=url_screenshots, url_manual=url_manual or "", - launchbox_metadata=build_launchbox_metadata( - local=local, - remote=remote, - images=images, - ), + launchbox_metadata=metadata, ) diff --git a/backend/handler/metadata/launchbox_handler/remote_source.py b/backend/handler/metadata/launchbox_handler/remote_source.py index c2b9e66d6..3952cdca9 100644 --- a/backend/handler/metadata/launchbox_handler/remote_source.py +++ b/backend/handler/metadata/launchbox_handler/remote_source.py @@ -5,6 +5,7 @@ from logger.logger import log from .platforms import get_platform from .types import ( + LAUNCHBOX_MAME_KEY, LAUNCHBOX_METADATA_ALTERNATE_NAME_KEY, LAUNCHBOX_METADATA_DATABASE_ID_KEY, LAUNCHBOX_METADATA_IMAGE_KEY, @@ -74,6 +75,32 @@ class RemoteSource: return None + async def get_mame_entry(self, file_name: str) -> dict | None: + """Resolve a MAME arcade filename to its LaunchBox MAME entry. + + LaunchBox's Mame.xml indexes `` records by `` — the + MAME short name without an extension (e.g. `wrlok_l3` for `wrlok_l3.zip`). + The entry carries `` — the full title to search for in Metadata.xml. + """ + file_name_clean = (file_name or "").strip() + if not file_name_clean: + return None + + # Try the raw filename first, then the stem (sans extension). + candidates: list[str] = [file_name_clean] + from pathlib import Path + + stem = Path(file_name_clean).stem + if stem and stem != file_name_clean: + candidates.append(stem) + + for candidate in candidates: + entry = await async_cache.hget(LAUNCHBOX_MAME_KEY, candidate) + if entry: + return json.loads(entry) + + return None + async def fetch_images( self, *, diff --git a/backend/handler/metadata/launchbox_handler/types.py b/backend/handler/metadata/launchbox_handler/types.py index bb3033400..864b6e39a 100644 --- a/backend/handler/metadata/launchbox_handler/types.py +++ b/backend/handler/metadata/launchbox_handler/types.py @@ -54,6 +54,8 @@ class LaunchboxMetadata(TypedDict): genres: NotRequired[list[str]] companies: NotRequired[list[str]] images: list[LaunchboxImage] + video_url: NotRequired[str] + video_path: NotRequired[str] class LaunchboxRom(BaseRom): diff --git a/backend/handler/metadata/launchbox_handler/utils.py b/backend/handler/metadata/launchbox_handler/utils.py index 8c73d0892..23bc8af53 100644 --- a/backend/handler/metadata/launchbox_handler/utils.py +++ b/backend/handler/metadata/launchbox_handler/utils.py @@ -20,7 +20,7 @@ def file_uri_for_local_path(path: Path) -> str | None: relative = path.resolve().relative_to(LAUNCHBOX_LOCAL_DIR.resolve()) except ValueError: return None - return f"file://{relative.as_posix()}" + return f"launchbox-file://{relative.as_posix()}" def coalesce(*values: object | None) -> str | None: diff --git a/backend/handler/scan_handler.py b/backend/handler/scan_handler.py index 03ab3d2dd..2ff79dbbc 100644 --- a/backend/handler/scan_handler.py +++ b/backend/handler/scan_handler.py @@ -29,6 +29,7 @@ from handler.metadata.gamelist_handler import GamelistRom from handler.metadata.hasheous_handler import HASHEOUS_PLATFORM_LIST, HasheousRom from handler.metadata.hltb_handler import HLTB_PLATFORM_LIST, HLTBRom from handler.metadata.igdb_handler import IGDB_PLATFORM_LIST, IGDBRom +from handler.metadata.launchbox_handler.media import populate_rom_specific_paths from handler.metadata.launchbox_handler.platforms import LAUNCHBOX_PLATFORM_LIST from handler.metadata.launchbox_handler.types import LaunchboxRom from handler.metadata.libretro_handler import LIBRETRO_PLATFORM_LIST, LibretroRom @@ -623,15 +624,24 @@ async def scan_rom( and rom.launchbox_id and launchbox_remote_enabled ): - return await meta_launchbox_handler.get_rom_by_id( - rom.launchbox_id, remote_enabled=True + launchbox_rom = await meta_launchbox_handler.get_rom_by_id( + rom.launchbox_id, + remote_enabled=True, + fs_name=rom_attrs["fs_name"], + platform_slug=platform_slug, + ) + else: + launchbox_rom = await meta_launchbox_handler.get_rom( + rom_attrs["fs_name"], + platform_slug, + remote_enabled=launchbox_remote_enabled, ) - return await meta_launchbox_handler.get_rom( - rom_attrs["fs_name"], - platform_slug, - remote_enabled=launchbox_remote_enabled, - ) + metadata = launchbox_rom.get("launchbox_metadata") + if metadata: + populate_rom_specific_paths(metadata, rom) + + return launchbox_rom return LaunchboxRom(launchbox_id=None) diff --git a/backend/models/rom.py b/backend/models/rom.py index 88fd257d8..9b5035e4b 100644 --- a/backend/models/rom.py +++ b/backend/models/rom.py @@ -357,6 +357,7 @@ class Rom(BaseModel): (self.ss_metadata or {}).get("video_path") or (self.ss_metadata or {}).get("video_normalized_path") or (self.gamelist_metadata or {}).get("video_path") + or (self.launchbox_metadata or {}).get("video_path") ) @property diff --git a/backend/tests/handler/metadata/test_launchbox_handler.py b/backend/tests/handler/metadata/test_launchbox_handler.py index cb97207b0..dfaf5408d 100644 --- a/backend/tests/handler/metadata/test_launchbox_handler.py +++ b/backend/tests/handler/metadata/test_launchbox_handler.py @@ -19,15 +19,24 @@ from defusedxml import ElementTree as ET from handler.metadata.launchbox_handler.handler import LaunchboxHandler from handler.metadata.launchbox_handler.local_source import LocalSource -from handler.metadata.launchbox_handler.media import build_launchbox_metadata, build_rom +from handler.metadata.launchbox_handler.media import ( + _get_video, + build_launchbox_metadata, + build_rom, + populate_rom_specific_paths, + remote_media_req, +) from handler.metadata.launchbox_handler.platforms import get_platform from handler.metadata.launchbox_handler.remote_source import RemoteSource from handler.metadata.launchbox_handler.types import ( + LAUNCHBOX_MAME_KEY, LAUNCHBOX_METADATA_ALTERNATE_NAME_KEY, LAUNCHBOX_METADATA_DATABASE_ID_KEY, LAUNCHBOX_METADATA_IMAGE_KEY, LAUNCHBOX_METADATA_NAME_KEY, LaunchboxImage, + LaunchboxMetadata, + MediaRequest, ) from handler.metadata.launchbox_handler.utils import ( coalesce, @@ -520,6 +529,67 @@ class TestRemoteSourceGetRom: assert result is None +class TestRemoteSourceGetMameEntry: + @pytest.fixture + def source(self) -> RemoteSource: + return RemoteSource() + + async def test_cache_miss_returns_none(self, source: RemoteSource): + with patch.object( + async_cache, "hget", new_callable=AsyncMock, return_value=None + ): + result = await source.get_mame_entry("pacman.zip") + assert result is None + + async def test_cache_hit_returns_dict(self, source: RemoteSource): + # Real LaunchBox Mame.xml indexes by stem (e.g. `wrlok_l3`, no ext) + # and carries `` as the full title. + mame_entry = { + "FileName": "wrlok_l3", + "Name": "Warlok", + "Year": "1982", + } + with patch.object( + async_cache, + "hget", + new_callable=AsyncMock, + return_value=json.dumps(mame_entry), + ) as mock_hget: + result = await source.get_mame_entry("wrlok_l3") + mock_hget.assert_called_once_with(LAUNCHBOX_MAME_KEY, "wrlok_l3") + assert result == mame_entry + + async def test_falls_back_to_stem_when_given_filename_with_ext( + self, source: RemoteSource + ): + mame_entry = {"FileName": "wrlok_l3", "Name": "Warlok"} + + async def fake_hget(_key, field): + return json.dumps(mame_entry) if field == "wrlok_l3" else None + + with patch.object( + async_cache, "hget", new_callable=AsyncMock, side_effect=fake_hget + ) as mock_hget: + result = await source.get_mame_entry("wrlok_l3.zip") + # First lookup by raw filename (miss), then by stem (hit). + assert mock_hget.call_count == 2 + assert mock_hget.call_args_list[0].args == (LAUNCHBOX_MAME_KEY, "wrlok_l3.zip") + assert mock_hget.call_args_list[1].args == (LAUNCHBOX_MAME_KEY, "wrlok_l3") + assert result == mame_entry + + async def test_empty_input_returns_none(self, source: RemoteSource): + result = await source.get_mame_entry("") + assert result is None + + async def test_whitespace_stripped(self, source: RemoteSource): + with patch.object( + async_cache, "hget", new_callable=AsyncMock, return_value=None + ) as mock_hget: + await source.get_mame_entry(" wrlok_l3 ") + # First call uses the trimmed filename. + assert mock_hget.call_args_list[0].args == (LAUNCHBOX_MAME_KEY, "wrlok_l3") + + class TestRemoteSourceFetchImages: @pytest.fixture def source(self) -> RemoteSource: @@ -636,6 +706,255 @@ class TestBuildLaunchboxMetadata: assert meta.get("images", []) == images +class TestGetVideo: + @pytest.fixture + def videos_dir(self, tmp_path: Path, monkeypatch) -> Path: + videos_root = tmp_path / "Videos" + videos_root.mkdir() + monkeypatch.setattr( + "handler.metadata.launchbox_handler.media.LAUNCHBOX_VIDEOS_DIR", + videos_root, + ) + # file_uri_for_local_path is rooted at LAUNCHBOX_LOCAL_DIR: patch so our + # tmp videos sit under it and produce a well-formed URL. + monkeypatch.setattr( + "handler.metadata.launchbox_handler.utils.LAUNCHBOX_LOCAL_DIR", + tmp_path, + ) + return videos_root + + def _req( + self, fs_name: str, title: str = "", platform: str = "NES" + ) -> MediaRequest: + return MediaRequest( + platform_name=platform, + fs_name=fs_name, + title=title, + region_hint=None, + remote_images=None, + remote_enabled=False, + ) + + def test_no_platform_returns_none(self, videos_dir: Path): + req = MediaRequest( + platform_name=None, + fs_name="game.nes", + title="", + region_hint=None, + remote_images=None, + remote_enabled=False, + ) + assert _get_video(req) is None + + def test_missing_platform_dir_returns_none(self, videos_dir: Path): + # videos_dir has no "NES" subdirectory + assert _get_video(self._req("mario.nes")) is None + + def test_finds_mp4_by_fs_stem(self, videos_dir: Path): + platform_dir = videos_dir / "NES" + platform_dir.mkdir() + (platform_dir / "Mario.mp4").write_bytes(b"") + + url = _get_video(self._req("Mario.nes")) + assert url == "launchbox-file://Videos/NES/Mario.mp4" + + def test_falls_back_to_title_stem(self, videos_dir: Path): + platform_dir = videos_dir / "NES" + platform_dir.mkdir() + (platform_dir / "Super Mario Bros.webm").write_bytes(b"") + + url = _get_video(self._req("roms-mario-1.nes", title="Super Mario Bros")) + assert url == "launchbox-file://Videos/NES/Super Mario Bros.webm" + + def test_multiple_extensions_tried(self, videos_dir: Path): + platform_dir = videos_dir / "NES" + platform_dir.mkdir() + (platform_dir / "Mario.mkv").write_bytes(b"") + + url = _get_video(self._req("Mario.nes")) + assert url is not None + assert url.endswith(".mkv") + + def test_no_match_returns_none(self, videos_dir: Path): + platform_dir = videos_dir / "NES" + platform_dir.mkdir() + (platform_dir / "Zelda.mp4").write_bytes(b"") + + assert _get_video(self._req("Mario.nes")) is None + + +class TestPopulateRomSpecificPaths: + def _rom(self) -> MagicMock: + rom = MagicMock() + rom.platform_id = 7 + rom.id = 42 + return rom + + def test_no_video_url_is_noop(self): + metadata: LaunchboxMetadata = {"first_release_date": None, "images": []} + with patch( + "handler.metadata.launchbox_handler.media.get_preferred_media_types" + ) as mock_preferred: + from config.config_manager import MetadataMediaType + + mock_preferred.return_value = [MetadataMediaType.VIDEO] + populate_rom_specific_paths(metadata, self._rom()) + assert "video_path" not in metadata + + def test_video_url_populates_path(self): + metadata: LaunchboxMetadata = { + "first_release_date": None, + "images": [], + "video_url": "launchbox-file://Videos/NES/Mario.mp4", + } + with patch( + "handler.metadata.launchbox_handler.media.get_preferred_media_types" + ) as mock_preferred: + from config.config_manager import MetadataMediaType + + mock_preferred.return_value = [MetadataMediaType.VIDEO] + populate_rom_specific_paths(metadata, self._rom()) + path = metadata.get("video_path", "") + assert path.endswith("/video.mp4") + assert "7" in path and "42" in path + + def test_video_path_preserves_source_extension(self): + from config.config_manager import MetadataMediaType + + for src_ext, expected in ( + (".mkv", "/video.mkv"), + (".webm", "/video.webm"), + (".MOV", "/video.mov"), + ): + metadata: LaunchboxMetadata = { + "first_release_date": None, + "images": [], + "video_url": f"launchbox-file://Videos/NES/Mario{src_ext}", + } + with patch( + "handler.metadata.launchbox_handler.media.get_preferred_media_types" + ) as mock_preferred: + mock_preferred.return_value = [MetadataMediaType.VIDEO] + populate_rom_specific_paths(metadata, self._rom()) + assert metadata.get("video_path", "").endswith(expected) + + def test_video_not_in_preferred_media_skips(self): + metadata: LaunchboxMetadata = { + "first_release_date": None, + "images": [], + "video_url": "launchbox-file://Videos/NES/Mario.mp4", + } + with patch( + "handler.metadata.launchbox_handler.media.get_preferred_media_types" + ) as mock_preferred: + mock_preferred.return_value = [] + populate_rom_specific_paths(metadata, self._rom()) + assert "video_path" not in metadata + + +class TestRemoteMediaReq: + def test_explicit_platform_name_wins(self): + req = remote_media_req( + remote={"Name": "Super Mario Bros.", "Platform": "Wii"}, + remote_images=None, + remote_enabled=True, + platform_name="Nintendo Entertainment System", + fs_name="mario.nes", + ) + assert req.platform_name == "Nintendo Entertainment System" + assert req.fs_name == "mario.nes" + assert req.title == "Super Mario Bros." + + def test_falls_back_to_remote_platform(self): + req = remote_media_req( + remote={"Name": "H.E.R.O.", "Platform": "Atari 2600"}, + remote_images=None, + remote_enabled=True, + ) + assert req.platform_name == "Atari 2600" + assert req.fs_name == "" + assert req.title == "H.E.R.O." + + def test_no_platform_available_is_none(self): + req = remote_media_req( + remote={"Name": "Some Game"}, + remote_images=None, + remote_enabled=True, + ) + assert req.platform_name is None + + +class TestRemoteMatchLocalImages: + """Regression test for bug where remote-matched roms skipped local image lookup. + + When a ROM matches only via the remote Metadata.xml (no local XML entry), + the handler previously passed platform_name=None to remote_media_req, which + caused _build_local_media_context to bail and never search on-disk images. + """ + + async def test_remote_match_finds_local_images( + self, tmp_path: Path, monkeypatch + ) -> None: + # Arrange: images on disk, no local XML match, remote returns a hit. + # Use "Clear Logo" + "Box - Front" to exercise both _get_images and + # _get_cover through the same remote-match path. + lb_root = tmp_path / "launchbox" + images_root = lb_root / "Images" + logo_dir = images_root / "Atari 2600" / "Clear Logo" + box_dir = images_root / "Atari 2600" / "Box - Front" + logo_dir.mkdir(parents=True) + box_dir.mkdir(parents=True) + (logo_dir / "H.E.R.O-01.png").write_bytes(b"") + (box_dir / "H.E.R.O-01.png").write_bytes(b"") + + monkeypatch.setattr( + "handler.metadata.launchbox_handler.media.LAUNCHBOX_IMAGES_DIR", + images_root, + ) + monkeypatch.setattr( + "handler.metadata.launchbox_handler.utils.LAUNCHBOX_LOCAL_DIR", + lb_root, + ) + + h = LaunchboxHandler() + h._local = MagicMock(spec=LocalSource) + h._remote = MagicMock(spec=RemoteSource) + h._local.get_rom = AsyncMock(return_value=None) # type: ignore[method-assign] + h._remote.get_mame_entry = AsyncMock(return_value=None) # type: ignore[method-assign] + h._remote.get_rom = AsyncMock( # type: ignore[method-assign] + return_value={ + "DatabaseID": "42", + "Name": "H.E.R.O.", + "Platform": "Atari 2600", + } + ) + h._remote.fetch_images = AsyncMock(return_value=None) # type: ignore[method-assign] + monkeypatch.setattr(LaunchboxHandler, "is_enabled", lambda *_: True) + monkeypatch.setattr(async_cache, "exists", AsyncMock(return_value=True)) + + with patch( + "handler.metadata.launchbox_handler.handler.fs_rom_handler" + ) as mock_fs: + mock_fs.get_file_name_with_no_tags.return_value = "hero" + result = await h.get_rom("hero.a26", "atari2600") + + # Assert: both the clear logo and box-front cover resolved to + # launchbox-file:// URLs, even though the local XML never matched. + assert "launchbox_metadata" in result + images = result["launchbox_metadata"]["images"] + assert len(images) == 1 + assert "type" in images[0] + assert "url" in images[0] + assert images[0]["type"] == "Clear Logo" + assert images[0]["url"] == ( + "launchbox-file://Images/Atari 2600/Clear Logo/H.E.R.O-01.png" + ) + assert "url_cover" in result + assert result["url_cover"] == ( + "launchbox-file://Images/Atari 2600/Box - Front/H.E.R.O-01.png" + ) + + class TestBuildRom: def test_name_from_local_title(self): local = {"Title": "Super Mario Bros.", "Notes": "Classic platformer"} @@ -739,6 +1058,7 @@ class TestLaunchboxHandlerGetRom: h._local.get_rom = AsyncMock(return_value=None) # type: ignore[method-assign] h._remote.get_rom = AsyncMock(return_value=None) # type: ignore[method-assign] h._remote.get_by_id = AsyncMock(return_value=None) # type: ignore[method-assign] + h._remote.get_mame_entry = AsyncMock(return_value=None) # type: ignore[method-assign] h._remote.fetch_images = AsyncMock(return_value=None) # type: ignore[method-assign] monkeypatch.setattr(LaunchboxHandler, "is_enabled", lambda *_: True) monkeypatch.setattr(async_cache, "exists", AsyncMock(return_value=True)) @@ -868,6 +1188,92 @@ class TestLaunchboxHandlerGetRom: # fs_rom_handler.get_file_name_with_no_tags should NOT be called mock_fs.get_file_name_with_no_tags.assert_not_called() + async def test_arcade_mame_resolves_shortname_to_full_title( + self, handler: LaunchboxHandler + ): + mame_entry = {"FileName": "wrlok_l3", "Name": "Warlok"} + remote_entry = {"DatabaseID": "999", "Name": "Warlok"} + with ( + patch.object( + handler._remote, + "get_mame_entry", + new=AsyncMock(return_value=mame_entry), + ) as mock_mame, + patch.object( + handler._remote, + "get_rom", + new=AsyncMock(return_value=remote_entry), + ) as mock_get_rom, + patch( + "handler.metadata.launchbox_handler.handler.fs_rom_handler" + ) as mock_fs, + ): + mock_fs.get_file_name_with_no_tags.return_value = "wrlok_l3" + result = await handler.get_rom("wrlok_l3.zip", "arcade") + + mock_mame.assert_called_once_with("wrlok_l3.zip") + # Search term should be the MAME Name, lowercased. + assert mock_get_rom.call_args.args[0] == "warlok" + assert result.get("name", None) == "Warlok" + assert result.get("launchbox_id", None) == 999 + + async def test_arcade_mame_miss_falls_back_to_filename_search( + self, handler: LaunchboxHandler + ): + with ( + patch.object( + handler._remote, + "get_mame_entry", + new=AsyncMock(return_value=None), + ) as mock_mame, + patch( + "handler.metadata.launchbox_handler.handler.fs_rom_handler" + ) as mock_fs, + ): + mock_fs.get_file_name_with_no_tags.return_value = "wrlok_l3" + result = await handler.get_rom("wrlok_l3.zip", "arcade") + + mock_mame.assert_called_once_with("wrlok_l3.zip") + assert result["launchbox_id"] is None + + async def test_arcade_mame_only_match_sets_fallback_name( + self, handler: LaunchboxHandler + ): + # MAME entry exists but Metadata.xml has no matching game: still surface + # the MAME name as the rom name. + mame_entry = {"FileName": "wrlok_l3", "Name": "Warlok"} + with ( + patch.object( + handler._remote, + "get_mame_entry", + new=AsyncMock(return_value=mame_entry), + ), + patch( + "handler.metadata.launchbox_handler.handler.fs_rom_handler" + ) as mock_fs, + ): + mock_fs.get_file_name_with_no_tags.return_value = "wrlok_l3" + result = await handler.get_rom("wrlok_l3.zip", "arcade") + + assert result["launchbox_id"] is None + assert result.get("name", None) == "Warlok" + + async def test_non_arcade_platform_skips_mame_lookup( + self, handler: LaunchboxHandler + ): + with ( + patch.object( + handler._remote, "get_mame_entry", new=AsyncMock() + ) as mock_mame, + patch( + "handler.metadata.launchbox_handler.handler.fs_rom_handler" + ) as mock_fs, + ): + mock_fs.get_file_name_with_no_tags.return_value = "wrlok_l3" + await handler.get_rom("wrlok_l3.zip", "nes") + + mock_mame.assert_not_called() + class TestLaunchboxHandlerGetRomById: @pytest.fixture @@ -923,6 +1329,7 @@ class TestLaunchboxHandlerSearch: h._local.get_rom = AsyncMock(return_value=None) # type: ignore[method-assign] h._remote.get_rom = AsyncMock(return_value=None) # type: ignore[method-assign] h._remote.get_by_id = AsyncMock(return_value=None) # type: ignore[method-assign] + h._remote.get_mame_entry = AsyncMock(return_value=None) # type: ignore[method-assign] h._remote.fetch_images = AsyncMock(return_value=None) # type: ignore[method-assign] monkeypatch.setattr(LaunchboxHandler, "is_enabled", lambda *_: True) monkeypatch.setattr(async_cache, "exists", AsyncMock(return_value=True))