214 lines
7.0 KiB
Python
214 lines
7.0 KiB
Python
"""Read identifying metadata from PSP .iso images (ISO9660 + PARAM.SFO).
|
|
|
|
PSP UMD dumps are ISO9660 images containing ``PSP_GAME/PARAM.SFO`` (games) or
|
|
``UMD_VIDEO/PARAM.SFO`` (videos). PARAM.SFO is a key/value blob that carries
|
|
``DISC_ID`` (the catalogue serial, e.g. ``ULUS-10027``) and ``TITLE``.
|
|
"""
|
|
import re
|
|
import struct
|
|
|
|
|
|
class IsoError(Exception):
|
|
"""Raised when an ISO cannot be parsed or lacks the expected files."""
|
|
|
|
|
|
SECTOR_SIZE = 2048
|
|
PVD_SECTOR = 16
|
|
ROOT_RECORD_OFFSET = 156
|
|
|
|
|
|
def _read_sector(f, lba):
|
|
f.seek(lba * SECTOR_SIZE)
|
|
data = f.read(SECTOR_SIZE)
|
|
if len(data) < SECTOR_SIZE:
|
|
raise IsoError(f"short read at LBA {lba}")
|
|
return data
|
|
|
|
|
|
def _parse_dir_record(buf, offset):
|
|
"""Parse one ISO9660 directory record at ``offset`` in ``buf``.
|
|
|
|
Returns ``(entry, next_offset)`` where ``entry`` is
|
|
``(name, is_dir, extent_lba, size)`` or ``None`` when ``offset`` points at
|
|
padding.
|
|
"""
|
|
if offset >= len(buf):
|
|
return None, offset
|
|
length = buf[offset]
|
|
if length == 0:
|
|
return None, offset + 1
|
|
if offset + 33 > len(buf) or offset + length > len(buf):
|
|
raise IsoError("directory record overruns buffer")
|
|
name_len = buf[offset + 32]
|
|
flags = buf[offset + 25]
|
|
extent = struct.unpack_from("<I", buf, offset + 2)[0]
|
|
size = struct.unpack_from("<I", buf, offset + 10)[0]
|
|
raw_name = buf[offset + 33 : offset + 33 + name_len]
|
|
name = raw_name.decode("ascii", "replace").rstrip("\x00")
|
|
is_dir = bool(flags & 0x02)
|
|
return (name, is_dir, extent, size), offset + length
|
|
|
|
|
|
def _list_dir(f, extent, size):
|
|
f.seek(extent * SECTOR_SIZE)
|
|
data = f.read(size)
|
|
if len(data) < size:
|
|
raise IsoError("short directory read")
|
|
records = []
|
|
offset = 0
|
|
while offset < len(data):
|
|
entry, next_offset = _parse_dir_record(data, offset)
|
|
if entry is None:
|
|
# Padding: skip to the start of the next 2048-byte sector.
|
|
offset = ((offset // SECTOR_SIZE) + 1) * SECTOR_SIZE
|
|
continue
|
|
records.append(entry)
|
|
offset = next_offset
|
|
return records
|
|
|
|
|
|
def _basename(name):
|
|
return name.split(";")[0].strip()
|
|
|
|
|
|
def _find_entry(f, extent, size, target):
|
|
for name, is_dir, entry_extent, entry_size in _list_dir(f, extent, size):
|
|
if _basename(name) == target:
|
|
return is_dir, entry_extent, entry_size
|
|
return None
|
|
|
|
|
|
def _root_record(f):
|
|
pvd = _read_sector(f, PVD_SECTOR)
|
|
if pvd[0] != 1:
|
|
raise IsoError("sector 16 is not a primary volume descriptor")
|
|
entry, _ = _parse_dir_record(pvd, ROOT_RECORD_OFFSET)
|
|
if entry is None:
|
|
raise IsoError("missing root directory record")
|
|
_, _, extent, size = entry
|
|
return extent, size
|
|
|
|
|
|
def _read_sfo_file(f, dir_name):
|
|
root_extent, root_size = _root_record(f)
|
|
found = _find_entry(f, root_extent, root_size, dir_name)
|
|
if not found:
|
|
raise IsoError(f"{dir_name} directory not found")
|
|
is_dir, extent, size = found
|
|
if not is_dir:
|
|
raise IsoError(f"{dir_name} is not a directory")
|
|
sfo = _find_entry(f, extent, size, "PARAM.SFO")
|
|
if not sfo:
|
|
raise IsoError(f"{dir_name}/PARAM.SFO not found")
|
|
_, sfo_extent, sfo_size = sfo
|
|
f.seek(sfo_extent * SECTOR_SIZE)
|
|
data = f.read(sfo_size)
|
|
if len(data) < sfo_size:
|
|
raise IsoError("short PARAM.SFO read")
|
|
return data
|
|
|
|
|
|
def read_file(path, dir_name, filename):
|
|
"""Read the bytes of ``dir_name/filename`` from an ISO image.
|
|
|
|
Raises ``IsoError`` when the directory or file is not found."""
|
|
with open(path, "rb") as f:
|
|
root_extent, root_size = _root_record(f)
|
|
found = _find_entry(f, root_extent, root_size, dir_name)
|
|
if not found:
|
|
raise IsoError(f"{dir_name} directory not found")
|
|
is_dir, extent, size = found
|
|
if not is_dir:
|
|
raise IsoError(f"{dir_name} is not a directory")
|
|
entry = _find_entry(f, extent, size, filename)
|
|
if not entry:
|
|
raise IsoError(f"{dir_name}/{filename} not found")
|
|
_, f_extent, f_size = entry
|
|
f.seek(f_extent * SECTOR_SIZE)
|
|
data = f.read(f_size)
|
|
if len(data) < f_size:
|
|
raise IsoError(f"short read for {dir_name}/{filename}")
|
|
return data
|
|
|
|
|
|
def parse_sfo(data):
|
|
"""Parse a PARAM.SFO blob into a ``{key: value}`` dict.
|
|
|
|
SFO layout: 20-byte header, then the index table (``count * 16`` bytes at
|
|
offset ``key_start - count * 16``), then the key table, then the data
|
|
table. ``fmt`` 0x0004 / 0x0404 is a uint32, anything else is a string."""
|
|
if len(data) < 20 or data[:4] != b"\x00PSF":
|
|
raise IsoError("bad PARAM.SFO magic")
|
|
_version, key_start, data_start, count = struct.unpack_from("<IIII", data, 4)
|
|
if count <= 0:
|
|
return {}
|
|
size = len(data)
|
|
index_start = key_start - count * 16
|
|
if index_start < 20:
|
|
index_start = 20
|
|
if index_start + count * 16 > size:
|
|
raise IsoError("bad PARAM.SFO index table")
|
|
fields = {}
|
|
for i in range(count):
|
|
off = index_start + i * 16
|
|
key_off, fmt, param_len, _max_len, data_off = struct.unpack_from("<HHIII", data, off)
|
|
key = data[key_start + key_off :].split(b"\x00", 1)[0].decode("utf-8", "replace")
|
|
value_start = data_start + data_off
|
|
if value_start + param_len > size:
|
|
continue
|
|
raw = data[value_start : value_start + param_len]
|
|
if fmt in (0x0004, 0x0404) and param_len == 4: # uint32
|
|
value = struct.unpack("<I", raw)[0]
|
|
else: # UTF-8 string
|
|
value = raw.rstrip(b"\x00").decode("utf-8", "replace")
|
|
fields[key] = value
|
|
return fields
|
|
|
|
|
|
_SERIAL_RE = re.compile(r"^([A-Za-z]{4})-?(\d{4,5})(.*)$")
|
|
|
|
|
|
def normalize_serial(value):
|
|
"""Normalize a PARAM.SFO DISC_ID (e.g. ULUS10021 -> ULUS-10021)."""
|
|
if not value:
|
|
return value
|
|
s = str(value).strip().upper()
|
|
m = _SERIAL_RE.match(s)
|
|
if m:
|
|
return f"{m.group(1)}-{m.group(2)}{m.group(3)}"
|
|
return s
|
|
|
|
|
|
def looks_like_serial(value):
|
|
"""True if a value looks like a real PSP catalogue number (e.g. ULES-00011)."""
|
|
if not value:
|
|
return False
|
|
return bool(re.match(r"^[A-Za-z]{3,4}-?\d{3,5}(-.*)?$", str(value).strip()))
|
|
|
|
|
|
def read_identity(path):
|
|
"""Return ``{"serial": ..., "title": ..., "sfo": {...}, "kind": ...}``.
|
|
|
|
``kind`` is ``"game"`` when the disc has ``PSP_GAME/PARAM.SFO`` or
|
|
``"video"`` when it has ``UMD_VIDEO/PARAM.SFO``.
|
|
"""
|
|
sfo = None
|
|
kind = None
|
|
with open(path, "rb") as f:
|
|
for dir_name in ("PSP_GAME", "UMD_VIDEO"):
|
|
try:
|
|
sfo = parse_sfo(_read_sfo_file(f, dir_name))
|
|
kind = "game" if dir_name == "PSP_GAME" else "video"
|
|
break
|
|
except IsoError:
|
|
continue
|
|
if sfo is None:
|
|
raise IsoError("no PARAM.SFO found (tried PSP_GAME and UMD_VIDEO)")
|
|
title = sfo.get("TITLE") or sfo.get("TITLE_0") or sfo.get("TITLE_00")
|
|
return {
|
|
"serial": normalize_serial(sfo.get("DISC_ID")),
|
|
"title": title,
|
|
"sfo": sfo,
|
|
"kind": kind,
|
|
}
|