feat: Physical games

backend/alembic/versions/0099_physical_roms.py

@@ -0,0 +1,35 @@
+"""Add physical-game columns to roms.
+
+Revision ID: 0095_physical_roms
+Revises: 0094_track_meta_table
+Create Date: 2026-07-04 00:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0095_physical_roms"
+down_revision = "0094_track_meta_table"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("roms", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "is_physical",
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.false(),
+            )
+        )
+        batch_op.add_column(sa.Column("upc", sa.String(length=64), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("roms", schema=None) as batch_op:
+        batch_op.drop_column("upc")
+        batch_op.drop_column("is_physical")

backend/config/__init__.py

@@ -128,6 +128,18 @@ def _get_env(var: str, fallback: str | None = None) -> str | None:
 # HOWLONGTOBEAT
 HLTB_API_ENABLED: Final[bool] = safe_str_to_bool(_get_env("HLTB_API_ENABLED"))
 
+# UPC LOOKUP (barcode -> title, used when adding physical games by UPC).
+# Defaults to the free UPCitemdb trial endpoint, which works without a key but
+# is heavily rate-limited. Set UPC_LOOKUP_API_KEY (and optionally a base URL) to
+# use the paid tier.
+UPC_LOOKUP_ENABLED: Final[bool] = safe_str_to_bool(
+    _get_env("UPC_LOOKUP_ENABLED", "true")
+)
+UPC_LOOKUP_API_KEY: Final[str | None] = _get_env("UPC_LOOKUP_API_KEY")
+UPC_LOOKUP_BASE_URL: Final[str] = _get_env(
+    "UPC_LOOKUP_BASE_URL", "https://api.upcitemdb.com/prod/trial"
+)
+
 # AUTH
 ROMM_AUTH_SECRET_KEY: Final[str] = _get_env("ROMM_AUTH_SECRET_KEY", "")
 if not ROMM_AUTH_SECRET_KEY:

backend/endpoints/responses/rom.py

@@ -339,6 +339,8 @@ class RomSchema(BaseModel):
     created_at: UTCDatetime
     updated_at: UTCDatetime
     missing_from_fs: bool
+    is_physical: bool
+    upc: str | None
     has_notes: bool
 
     rom_user: RomUserSchema

backend/endpoints/roms/__init__.py

@@ -52,10 +52,11 @@
     assert_rom_visible,
     get_permissions,
 )
-from handler.database import db_rom_handler, db_save_handler
+from handler.database import db_platform_handler, db_rom_handler, db_save_handler
 from handler.database.base_handler import sync_session
 from handler.filesystem import fs_resource_handler, fs_rom_handler
 from handler.filesystem.assets_handler import validate_image_upload
+from handler.filesystem.roms_handler import FSRom
 from handler.metadata import (
     meta_flashpoint_handler,
     meta_igdb_handler,
@@ -64,10 +65,19 @@
     meta_playmatch_handler,
     meta_ra_handler,
     meta_ss_handler,
+    meta_upc_handler,
 )
 from handler.metadata.launchbox_handler.media import populate_rom_specific_paths
 from handler.metadata.ss_handler import add_ss_auth_to_url, get_preferred_media_types
 from handler.rom_conversion import promote_single_file_to_folder
+from handler.scan_handler import (
+    MetadataSource,
+    ScanType,
+    build_physical_fs_name,
+    build_physical_fs_path,
+    download_rom_resources,
+    scan_rom,
+)
 from logger.formatter import BLUE
 from logger.formatter import highlight as hl
 from logger.logger import log
@@ -1440,6 +1450,127 @@ async def build_zip_in_memory() -> bytes:
     )
 
 
+class PhysicalRomCreateForm(BaseModel):
+    platform_id: int = Field(..., ge=1, description="Platform the game belongs to.")
+    name: str | None = Field(
+        default=None, description="Game name to match metadata against."
+    )
+    upc: str | None = Field(
+        default=None, description="UPC/EAN/barcode of the physical copy."
+    )
+    metadata_sources: list[str] | None = Field(
+        default=None,
+        description="Metadata providers to match against; defaults to all enabled.",
+    )
+
+
+@protected_route(
+    router.post,
+    "/physical",
+    [Scope.ROMS_WRITE],
+    responses={status.HTTP_404_NOT_FOUND: {}},
+)
+async def create_physical_rom(
+    request: Request,
+    form_data: Annotated[PhysicalRomCreateForm, Body()],
+) -> DetailedRomSchema:
+    """Manually add a physical game and auto-link its metadata (a single quick scan).
+
+    The game has no file on disk; it is stored as a `Rom` with `is_physical=True`
+    and a synthetic filesystem name, then matched by name against the metadata
+    providers exactly like a one-title scan.
+    """
+    platform = db_platform_handler.get_platform(form_data.platform_id)
+    if not platform:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Platform with id {form_data.platform_id} not found",
+        )
+
+    match_name = (form_data.name or "").strip()
+    if not match_name and form_data.upc:
+        match_name = (await meta_upc_handler.resolve_upc_to_title(form_data.upc)) or ""
+        if not match_name:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Could not resolve the provided UPC to a game title",
+            )
+
+    if not match_name:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="A name or a resolvable UPC is required",
+        )
+
+    fs_name = build_physical_fs_name(platform.id, match_name)
+    fs_path = build_physical_fs_path(platform)
+
+    try:
+        rom = db_rom_handler.add_rom(
+            Rom(
+                platform_id=platform.id,
+                fs_name=fs_name,
+                fs_path=fs_path,
+                fs_size_bytes=0,
+                name=match_name,
+                is_physical=True,
+                upc=form_data.upc,
+                url_cover="",
+                url_manual="",
+                url_screenshots=[],
+            )
+        )
+    except IntegrityError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f"A game named {match_name!r} already exists on this platform",
+        ) from exc
+
+    metadata_sources = form_data.metadata_sources or [s.value for s in MetadataSource]
+
+    fs_rom: FSRom = {
+        "fs_name": fs_name,
+        "flat": True,
+        "nested": False,
+        "files": [],
+        "crc_hash": "",
+        "md5_hash": "",
+        "sha1_hash": "",
+        "ra_hash": "",
+    }
+
+    scanned_rom = await scan_rom(
+        scan_type=ScanType.QUICK,
+        platform=platform,
+        rom=rom,
+        fs_rom=fs_rom,
+        metadata_sources=metadata_sources,
+        newly_added=True,
+    )
+
+    # scan_rom returns a fresh Rom; re-assert the physical identity before persisting
+    # so a matched row can never be treated as a file-less-and-missing digital rom.
+    scanned_rom.is_physical = True
+    scanned_rom.upc = form_data.upc
+    scanned_rom.fs_path = fs_path
+
+    added_rom = db_rom_handler.add_rom(scanned_rom)
+
+    await download_rom_resources(
+        added_rom=added_rom,
+        previous_url_cover=rom.url_cover,
+        previous_url_manual=rom.url_manual,
+        previous_url_screenshots=rom.url_screenshots,
+        metadata_sources=metadata_sources,
+    )
+
+    db_rom_handler.invalidate_filter_values_cache()
+
+    return DetailedRomSchema.from_orm_with_request(
+        db_rom_handler.get_rom(added_rom.id), request
+    )
+
+
 @protected_route(
     router.put,
     "/{id}",

backend/endpoints/sockets/scan.py

@@ -5,7 +5,6 @@
 from itertools import batched
 from typing import Any, Final
 
-import pydash
 import socketio  # type: ignore
 from rq import Worker
 from rq.job import Job
@@ -36,11 +35,11 @@
 )
 from handler.filesystem.roms_handler import FSRom
 from handler.metadata import meta_gamelist_handler, meta_hltb_handler
-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 (
     MetadataSource,
     ScanType,
+    download_rom_resources,
     persist_soundtrack_cover,
     scan_firmware,
     scan_platform,
@@ -422,90 +421,14 @@ async def _identify_rom(
     if scan_type == ScanType.HASHES:
         return
 
-    path_cover_s, path_cover_l = await fs_resource_handler.get_cover(
-        entity=_added_rom,
-        overwrite=_added_rom.url_cover != rom.url_cover,
-        url_cover=add_ss_auth_to_url(_added_rom.url_cover),
-    )
-
-    path_manual = await fs_resource_handler.get_manual(
-        rom=_added_rom,
-        overwrite=_added_rom.url_manual != rom.url_manual,
-        url_manual=add_ss_auth_to_url(_added_rom.url_manual),
-    )
-
-    screenshots_changed = pydash.xor(
-        _added_rom.url_screenshots or [], rom.url_screenshots or []
-    )
-    url_screenshots = _added_rom.url_screenshots or []
-    path_screenshots = await fs_resource_handler.get_rom_screenshots(
-        rom=_added_rom,
-        overwrite=bool(screenshots_changed),
-        url_screenshots=[add_ss_auth_to_url(u) for u in url_screenshots],
-    )
-
-    _added_rom.path_cover_s = path_cover_s
-    _added_rom.path_cover_l = path_cover_l
-    _added_rom.path_screenshots = path_screenshots
-    _added_rom.path_manual = path_manual
-
-    # Update the scanned rom with the cover and screenshots paths and update database
-    db_rom_handler.update_rom(
-        _added_rom.id,
-        {
-            "path_cover_s": path_cover_s,
-            "path_cover_l": path_cover_l,
-            "path_screenshots": path_screenshots,
-            "path_manual": path_manual,
-        },
+    await download_rom_resources(
+        added_rom=_added_rom,
+        previous_url_cover=rom.url_cover,
+        previous_url_manual=rom.url_manual,
+        previous_url_screenshots=rom.url_screenshots,
+        metadata_sources=metadata_sources,
     )
 
-    # Handle special media files from Screenscraper
-    if _added_rom.ss_metadata and MetadataSource.SS in metadata_sources:
-        preferred_media_types = get_preferred_media_types()
-        for media_type in preferred_media_types:
-            media_path = _added_rom.ss_metadata.get(f"{media_type.value}_path")
-            media_url = _added_rom.ss_metadata.get(f"{media_type.value}_url")
-            if media_path and media_url:
-                await fs_resource_handler.store_media_file(
-                    add_ss_auth_to_url(media_url),
-                    media_path,
-                )
-
-    # Handle special media files from ES-DE gamelist.xml
-    if _added_rom.gamelist_metadata and MetadataSource.GAMELIST in metadata_sources:
-        preferred_media_types = get_preferred_media_types()
-        for media_type in preferred_media_types:
-            if _added_rom.gamelist_metadata.get(f"{media_type.value}_path"):
-                await fs_resource_handler.store_media_file(
-                    _added_rom.gamelist_metadata[f"{media_type.value}_url"],
-                    _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", []):
-            badge_url_lock = ach.get("badge_url_lock", None)
-            badge_path_lock = ach.get("badge_path_lock", None)
-            if badge_url_lock and badge_path_lock:
-                await fs_resource_handler.store_ra_badge(
-                    badge_url_lock, badge_path_lock
-                )
-            badge_url = ach.get("badge_url", None)
-            badge_path = ach.get("badge_path", None)
-            if badge_url and badge_path:
-                await fs_resource_handler.store_ra_badge(badge_url, badge_path)
-
     await socket_manager.emit(
         "scan:scanning_rom",
         SimpleRomSchema.from_orm_with_factory(_added_rom).model_dump(

backend/handler/database/roms_handler.py

@@ -628,7 +628,10 @@ def _filter_by_missing_from_fs(self, query: Query, value: bool) -> Query:
         predicate = Rom.missing_from_fs.isnot(False)
         if not value:
             predicate = not_(predicate)
-        return query.filter(predicate)
+            return query.filter(predicate)
+        # Physical games are never "missing"; exclude them so a stray flag can
+        # never make one eligible for the missing-roms cleanup that hard-deletes.
+        return query.filter(and_(predicate, Rom.is_physical.is_(False)))
 
     def _filter_by_verified(self, query: Query, value: bool) -> Query:
         keys_to_check = [
@@ -1570,9 +1573,13 @@ def mark_missing_roms(
         changes, so a re-scan of an unchanged platform issues no updates.
         """
         keep_set = set(fs_roms_to_keep)
+        # Physical games have no file on disk, so they must never be flagged missing.
         rows = session.execute(
             select(Rom.id, Rom.fs_name, Rom.missing_from_fs).where(
-                Rom.platform_id == platform_id
+                and_(
+                    Rom.platform_id == platform_id,
+                    Rom.is_physical.is_(False),
+                )
             )
         ).all()
 
@@ -1599,6 +1606,7 @@ def mark_missing_roms(
                     and_(
                         Rom.platform_id == platform_id,
                         Rom.missing_from_fs.is_(True),
+                        Rom.is_physical.is_(False),
                     )
                 )
                 .order_by(Rom.fs_name.asc())

backend/handler/metadata/__init__.py

@@ -11,6 +11,7 @@
 from .sgdb_handler import SGDBBaseHandler
 from .ss_handler import SSHandler
 from .tgdb_handler import TGDBHandler
+from .upc_handler import UPCHandler
 
 meta_igdb_handler = IGDBHandler()
 meta_moby_handler = MobyGamesHandler()
@@ -25,3 +26,4 @@
 meta_flashpoint_handler = FlashpointHandler()
 meta_gamelist_handler = GamelistHandler()
 meta_hltb_handler = HLTBHandler()
+meta_upc_handler = UPCHandler()