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

commit backen code

Browse Source
This commit is contained in:
Georges-Antoine Assi
2025-10-16 15:41:42 -04:00
parent 7e4e27db62
commit 05c5d79e5f
7 changed files with 42 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/endpoints/responses/__init__.py
View File

@@ -13,7 +13,7 @@ class ScanStats(TypedDict):
identified_platforms: int
scanned_roms: int
added_roms: int
metadata_roms: int
identified_roms: int
scanned_firmware: int
added_firmware: int

20
backend/endpoints/sockets/scan.py
View File

@@ -56,7 +56,7 @@ class ScanStats:
identified_platforms: int = 0
scanned_roms: int = 0
added_roms: int = 0
metadata_roms: int = 0
identified_roms: int = 0
scanned_firmware: int = 0
added_firmware: int = 0
@@ -76,7 +76,7 @@ class ScanStats:
"identified_platforms": self.identified_platforms,
"scanned_roms": self.scanned_roms,
"added_roms": self.added_roms,
"metadata_roms": self.metadata_roms,
"identified_roms": self.identified_roms,
"scanned_firmware": self.scanned_firmware,
"added_firmware": self.added_firmware,
}
@@ -248,7 +248,7 @@ async def _identify_rom(
scan_stats.update(
scanned_roms=scan_stats.scanned_roms + 1,
added_roms=scan_stats.added_roms + (1 if not rom else 0),
metadata_roms=scan_stats.metadata_roms
identified_roms=scan_stats.identified_roms
+ (1 if scanned_rom.is_identified else 0),
)
@@ -460,7 +460,7 @@ async def scan_platforms(
scan_type: ScanType = ScanType.QUICK,
roms_ids: list[int] | None = None,
metadata_sources: list[str] | None = None,
):
) -> ScanStats:
"""Scan all the listed platforms and fetch metadata from different sources
Args:
@@ -474,23 +474,21 @@ async def scan_platforms(
roms_ids = []
socket_manager = _get_socket_manager()
scan_stats = ScanStats()
if not metadata_sources:
log.error("No metadata sources provided")
await socket_manager.emit("scan:done_ko", "No metadata sources provided")
return None
return scan_stats
try:
fs_platforms: list[str] = await fs_platform_handler.get_platforms()
except FolderStructureNotMatchException as e:
log.error(e)
await socket_manager.emit("scan:done_ko", e.message)
return None
return scan_stats
scan_stats = ScanStats(total_platforms=len(fs_platforms))
# Set up meta update for the scan stats
update_job_meta({"scan_stats": scan_stats.to_dict()})
scan_stats.update(total_platforms=len(fs_platforms))
for platform in fs_platforms:
pl = Platform(fs_slug=platform)
@@ -548,6 +546,8 @@ async def scan_platforms(
# Re-raise the exception to be caught by the error handler
raise e
return scan_stats
@socket_handler.socket_server.on("scan") # type: ignore
async def scan_handler(_sid: str, options: dict[str, Any]):

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

@@ -2,7 +2,7 @@ from config import (
ENABLE_SCHEDULED_RESCAN,
SCHEDULED_RESCAN_CRON,
)
from endpoints.sockets.scan import scan_platforms
from endpoints.sockets.scan import ScanStats, scan_platforms
from handler.metadata import (
meta_flashpoint_handler,
meta_hasheous_handler,
@@ -33,10 +33,12 @@ class ScanLibraryTask(PeriodicTask):
)
async def run(self) -> dict[str, str]:
scan_stats = ScanStats()
if not ENABLE_SCHEDULED_RESCAN:
log.info("Scheduled library scan not enabled, unscheduling...")
self.unschedule()
return {"status": "skipped", "reason": "Scheduled library scan not enabled"}
return scan_stats.to_dict()
source_mapping: dict[str, bool] = {
MetadataSource.IGDB: meta_igdb_handler.is_enabled(),
@@ -55,15 +57,15 @@ class ScanLibraryTask(PeriodicTask):
if not metadata_sources:
log.warning("No metadata sources enabled, unscheduling library scan")
self.unschedule()
return {"status": "skipped", "reason": "No metadata sources enabled"}
return scan_stats.to_dict()
log.info("Scheduled library scan started...")
await scan_platforms(
scan_stats = await scan_platforms(
[], scan_type=ScanType.UNIDENTIFIED, metadata_sources=metadata_sources
)
log.info("Scheduled library scan done")
return {"status": "completed", "message": "Library scan completed successfully"}
return scan_stats.to_dict()
scan_library_task = ScanLibraryTask()

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

@@ -28,13 +28,11 @@ class SyncRetroAchievementsProgressTask(PeriodicTask):
@initialize_context()
async def run(self) -> dict[str, Any]:
update_stats = UpdateStats()
if not meta_ra_handler.is_enabled():
log.warning("RetroAchievements API is not enabled, skipping progress sync")
return {
"status": "skipped",
"reason": "RetroAchievements API not enabled",
"updated_users": 0,
}
return update_stats.to_dict()
log.info("Scheduled RetroAchievements progress sync started...")
@@ -43,7 +41,6 @@ class SyncRetroAchievementsProgressTask(PeriodicTask):
processed_users = 0
# Update initial progress
update_stats = UpdateStats()
update_stats.update(processed=processed_users, total=total_users)
for user in users:

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

@@ -42,16 +42,16 @@ class UpdateLaunchboxMetadataTask(RemoteFilePullTask):
@initialize_context()
async def run(self, force: bool = False) -> dict[str, Any]:
update_stats = UpdateStats()
if not meta_launchbox_handler.is_enabled():
log.warning("Launchbox API is not enabled, skipping metadata update")
return {"status": "skipped", "reason": "Launchbox API not enabled"}
return update_stats.to_dict()
content = await super().run(force)
if content is None:
log.warning("No content received from launchbox metadata update")
return {"status": "failed", "reason": "No content received"}
update_stats = UpdateStats()
return update_stats.to_dict()
try:
zip_file_bytes = BytesIO(content)

6
backend/tasks/scheduled/update_switch_titledb.py
View File

@@ -32,18 +32,18 @@ class UpdateSwitchTitleDBTask(RemoteFilePullTask):
@initialize_context()
async def run(self, force: bool = False) -> dict[str, Any]:
update_stats = UpdateStats()
content = await super().run(force)
if content is None:
return {"status": "failed", "reason": "No content received"}
return update_stats.to_dict()
index_json = json.loads(content)
relevant_data = {k: v for k, v in index_json.items() if k and v}
total_items = len(relevant_data)
processed_items = 0
# Update initial progress
update_stats = UpdateStats()
update_stats.update(processed=processed_items, total=total_items)
async with async_cache.pipeline() as pipe:

34
backend/tests/endpoints/sockets/test_scan.py
View File

@@ -14,7 +14,7 @@ def test_scan_stats():
assert stats.identified_platforms == 0
assert stats.scanned_roms == 0
assert stats.added_roms == 0
assert stats.metadata_roms == 0
assert stats.identified_roms == 0
assert stats.scanned_firmware == 0
assert stats.added_firmware == 0
@@ -23,7 +23,7 @@ def test_scan_stats():
stats.identified_platforms += 1
stats.scanned_roms += 1
stats.added_roms += 1
stats.metadata_roms += 1
stats.identified_roms += 1
stats.scanned_firmware += 1
stats.added_firmware += 1
@@ -32,7 +32,7 @@ def test_scan_stats():
assert stats.identified_platforms == 1
assert stats.scanned_roms == 1
assert stats.added_roms == 1
assert stats.metadata_roms == 1
assert stats.identified_roms == 1
assert stats.scanned_firmware == 1
assert stats.added_firmware == 1
@@ -44,37 +44,31 @@ def test_merging_scan_stats():
identified_platforms=3,
scanned_roms=4,
added_roms=5,
metadata_roms=6,
identified_roms=6,
scanned_firmware=7,
added_firmware=8,
)
stats2 = ScanStats(
scanned_platforms=10,
new_platforms=11,
identified_platforms=12,
scanned_roms=13,
added_roms=14,
metadata_roms=15,
scanned_firmware=16,
added_firmware=17,
stats.update(
scanned_platforms=stats.scanned_platforms + 10,
new_platforms=stats.new_platforms + 11,
identified_platforms=stats.identified_platforms + 12,
scanned_roms=stats.scanned_roms + 13,
added_roms=stats.added_roms + 14,
identified_roms=stats.identified_roms + 15,
scanned_firmware=stats.scanned_firmware + 16,
added_firmware=stats.added_firmware + 17,
)
stats += stats2
assert stats.scanned_platforms == 11
assert stats.new_platforms == 13
assert stats.identified_platforms == 15
assert stats.scanned_roms == 17
assert stats.added_roms == 19
assert stats.metadata_roms == 21
assert stats.identified_roms == 21
assert stats.scanned_firmware == 23
assert stats.added_firmware == 25
stats3: dict = {}
with pytest.raises(NotImplementedError):
stats += stats3
class TestShouldScanRom:
def test_new_platforms_scan_with_no_rom(self):