first commit
This commit is contained in:
@@ -0,0 +1,213 @@
|
||||
"""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,
|
||||
}
|
||||
Reference in New Issue
Block a user