Merge pull request #3286 from rommapp/local-lb-fix

Fix local LaunchBox handler: paths, arcade, videos, remote-match images

Author: gantoine

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", []):

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_rom_handler = FSRomsHandler()
 fs_resource_handler = FSResourcesHandler()
 fs_sync_handler = FSSyncHandler()
+fs_launchbox_handler = FSLaunchboxHandler()

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"))

backend/handler/filesystem/resources_handler.py

@@ -19,6 +19,32 @@
 
 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 @@ async def _store_cover(
         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 @@ async def _store_screenshot(self, rom: Rom, url_screenhot: str, idx: int):
         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 @@ async def _store_manual(self, rom: Rom, url_manual: str):
         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 @@ async def store_media_file(self, url_media: str, dest_path: str) -> None:
         # 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

backend/handler/metadata/launchbox_handler/handler.py

@@ -128,6 +128,16 @@ async def get_rom(
         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 @@ async def get_rom(
             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 @@ async def get_rom(
         )
 
     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 @@ async def get_rom_by_id(
         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,