132 lines
3.9 KiB
Python
132 lines
3.9 KiB
Python
"""Redump 7z store: scan archives and extract the disc image inside.
|
|
|
|
Store archives are ``Name (Region) (tags).7z`` with the ``.iso`` (or ``.cso``)
|
|
inside. Extraction shells out to the system ``7z`` binary.
|
|
"""
|
|
import logging
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
log = logging.getLogger("store")
|
|
|
|
_ARCHIVE_EXTS = {".7z", ".zip"}
|
|
_IMAGE_EXTS = {".iso", ".cso", ".img", ".bin"}
|
|
|
|
|
|
class TransferCancelled(Exception):
|
|
"""Raised inside a transfer when the user cancels it."""
|
|
|
|
|
|
def is_archive(path):
|
|
return Path(path).suffix.lower() in _ARCHIVE_EXTS
|
|
|
|
|
|
def display_name(name):
|
|
"""Display title of a store archive (Redump names are kept as-is)."""
|
|
return str(name).strip()
|
|
|
|
|
|
def region(name):
|
|
"""Extract the region tag (first trailing parenthetical) from a store name."""
|
|
s = str(name)
|
|
tags = []
|
|
while True:
|
|
m = re.search(r"\s*\(([^()]*)\)\s*$", s)
|
|
if not m:
|
|
break
|
|
tags.insert(0, m.group(1).strip())
|
|
s = s[: m.start()]
|
|
return tags[0] if tags else ""
|
|
|
|
|
|
def scan(root, limit=None):
|
|
"""Return ``[{path, name, size, mtime_ns}]`` for archives in the store."""
|
|
root = Path(root)
|
|
if not root.is_dir():
|
|
return []
|
|
items = []
|
|
for path in sorted(root.iterdir()):
|
|
if not path.is_file() or not is_archive(path):
|
|
continue
|
|
try:
|
|
stat = path.stat()
|
|
except OSError:
|
|
continue
|
|
items.append({
|
|
"path": str(path),
|
|
"name": path.stem,
|
|
"size": stat.st_size,
|
|
"mtime_ns": stat.st_mtime_ns,
|
|
})
|
|
if limit is not None and len(items) >= limit:
|
|
break
|
|
return items
|
|
|
|
|
|
def peek_archive(path, cancel_event=None):
|
|
"""Return the disc-image filename inside a 7z archive, or None."""
|
|
path = Path(path)
|
|
if not path.is_file():
|
|
return None
|
|
try:
|
|
proc = subprocess.Popen(
|
|
["7z", "l", "-slt", str(path)],
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
|
|
)
|
|
except OSError:
|
|
return None
|
|
name = None
|
|
for line in proc.stdout:
|
|
if cancel_event is not None and cancel_event.is_set():
|
|
proc.terminate()
|
|
try:
|
|
proc.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
proc.kill()
|
|
raise TransferCancelled()
|
|
if line.startswith("Path = "):
|
|
candidate = line[7:].strip()
|
|
if candidate.lower().endswith(tuple(_IMAGE_EXTS)):
|
|
name = candidate
|
|
code = proc.wait()
|
|
if code != 0:
|
|
return None
|
|
return name
|
|
|
|
|
|
def extract_iso(archive, iso_name, out_dir, progress_cb=None, cancel_event=None):
|
|
"""Extract one disc image from a 7z archive; return its path."""
|
|
archive = Path(archive)
|
|
out_dir = Path(out_dir)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
cmd = ["7z", "x", "-y", f"-o{out_dir}", str(archive), iso_name]
|
|
try:
|
|
proc = subprocess.Popen(
|
|
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
|
|
)
|
|
except OSError as exc:
|
|
raise RuntimeError(f"7z is not available: {exc}") from exc
|
|
|
|
last_pct = 0.0
|
|
for line in proc.stdout:
|
|
if cancel_event is not None and cancel_event.is_set():
|
|
proc.terminate()
|
|
try:
|
|
proc.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
proc.kill()
|
|
raise TransferCancelled()
|
|
m = re.search(r"(\d{1,3})\s*%", line)
|
|
if m:
|
|
last_pct = int(m.group(1)) / 100.0
|
|
if progress_cb is not None:
|
|
progress_cb("Extracting", last_pct, line.strip()[:80])
|
|
code = proc.wait()
|
|
if code != 0:
|
|
raise RuntimeError(f"7z extraction failed (exit {code})")
|
|
result = out_dir / iso_name
|
|
if not result.exists():
|
|
raise RuntimeError("7z extraction completed but the image was not found")
|
|
return result
|