[ROMM-403] Import metadata from Launchbox database

backend/config/__init__.py

@@ -108,6 +108,13 @@ def str_to_bool(value: str) -> bool:
     "SCHEDULED_UPDATE_SWITCH_TITLEDB_CRON",
     "0 4 * * *",  # At 4:00 AM every day
 )
+ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA: Final = str_to_bool(
+    os.environ.get("ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA", "false")
+)
+SCHEDULED_UPDATE_LAUNCHBOX_METADATA_CRON: Final = os.environ.get(
+    "SCHEDULED_UPDATE_LAUNCHBOX_METADATA_CRON",
+    "0 5 * * *",  # At 5:00 AM every day
+)
 
 # EMULATION
 DISABLE_EMULATOR_JS = str_to_bool(os.environ.get("DISABLE_EMULATOR_JS", "false"))

backend/endpoints/heartbeat.py

@@ -4,11 +4,13 @@
     DISABLE_USERPASS_LOGIN,
     ENABLE_RESCAN_ON_FILESYSTEM_CHANGE,
     ENABLE_SCHEDULED_RESCAN,
+    ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA,
     ENABLE_SCHEDULED_UPDATE_SWITCH_TITLEDB,
     OIDC_ENABLED,
     OIDC_PROVIDER,
     RESCAN_ON_FILESYSTEM_CHANGE_DELAY,
     SCHEDULED_RESCAN_CRON,
+    SCHEDULED_UPDATE_LAUNCHBOX_METADATA_CRON,
     SCHEDULED_UPDATE_SWITCH_TITLEDB_CRON,
     UPLOAD_TIMEOUT,
 )
@@ -64,6 +66,12 @@ def heartbeat() -> HeartbeatResponse:
                 "TITLE": "Scheduled Switch TitleDB update",
                 "MESSAGE": "Updates the Nintendo Switch TitleDB file",
             },
+            "LAUNCHBOX_METADATA": {
+                "ENABLED": ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA,
+                "CRON": SCHEDULED_UPDATE_LAUNCHBOX_METADATA_CRON,
+                "TITLE": "Scheduled LaunchBox metadata update",
+                "MESSAGE": "Updates the LaunchBox metadata",
+            },
         },
         "EMULATION": {
             "DISABLE_EMULATOR_JS": DISABLE_EMULATOR_JS,

backend/endpoints/responses/heartbeat.py

@@ -19,6 +19,7 @@ class TaskDict(WatcherDict):
 class SchedulerDict(TypedDict):
     RESCAN: TaskDict
     SWITCH_TITLEDB: TaskDict
+    LAUNCHBOX_METADATA: TaskDict
 
 
 class MetadataSourcesDict(TypedDict):

backend/endpoints/tasks.py

@@ -2,6 +2,7 @@
 from endpoints.responses import MessageResponse
 from fastapi import Request
 from handler.auth.constants import Scope
+from tasks.update_launchbox_metadata import update_launchbox_metadata_task
 from tasks.update_switch_titledb import update_switch_titledb_task
 from utils.router import APIRouter
 
@@ -19,20 +20,24 @@ async def run_tasks(request: Request) -> MessageResponse:
     """
 
     await update_switch_titledb_task.run()
+    await update_launchbox_metadata_task.run()
     return {"msg": "All tasks ran successfully!"}
 
 
 @protected_route(router.post, "/tasks/{task}/run", [Scope.TASKS_RUN])
 async def run_task(request: Request, task: str) -> MessageResponse:
-    """Run all tasks endpoint
+    """Run single tasks endpoint
 
     Args:
         request (Request): Fastapi Request object
     Returns:
         RunTasksResponse: Standard message response
     """
 
-    tasks = {"switch_titledb": update_switch_titledb_task}
+    tasks = {
+        "switch_titledb": update_switch_titledb_task,
+        "launchbox_metadata": update_launchbox_metadata_task,
+    }
 
     await tasks[task].run()
     return {"msg": f"Task {task} run successfully!"}

backend/endpoints/tests/test_heartbeat.py

@@ -27,4 +27,12 @@ def test_heartbeat(client):
         heartbeat.get("SCHEDULER").get("SWITCH_TITLEDB").get("TITLE")
         == "Scheduled Switch TitleDB update"
     )
+    assert heartbeat.get("SCHEDULER").get("LAUNCHBOX_METADATA").get("ENABLED")
+    assert (
+        heartbeat.get("SCHEDULER").get("LAUNCHBOX_METADATA").get("CRON") == "0 5 * * *"
+    )
+    assert (
+        heartbeat.get("SCHEDULER").get("LAUNCHBOX_METADATA").get("TITLE")
+        == "Scheduled LaunchBox metadata update"
+    )
     assert heartbeat.get("FRONTEND").get("UPLOAD_TIMEOUT") == 20

backend/handler/filesystem/roms_handler.py

@@ -8,7 +8,7 @@
 import zipfile
 from collections.abc import Callable, Iterator
 from pathlib import Path
-from typing import Any, Final, TypedDict
+from typing import IO, Any, Final, Literal, TypedDict
 
 import magic
 import py7zr
@@ -76,25 +76,28 @@ def is_compressed_file(file_path: str) -> bool:
     )
 
 
-def read_basic_file(file_path: Path) -> Iterator[bytes]:
+def read_basic_file(file_path: os.PathLike[str]) -> Iterator[bytes]:
     with open(file_path, "rb") as f:
         while chunk := f.read(FILE_READ_CHUNK_SIZE):
             yield chunk
 
 
-def read_zip_file(file_path: Path) -> Iterator[bytes]:
+def read_zip_file(file: str | os.PathLike[str] | IO[bytes]) -> Iterator[bytes]:
     try:
-        with zipfile.ZipFile(file_path, "r") as z:
+        with zipfile.ZipFile(file, "r") as z:
             for file in z.namelist():
                 with z.open(file, "r") as f:
                     while chunk := f.read(FILE_READ_CHUNK_SIZE):
                         yield chunk
     except zipfile.BadZipFile:
-        for chunk in read_basic_file(file_path):
-            yield chunk
+        if isinstance(file, Path):
+            for chunk in read_basic_file(file):
+                yield chunk
 
 
-def read_tar_file(file_path: Path, mode: str = "r") -> Iterator[bytes]:
+def read_tar_file(
+    file_path: Path, mode: Literal["r", "r:*", "r:", "r:gz", "r:bz2", "r:xz"] = "r"
+) -> Iterator[bytes]:
     try:
         with tarfile.open(file_path, mode) as f:
             for member in f.getmembers():

backend/scheduler.py

@@ -3,6 +3,7 @@
 from logger.logger import log
 from tasks.scan_library import scan_library_task
 from tasks.tasks import tasks_scheduler
+from tasks.update_launchbox_metadata import update_launchbox_metadata_task
 from tasks.update_switch_titledb import update_switch_titledb_task
 from utils import get_version
 
@@ -15,6 +16,7 @@
     # Initialize the tasks
     scan_library_task.init()
     update_switch_titledb_task.init()
+    update_launchbox_metadata_task.init()
 
     log.info("Starting scheduler")
 

backend/tasks/fixtures/launchbox/files.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" standalone="yes"?>
+<LaunchBox>
+  <File>
+    <Platform>3DO Interactive Multiplayer</Platform>
+    <FileName>20th Century Video Almanac (USA)</FileName>
+    <GameName>20th Century Video Almanac (USA)</GameName>
+  </File>
+</LaunchBox>

backend/tasks/fixtures/launchbox/mame.xml

@@ -0,0 +1,41 @@
+<?xml version="1.0" standalone="yes"?>
+<LaunchBox>
+  <MameFile>
+    <FileName>wrlok_l3</FileName>
+    <Name>Warlok</Name>
+    <Status>preliminary</Status>
+    <Developer />
+    <Publisher>Williams</Publisher>
+    <Year>1982</Year>
+    <IsMechanical>true</IsMechanical>
+    <Version>(L-3)</Version>
+    <Region>North America</Region>
+    <IsBootleg>false</IsBootleg>
+    <IsPrototype>false</IsPrototype>
+    <IsHack>false</IsHack>
+    <IsMature>false</IsMature>
+    <IsQuiz>false</IsQuiz>
+    <IsFruit>false</IsFruit>
+    <IsCasino>false</IsCasino>
+    <IsRhythm>false</IsRhythm>
+    <IsTableTop>false</IsTableTop>
+    <IsPlayChoice>false</IsPlayChoice>
+    <IsMahjong>false</IsMahjong>
+    <IsNonArcade>false</IsNonArcade>
+    <Genre>Electromechanical / Pinball</Genre>
+    <PlayMode>Pinball</PlayMode>
+    <Language>English</Language>
+    <Source>pinball/s7.cpp</Source>
+  </MameFile>
+  <ControllerSupport>
+    <ControllerName>Generic Keypad</ControllerName>
+    <ControllerCategory>Keyboard</ControllerCategory>
+    <FileName>wrlok_l3</FileName>
+    <Required>true</Required>
+  </ControllerSupport>
+  <MameListItem>
+    <FileName>wrlok_l3</FileName>
+    <GameName>Warlok (L-3)</GameName>
+    <ListName>Williams Classics</ListName>
+  </MameListItem>
+</LaunchBox>