Add check for productID in file name

.gitignore

@@ -55,4 +55,5 @@ frontend/dev-dist
 
 # outside data
 switch_titledb.json
+switch_product_ids.json
 mame.xml

backend/handler/igdb_handler.py

@@ -30,11 +30,16 @@
     os.path.dirname(__file__), "fixtures", "ps2_opl_index.json"
 )
 
-SWITCH_TITLEDB_REGEX: Final = r"^(70[0-9]{12})$"
+SWITCH_TITLEDB_REGEX: Final = r"(70[0-9]{12})"
 SWITCH_TITLEDB_INDEX_FILE: Final = os.path.join(
     os.path.dirname(__file__), "fixtures", "switch_titledb.json"
 )
 
+SWITCH_PRODUCT_ID_REGEX: Final = r"(0100[0-9A-F]{12})"
+SWITCH_PRODUCT_ID_FILE: Final = os.path.join(
+    os.path.dirname(__file__), "fixtures", "switch_product_ids.json"
+)
+
 MAME_XML_FILE: Final = os.path.join(os.path.dirname(__file__), "fixtures", "mame.xml")
 
 
@@ -72,7 +77,7 @@ def _request(self, url: str, data: str, timeout: int = 120) -> list:
             if err.response.status_code != 401:
                 log.error(err)
                 return []  # All requests to the IGDB API return a list
-            
+
             # Attempt to force a token refresh if the token is invalid
             log.warning("Twitch token invalid: fetching a new one...")
             token = self.twitch_auth._update_twitch_token()
@@ -122,7 +127,11 @@ def _search_cover(self, rom_id: int) -> str:
         )
 
         cover = pydash.get(covers, "[0]", None)
-        return DEFAULT_URL_COVER_L if not cover else self._normalize_cover_url(cover["url"])
+        return (
+            DEFAULT_URL_COVER_L
+            if not cover
+            else self._normalize_cover_url(cover["url"])
+        )
 
     def _search_screenshots(self, rom_id: int) -> list:
         screenshots = self._request(
@@ -161,8 +170,8 @@ def get_platform(self, p_slug: str):
     async def get_rom(self, file_name: str, p_igdb_id: int):
         search_term = get_search_term(file_name)
 
-        # Patch support for PS2 OPL flename format
-        match = re.match(PS2_OPL_REGEX, search_term)
+        # Support for PS2 OPL flename format
+        match = re.match(PS2_OPL_REGEX, file_name)
         if p_igdb_id == PS2_IGDB_ID and match:
             serial_code = match.group(1)
 
@@ -172,8 +181,8 @@ async def get_rom(self, file_name: str, p_igdb_id: int):
                 if index_entry:
                     search_term = index_entry["Name"]  # type: ignore
 
-        # Patch support for switch titleID filename format
-        match = re.match(SWITCH_TITLEDB_REGEX, search_term)
+        # Support for switch titleID filename format
+        match = re.search(SWITCH_TITLEDB_REGEX, file_name)
         if p_igdb_id == SWITCH_IGDB_ID and match:
             title_id = match.group(1)
             titledb_index = {}
@@ -195,8 +204,32 @@ async def get_rom(self, file_name: str, p_igdb_id: int):
                 if index_entry:
                     search_term = index_entry["name"]  # type: ignore
 
+        # Support for switch productID filename format
+        match = re.search(SWITCH_PRODUCT_ID_REGEX, file_name)
+        if p_igdb_id == SWITCH_IGDB_ID and match:
+            product_id = match.group(1)
+            product_id_index = {}
+
+            try:
+                with open(SWITCH_PRODUCT_ID_FILE, "r") as index_json:
+                    product_id_index = json.loads(index_json.read())
+            except FileNotFoundError:
+                log.warning("Fetching the Switch titleDB index file...")
+                await update_switch_titledb_task.run(force=True)
+
+                try:
+                    with open(SWITCH_PRODUCT_ID_FILE, "r") as index_json:
+                        product_id_index = json.loads(index_json.read())
+                except FileNotFoundError:
+                    log.error("Could not fetch the Switch titleDB index file")
+            finally:
+                index_entry = product_id_index.get(product_id, None)
+                if index_entry:
+                    search_term = index_entry["name"]  # type: ignore
+
+        # Support for MAME arcade filename format
         if p_igdb_id == ARCADE_IGDB_ID:
-            mame_index = { "menu": { "game": [] } }
+            mame_index = {"menu": {"game": []}}
 
             try:
                 with open(MAME_XML_FILE, "r") as index_xml:
@@ -219,6 +252,14 @@ async def get_rom(self, file_name: str, p_igdb_id: int):
                 if index_entry:
                     search_term = index_entry[0].get("description", search_term)
 
+        search_term = (
+            search_term.replace("\u2122", "") # Remove trademark symbol
+            .replace("\u00ae", "") # Remove registered symbol
+            .replace("\u00a9", "") # Remove copywrite symbol
+            .replace("\u2120", "") # Remove service mark symbol
+            .strip() # Remove leading and trailing spaces
+        )
+
         res = (
             self._search_rom(uc(search_term), p_igdb_id, MAIN_GAME_CATEGORY)
             or self._search_rom(uc(search_term), p_igdb_id, EXPANDED_GAME_CATEGORY)

backend/tasks/update_switch_titledb.py

@@ -1,4 +1,5 @@
 import os
+import json
 from pathlib import Path
 
 from typing import Final
@@ -15,6 +16,13 @@
     / "switch_titledb.json"
 )
 
+SWITCH_PRODUCT_ID_FILE_PATH: Final = (
+    Path(os.path.dirname(__file__)).parent
+    / "handler"
+    / "fixtures"
+    / "switch_product_ids.json"
+)
+
 
 class UpdateSwitchTitleDBTask(RemoteFilePullTask):
     def __init__(self):
@@ -27,5 +35,16 @@ def __init__(self):
             file_path=FIXTURE_FILE_PATH,
         )
 
+    async def run(self, force: bool = False):
+        content = await super().run(force)
+        if content is None:
+            return
+
+        index_json = json.loads(content)
+        product_ids = dict((v["id"], v) for _k, v in index_json.items())
+
+        with open(SWITCH_PRODUCT_ID_FILE_PATH, "wb") as fixture:
+            fixture.write(json.dumps(product_ids).encode())
+
 
 update_switch_titledb_task = UpdateSwitchTitleDBTask()

backend/tasks/utils.py

@@ -83,11 +83,11 @@ def __init__(self, *args, url: str, file_path: str, **kwargs):
         self.url = url
         self.file_path = file_path
 
-    async def run(self, force: bool = False):
+    async def run(self, force: bool = False) -> bytes | None:
         if not self.enabled and not force:
             log.info(f"Scheduled {self.description} not enabled, unscheduling...")
             self.unschedule()
-            return
+            return None
 
         log.info(f"Scheduled {self.description} started...")
 
@@ -99,6 +99,8 @@ async def run(self, force: bool = False):
                 fixture.write(response.content)
 
             log.info(f"Scheduled {self.description} done")
+            return response.content
         except requests.exceptions.RequestException as e:
             log.error(f"Scheduled {self.description} failed", exc_info=True)
             log.error(e)
+            return None

backend/utils/__init__.py

@@ -59,7 +59,7 @@
 REGIONS_NAME_KEYS = [region[1].lower() for region in REGIONS]
 
 TAG_REGEX = r"\(([^)]+)\)|\[([^]]+)\]"
-EXTENSION_REGEX = r"\.(\w+(\.\w+)*)$"
+EXTENSION_REGEX = r"\.(([a-z]+\.)*\w+)$"
 
 
 def parse_tags(file_name: str) -> tuple:

backend/utils/tests/test_utils.py

@@ -58,7 +58,7 @@ def test_get_file_name_with_no_tags():
 
     # This is expected behavior, since the regex is aggressive
     file_name = "Battle Stadium D.O.N.zip"
-    assert gfnwt(file_name) == "Battle Stadium D"
+    assert gfnwt(file_name) == "Battle Stadium D.O.N"
 
 
 def test_get_file_extension():