Merge pull request #84 from zurdi15/refactor/codebase

Refactor/codebase

.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/db_handler.py

@@ -1,4 +1,5 @@
 import functools
+import json
 
 from fastapi import HTTPException
 from sqlalchemy import select
@@ -43,22 +44,22 @@ def add_rom(self, rom: Rom) -> None:
 
     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).order_by(Rom.filename.asc())).all()
+            return session.scalars(select(Rom).filter_by(p_slug=p_slug).order_by(Rom.file_name.asc())).all()
 
-    def get_rom(self, p_slug: str, filename: str) -> Rom:
+    def get_rom(self, p_slug: str, file_name: str) -> 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, 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[str]) -> None:
@@ -72,5 +73,5 @@ 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
 
 
@@ -44,8 +44,8 @@ def get_platform_details(self, slug: str) -> tuple:
 
 
     @check_twitch_token
-    def get_rom_details(self, filename: str, p_igdb_id: int, r_igdb_id_search: str) -> dict:
-        filename_no_ext: str = filename.split('.')[0]
+    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 = ""
@@ -65,11 +65,10 @@ def get_rom_details(self, filename: str, p_igdb_id: int, r_igdb_id_search: str)
 
         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]
+                                                      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']
@@ -80,7 +79,7 @@ def get_rom_details(self, filename: str, p_igdb_id: int, r_igdb_id_search: str)
                 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]
+                                                          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']
@@ -91,7 +90,7 @@ def get_rom_details(self, filename: str, p_igdb_id: int, r_igdb_id_search: str)
                     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]
+                                                              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']
@@ -100,31 +99,33 @@ def get_rom_details(self, filename: str, p_igdb_id: int, r_igdb_id_search: str)
                             except KeyError:
                                 pass
                         except IndexError:
-                            log.warning(f"{filename} rom not found in igdb")
+                            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={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 (r_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
 
 

backend/src/main.py

@@ -1,9 +1,10 @@
+import os
 from fastapi import FastAPI, Request
 import uvicorn
 
 from logger.logger import log
 from handler import igdbh, dbh
-from config.config import DEV_PORT, DEV_HOST
+from config import DEV_PORT, DEV_HOST
 from models.platform import Platform
 from utils import fs, fastapi
 
@@ -12,47 +13,53 @@
 fastapi.allow_cors(app)
 
 
-@app.patch("/platforms/{p_slug}/roms/{filename}")
-async def updateRom(req: Request, p_slug: str, filename: str) -> dict:
+@app.get("/platforms/{p_slug}/roms/{file_name}")
+async def rom(p_slug: str, file_name: str) -> dict:
+    """Returns one rom data of the desired platform"""
+
+    return {'data':  dbh.get_rom(p_slug, file_name)}
+
+
+@app.patch("/platforms/{p_slug}/roms")
+async def updateRom(req: Request, p_slug: str) -> dict:
     """Updates rom details"""
 
     data: dict = await req.json()
-    if 'r_igdb_id' in data:
-        r_igdb_id, filename_no_ext, r_slug, r_name, summary, url_cover = igdbh.get_rom_details(filename, data['p_igdb_id'], data['r_igdb_id'])
-        path_cover_s, path_cover_l, has_cover = fs.get_cover_details(True, p_slug, filename_no_ext, url_cover)
-        data['r_igdb_id'] = r_igdb_id
-        data['filename_no_ext'] = filename_no_ext
-        data['r_slug'] = r_slug
-        data['name'] = r_name
-        data['summary'] = summary
-        data['path_cover_s'] = path_cover_s
-        data['path_cover_l'] = path_cover_l
-        data['has_cover'] = has_cover
-        data['p_slug'] = p_slug
-    else:
-        fs.rename_rom(p_slug, filename, data)
-        data['filename_no_ext'] = data['filename'].split('.')[0]
-    dbh.update_rom(p_slug, filename, data)
-    return {'data': data}
-
-
-@app.delete("/platforms/{p_slug}/roms/{filename}")
-async def delete_rom(p_slug: str, filename: str, filesystem: bool=False) -> dict:
+    rom: dict = data['rom']
+    updatedRom: dict = data['updatedRom']
+    r_igdb_id, file_name_no_tags, r_slug, r_name, summary, url_cover = igdbh.get_rom_details(updatedRom['file_name'], rom['p_igdb_id'], updatedRom['r_igdb_id'])
+    path_cover_s, path_cover_l, has_cover = fs.get_cover_details(True, p_slug, updatedRom['file_name'], url_cover)
+    updatedRom['file_name_no_tags'] = file_name_no_tags
+    updatedRom['r_igdb_id'] = r_igdb_id
+    updatedRom['p_igdb_id'] = rom['p_igdb_id']
+    updatedRom['r_slug'] = r_slug
+    updatedRom['p_slug'] = p_slug
+    updatedRom['name'] = r_name
+    updatedRom['summary'] = summary
+    updatedRom['path_cover_s'] = path_cover_s
+    updatedRom['path_cover_l'] = path_cover_l
+    updatedRom['has_cover'] = has_cover
+    updatedRom['file_path'] = rom['file_path']
+    updatedRom['file_size'] = rom['file_size']
+    updatedRom['file_extension'] = updatedRom['file_name'].split('.')[-1] if '.' in updatedRom['file_name'] else ""
+    reg, rev, other_tags = fs.parse_tags(updatedRom['file_name'])
+    updatedRom.update({'region': reg, 'revision': rev, 'tags': other_tags})
+    if 'url_cover' in updatedRom.keys(): del updatedRom['url_cover']
+    fs.rename_rom(p_slug, rom['file_name'], updatedRom['file_name'])
+    dbh.update_rom(p_slug, rom['file_name'], updatedRom)
+    return {'data': updatedRom}
+
+
+@app.delete("/platforms/{p_slug}/roms")
+async def delete_rom(p_slug: str, file_name: str, filesystem: bool=False) -> dict:
     """Detele rom from filesystem and database"""
 
     log.info("deleting rom...")
-    if filesystem: fs.delete_rom(p_slug, filename)
-    dbh.delete_rom(p_slug, filename)
+    if filesystem: fs.delete_rom(p_slug, file_name)
+    dbh.delete_rom(p_slug, file_name)
     return {'msg': 'success'}
 
 
-@app.get("/platforms/{p_slug}/roms/{filename}")
-async def rom(p_slug: str, filename: str) -> dict:
-    """Returns one rom data of the desired platform"""
-
-    return {'data':  dbh.get_rom(p_slug, filename)}
-
-
 @app.get("/platforms/{p_slug}/roms")
 async def roms(p_slug: str) -> dict:
     """Returns all roms of the desired platform"""
@@ -90,8 +97,8 @@ async def search_rom_igdb(req: Request) -> dict:
     """Get all the roms matched from igdb."""
 
     data: dict = await req.json()
-    log.info(f"getting {data['filename']} roms from {data['p_igdb_id']} igdb ...")
-    return {'data': igdbh.get_matched_roms(data['filename'], data['p_igdb_id'])}
+    log.info(f"getting {data['file_name']} roms from {data['p_igdb_id']} igdb ...")
+    return {'data': igdbh.get_matched_roms(data['file_name'], data['p_igdb_id'])}
 
 
 if __name__ == '__main__':

backend/src/models/base.py

@@ -2,10 +2,9 @@
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker
 
-from config.config import get_db_engine
+from config.config_loader import get_db_engine
 
 BaseModel = declarative_base()
 
-# engine = create_engine(DB_DRIVERS[ROMM_DB_DRIVER], pool_pre_ping=True)
 engine = create_engine(get_db_engine(), pool_pre_ping=True)
 Session = sessionmaker(bind=engine, expire_on_commit=False)

backend/src/models/platform.py

@@ -1,14 +1,17 @@
 from sqlalchemy import Column, String, Integer, Text
 
-from config.config import DEFAULT_PATH_LOGO
+from config import DEFAULT_PATH_COVER_S
 from models.base import BaseModel
 
 
 class Platform(BaseModel):
     __tablename__ = 'platforms'
-    igdb_id = Column(String(length=50), default="")
-    sgdb_id = Column(String(length=50), default="")
-    slug = Column(String(length=500), primary_key=True)
-    name = Column(String(length=500), default="")
-    path_logo = Column(Text, default=DEFAULT_PATH_LOGO)
+    igdb_id = Column(String(length=10), default="")
+    sgdb_id = Column(String(length=10), default="")
+
+    slug = Column(String(length=50), primary_key=True)
+    name = Column(String(length=400), default="")
+
+    logo_path = Column(String(length=1000), default=DEFAULT_PATH_COVER_S)
+
     n_roms = Column(Integer, default=0)