Action to mass rename all ROMs for a platform

backend/endpoints/rom.py

@@ -1,10 +1,10 @@
+import emoji
 from fastapi import APIRouter, Request, status, HTTPException
 
 from logger.logger import log
 from handler import dbh
 from utils import fs, get_file_name_with_no_tags
 from utils.exceptions import RomNotFoundError, RomAlreadyExistsException
-from models.rom import Rom
 from models.platform import Platform
 
 router = APIRouter()
@@ -14,51 +14,100 @@
 def rom(id: int) -> dict:
     """Returns one rom data of the desired platform"""
 
-    return {'data': dbh.get_rom(id)}
+    return {"data": dbh.get_rom(id)}
 
 
 @router.get("/platforms/{p_slug}/roms", status_code=200)
 def roms(p_slug: str) -> dict:
-    """Returns all roms of the desired platform"""
+    """Returns all roms of the given platform"""
 
-    return {'data':  dbh.get_roms(p_slug)}
+    return {"data": dbh.get_roms(p_slug)}
+
+
+async def rename_all_roms(_sid: str, platform_slug: str, sm=None) -> dict:
+    """Renames all ROMs of the given platform to IGDB format"""
+
+    platform = dbh.get_platform(platform_slug)
+    platform_roms = [r for r in dbh.get_roms(platform_slug) if r.r_igdb_id]
+    log.info(emoji.emojize(f":pencil: Renaming all {platform.name} ROMs"))
+
+    for rom in platform_roms:
+        try:
+            ext = rom.file_name.split(".")[-1]
+            new_filename = rom.r_name
+            if rom.region:
+                new_filename += f" [{rom.region}]"
+            if rom.revision:
+                new_filename += f" [Rev {rom.revision}]"
+            new_filename += f".{ext}"
+
+            fs.rename_rom(platform.fs_slug, rom.file_name, new_filename)
+            dbh.update_rom(
+                rom.id,
+                {
+                    "file_name": new_filename,
+                    "file_name_no_tags": get_file_name_with_no_tags(new_filename),
+                },
+            )
+            log.info(f"Renamed {rom.file_name} to {new_filename}")
+        except RomAlreadyExistsException as e:
+            log.warning(str(e))
+
+    await sm.emit("mass_rename:done")
 
 
 @router.patch("/platforms/{p_slug}/roms/{id}", status_code=200)
 async def updateRom(req: Request, p_slug: str, id: int) -> dict:
     """Updates rom details"""
 
-    data: dict = await req.json()
-    updated_rom: dict = data['updatedRom']
-    db_rom: Rom = dbh.get_rom(id)
-    platform: Platform = dbh.get_platform(p_slug)
+    data = await req.json()
+    updated_rom: dict = data["updatedRom"]
+    db_rom = dbh.get_rom(id)
+    platform = dbh.get_platform(p_slug)
+
     try:
-        fs.rename_rom(platform.fs_slug, db_rom.file_name, updated_rom['file_name'])
+        fs.rename_rom(platform.fs_slug, db_rom.file_name, updated_rom["file_name"])
     except RomAlreadyExistsException as e:
-        error: str = f"{e}"
-        log.error(error)
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error)
-    updated_rom['file_name_no_tags'] = get_file_name_with_no_tags(updated_rom['file_name'])
-    updated_rom.update(fs.get_cover(True, p_slug, updated_rom['file_name'], updated_rom['url_cover']))
-    updated_rom.update(fs.get_screenshots(p_slug, updated_rom['file_name'], updated_rom['url_screenshots']))
+        log.error(str(e))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)
+        )
+
+    updated_rom["file_name_no_tags"] = get_file_name_with_no_tags(
+        updated_rom["file_name"]
+    )
+    updated_rom.update(
+        fs.get_cover(True, p_slug, updated_rom["file_name"], updated_rom["url_cover"])
+    )
+    updated_rom.update(
+        fs.get_screenshots(
+            p_slug, updated_rom["file_name"], updated_rom["url_screenshots"]
+        )
+    )
     dbh.update_rom(id, updated_rom)
-    return {'data': dbh.get_rom(id), 'msg': f"{updated_rom['file_name']} updated successfully!"}
+
+    return {
+        "data": dbh.get_rom(id),
+        "msg": f"{updated_rom['file_name']} updated successfully!",
+    }
 
 
 @router.delete("/platforms/{p_slug}/roms/{id}", status_code=200)
-def delete_rom(p_slug: str, id: int, filesystem: bool=False) -> dict:
+def delete_rom(p_slug: str, id: int, filesystem: bool = False) -> dict:
     """Detele rom from database [and filesystem]"""
 
-    rom: Rom = dbh.get_rom(id)
+    rom = dbh.get_rom(id)
     log.info(f"Deleting {rom.file_name} from database")
     dbh.delete_rom(id)
+
     if filesystem:
         log.info(f"Deleting {rom.file_name} from filesystem")
         try:
             platform: Platform = dbh.get_platform(p_slug)
             fs.remove_rom(platform.fs_slug, rom.file_name)
         except RomNotFoundError as e:
-            error: str = f"{e}. Couldn't delete from filesystem."
+            error = f"{str(e)}. Couldn't delete from filesystem."
             log.error(error)
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=error)
-    return {'msg': f'{rom.file_name} deleted successfully!'}
+
+    return {"msg": f"{rom.file_name} deleted successfully!"}

backend/endpoints/scan.py

@@ -9,7 +9,7 @@
 from models.rom import Rom
 
 
-async def scan(sid: str, platforms: str, complete_rescan: bool=True, sm=None):
+async def scan(_sid: str, platforms: str, complete_rescan: bool = True, sm = None):
     """Scan platforms and roms and write them in database."""
 
     log.info(emoji.emojize(":magnifying_glass_tilted_right: Scanning "))
@@ -18,7 +18,7 @@ async def scan(sid: str, platforms: str, complete_rescan: bool=True, sm=None):
     try: # Scanning platforms
         fs_platforms: list[str] = fs.get_platforms()
     except PlatformsNotFoundException as e:
-        log.error(e); await sm.emit('done_ko', e.message); return
+        log.error(e); await sm.emit('scan:done_ko', e.message); return
 
     platforms: list[str] = json.loads(platforms) if len(json.loads(platforms)) > 0 else fs_platforms
     log.info(f"Platforms to be scanned: {', '.join(platforms)}")
@@ -28,7 +28,7 @@ async def scan(sid: str, platforms: str, complete_rescan: bool=True, sm=None):
             scanned_platform: Platform = fastapi.scan_platform(platform)
         except RomsNotFoundException as e:
             log.error(e); continue
-        await sm.emit('scanning_platform', [scanned_platform.name, scanned_platform.slug]); await sm.emit('') # Workaround to emit in real-time
+        await sm.emit('scan:scanning_platform', [scanned_platform.name, scanned_platform.slug]); await sm.emit('') # Workaround to emit in real-time
         if platform != str(scanned_platform): log.info(f"Identified as {COLORS['blue']}{scanned_platform}{COLORS['reset']}")
         dbh.add_platform(scanned_platform)
 
@@ -37,12 +37,12 @@ async def scan(sid: str, platforms: str, complete_rescan: bool=True, sm=None):
         for rom in fs_roms:
             rom_id: int = dbh.rom_exists(scanned_platform.slug, rom['file_name'])
             if rom_id and not complete_rescan: continue
-            await sm.emit('scanning_rom', rom['file_name']); await sm.emit('') # Workaround to emit in real-time
+            await sm.emit('scan:scanning_rom', rom['file_name']); await sm.emit('') # Workaround to emit in real-time
             log.info(f"Scanning {COLORS['orange']}{rom['file_name']}{COLORS['reset']}")
             if rom['multi']: [log.info(f"\t - {COLORS['orange_i']}{file}{COLORS['reset']}") for file in rom['files']]
             scanned_rom: Rom = fastapi.scan_rom(scanned_platform, rom)
             if rom_id: scanned_rom.id = rom_id
             dbh.add_rom(scanned_rom)
         dbh.purge_roms(scanned_platform.slug, [rom['file_name'] for rom in fs_roms])
     dbh.purge_platforms(fs_platforms)
-    await sm.emit('done')
+    await sm.emit('scan:done')

backend/handler/db_handler.py

@@ -27,9 +27,8 @@ def wrapper(*args):
 
     @staticmethod
     def raise_error(e: Exception) -> None:
-        error: str = f"{e}"
-        log.critical(error)
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error)
+        log.critical(str(e))
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
 
 
     # ========= Platforms =========

backend/handler/igdb_handler.py

@@ -191,4 +191,4 @@ def get_oauth_token(self) -> str:
         if not self._is_token_valid():
             log.warning("Twitch token invalid: fetching a new one")
             self._update_twitch_token()
-        return self.token
+        return self.token

backend/main.py

@@ -21,8 +21,17 @@
 sm = SocketManager()
 sm.mount_to("/ws", app)
 
-async def scan_handler(*args): await scan.scan(*args, sm)
-sm.on('scan', handler=scan_handler)
+
+async def scan_handler(*args):
+    await scan.scan(*args, sm)
+
+
+async def mass_rename_handler(*args):
+    await rom.rename_all_roms(*args, sm)
+
+
+sm.on("scan", handler=scan_handler)
+sm.on("mass_rename", handler=mass_rename_handler)
 
 
 @app.on_event("startup")
@@ -31,6 +40,6 @@ def startup() -> None:
     pass
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     uvicorn.run("main:app", host=DEV_HOST, port=DEV_PORT, reload=True)
     # uvicorn.run("main:app", host=DEV_HOST, port=DEV_PORT, reload=False, workers=2)

backend/utils/__init__.py

@@ -1,26 +1,84 @@
 import re
 
+LANGUAGES = [
+    "Ar",
+    "Da",
+    "De",
+    "En",
+    "En-US",
+    "Es",
+    "Fi",
+    "Fr",
+    "It",
+    "Ja",
+    "Ko",
+    "Nl",
+    "Pl",
+    "Pt",
+    "Pt-BR",
+    "Ru",
+    "Sv",
+    "Zh",
+    "Zh-Hans",
+    "Zh-Hant",
+    "nolang"
+]
+
+REGIONS = [
+    ("U", "USA"),
+    ("E", "Europe"),
+    ("J", "Japan"),
+    ("K", "Korea"),
+    ("T", "Taiwan"),
+    ("G", "Germany"),
+    ("B", "Brazil"),
+    ("A", "Australia"),
+    ("CH", "China"),
+    ("NL", "Netherlands"),
+    ("PD", "Public Domain"),
+    ("F", "France"),
+    ("S", "Spain"),
+    ("W", "World"),
+    ("C", "Canada"),
+    ("SW", "Sweden"),
+    ("FN", "Finland"),
+    ("UK", "England"),
+    ("GR", "Greece"),
+    ("UNK", "Unknown"),
+    ("HK", "Hong Kong"),
+    ("I", "Italy"),
+    ("H", "Holland"),
+    ("UNL", "Unlicensed"),
+    ("AS", "Asia"),
+    ("R", "Russia"),
+    ("NO", "Norway"),
+]
+
+REGIONS_BY_SHORTCODE = {region[0].lower(): region[1] for region in REGIONS}
+REGIONS_NAME_KEYS = [region[1].lower() for region in REGIONS]
+
 
 def parse_tags(file_name: str) -> tuple:
-    reg: str = ''
-    rev: str = ''
-    other_tags: list = []
-    tags: list = re.findall('\(([^)]+)', file_name)
+    reg = ""
+    rev = ""
+    other_tags = []
+    tags = re.findall("\(([^)]+)", file_name)
+
     for tag in tags:
-        if tag.split('-')[0].lower() == 'reg':
-            try: reg = tag.split('-', 1)[1]
-            except IndexError: pass
-        elif tag.split('-')[0].lower() == 'rev':
-            try: rev = tag.split('-', 1)[1]
-            except IndexError: pass
+        if tag.lower() in REGIONS_BY_SHORTCODE.keys():
+            reg = REGIONS_BY_SHORTCODE[tag.lower()]
+        elif tag.lower() in REGIONS_NAME_KEYS:
+            reg = tag
+        elif "rev" in tag.lower():
+            rev = tag.split(" ")[1]
         else:
             other_tags.append(tag)
     return reg, rev, other_tags
 
 
 def get_file_name_with_no_tags(file_name: str) -> str:
-    return re.sub('[\(\[].*?[\)\]]', '', file_name.split('.')[0])
+    return re.sub("[\(\[].*?[\)\]]", "", file_name.split(".")[0])
 
 
 def get_file_extension(rom: dict) -> str:
-    return rom['file_name'].split('.')[-1] if not rom['multi'] else ''
+    return rom["file_name"].split(".")[-1] if not rom["multi"] else ""

backend/utils/fs.py

@@ -210,7 +210,7 @@ def _rom_exists(p_slug: str, file_name: str) -> bool:
 
 
 def rename_rom(p_slug: str, old_name: str, new_name: str) -> None:
-    if new_name != old_name:
+    if new_name and new_name != old_name:
         rom_path = get_roms_structure(p_slug)
         if _rom_exists(p_slug, new_name):
             raise RomAlreadyExistsException(new_name)

frontend/.eslintignore

@@ -0,0 +1,15 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+.DS_Store
+dist
+dist-ssr
+coverage
+*.local

frontend/.eslintrc.json

@@ -0,0 +1,10 @@
+{
+   "env": {
+      "node": true,
+      "browser": true
+   },
+   "extends": [
+      "eslint:recommended",
+      "plugin:vue/vue3-recommended"
+   ]
+}