mirror of
https://github.com/rommapp/romm.git
synced 2026-06-28 14:56:01 +00:00
Merge pull request #3286 from rommapp/local-lb-fix
Fix local LaunchBox handler: paths, arcade, videos, remote-match images
This commit is contained in:
@@ -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", []):
|
||||
|
||||
@@ -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()
|
||||
|
||||
9
backend/handler/filesystem/launchbox_handler.py
Normal file
9
backend/handler/filesystem/launchbox_handler.py
Normal file
@@ -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"))
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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/<Platform>/<GameStem>.<ext>`
|
||||
(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,
|
||||
)
|
||||
|
||||
@@ -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 `<MameFile>` records by `<FileName>` — the
|
||||
MAME short name without an extension (e.g. `wrlok_l3` for `wrlok_l3.zip`).
|
||||
The entry carries `<Name>` — 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,
|
||||
*,
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 `<Name>` 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))
|
||||
|
||||
Reference in New Issue
Block a user