feat: add Pegasus Frontend metadata export support

Add metadata.pegasus.txt export alongside the existing gamelist.xml
export. Restructure the export system: rename the gamelist endpoint to
a general-purpose export endpoint (`/api/export/`) with sub-routes for
each format (`/gamelist-xml`, `/pegasus`). Move config from flat
`scan.export_gamelist` to nested `scan.export.gamelist_xml` and
`scan.export.pegasus` for auto-export on scan.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Georges-Antoine Assi
2026-03-21 16:09:20 -04:00
parent c52bdf9b81
commit 770b8f94ac
8 changed files with 327 additions and 30 deletions

View File

@@ -45,6 +45,7 @@ DEFAULT_EXCLUDED_FILES: Final = [
".stfolder",
"@SynoResource",
"gamelist.xml",
"metadata.pegasus.txt",
]
DEFAULT_EXCLUDED_DIRS: Final = [
"@eaDir",
@@ -106,7 +107,8 @@ class Config:
EXCLUDED_MULTI_FILES: list[str]
EXCLUDED_MULTI_PARTS_EXT: list[str]
EXCLUDED_MULTI_PARTS_FILES: list[str]
GAMELIST_AUTO_EXPORT_ON_SCAN: bool
SCAN_EXPORT_GAMELIST_XML: bool
SCAN_EXPORT_PEGASUS: bool
PLATFORMS_BINDING: dict[str, str]
PLATFORMS_VERSIONS: dict[str, str]
ROMS_FOLDER_NAME: str
@@ -266,8 +268,11 @@ class ConfigManager:
FIRMWARE_FOLDER_NAME=pydash.get(
self._raw_config, "filesystem.firmware_folder", "bios"
),
GAMELIST_AUTO_EXPORT_ON_SCAN=pydash.get(
self._raw_config, "scan.export_gamelist", False
SCAN_EXPORT_GAMELIST_XML=pydash.get(
self._raw_config, "scan.export.gamelist_xml", False
),
SCAN_EXPORT_PEGASUS=pydash.get(
self._raw_config, "scan.export.pegasus", False
),
SKIP_HASH_CALCULATION=pydash.get(
self._raw_config, "filesystem.skip_hash_calculation", False
@@ -410,8 +415,14 @@ class ConfigManager:
"Invalid config.yml: exclude.roms.multi_file.parts.names must be a list"
)
sys.exit(3)
if not isinstance(self.config.GAMELIST_AUTO_EXPORT_ON_SCAN, bool):
log.critical("Invalid config.yml: scan.export_gamelist must be a boolean")
if not isinstance(self.config.SCAN_EXPORT_GAMELIST_XML, bool):
log.critical(
"Invalid config.yml: scan.export.gamelist_xml must be a boolean"
)
sys.exit(3)
if not isinstance(self.config.SCAN_EXPORT_PEGASUS, bool):
log.critical("Invalid config.yml: scan.export.pegasus must be a boolean")
sys.exit(3)
if not isinstance(self.config.PLATFORMS_BINDING, dict):
@@ -619,7 +630,10 @@ class ConfigManager:
"language": self.config.SCAN_LANGUAGE_PRIORITY,
},
"media": self.config.SCAN_MEDIA,
"export_gamelist": self.config.GAMELIST_AUTO_EXPORT_ON_SCAN,
"export": {
"gamelist_xml": self.config.SCAN_EXPORT_GAMELIST_XML,
"pegasus": self.config.SCAN_EXPORT_PEGASUS,
},
},
}

View File

@@ -9,16 +9,17 @@ from logger.formatter import BLUE
from logger.formatter import highlight as hl
from logger.logger import log
from utils.gamelist_exporter import GamelistExporter
from utils.pegasus_exporter import PegasusExporter
from utils.router import APIRouter
router = APIRouter(
prefix="/gamelist",
tags=["gamelist"],
prefix="/export",
tags=["export"],
)
@protected_route(router.post, "/export", [Scope.ROMS_READ])
async def export_gamelist(
@protected_route(router.post, "/gamelist-xml", [Scope.ROMS_READ])
async def export_gamelist_xml(
request: Request,
platform_ids: Annotated[
List[int], Query(description="List of platform IDs to export")
@@ -38,7 +39,6 @@ async def export_gamelist(
exporter = GamelistExporter(local_export=local_export)
files_written = []
# Export each platform to its respective directory
for platform_id in platform_ids:
success = await exporter.export_platform_to_file(
platform_id,
@@ -69,3 +69,58 @@ async def export_gamelist(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to export gamelist",
) from e
@protected_route(router.post, "/pegasus", [Scope.ROMS_READ])
async def export_pegasus(
request: Request,
platform_ids: Annotated[
List[int], Query(description="List of platform IDs to export")
],
local_export: Annotated[
bool, Query(description="Use local paths instead of URLs")
] = False,
) -> Response:
"""Export platforms/ROMs to Pegasus metadata.pegasus.txt format and write to platform directories"""
if not platform_ids:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="At least one platform ID must be provided",
)
try:
exporter = PegasusExporter(local_export=local_export)
files_written = []
for platform_id in platform_ids:
success = await exporter.export_platform_to_file(
platform_id,
request,
)
if success:
files_written.append(f"metadata_pegasus_{platform_id}.txt")
else:
log.warning(
f"Failed to write Pegasus metadata for platform {platform_id}"
)
if not files_written:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to write any Pegasus metadata files",
)
log.info(
f"Exported Pegasus metadata for {hl(str(len(files_written)), color=BLUE)} platform(s):"
)
for file in files_written:
log.info(f"\t{file}")
return Response(status_code=status.HTTP_200_OK)
except ValueError as e:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) from e
except Exception as e:
log.error(f"Failed to export Pegasus metadata: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to export Pegasus metadata",
) from e

View File

@@ -51,6 +51,7 @@ from tasks.tasks import update_job_meta
from utils import emoji
from utils.context import initialize_context
from utils.gamelist_exporter import GamelistExporter
from utils.pegasus_exporter import PegasusExporter
STOP_SCAN_FLAG: Final = "scan:stop"
@@ -699,14 +700,16 @@ async def scan_platforms(
log.info(f"{emoji.EMOJI_CHECK_MARK} Scan completed")
# Export gamelist.xml if enabled in config
# Export metadata files if enabled in config
config = cm.get_config()
if config.GAMELIST_AUTO_EXPORT_ON_SCAN:
log.info("Auto-exporting gamelist.xml for all platforms...")
gamelist_exporter = GamelistExporter(local_export=True)
if config.SCAN_EXPORT_GAMELIST_XML or config.SCAN_EXPORT_PEGASUS:
platforms_by_slug = {
p.fs_slug: p for p in db_platform_handler.get_platforms()
}
if config.SCAN_EXPORT_GAMELIST_XML:
log.info("Auto-exporting gamelist.xml for all platforms...")
gamelist_exporter = GamelistExporter(local_export=True)
for platform_slug in platform_list:
platform = platforms_by_slug.get(platform_slug)
if platform:
@@ -724,6 +727,26 @@ async def scan_platforms(
)
log.info("Gamelist.xml auto-export completed.")
if config.SCAN_EXPORT_PEGASUS:
log.info("Auto-exporting metadata.pegasus.txt for all platforms...")
pegasus_exporter = PegasusExporter(local_export=True)
for platform_slug in platform_list:
platform = platforms_by_slug.get(platform_slug)
if platform:
export_success = await pegasus_exporter.export_platform_to_file(
platform.id,
request=None,
)
if export_success:
log.info(
f"Auto-exported metadata.pegasus.txt for platform {platform.name} after scan"
)
else:
log.warning(
f"Failed to auto-export metadata.pegasus.txt for platform {platform.name} after scan"
)
log.info("Pegasus metadata auto-export completed.")
await socket_manager.emit("scan:done", scan_stats.to_dict())
except ScanStoppedException:
await stop_scan()

View File

@@ -29,9 +29,9 @@ from endpoints.client_tokens import router as client_tokens_router
from endpoints.collections import router as collections_router
from endpoints.configs import router as configs_router
from endpoints.device import router as device_router
from endpoints.export import router as export_router
from endpoints.feeds import router as feeds_router
from endpoints.firmware import router as firmware_router
from endpoints.gamelist import router as gamelist_router
from endpoints.heartbeat import router as heartbeat_router
from endpoints.netplay import router as netplay_router
from endpoints.platform import router as platform_router
@@ -139,7 +139,7 @@ app.include_router(raw_router, prefix="/api")
app.include_router(screenshots_router, prefix="/api")
app.include_router(firmware_router, prefix="/api")
app.include_router(collections_router, prefix="/api")
app.include_router(gamelist_router, prefix="/api")
app.include_router(export_router, prefix="/api")
app.include_router(netplay_router, prefix="/api")
app.mount("/ws", socket_handler.socket_app)

View File

@@ -0,0 +1,196 @@
from datetime import datetime
from fastapi import Request
from handler.database import db_platform_handler, db_rom_handler
from handler.filesystem import fs_platform_handler
from logger.logger import log
from models.rom import Rom
class PegasusExporter:
"""Export RomM collections to Pegasus Frontend metadata.pegasus.txt format"""
def __init__(self, local_export: bool = False):
self.local_export = local_export
def _format_release_date(self, timestamp: int) -> str:
"""Format release date to YYYY-MM-DD format"""
return datetime.fromtimestamp(timestamp / 1000).strftime("%Y-%m-%d")
def _format_rating(self, average_rating: float) -> str:
"""Format rating as percentage (0-100%). Input is on 0-10 scale."""
return f"{int(average_rating * 10)}%"
def _escape_multiline(self, text: str) -> str:
"""Indent continuation lines for multi-line values in Pegasus format"""
lines = text.strip().splitlines()
if len(lines) <= 1:
return text.strip()
# First line is on the same line as the key, subsequent lines are indented
return (
lines[0]
+ "\n"
+ "\n".join(f" {line}" if line.strip() else " ." for line in lines[1:])
)
def _create_game_entry(self, rom: Rom, request: Request | None) -> str:
"""Create a game entry for a ROM in Pegasus metadata format"""
lines: list[str] = []
# Game title (required)
lines.append(f"game: {rom.name or rom.fs_name}")
# File path
if self.local_export:
lines.append(f"file: {rom.fs_name}")
else:
if request is None:
raise ValueError(
"Request object must be provided for non-local exports"
)
lines.append(
f"file: {request.url_for('get_rom_content', id=rom.id, file_name=rom.fs_name)}"
)
# Sort title (use fs_name_no_tags if different from name)
if rom.name and rom.fs_name_no_tags and rom.name != rom.fs_name_no_tags:
lines.append(f"sort-by: {rom.fs_name_no_tags}")
# Developers and publishers
if rom.metadatum and rom.metadatum.companies:
if len(rom.metadatum.companies) > 0:
lines.append(f"developer: {rom.metadatum.companies[0]}")
if len(rom.metadatum.companies) > 1:
lines.append(f"publisher: {rom.metadatum.companies[1]}")
# Genres
if rom.metadatum and rom.metadatum.genres:
for genre in rom.metadatum.genres:
lines.append(f"genre: {genre}")
# Tags (rom tags like region, language info)
if rom.tags:
for tag in rom.tags:
lines.append(f"tag: {tag}")
# Player count
player_count = None
if rom.gamelist_metadata and rom.gamelist_metadata.get("player_count"):
player_count = rom.gamelist_metadata["player_count"]
elif rom.metadatum and rom.metadatum.player_count:
player_count = rom.metadatum.player_count
if player_count:
lines.append(f"players: {player_count}")
# Summary / description
if rom.summary:
lines.append(f"description: {self._escape_multiline(rom.summary)}")
# Release date
if rom.metadatum and rom.metadatum.first_release_date is not None:
lines.append(
f"release: {self._format_release_date(rom.metadatum.first_release_date)}"
)
# Rating
if rom.metadatum and rom.metadatum.average_rating is not None:
lines.append(f"rating: {self._format_rating(rom.metadatum.average_rating)}")
# RomM-specific extensions (x-* fields)
if rom.regions:
lines.append(f"x-region: {', '.join(rom.regions)}")
if rom.languages:
lines.append(f"x-language: {', '.join(rom.languages)}")
if rom.gamelist_id:
lines.append(f"x-romm-gamelist-id: {rom.gamelist_id}")
return "\n".join(lines)
def export_platform_to_pegasus(
self, platform_id: int, request: Request | None
) -> str:
"""Export a platform's ROMs to metadata.pegasus.txt format
Args:
platform_id: Platform ID to export
request: FastAPI request object for URL generation
Returns:
String content in Pegasus metadata format
"""
platform = db_platform_handler.get_platform(platform_id)
if not platform:
raise ValueError(f"Platform with ID {platform_id} not found")
roms = db_rom_handler.get_roms_scalar(platform_ids=[platform_id])
lines: list[str] = []
# Collection header
lines.append(f"collection: {platform.custom_name or platform.name}")
lines.append(f"shortname: {platform.slug}")
lines.append("")
# Game entries
game_count = 0
for rom in roms:
if (
rom
and not rom.missing_from_fs
and rom.fs_name
not in (
"gamelist.xml",
"metadata.pegasus.txt",
)
):
if game_count > 0:
lines.append("")
lines.append(self._create_game_entry(rom, request=request))
game_count += 1
log.info(f"Exported {game_count} ROMs for platform {platform.name}")
return "\n".join(lines) + "\n"
async def export_platform_to_file(
self,
platform_id: int,
request: Request | None,
) -> bool:
"""Export platform ROMs to metadata.pegasus.txt file in the platform's directory
Args:
platform_id: Platform ID to export
request: FastAPI request object for URL generation
Returns:
True if successful, False otherwise
"""
try:
platform = db_platform_handler.get_platform(platform_id)
if not platform:
log.error(f"Platform with ID {platform_id} not found")
return False
platform_fs_structure = fs_platform_handler.get_platform_fs_structure(
platform.fs_slug
)
content = self.export_platform_to_pegasus(platform_id, request=request)
await fs_platform_handler.write_file(
content.encode("utf-8"),
platform_fs_structure,
"metadata.pegasus.txt",
)
log.info(
f"Exported metadata.pegasus.txt to {platform_fs_structure}/metadata.pegasus.txt"
)
return True
except Exception as e:
log.error(
f"Failed to export metadata.pegasus.txt for platform {platform_id}: {e}"
)
return False

View File

@@ -134,6 +134,10 @@
# - logo # Transparent logo
# # Other media assets (might be used in the future)
# - marquee # Custom marquee
# # Auto-export metadata files after scan completes
# export:
# gamelist_xml: false # Export gamelist.xml (ES-DE format)
# pegasus: false # Export metadata.pegasus.txt (Pegasus Frontend format)
# EmulatorJS per-core options
# emulatorjs:

View File

@@ -0,0 +1,18 @@
import api from "@/services/api";
async function exportGamelistXml({ platformIds }: { platformIds: number[] }) {
const params = new URLSearchParams();
platformIds.forEach((id) => params.append("platform_ids", id.toString()));
await api.post(`/export/gamelist-xml?${params.toString()}`);
}
async function exportPegasus({ platformIds }: { platformIds: number[] }) {
const params = new URLSearchParams();
platformIds.forEach((id) => params.append("platform_ids", id.toString()));
await api.post(`/export/pegasus?${params.toString()}`);
}
export default {
exportGamelistXml,
exportPegasus,
};

View File

@@ -1,13 +0,0 @@
import api from "@/services/api";
export const gamelistApi = api;
async function exportGamelist({ platformIds }: { platformIds: number[] }) {
const params = new URLSearchParams();
platformIds.forEach((id) => params.append("platform_ids", id.toString()));
await api.post(`/gamelist/export?${params.toString()}`);
}
export default {
exportGamelist,
};