[ROMM-2762] Add get_rom_by_hash endpoint

This commit is contained in:
Georges-Antoine Assi
2025-12-12 16:52:17 -05:00
parent 109bc113b0
commit 8a66ac8127
4 changed files with 139 additions and 72 deletions

View File

@@ -454,67 +454,95 @@ async def download_roms(
[] if DISABLE_DOWNLOAD_ENDPOINT_AUTH else [Scope.ROMS_READ],
responses={status.HTTP_404_NOT_FOUND: {}},
)
def get_rom_by_metadata(
def get_rom_by_metadata_provider(
request: Request,
igdb: Annotated[int | None, Query(description="IGDB ID to search by")] = None,
moby: Annotated[int | None, Query(description="MobyGames ID to search by")] = None,
ss: Annotated[
igdb_id: Annotated[int | None, Query(description="IGDB ID to search by")] = None,
moby_id: Annotated[
int | None, Query(description="MobyGames ID to search by")
] = None,
ss_id: Annotated[
int | None, Query(description="ScreenScraper ID to search by")
] = None,
ra: Annotated[
ra_id: Annotated[
int | None, Query(description="RetroAchievements ID to search by")
] = None,
launchbox: Annotated[
launchbox_id: Annotated[
int | None, Query(description="LaunchBox ID to search by")
] = None,
hasheous: Annotated[
hasheous_id: Annotated[
int | None, Query(description="Hasheous ID to search by")
] = None,
tgdb: Annotated[int | None, Query(description="TGDB ID to search by")] = None,
flashpoint: Annotated[
tgdb_id: Annotated[int | None, Query(description="TGDB ID to search by")] = None,
flashpoint_id: Annotated[
str | None, Query(description="Flashpoint ID to search by")
] = None,
hltb: Annotated[int | None, Query(description="HLTB ID to search by")] = None,
hltb_id: Annotated[int | None, Query(description="HLTB ID to search by")] = None,
) -> DetailedRomSchema:
"""Retrieve a rom by metadata ID."""
if (
not igdb_id
and not moby_id
and not ss_id
and not ra_id
and not launchbox_id
and not hasheous_id
and not tgdb_id
and not flashpoint_id
and not hltb_id
):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="At least one metadata ID must be provided",
)
rom = db_rom_handler.get_rom_by_metadata_id(
igdb=igdb,
moby=moby,
ss=ss,
ra=ra,
launchbox=launchbox,
hasheous=hasheous,
tgdb=tgdb,
flashpoint=flashpoint,
hltb=hltb,
igdb_id=igdb_id,
moby_id=moby_id,
ss_id=ss_id,
ra_id=ra_id,
launchbox_id=launchbox_id,
hasheous_id=hasheous_id,
tgdb_id=tgdb_id,
flashpoint_id=flashpoint_id,
hltb_id=hltb_id,
)
if not rom:
provided_ids = {
"igdb_id": igdb,
"moby_id": moby,
"ss_id": ss,
"ra_id": ra,
"launchbox_id": launchbox,
"hasheous_id": hasheous,
"tgdb_id": tgdb,
"flashpoint_id": flashpoint,
"hltb_id": hltb,
}
metadata_info = [
f"{key}={value}" for key, value in provided_ids.items() if value is not None
]
if not metadata_info:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="At least one metadata ID must be provided",
)
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"ROM not found with metadata: {', '.join(metadata_info)}",
detail=f"ROM not found with given metadata IDs",
)
return DetailedRomSchema.from_orm_with_request(rom, request)
@protected_route(
router.get,
"/by-hash",
[] if DISABLE_DOWNLOAD_ENDPOINT_AUTH else [Scope.ROMS_READ],
responses={status.HTTP_404_NOT_FOUND: {}},
)
def get_rom_by_hash(
request: Request,
crc_hash: Annotated[str | None, Query(description="CRC hash value")] = None,
md5_hash: Annotated[str | None, Query(description="MD5 hash value")] = None,
sha1_hash: Annotated[str | None, Query(description="SHA1 hash value")] = None,
):
if not crc_hash and not md5_hash and not sha1_hash:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="At least one metadata hash value must be provided",
)
rom = db_rom_handler.get_rom_by_hash(
crc_hash=crc_hash, md5_hash=md5_hash, sha1_hash=sha1_hash
)
if not rom:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"No ROM or file found with given hash values",
)
return DetailedRomSchema.from_orm_with_request(rom, request)

View File

@@ -982,39 +982,79 @@ class DBRomsHandler(DBBaseHandler):
@with_details
def get_rom_by_metadata_id(
self,
igdb: int | None = None,
moby: int | None = None,
ss: int | None = None,
ra: int | None = None,
launchbox: int | None = None,
hasheous: int | None = None,
tgdb: int | None = None,
flashpoint: str | None = None,
hltb: int | None = None,
igdb_id: int | None = None,
moby_id: int | None = None,
ss_id: int | None = None,
ra_id: int | None = None,
launchbox_id: int | None = None,
hasheous_id: int | None = None,
tgdb_id: int | None = None,
flashpoint_id: str | None = None,
hltb_id: int | None = None,
*,
query: Query = None,
session: Session = None,
) -> Rom | None:
"""Get a ROM by any metadata ID."""
filters = []
param_map = [
(igdb, Rom.igdb_id),
(moby, Rom.moby_id),
(ss, Rom.ss_id),
(ra, Rom.ra_id),
(launchbox, Rom.launchbox_id),
(hasheous, Rom.hasheous_id),
(tgdb, Rom.tgdb_id),
(flashpoint, Rom.flashpoint_id),
(hltb, Rom.hltb_id),
]
"""
Get a ROM by any metadata ID.
for value, column in param_map:
if value is not None:
filters.append(column == value)
Returns the first ROM that matches any of the provided metadata IDs.
"""
# Build filters for non-nil IDs
filters = [
column == value
for value, column in [
(igdb_id, Rom.igdb_id),
(moby_id, Rom.moby_id),
(ss_id, Rom.ss_id),
(ra_id, Rom.ra_id),
(launchbox_id, Rom.launchbox_id),
(hasheous_id, Rom.hasheous_id),
(tgdb_id, Rom.tgdb_id),
(flashpoint_id, Rom.flashpoint_id),
(hltb_id, Rom.hltb_id),
]
if value is not None
]
if not filters:
return None
# Use OR to find ROM matching any of the provided metadata IDs
# Return the first ROM matching any of the provided metadata IDs
return session.scalar(query.filter(or_(*filters)).limit(1))
@begin_session
@with_details
def get_rom_by_hash(
self,
crc_hash: str | None,
md5_hash: str | None,
sha1_hash: str | None,
*,
query: Query = None,
session: Session = None,
):
"""
Get a ROM by calculated hash value.
Returns the first ROM that matches any of the provided hash values.
"""
# Build filters for non-nil IDs
filters = [
column == value
for value, column in [
(crc_hash, Rom.crc_hash),
(md5_hash, Rom.md5_hash),
(sha1_hash, Rom.sha1_hash),
(crc_hash, RomFile.crc_hash),
(md5_hash, RomFile.md5_hash),
(sha1_hash, RomFile.sha1_hash),
]
if value is not None
]
if not filters:
return None
# Return the first ROM matching any of the provided metadata IDs
return session.scalar(query.filter(or_(*filters)).limit(1))

View File

@@ -30,7 +30,7 @@ const linkTarget = computed(() => {
onMounted(async () => {
await romApi
.getRomByMetadataProvider({ provider: "igdb", id: props.game.id })
.getRomByMetadataProvider({ field: "igdb_id", id: props.game.id })
.then((response) => {
console.log("Fetched ROM by metadata provider:", response.data);
romId.value = response.data.id;

View File

@@ -183,15 +183,14 @@ async function getRom({
}
async function getRomByMetadataProvider({
provider,
field,
id,
}: {
provider: string;
field: Partial<keyof DetailedRom>;
id: number;
}): Promise<{ data: DetailedRom }> {
const params = { [provider]: id };
return api.get(`/roms/by-metadata-provider/`, {
params,
params: { [field]: id },
});
}