v1.5

.github/actions/build.sh

@@ -1,6 +1,6 @@
-if [[ $GIT_BRANCH = 'develop' ]]; then
+if [[ $GIT_BRANCH != 'master' ]]; then
     docker buildx build --push\
-        --tag zurdi15/romm:dev-${VERSION}\
+        --tag zurdi15/romm:dev-latest --tag zurdi15/romm:dev-${VERSION}\
         --platform linux/arm64 . --file ./docker/Dockerfile
 else
     docker buildx build --push\

.gitignore

@@ -40,4 +40,7 @@ envs.env
 mariadb
 
 # data test
-library
+library
+
+# config test
+romm

backend/requirements.txt

@@ -2,4 +2,5 @@ requests==2.28.2
 fastapi==0.92.0
 uvicorn==0.20.0
 mariadb==1.1.6
-SQLAlchemy==2.0.7
+SQLAlchemy==2.0.7
+PyYAML==6.0

backend/src/config/__init__.py

@@ -0,0 +1,39 @@
+import os
+import pathlib
+import yaml
+from yaml.loader import SafeLoader
+
+
+# Uvicorn
+DEV_PORT: int = 5000
+DEV_HOST: str = "0.0.0.0"
+
+# PATHS
+LIBRARY_BASE_PATH: str = f"{pathlib.Path(__file__).parent.parent.parent.parent.resolve()}/library"
+ROMM_USER_CONFIG_PATH: str = f"{pathlib.Path(__file__).parent.parent.parent.parent.resolve()}/romm/config.yml"
+
+# ROMM RESERVED FOLDERS
+RESERVED_FOLDERS: list = ['resources', 'database']
+
+# DEFAULT RESOURCES
+DEFAULT_URL_COVER_L: str = "https://images.igdb.com/igdb/image/upload/t_cover_big/nocover.png"
+DEFAULT_PATH_COVER_L: str = f"/assets/library/resources/default/cover_l.png"
+DEFAULT_URL_COVER_S: str = "https://images.igdb.com/igdb/image/upload/t_cover_small/nocover.png"
+DEFAULT_PATH_COVER_S: str = f"/assets/library/resources/default/cover_s.png"
+
+# IGDB
+CLIENT_ID: str = os.getenv('CLIENT_ID')
+CLIENT_SECRET: str = os.getenv('CLIENT_SECRET')
+# STEAMGRIDDB
+STEAMGRIDDB_API_KEY: str = os.getenv('STEAMGRIDDB_API_KEY')
+
+# USER CONFIG
+try:
+    with open(ROMM_USER_CONFIG_PATH) as config: config = yaml.load(config, Loader=SafeLoader)
+except FileNotFoundError:
+    config = None
+user_config: dict = {} if not config else config
+
+# DB DRIVERS
+SUPPORTED_DB_DRIVERS: list = ['sqlite', 'mariadb']
+ROMM_DB_DRIVER: str = os.getenv('ROMM_DB_DRIVER', 'sqlite')

backend/src/config/config.py → backend/src/config/config_loader.py

@@ -1,38 +1,9 @@
 import os
 import sys
-import pathlib
-
 from urllib.parse import quote_plus
-from logger.logger import log
-
-# Uvicorn
-DEV_PORT: int = 5000
-DEV_HOST: str = "0.0.0.0"
-
-# PATHS
-LIBRARY_BASE_PATH: str = f"{pathlib.Path(__file__).parent.parent.parent.parent.resolve()}/library"
-
-DEFAULT_URL_LOGO: str = "https://images.igdb.com/igdb/image/upload/t_cover_big/nocover.png"
-DEFAULT_PATH_LOGO: str = f"/assets/library/resources/default/logo_l.png"
-
-DEFAULT_URL_COVER_L: str = "https://images.igdb.com/igdb/image/upload/t_cover_big/nocover.png"
-DEFAULT_PATH_COVER_L: str = f"/assets/library/resources/default/cover_l.png"
-DEFAULT_URL_COVER_S: str = "https://images.igdb.com/igdb/image/upload/t_cover_small/nocover.png"
-DEFAULT_PATH_COVER_S: str = f"/assets/library/resources/default/cover_s.png"
 
-# IGDB
-CLIENT_ID: str = os.getenv('CLIENT_ID')
-CLIENT_SECRET: str = os.getenv('CLIENT_SECRET')
-# STEAMGRIDDB
-STEAMGRIDDB_API_KEY: str = os.getenv('STEAMGRIDDB_API_KEY')
-
-
-RESERVED_FOLDERS: list = ['resources', 'database']
-
-
-# DB DRIVERS
-SUPPORTED_DB_DRIVERS: list = ['sqlite', 'mariadb']
-ROMM_DB_DRIVER: str = os.getenv('ROMM_DB_DRIVER', 'sqlite')
+from config import ROMM_DB_DRIVER, SUPPORTED_DB_DRIVERS, LIBRARY_BASE_PATH
+from logger.logger import log
 
 
 def get_db_engine():

backend/src/handler/__init__.py

@@ -0,0 +1,7 @@
+from handler.db_handler import DBHandler
+from handler.igdb_handler import IGDBHandler
+from handler.sgdb_handler import SGDBHandler
+
+igdbh: IGDBHandler = IGDBHandler()
+sgdbh: SGDBHandler = SGDBHandler()
+dbh: DBHandler = DBHandler()

backend/src/handler/db_handler.py

@@ -1,4 +1,5 @@
 import functools
+import json
 
 from fastapi import HTTPException
 from sqlalchemy import select
@@ -23,70 +24,54 @@ def wrapper(*args):
         return wrapper
 
 
-    def add_platform(self, **kargs) -> None:
-        with Session.begin() as session:
-            session.merge(Platform(**kargs))
-
-
-    def add_platform(self, **kargs) -> None:
-        with Session.begin() as session:
-            session.merge(Platform(**kargs))
-
-
-    def get_platforms(self) -> list[Platform]:
+    def add_platform(self, Platform: Platform) -> None:
         try:
             with Session.begin() as session:
-                return session.scalars(select(Platform).order_by(Platform.slug.asc())).all()
+                session.merge(Platform)
         except ProgrammingError as e:
             raise HTTPException(status_code=404, detail=f"Platforms table not found: {e}")
-
 
-    def add_platform(self, **kargs) -> None:
+    def get_platforms(self) -> list[Platform]:
         try:
             with Session.begin() as session:
-                session.merge(Platform(**kargs))
+                return session.scalars(select(Platform).order_by(Platform.slug.asc())).all()
         except ProgrammingError as e:
-            raise HTTPException(status_code=404, detail=f"Roms table not found: {e}")
-
+            raise HTTPException(status_code=404, detail=f"Platforms table not found: {e}")
 
-    def get_roms(self, p_slug: str) -> list[Rom]:
+    def add_rom(self, rom: Rom) -> None:
         with Session.begin() as session:
-            return session.scalars(select(Rom).filter_by(p_slug=p_slug).order_by(Rom.filename.asc())).all()
+            session.merge(rom)
 
-
-    def get_rom(self, p_slug: str, filename: str) -> Rom:
+    def get_roms(self, p_slug: str) -> list[Rom]:
         with Session.begin() as session:
-            return session.scalars(select(Rom).filter_by(p_slug=p_slug, filename=filename)).first()
+            return session.scalars(select(Rom).filter_by(p_slug=p_slug).order_by(Rom.file_name.asc())).all()
 
-
-    def add_rom(self, **kargs) -> None:
+    def get_rom(self, p_slug: str, file_name: str) -> Rom:
         with Session.begin() as session:
-            session.merge(Rom(**kargs))
-
+            return session.scalars(select(Rom).filter_by(p_slug=p_slug, file_name=file_name)).first()
 
-    def update_rom(self, p_slug: str, filename: str, data: dict) -> None:
+    def update_rom(self, p_slug: str, file_name: str, data: dict) -> None:
         with Session.begin() as session:
             session.query(Rom) \
-                .filter(Rom.p_slug==p_slug, Rom.filename==filename) \
+                .filter(Rom.p_slug==p_slug, Rom.file_name==file_name) \
                 .update(data, synchronize_session='evaluate')
 
-
-    def delete_rom(self, p_slug: str, filename: str) -> None:
+    def delete_rom(self, p_slug: str, file_name: str) -> None:
         with Session.begin() as session:
             session.query(Rom) \
-                .filter(Rom.p_slug==p_slug, Rom.filename==filename) \
+                .filter(Rom.p_slug==p_slug, Rom.file_name==file_name) \
                 .delete(synchronize_session='evaluate')
 
-
-    def purge_platforms(self, platforms: list) -> None:
+    def purge_platforms(self, platforms: list[str]) -> None:
+        log.info("Purging platforms")
         with Session.begin() as session:
             session.query(Platform) \
                 .filter(Platform.slug.not_in(platforms)) \
                 .delete(synchronize_session='evaluate')
 
-
-    def purge_roms(self, p_slug: str, roms: list) -> None:
+    def purge_roms(self, p_slug: str, roms: list[dict]) -> None:
+        log.info(f"Purging {p_slug} roms")
         with Session.begin() as session:
             session.query(Rom) \
-                .filter(Rom.p_slug==p_slug, Rom.filename.not_in([rom['filename'] for rom in roms])) \
+                .filter(Rom.p_slug==p_slug, Rom.file_name.not_in([rom['file_name'] for rom in roms])) \
                 .delete(synchronize_session='evaluate')

backend/src/handler/igdb_handler.py

@@ -5,7 +5,7 @@
 
 import requests
 
-from config.config import CLIENT_ID, CLIENT_SECRET, DEFAULT_URL_COVER_L
+from config import CLIENT_ID, CLIENT_SECRET, DEFAULT_URL_COVER_L
 from logger.logger import log
 
 
@@ -32,31 +32,30 @@ def wrapper(*args):
     def get_platform_details(self, slug: str) -> tuple:
         igdb_id: str = ""
         name: str = ""
-        url_logo: str = ""
         try:
             res_details: dict = requests.post("https://api.igdb.com/v4/platforms/", headers=self.headers,
-                                              data=f"fields id, name, platform_logo; where slug=\"{slug}\";").json()[0]
+                                              data=f"fields id, name; where slug=\"{slug}\";").json()[0]
             igdb_id = res_details['id']
             name = res_details['name']
         except IndexError:
             log.warning("platform not found in igdb")
         if not name: name = slug
-        return igdb_id, name, url_logo
+        return {'igdb_id': igdb_id, 'name': name}
 
 
     @check_twitch_token
-    def get_rom_details(self, filename: str, p_igdb_id: int, r_igdb_id: str) -> dict:
-        filename_no_ext: str = filename.split('.')[0]
-        igdb_id: str = ""
+    def get_rom_details(self, file_name: str, p_igdb_id: int, r_igdb_id_search: str) -> dict:
+        file_name_no_tags: str = re.sub('[\(\[].*?[\)\]]', '', file_name.split('.')[0])
+        r_igdb_id: str = ""
         slug: str = ""
         name: str = ""
         summary: str = ""
         url_cover: str = ""
 
-        if r_igdb_id:
+        if r_igdb_id_search:
             res_details: dict = requests.post("https://api.igdb.com/v4/games/", headers=self.headers,
-                                              data=f"fields id, slug, name, summary; where id={r_igdb_id};").json()[0]
-            igdb_id = res_details['id']
+                                              data=f"fields id, slug, name, summary; where id={r_igdb_id_search};").json()[0]
+            r_igdb_id = res_details['id']
             slug = res_details['slug']
             name = res_details['name']
             try:
@@ -66,12 +65,11 @@ def get_rom_details(self, filename: str, p_igdb_id: int, r_igdb_id: str) -> dict
 
         else:
             if p_igdb_id:
-                search_term: str = re.sub('[\(\[].*?[\)\]]', '', filename_no_ext)
                 try:
 
                     res_details: dict = requests.post("https://api.igdb.com/v4/games/", headers=self.headers,
-                                                      data=f"search \"{search_term}\";fields id, slug, name, summary; where platforms=[{p_igdb_id}] & category=0;").json()[0]
-                    igdb_id = res_details['id']
+                                                      data=f"search \"{file_name_no_tags}\";fields id, slug, name, summary; where platforms=[{p_igdb_id}] & category=0;").json()[0]
+                    r_igdb_id = res_details['id']
                     slug = res_details['slug']
                     name = res_details['name']
                     try:
@@ -81,8 +79,8 @@ def get_rom_details(self, filename: str, p_igdb_id: int, r_igdb_id: str) -> dict
                 except IndexError:
                     try:
                         res_details: dict = requests.post("https://api.igdb.com/v4/games/", headers=self.headers,
-                                                          data=f"search \"{search_term}\";fields name, id, slug, summary; where platforms=[{p_igdb_id}] & category=10;").json()[0]
-                        igdb_id = res_details['id']
+                                                          data=f"search \"{file_name_no_tags}\";fields name, id, slug, summary; where platforms=[{p_igdb_id}] & category=10;").json()[0]
+                        r_igdb_id = res_details['id']
                         slug = res_details['slug']
                         name = res_details['name']
                         try:
@@ -92,40 +90,42 @@ def get_rom_details(self, filename: str, p_igdb_id: int, r_igdb_id: str) -> dict
                     except IndexError:
                         try:
                             res_details: dict = requests.post("https://api.igdb.com/v4/games/", headers=self.headers,
-                                                              data=f"search \"{search_term}\";fields name, id, slug, summary; where platforms=[{p_igdb_id}];").json()[0]
-                            igdb_id = res_details['id']
+                                                              data=f"search \"{file_name_no_tags}\";fields name, id, slug, summary; where platforms=[{p_igdb_id}];").json()[0]
+                            r_igdb_id = res_details['id']
                             slug = res_details['slug']
                             name = res_details['name']
                             try:
                                 summary = res_details['summary']
                             except KeyError:
                                 pass
                         except IndexError:
-                            log.warning(f"{filename} rom not found in igdb")
-        if igdb_id:
+                            log.warning(f"{file_name} rom not found in igdb")
+        if r_igdb_id:
             try:
                 res_details: dict = requests.post("https://api.igdb.com/v4/covers/", headers=self.headers,
-                                                  data=f"fields url; where game={igdb_id};").json()[0]
+                                                  data=f"fields url; where game={r_igdb_id};").json()[0]
                 url_cover: str = f"https:{res_details['url']}"
             except IndexError:
                 log.warning(f"{name} cover not found in igdb")
-        if not name: name = filename_no_ext
-        return (igdb_id, filename_no_ext, slug, name, summary, url_cover)
+        if not name: name = file_name_no_tags
+        return (r_igdb_id, file_name_no_tags, slug, name, summary, url_cover)
 
 
     @check_twitch_token
-    def get_matched_roms(self, filename: str, p_igdb_id: int) -> list:
-        search_term: str = re.sub('[\(\[].*?[\)\]]', '', filename.split('.')[0])
+    def get_matched_roms(self, file_name: str, p_igdb_id: int) -> list:
+        search_term: str = re.sub('[\(\[].*?[\)\]]', '', file_name.split('.')[0])
         matched_roms: list = requests.post("https://api.igdb.com/v4/games/", headers=self.headers,
                                            data=f"search \"{search_term}\";fields name, id, slug, summary; where platforms=[{p_igdb_id}];").json()
-        log.info(f"Matched roms for {filename}: {matched_roms}")
+        log.info(f"Matched roms for {file_name}: {matched_roms}")
         for rom in matched_roms:
             try:
                 res_details: dict = requests.post("https://api.igdb.com/v4/covers/", headers=self.headers,
                                                   data=f"fields url; where game={rom['id']};").json()[0]
                 rom['url_cover'] = f"https:{res_details['url']}".replace('t_thumb', f't_cover_big')
             except IndexError:
                 rom['url_cover'] = DEFAULT_URL_COVER_L
+            rom['r_igdb_id'] = rom.pop('id')
+            rom['r_slug'] = rom.pop('slug')
         return matched_roms
 
 

backend/src/handler/sgdb_handler.py

@@ -1,6 +1,6 @@
 import requests
 
-from config.config import STEAMGRIDDB_API_KEY
+from config import STEAMGRIDDB_API_KEY
 from logger.logger import log