rommapp/romm
1
0
Fork 0
mirror of https://github.com/rommapp/romm.git synced 2026-02-18 00:27:41 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2370 from rommapp/misc/metadata-handler-is-enabled

Browse Source 
misc: Add MetadataHandler's is_enabled method
This commit is contained in:
Michael Manganiello
2025-09-04 09:47:42 -03:00
committed by GitHub
parent 990c456f97 d216bad78b
commit c66a6eb5b7
21 changed files with 236 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
backend/config/__init__.py
View File

@@ -55,46 +55,46 @@ REDIS_URL: Final = yarl.URL.build(
)
# IGDB
IGDB_CLIENT_ID: Final = os.environ.get(
IGDB_CLIENT_ID: Final[str] = os.environ.get(
"IGDB_CLIENT_ID", os.environ.get("CLIENT_ID", "")
).strip()
IGDB_CLIENT_SECRET: Final = os.environ.get(
IGDB_CLIENT_SECRET: Final[str] = os.environ.get(
"IGDB_CLIENT_SECRET", os.environ.get("CLIENT_SECRET", "")
).strip()
# MOBYGAMES
MOBYGAMES_API_KEY: Final = os.environ.get("MOBYGAMES_API_KEY", "").strip()
MOBYGAMES_API_KEY: Final[str] = os.environ.get("MOBYGAMES_API_KEY", "").strip()
# SCREENSCRAPER
SCREENSCRAPER_USER: Final = os.environ.get("SCREENSCRAPER_USER", "")
SCREENSCRAPER_PASSWORD: Final = os.environ.get("SCREENSCRAPER_PASSWORD", "")
SCREENSCRAPER_USER: Final[str] = os.environ.get("SCREENSCRAPER_USER", "")
SCREENSCRAPER_PASSWORD: Final[str] = os.environ.get("SCREENSCRAPER_PASSWORD", "")
# STEAMGRIDDB
STEAMGRIDDB_API_KEY: Final = os.environ.get("STEAMGRIDDB_API_KEY", "").strip()
STEAMGRIDDB_API_KEY: Final[str] = os.environ.get("STEAMGRIDDB_API_KEY", "").strip()
# RETROACHIEVEMENTS
RETROACHIEVEMENTS_API_KEY: Final = os.environ.get("RETROACHIEVEMENTS_API_KEY", "")
REFRESH_RETROACHIEVEMENTS_CACHE_DAYS: Final = int(
RETROACHIEVEMENTS_API_KEY: Final[str] = os.environ.get("RETROACHIEVEMENTS_API_KEY", "")
REFRESH_RETROACHIEVEMENTS_CACHE_DAYS: Final[int] = int(
os.environ.get("REFRESH_RETROACHIEVEMENTS_CACHE_DAYS", 30)
)
# LAUNCHBOX
LAUNCHBOX_API_ENABLED: Final = str_to_bool(
LAUNCHBOX_API_ENABLED: Final[bool] = str_to_bool(
os.environ.get("LAUNCHBOX_API_ENABLED", "false")
)
# PLAYMATCH
PLAYMATCH_API_ENABLED: Final = str_to_bool(
PLAYMATCH_API_ENABLED: Final[bool] = str_to_bool(
os.environ.get("PLAYMATCH_API_ENABLED", "false")
)
# HASHEOUS
HASHEOUS_API_ENABLED: Final = str_to_bool(
HASHEOUS_API_ENABLED: Final[bool] = str_to_bool(
os.environ.get("HASHEOUS_API_ENABLED", "false")
)
# THEGAMESDB
TGDB_API_ENABLED: Final = str_to_bool(os.environ.get("TGDB_API_ENABLED", "false"))
TGDB_API_ENABLED: Final[bool] = str_to_bool(os.environ.get("TGDB_API_ENABLED", "false"))
# AUTH
ROMM_AUTH_SECRET_KEY: Final = os.environ.get("ROMM_AUTH_SECRET_KEY")

54
backend/endpoints/heartbeat.py
View File

@@ -6,27 +6,29 @@ from config import (
ENABLE_SCHEDULED_RESCAN,
ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA,
ENABLE_SCHEDULED_UPDATE_SWITCH_TITLEDB,
HASHEOUS_API_ENABLED,
LAUNCHBOX_API_ENABLED,
OIDC_ENABLED,
OIDC_PROVIDER,
PLAYMATCH_API_ENABLED,
SCHEDULED_CONVERT_IMAGES_TO_WEBP_CRON,
SCHEDULED_RESCAN_CRON,
SCHEDULED_UPDATE_LAUNCHBOX_METADATA_CRON,
SCHEDULED_UPDATE_SWITCH_TITLEDB_CRON,
TGDB_API_ENABLED,
UPLOAD_TIMEOUT,
YOUTUBE_BASE_URL,
)
from endpoints.responses.heartbeat import HeartbeatResponse
from handler.database import db_user_handler
from handler.filesystem import fs_platform_handler
from handler.metadata.igdb_handler import IGDB_API_ENABLED
from handler.metadata.moby_handler import MOBY_API_ENABLED
from handler.metadata.ra_handler import RA_API_ENABLED
from handler.metadata.sgdb_handler import STEAMGRIDDB_API_ENABLED
from handler.metadata.ss_handler import SS_API_ENABLED
from handler.metadata import (
meta_hasheous_handler,
meta_igdb_handler,
meta_launchbox_handler,
meta_moby_handler,
meta_playmatch_handler,
meta_ra_handler,
meta_sgdb_handler,
meta_ss_handler,
meta_tgdb_handler,
)
from utils import get_version
from utils.router import APIRouter
@@ -49,22 +51,24 @@ async def heartbeat() -> HeartbeatResponse:
"SHOW_SETUP_WIZARD": len(db_user_handler.get_admin_users()) == 0,
},
"METADATA_SOURCES": {
"ANY_SOURCE_ENABLED": IGDB_API_ENABLED
or SS_API_ENABLED
or MOBY_API_ENABLED
or RA_API_ENABLED
or LAUNCHBOX_API_ENABLED
or HASHEOUS_API_ENABLED
or TGDB_API_ENABLED,
"IGDB_API_ENABLED": IGDB_API_ENABLED,
"SS_API_ENABLED": SS_API_ENABLED,
"MOBY_API_ENABLED": MOBY_API_ENABLED,
"STEAMGRIDDB_API_ENABLED": STEAMGRIDDB_API_ENABLED,
"RA_API_ENABLED": RA_API_ENABLED,
"LAUNCHBOX_API_ENABLED": LAUNCHBOX_API_ENABLED,
"HASHEOUS_API_ENABLED": HASHEOUS_API_ENABLED,
"PLAYMATCH_API_ENABLED": PLAYMATCH_API_ENABLED,
"TGDB_API_ENABLED": TGDB_API_ENABLED,
"ANY_SOURCE_ENABLED": (
meta_igdb_handler.is_enabled()
or meta_ss_handler.is_enabled()
or meta_moby_handler.is_enabled()
or meta_ra_handler.is_enabled()
or meta_launchbox_handler.is_enabled()
or meta_hasheous_handler.is_enabled()
or meta_tgdb_handler.is_enabled()
),
"IGDB_API_ENABLED": meta_igdb_handler.is_enabled(),
"SS_API_ENABLED": meta_ss_handler.is_enabled(),
"MOBY_API_ENABLED": meta_moby_handler.is_enabled(),
"STEAMGRIDDB_API_ENABLED": meta_sgdb_handler.is_enabled(),
"RA_API_ENABLED": meta_ra_handler.is_enabled(),
"LAUNCHBOX_API_ENABLED": meta_launchbox_handler.is_enabled(),
"HASHEOUS_API_ENABLED": meta_hasheous_handler.is_enabled(),
"PLAYMATCH_API_ENABLED": meta_playmatch_handler.is_enabled(),
"TGDB_API_ENABLED": meta_tgdb_handler.is_enabled(),
},
"FILESYSTEM": {
"FS_PLATFORMS": await fs_platform_handler.get_platforms(),

16
backend/endpoints/search.py
View File

@@ -12,10 +12,10 @@ from handler.metadata import (
meta_sgdb_handler,
meta_ss_handler,
)
from handler.metadata.igdb_handler import IGDB_API_ENABLED, IGDBRom
from handler.metadata.moby_handler import MOBY_API_ENABLED, MobyGamesRom
from handler.metadata.sgdb_handler import STEAMGRIDDB_API_ENABLED, SGDBRom
from handler.metadata.ss_handler import SS_API_ENABLED, SSRom
from handler.metadata.igdb_handler import IGDBRom
from handler.metadata.moby_handler import MobyGamesRom
from handler.metadata.sgdb_handler import SGDBRom
from handler.metadata.ss_handler import SSRom
from handler.scan_handler import get_main_platform_igdb_id
from logger.formatter import BLUE, CYAN
from logger.formatter import highlight as hl
@@ -50,7 +50,11 @@ async def search_rom(
list[SearchRomSchema]: List of matched roms
"""
if not IGDB_API_ENABLED and not SS_API_ENABLED and not MOBY_API_ENABLED:
if (
not meta_igdb_handler.is_enabled()
and not meta_ss_handler.is_enabled()
and not meta_moby_handler.is_enabled()
):
log.error("Search error: No metadata providers enabled")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
@@ -182,7 +186,7 @@ async def search_cover(
search_term: str = "",
) -> list[SearchCoverSchema]:
if not STEAMGRIDDB_API_ENABLED:
if not meta_sgdb_handler.is_enabled():
log.error("Search error: No SteamGridDB enabled")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,

8
backend/handler/metadata/base_hander.py
View File

@@ -1,3 +1,4 @@
import abc
import enum
import json
import re
@@ -79,10 +80,15 @@ def _normalize_search_term(
return name.strip()
class MetadataHandler:
class MetadataHandler(abc.ABC):
SEARCH_TERM_SPLIT_PATTERN = re.compile(r"[\:\-\/]")
SEARCH_TERM_NORMALIZER = re.compile(r"\s*[:-]\s*")
@classmethod
@abc.abstractmethod
def is_enabled(cls) -> bool:
"""Return whether this metadata handler is enabled."""
def normalize_cover_url(self, url: str) -> str:
return url if not url else f"https:{url.replace('https:', '')}"

11
backend/handler/metadata/hasheous_handler.py
View File

@@ -123,6 +123,11 @@ class HasheousHandler(MetadataHandler):
else "JNoFBA-jEh4HbxuxEHM6MVzydKoAXs9eCcp2dvcg5LRCnpp312voiWmjuaIssSzS"
)
@classmethod
def is_enabled(cls) -> bool:
"""Return whether this metadata handler is enabled."""
return HASHEOUS_API_ENABLED
async def _request(
self,
url: str,
@@ -213,7 +218,7 @@ class HasheousHandler(MetadataHandler):
hasheous_id=None, igdb_id=None, tgdb_id=None, ra_id=None
)
if not HASHEOUS_API_ENABLED:
if not self.is_enabled():
return fallback_rom
filtered_files = [
@@ -314,7 +319,7 @@ class HasheousHandler(MetadataHandler):
)
async def get_igdb_game(self, hasheous_rom: HasheousRom) -> HasheousRom:
if not HASHEOUS_API_ENABLED:
if not self.is_enabled():
return hasheous_rom
igdb_id = hasheous_rom.get("igdb_id", None)
@@ -358,7 +363,7 @@ class HasheousHandler(MetadataHandler):
)
async def get_ra_game(self, hasheous_rom: HasheousRom) -> HasheousRom:
if not HASHEOUS_API_ENABLED:
if not self.is_enabled():
return hasheous_rom
ra_id = hasheous_rom.get("ra_id", None)

23
backend/handler/metadata/igdb_handler.py
View File

@@ -26,9 +26,6 @@ from .base_hander import (
)
from .base_hander import UniversalPlatformSlug as UPS
# Used to display the IGDB API status in the frontend
IGDB_API_ENABLED: Final = bool(IGDB_CLIENT_ID) and bool(IGDB_CLIENT_SECRET)
PS1_IGDB_ID: Final = 7
PS2_IGDB_ID: Final = 8
PSP_IGDB_ID: Final = 38
@@ -213,6 +210,10 @@ class IGDBHandler(MetadataHandler):
self.igdb_service = IGDBService(twitch_auth=TwitchAuth())
self.pagination_limit = 200
@classmethod
def is_enabled(cls) -> bool:
return bool(IGDB_CLIENT_ID and IGDB_CLIENT_SECRET)
async def _search_rom(
self, search_term: str, platform_igdb_id: int, with_game_type: bool = False
) -> Game | None:
@@ -337,7 +338,7 @@ class IGDBHandler(MetadataHandler):
async def get_rom(self, fs_name: str, platform_igdb_id: int) -> IGDBRom:
from handler.filesystem import fs_rom_handler
if not IGDB_API_ENABLED:
if not self.is_enabled():
return IGDBRom(igdb_id=None)
if not platform_igdb_id:
@@ -432,7 +433,7 @@ class IGDBHandler(MetadataHandler):
)
async def get_rom_by_id(self, igdb_id: int) -> IGDBRom:
if not IGDB_API_ENABLED:
if not self.is_enabled():
return IGDBRom(igdb_id=None)
roms = await self.igdb_service.list_games(
@@ -463,7 +464,7 @@ class IGDBHandler(MetadataHandler):
)
async def get_matched_rom_by_id(self, igdb_id: int) -> IGDBRom | None:
if not IGDB_API_ENABLED:
if not self.is_enabled():
return None
rom = await self.get_rom_by_id(igdb_id)
@@ -472,7 +473,7 @@ class IGDBHandler(MetadataHandler):
async def get_matched_roms_by_name(
self, search_term: str, platform_igdb_id: int | None
) -> list[IGDBRom]:
if not IGDB_API_ENABLED:
if not self.is_enabled():
return []
if not platform_igdb_id:
@@ -561,8 +562,12 @@ class TwitchAuth(MetadataHandler):
self.masked_params = self._mask_sensitive_values(self.params)
self.timeout = 10
@classmethod
def is_enabled(cls) -> bool:
return IGDBHandler.is_enabled()
async def _update_twitch_token(self) -> str:
if not IGDB_API_ENABLED:
if not self.is_enabled():
return ""
token = None
@@ -608,7 +613,7 @@ class TwitchAuth(MetadataHandler):
if IS_PYTEST_RUN:
return "test_token"
if not IGDB_API_ENABLED:
if not self.is_enabled():
return ""
# Fetch the token cache

34
backend/handler/metadata/launchbox_handler.py
View File

@@ -1,22 +1,25 @@
import json
from datetime import datetime
from typing import NotRequired, TypedDict
from typing import Final, NotRequired, TypedDict
import pydash
from config import LAUNCHBOX_API_ENABLED, str_to_bool
from handler.redis_handler import async_cache
from logger.logger import log
from tasks.scheduled.update_launchbox_metadata import ( # LAUNCHBOX_MAME_KEY,
LAUNCHBOX_METADATA_ALTERNATE_NAME_KEY,
LAUNCHBOX_METADATA_DATABASE_ID_KEY,
LAUNCHBOX_METADATA_IMAGE_KEY,
LAUNCHBOX_METADATA_NAME_KEY,
update_launchbox_metadata_task,
)
from .base_hander import BaseRom, MetadataHandler
from .base_hander import UniversalPlatformSlug as UPS
LAUNCHBOX_PLATFORMS_KEY: Final[str] = "romm:launchbox_platforms"
LAUNCHBOX_METADATA_DATABASE_ID_KEY: Final[str] = "romm:launchbox_metadata_database_id"
LAUNCHBOX_METADATA_NAME_KEY: Final[str] = "romm:launchbox_metadata_name"
LAUNCHBOX_METADATA_ALTERNATE_NAME_KEY: Final[str] = (
"romm:launchbox_metadata_alternate_name"
)
LAUNCHBOX_METADATA_IMAGE_KEY: Final[str] = "romm:launchbox_metadata_image"
LAUNCHBOX_MAME_KEY: Final[str] = "romm:launchbox_mame"
LAUNCHBOX_FILES_KEY: Final[str] = "romm:launchbox_files"
class LaunchboxPlatform(TypedDict):
slug: str
@@ -115,11 +118,20 @@ def extract_metadata_from_launchbox_rom(
class LaunchboxHandler(MetadataHandler):
@classmethod
def is_enabled(cls) -> bool:
return LAUNCHBOX_API_ENABLED
async def _get_rom_from_metadata(
self, file_name: str, platform_slug: str
) -> dict | None:
if not (await async_cache.exists(LAUNCHBOX_METADATA_NAME_KEY)):
log.info("Fetching the Launchbox Metadata.xml file...")
from tasks.scheduled.update_launchbox_metadata import (
update_launchbox_metadata_task,
)
await update_launchbox_metadata_task.run(force=True)
if not (await async_cache.exists(LAUNCHBOX_METADATA_NAME_KEY)):
@@ -215,7 +227,7 @@ class LaunchboxHandler(MetadataHandler):
fallback_rom = LaunchboxRom(launchbox_id=None)
if not LAUNCHBOX_API_ENABLED:
if not self.is_enabled():
return fallback_rom
# We replace " - " with ": " to match Launchbox's naming convention
@@ -254,7 +266,7 @@ class LaunchboxHandler(MetadataHandler):
return LaunchboxRom({k: v for k, v in rom.items() if v}) # type: ignore[misc]
async def get_rom_by_id(self, database_id: int) -> LaunchboxRom:
if not LAUNCHBOX_API_ENABLED:
if not self.is_enabled():
return LaunchboxRom(launchbox_id=None)
metadata_database_index_entry = await async_cache.hget(
@@ -281,7 +293,7 @@ class LaunchboxHandler(MetadataHandler):
return LaunchboxRom({k: v for k, v in rom.items() if v}) # type: ignore[misc]
async def get_matched_rom_by_id(self, database_id: int) -> LaunchboxRom | None:
if not LAUNCHBOX_API_ENABLED:
if not self.is_enabled():
return None
return await self.get_rom_by_id(database_id)

15
backend/handler/metadata/moby_handler.py
View File

@@ -19,9 +19,6 @@ from .base_hander import (
)
from .base_hander import UniversalPlatformSlug as UPS
# Used to display the Mobygames API status in the frontend
MOBY_API_ENABLED: Final = bool(MOBYGAMES_API_KEY)
PS1_MOBY_ID: Final = 6
PS2_MOBY_ID: Final = 7
PSP_MOBY_ID: Final = 46
@@ -77,6 +74,10 @@ class MobyGamesHandler(MetadataHandler):
self.moby_service = MobyGamesService()
self.min_similarity_score = 0.6
@classmethod
def is_enabled(cls) -> bool:
return bool(MOBYGAMES_API_KEY)
async def _search_rom(
self, search_term: str, platform_moby_id: int, split_game_name: bool = False
) -> MobyGame | None:
@@ -128,7 +129,7 @@ class MobyGamesHandler(MetadataHandler):
async def get_rom(self, fs_name: str, platform_moby_id: int) -> MobyGamesRom:
from handler.filesystem import fs_rom_handler
if not MOBY_API_ENABLED:
if not self.is_enabled():
return MobyGamesRom(moby_id=None)
if not platform_moby_id:
@@ -222,7 +223,7 @@ class MobyGamesHandler(MetadataHandler):
return MobyGamesRom({k: v for k, v in rom.items() if v}) # type: ignore[misc]
async def get_rom_by_id(self, moby_id: int) -> MobyGamesRom:
if not MOBY_API_ENABLED:
if not self.is_enabled():
return MobyGamesRom(moby_id=None)
roms = await self.moby_service.list_games(game_id=moby_id)
@@ -242,7 +243,7 @@ class MobyGamesHandler(MetadataHandler):
return MobyGamesRom({k: v for k, v in rom.items() if v}) # type: ignore[misc]
async def get_matched_rom_by_id(self, moby_id: int) -> MobyGamesRom | None:
if not MOBY_API_ENABLED:
if not self.is_enabled():
return None
rom = await self.get_rom_by_id(moby_id)
@@ -251,7 +252,7 @@ class MobyGamesHandler(MetadataHandler):
async def get_matched_roms_by_name(
self, search_term: str, platform_moby_id: int | None
) -> list[MobyGamesRom]:
if not MOBY_API_ENABLED:
if not self.is_enabled():
return []
if not platform_moby_id:

9
backend/handler/metadata/playmatch_handler.py
View File

@@ -6,6 +6,7 @@ import httpx
import yarl
from config import PLAYMATCH_API_ENABLED
from fastapi import HTTPException, status
from handler.metadata.base_hander import MetadataHandler
from logger.logger import log
from models.rom import RomFile
from utils import get_version
@@ -38,7 +39,7 @@ class PlaymatchRomMatch(TypedDict):
igdb_id: int | None
class PlaymatchHandler:
class PlaymatchHandler(MetadataHandler):
"""
Handler for [Playmatch](https://github.com/RetroRealm/playmatch), a service for matching Roms by Hashes.
"""
@@ -47,6 +48,10 @@ class PlaymatchHandler:
self.base_url = "https://playmatch.retrorealm.dev/api"
self.identify_url = f"{self.base_url}/identify/ids"
@classmethod
def is_enabled(cls) -> bool:
return PLAYMATCH_API_ENABLED
async def _request(self, url: str, query: dict) -> dict:
"""
Sends a Request to Playmatch API.
@@ -100,7 +105,7 @@ class PlaymatchHandler:
:return: A PlaymatchRomMatch objects containing the matched ROM information.
:raises HTTPException: If the request fails or the service is unavailable.
"""
if not PLAYMATCH_API_ENABLED:
if not self.is_enabled():
return PlaymatchRomMatch(igdb_id=None)
first_file = next(

9
backend/handler/metadata/ra_handler.py
View File

@@ -2,7 +2,7 @@ import json
import os
import time
from datetime import datetime
from typing import Final, NotRequired, TypedDict
from typing import NotRequired, TypedDict
import pydash
from adapters.services.retroachievements import RetroAchievementsService
@@ -20,9 +20,6 @@ from models.rom import Rom
from .base_hander import BaseRom, MetadataHandler
from .base_hander import UniversalPlatformSlug as UPS
# Used to display the Retroachievements API status in the frontend
RA_API_ENABLED: Final = bool(RETROACHIEVEMENTS_API_KEY)
class RAGamesPlatform(TypedDict):
slug: str
@@ -125,6 +122,10 @@ class RAHandler(MetadataHandler):
self.ra_service = RetroAchievementsService()
self.HASHES_FILE_NAME = "ra_hashes.json"
@classmethod
def is_enabled(cls) -> bool:
return bool(RETROACHIEVEMENTS_API_KEY)
def _get_hashes_file_path(self, platform_id: int) -> str:
platform_resources_path = fs_resource_handler.get_platform_resources_path(
platform_id

11
backend/handler/metadata/sgdb_handler.py
View File

@@ -8,9 +8,6 @@ from logger.logger import log
from .base_hander import MetadataHandler
# Used to display the Mobygames API status in the frontend
STEAMGRIDDB_API_ENABLED: Final = bool(STEAMGRIDDB_API_KEY)
class SGDBResource(TypedDict):
thumb: str
@@ -33,8 +30,12 @@ class SGDBBaseHandler(MetadataHandler):
self.sgdb_service = SteamGridDBService()
self.min_similarity_score: Final = 0.98
@classmethod
def is_enabled(cls) -> bool:
return bool(STEAMGRIDDB_API_KEY)
async def get_details(self, search_term: str) -> list[SGDBResult]:
if not STEAMGRIDDB_API_ENABLED:
if not self.is_enabled():
return []
games = await self.sgdb_service.search_games(term=search_term)
@@ -51,7 +52,7 @@ class SGDBBaseHandler(MetadataHandler):
return list(filter(None, results))
async def get_details_by_names(self, game_names: list[str]) -> SGDBRom:
if not STEAMGRIDDB_API_ENABLED:
if not self.is_enabled():
return SGDBRom(sgdb_id=None)
for game_name in game_names:

14
backend/handler/metadata/ss_handler.py
View File

@@ -21,8 +21,6 @@ from .base_hander import (
)
from .base_hander import UniversalPlatformSlug as UPS
# Used to display the Screenscraper API status in the frontend
SS_API_ENABLED: Final = bool(SCREENSCRAPER_USER) and bool(SCREENSCRAPER_PASSWORD)
SS_DEV_ID: Final = base64.b64decode("enVyZGkxNQ==").decode()
SS_DEV_PASSWORD: Final = base64.b64decode("eFRKd29PRmpPUUc=").decode()
@@ -276,6 +274,10 @@ class SSHandler(MetadataHandler):
def __init__(self) -> None:
self.ss_service = ScreenScraperService()
@classmethod
def is_enabled(cls) -> bool:
return bool(SCREENSCRAPER_USER and SCREENSCRAPER_PASSWORD)
async def _search_rom(
self, search_term: str, platform_ss_id: int, split_game_name: bool = False
) -> SSGame | None:
@@ -323,7 +325,7 @@ class SSHandler(MetadataHandler):
async def get_rom(self, file_name: str, platform_ss_id: int) -> SSRom:
from handler.filesystem import fs_rom_handler
if not SS_API_ENABLED:
if not self.is_enabled():
return SSRom(ss_id=None)
if not platform_ss_id:
@@ -411,7 +413,7 @@ class SSHandler(MetadataHandler):
return build_ss_rom(res)
async def get_rom_by_id(self, ss_id: int) -> SSRom:
if not SS_API_ENABLED:
if not self.is_enabled():
return SSRom(ss_id=None)
res = await self.ss_service.get_game_info(game_id=ss_id)
@@ -421,7 +423,7 @@ class SSHandler(MetadataHandler):
return build_ss_rom(res)
async def get_matched_rom_by_id(self, ss_id: int) -> SSRom | None:
if not SS_API_ENABLED:
if not self.is_enabled():
return None
rom = await self.get_rom_by_id(ss_id)
@@ -430,7 +432,7 @@ class SSHandler(MetadataHandler):
async def get_matched_roms_by_name(
self, search_term: str, platform_ss_id: int | None
) -> list[SSRom]:
if not SS_API_ENABLED:
if not self.is_enabled():
return []
if not platform_ss_id:

6
backend/handler/metadata/tgdb_handler.py
View File

@@ -1,5 +1,7 @@
from typing import NotRequired, TypedDict
from config import TGDB_API_ENABLED
from .base_hander import MetadataHandler
from .base_hander import UniversalPlatformSlug as UPS
@@ -22,6 +24,10 @@ class TGDBHandler(MetadataHandler):
self.platform_endpoint = f"{self.BASE_URL}/Lookup/Platforms"
self.games_endpoint = f"{self.BASE_URL}/Lookup/ByHash"
@classmethod
def is_enabled(cls) -> bool:
return TGDB_API_ENABLED
def get_platform(self, slug: str) -> TGDBPlatform:
if slug not in TGDB_PLATFORM_LIST:
return TGDBPlatform(tgdb_id=None, slug=slug)

30
backend/tasks/scheduled/scan_library.py
View File

@@ -1,15 +1,17 @@
from config import (
ENABLE_SCHEDULED_RESCAN,
HASHEOUS_API_ENABLED,
LAUNCHBOX_API_ENABLED,
SCHEDULED_RESCAN_CRON,
)
from endpoints.sockets.scan import scan_platforms
from handler.metadata.igdb_handler import IGDB_API_ENABLED
from handler.metadata.moby_handler import MOBY_API_ENABLED
from handler.metadata.ra_handler import RA_API_ENABLED
from handler.metadata.sgdb_handler import STEAMGRIDDB_API_ENABLED
from handler.metadata.ss_handler import SS_API_ENABLED
from handler.metadata import (
meta_hasheous_handler,
meta_igdb_handler,
meta_launchbox_handler,
meta_moby_handler,
meta_ra_handler,
meta_sgdb_handler,
meta_ss_handler,
)
from handler.scan_handler import MetadataSource, ScanType
from logger.logger import log
from tasks.tasks import PeriodicTask
@@ -33,13 +35,13 @@ class ScanLibraryTask(PeriodicTask):
return None
source_mapping: dict[str, bool] = {
MetadataSource.IGDB: IGDB_API_ENABLED,
MetadataSource.SS: SS_API_ENABLED,
MetadataSource.MOBY: MOBY_API_ENABLED,
MetadataSource.RA: RA_API_ENABLED,
MetadataSource.LB: LAUNCHBOX_API_ENABLED,
MetadataSource.HASHEOUS: HASHEOUS_API_ENABLED,
MetadataSource.SGDB: STEAMGRIDDB_API_ENABLED,
MetadataSource.IGDB: meta_igdb_handler.is_enabled(),
MetadataSource.SS: meta_ss_handler.is_enabled(),
MetadataSource.MOBY: meta_moby_handler.is_enabled(),
MetadataSource.RA: meta_ra_handler.is_enabled(),
MetadataSource.LB: meta_launchbox_handler.is_enabled(),
MetadataSource.HASHEOUS: meta_hasheous_handler.is_enabled(),
MetadataSource.SGDB: meta_sgdb_handler.is_enabled(),
}
metadata_sources = [source for source, flag in source_mapping.items() if flag]

4
backend/tasks/scheduled/sync_retroachievements_progress.py
View File

@@ -6,7 +6,7 @@ from config import (
)
from handler.database import db_user_handler
from handler.metadata import meta_ra_handler
from handler.metadata.ra_handler import RA_API_ENABLED, RAUserProgression
from handler.metadata.ra_handler import RAUserProgression
from logger.logger import log
from tasks.tasks import PeriodicTask
from utils.context import initialize_context
@@ -25,7 +25,7 @@ class SyncRetroAchievementsProgressTask(PeriodicTask):
@initialize_context()
async def run(self) -> None:
if not RA_API_ENABLED:
if not meta_ra_handler.is_enabled():
log.warning("RetroAchievements API is not enabled, skipping progress sync")
return None

23
backend/tasks/scheduled/update_launchbox_metadata.py
View File

@@ -1,27 +1,28 @@
import json
import zipfile
from io import BytesIO
from typing import Any, Final
from typing import Any
from config import (
ENABLE_SCHEDULED_UPDATE_LAUNCHBOX_METADATA,
LAUNCHBOX_API_ENABLED,
SCHEDULED_UPDATE_LAUNCHBOX_METADATA_CRON,
)
from defusedxml import ElementTree as ET
from handler.metadata import meta_launchbox_handler
from handler.metadata.launchbox_handler import (
LAUNCHBOX_FILES_KEY,
LAUNCHBOX_MAME_KEY,
LAUNCHBOX_METADATA_ALTERNATE_NAME_KEY,
LAUNCHBOX_METADATA_DATABASE_ID_KEY,
LAUNCHBOX_METADATA_IMAGE_KEY,
LAUNCHBOX_METADATA_NAME_KEY,
LAUNCHBOX_PLATFORMS_KEY,
)
from handler.redis_handler import async_cache
from logger.logger import log
from tasks.tasks import RemoteFilePullTask
from utils.context import initialize_context
LAUNCHBOX_PLATFORMS_KEY: Final = "romm:launchbox_platforms"
LAUNCHBOX_METADATA_DATABASE_ID_KEY: Final = "romm:launchbox_metadata_database_id"
LAUNCHBOX_METADATA_NAME_KEY: Final = "romm:launchbox_metadata_name"
LAUNCHBOX_METADATA_ALTERNATE_NAME_KEY: Final = "romm:launchbox_metadata_alternate_name"
LAUNCHBOX_METADATA_IMAGE_KEY: Final = "romm:launchbox_metadata_image"
LAUNCHBOX_MAME_KEY: Final = "romm:launchbox_mame"
LAUNCHBOX_FILES_KEY: Final = "romm:launchbox_files"
class UpdateLaunchboxMetadataTask(RemoteFilePullTask):
def __init__(self):
@@ -37,7 +38,7 @@ class UpdateLaunchboxMetadataTask(RemoteFilePullTask):
@initialize_context()
async def run(self, force: bool = False) -> None:
if not LAUNCHBOX_API_ENABLED:
if not meta_launchbox_handler.is_enabled():
log.warning("Launchbox API is not enabled, skipping metadata update")
return None

8
backend/tests/handler/metadata/test_base_handler.py
View File

@@ -24,6 +24,12 @@ from handler.metadata.base_hander import (
from handler.redis_handler import async_cache
class ExampleMetadataHandler(MetadataHandler):
@classmethod
def is_enabled(cls) -> bool:
return True
class TestNormalizeSearchTerm:
"""Test the _normalize_search_term function."""
@@ -102,7 +108,7 @@ class TestMetadataHandlerMethods:
@pytest.fixture
def handler(self):
return MetadataHandler()
return ExampleMetadataHandler()
def test_normalize_cover_url_with_url(self, handler: MetadataHandler):
"""Test URL normalization with valid URL."""

43
backend/tests/tasks/test_scan_library.py
View File

@@ -1,6 +1,13 @@
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import AsyncMock, MagicMock
import pytest
from handler.metadata.hasheous_handler import HasheousHandler
from handler.metadata.igdb_handler import IGDBHandler
from handler.metadata.launchbox_handler import LaunchboxHandler
from handler.metadata.moby_handler import MobyGamesHandler
from handler.metadata.ra_handler import RAHandler
from handler.metadata.sgdb_handler import SGDBBaseHandler
from handler.metadata.ss_handler import SSHandler
from handler.scan_handler import MetadataSource, ScanType
from tasks.scheduled.scan_library import ScanLibraryTask, scan_library_task
@@ -15,18 +22,20 @@ class TestScanLibraryTask:
assert task.func == "tasks.scheduled.scan_library.scan_library_task.run"
assert task.description == "Rescans the entire library"
@patch("tasks.scheduled.scan_library.ENABLE_SCHEDULED_RESCAN", True)
@patch("tasks.scheduled.scan_library.IGDB_API_ENABLED", False)
@patch("tasks.scheduled.scan_library.SS_API_ENABLED", False)
@patch("tasks.scheduled.scan_library.MOBY_API_ENABLED", False)
@patch("tasks.scheduled.scan_library.RA_API_ENABLED", True)
@patch("tasks.scheduled.scan_library.LAUNCHBOX_API_ENABLED", True)
@patch("tasks.scheduled.scan_library.HASHEOUS_API_ENABLED", False)
@patch("tasks.scheduled.scan_library.STEAMGRIDDB_API_ENABLED", False)
@patch("tasks.scheduled.scan_library.scan_platforms")
@patch("tasks.scheduled.scan_library.log")
async def test_run_enabled(self, mock_log, mock_scan_platforms, task):
async def test_run_enabled(self, task, mocker):
"""Test run when scheduled rescan is enabled"""
mocker.patch.object(HasheousHandler, "is_enabled", return_value=False)
mocker.patch.object(IGDBHandler, "is_enabled", return_value=False)
mocker.patch.object(LaunchboxHandler, "is_enabled", return_value=True)
mocker.patch.object(MobyGamesHandler, "is_enabled", return_value=False)
mocker.patch.object(RAHandler, "is_enabled", return_value=True)
mocker.patch.object(SGDBBaseHandler, "is_enabled", return_value=False)
mocker.patch.object(SSHandler, "is_enabled", return_value=False)
mocker.patch("tasks.scheduled.scan_library.ENABLE_SCHEDULED_RESCAN", True)
mock_scan_platforms = mocker.patch(
"tasks.scheduled.scan_library.scan_platforms"
)
mock_log = mocker.patch("tasks.scheduled.scan_library.log")
mock_scan_platforms.return_value = AsyncMock()
await task.run()
@@ -39,11 +48,13 @@ class TestScanLibraryTask:
)
mock_log.info.assert_any_call("Scheduled library scan done")
@patch("tasks.scheduled.scan_library.ENABLE_SCHEDULED_RESCAN", False)
@patch("tasks.scheduled.scan_library.scan_platforms")
@patch("tasks.scheduled.scan_library.log")
async def test_run_disabled(self, mock_log, mock_scan_platforms, task):
async def test_run_disabled(self, task, mocker):
"""Test run when scheduled rescan is disabled"""
mocker.patch("tasks.scheduled.scan_library.ENABLE_SCHEDULED_RESCAN", False)
mock_scan_platforms = mocker.patch(
"tasks.scheduled.scan_library.scan_platforms"
)
mock_log = mocker.patch("tasks.scheduled.scan_library.log")
task.unschedule = MagicMock()
await task.run()

9
backend/tests/tasks/test_sync_retroachievements_progress.py
View File

@@ -1,4 +1,4 @@
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock
import pytest
from handler.database.users_handler import DBUsersHandler
@@ -25,10 +25,11 @@ class TestSyncRetroAchievementsProgressTask:
)
assert task.description == "Updates RetroAchievements progress for all users"
@patch("tasks.scheduled.sync_retroachievements_progress.RA_API_ENABLED", False)
@patch("tasks.scheduled.sync_retroachievements_progress.log")
async def test_run_when_retroachievements_api_disabled(self, mock_log, task):
async def test_run_when_retroachievements_api_disabled(self, task, mocker):
"""Test run method when RetroAchievements API is disabled."""
mocker.patch.object(RAHandler, "is_enabled", return_value=False)
mock_log = mocker.patch("tasks.scheduled.sync_retroachievements_progress.log")
await task.run()
mock_log.warning.assert_called_once_with(

12
backend/tests/tasks/test_update_launchbox_metadata.py
View File

@@ -3,7 +3,7 @@ from unittest.mock import AsyncMock, patch
import anyio
import pytest
from tasks.scheduled.update_launchbox_metadata import (
from handler.metadata.launchbox_handler import (
LAUNCHBOX_FILES_KEY,
LAUNCHBOX_MAME_KEY,
LAUNCHBOX_METADATA_ALTERNATE_NAME_KEY,
@@ -11,6 +11,9 @@ from tasks.scheduled.update_launchbox_metadata import (
LAUNCHBOX_METADATA_IMAGE_KEY,
LAUNCHBOX_METADATA_NAME_KEY,
LAUNCHBOX_PLATFORMS_KEY,
LaunchboxHandler,
)
from tasks.scheduled.update_launchbox_metadata import (
UpdateLaunchboxMetadataTask,
update_launchbox_metadata_task,
)
@@ -61,10 +64,11 @@ class TestUpdateLaunchboxMetadataTask:
mock_super_run.assert_called_once_with(True)
@patch("tasks.scheduled.update_launchbox_metadata.LAUNCHBOX_API_ENABLED", False)
@patch("tasks.scheduled.update_launchbox_metadata.log")
async def test_run_when_launchbox_api_disabled(self, mock_log, task):
async def test_run_when_launchbox_api_disabled(self, task, mocker):
"""Test run method when Launchbox API is disabled"""
mocker.patch.object(LaunchboxHandler, "is_enabled", return_value=False)
mock_log = mocker.patch("tasks.scheduled.update_launchbox_metadata.log")
await task.run(force=True)
mock_log.warning.assert_called_once_with(

30
backend/watcher.py
View File

@@ -8,8 +8,6 @@ from typing import cast
import sentry_sdk
from config import (
ENABLE_RESCAN_ON_FILESYSTEM_CHANGE,
HASHEOUS_API_ENABLED,
LAUNCHBOX_API_ENABLED,
LIBRARY_BASE_PATH,
RESCAN_ON_FILESYSTEM_CHANGE_DELAY,
SCAN_TIMEOUT,
@@ -18,11 +16,15 @@ from config import (
from config.config_manager import config_manager as cm
from endpoints.sockets.scan import scan_platforms
from handler.database import db_platform_handler
from handler.metadata.igdb_handler import IGDB_API_ENABLED
from handler.metadata.moby_handler import MOBY_API_ENABLED
from handler.metadata.ra_handler import RA_API_ENABLED
from handler.metadata.sgdb_handler import STEAMGRIDDB_API_ENABLED
from handler.metadata.ss_handler import SS_API_ENABLED
from handler.metadata import (
meta_hasheous_handler,
meta_igdb_handler,
meta_launchbox_handler,
meta_moby_handler,
meta_ra_handler,
meta_sgdb_handler,
meta_ss_handler,
)
from handler.scan_handler import MetadataSource, ScanType
from logger.formatter import CYAN
from logger.formatter import highlight as hl
@@ -96,13 +98,13 @@ def process_changes(changes: Sequence[Change]) -> None:
# Check whether any metadata source is enabled.
source_mapping: dict[str, bool] = {
MetadataSource.IGDB: IGDB_API_ENABLED,
MetadataSource.SS: SS_API_ENABLED,
MetadataSource.MOBY: MOBY_API_ENABLED,
MetadataSource.RA: RA_API_ENABLED,
MetadataSource.LB: LAUNCHBOX_API_ENABLED,
MetadataSource.HASHEOUS: HASHEOUS_API_ENABLED,
MetadataSource.SGDB: STEAMGRIDDB_API_ENABLED,
MetadataSource.IGDB: meta_igdb_handler.is_enabled(),
MetadataSource.SS: meta_ss_handler.is_enabled(),
MetadataSource.MOBY: meta_moby_handler.is_enabled(),
MetadataSource.RA: meta_ra_handler.is_enabled(),
MetadataSource.LB: meta_launchbox_handler.is_enabled(),
MetadataSource.HASHEOUS: meta_hasheous_handler.is_enabled(),
MetadataSource.SGDB: meta_sgdb_handler.is_enabled(),
}
metadata_sources = [source for source, flag in source_mapping.items() if flag]
if not metadata_sources: