quoije
d02152ca06
- PSP_GAMES.tsv and PSX_GAMES.tsv use different headers (PSP adds Type/RAP columns, PSX adds Original Name). Parse the header row and map columns by name instead of hardcoding positions — PSP names were showing as 'Minis'/ 'PC Engine' (the Type column) with broken download URLs. - GameDetailsPanel.set_item referenced is_ps1 before it was assigned, raising UnboundLocalError and leaving the detail panel stale.
219 lines
7.4 KiB
Python
219 lines
7.4 KiB
Python
"""NoPayStation (.pkg) browsing and extraction.
|
|
|
|
NoPayStation publishes per-platform TSV lists (``https://nopaystation.com/tsv/``)
|
|
with one row per downloadable PSN item. Each row links to an encrypted PSN
|
|
``.pkg`` (a ``\\x7FPKG`` container, not a ZIP) that must be decrypted with the
|
|
``pkg2zip`` tool before it can be installed as a ``PSP/GAME/<TITLE_ID>/``
|
|
folder.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import shutil
|
|
import subprocess
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
from . import config
|
|
|
|
log = logging.getLogger("nopaystation")
|
|
|
|
_TSV_NAMES = {"psp": "PSP_GAMES.tsv", "psx": "PSX_GAMES.tsv"}
|
|
|
|
# NoPayStation TSVs are tab-separated with a header row whose exact columns
|
|
# differ per platform (PSP has "Type"/"RAP" columns, PSX has "Original Name").
|
|
# Map header names to canonical keys instead of relying on fixed positions.
|
|
_HEADER_ALIASES = {
|
|
"title id": "title_id",
|
|
"region": "region",
|
|
"name": "name",
|
|
"pkg direct link": "pkg_url",
|
|
"content id": "content_id",
|
|
"last modification date": "date",
|
|
"file size": "size",
|
|
"sha256": "sha256",
|
|
"zrif": "zrif",
|
|
"rap": "rap",
|
|
"download .rap file": "rap_url",
|
|
"type": "content_kind",
|
|
"original name": "original_name",
|
|
}
|
|
|
|
|
|
def tsv_url(content_type: str, base: str = "") -> str:
|
|
"""TSV URL for a content type (``psp`` / ``psx``)."""
|
|
base = (base or config.NPS_TSV_BASE).rstrip("/")
|
|
return f"{base}/{_TSV_NAMES.get(content_type, _TSV_NAMES['psp'])}"
|
|
|
|
|
|
def _tsv_dir() -> Path:
|
|
d = config.DATA_DIR / "nps"
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return d
|
|
|
|
|
|
def _parse_date(value: str) -> str:
|
|
value = (value or "").strip()
|
|
if not value:
|
|
return ""
|
|
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
|
|
try:
|
|
return datetime.strptime(value, fmt).strftime("%Y-%m-%d")
|
|
except ValueError:
|
|
continue
|
|
return value[:10]
|
|
|
|
|
|
def parse_tsv(text: str, content_type: str) -> List[dict]:
|
|
"""Parse a NoPayStation TSV into store item dicts.
|
|
|
|
The header row drives column mapping (PSP and PSX TSVs differ). Each item
|
|
carries the fields the Store page expects (``name``, ``size``,
|
|
``release_date``, ``mtime_ns``) plus ``title_id``, ``pkg_url``, ``sha256``,
|
|
``content_type`` and the cover ``system`` ("games" for PSP, "ps1" for PSX).
|
|
"""
|
|
system = "games" if content_type == "psp" else "ps1"
|
|
items: List[dict] = []
|
|
header = None
|
|
for line in text.splitlines():
|
|
line = line.rstrip("\r\n")
|
|
if not line.strip():
|
|
continue
|
|
cols = line.split("\t")
|
|
if header is None:
|
|
header = [c.strip().lower() for c in cols]
|
|
continue
|
|
row = {}
|
|
for i, key in enumerate(header):
|
|
canon = _HEADER_ALIASES.get(key)
|
|
if canon and i < len(cols):
|
|
row[canon] = cols[i].strip()
|
|
name = row.get("name") or ""
|
|
pkg_url = row.get("pkg_url") or ""
|
|
if not name or not pkg_url:
|
|
continue
|
|
title_id = row.get("title_id") or ""
|
|
try:
|
|
size = int(float(row.get("size") or 0))
|
|
except (TypeError, ValueError):
|
|
size = 0
|
|
date = _parse_date(row.get("date") or "")
|
|
try:
|
|
mtime_ns = int(datetime.strptime(date, "%Y-%m-%d").timestamp() * 1_000_000_000) if date else 0
|
|
except ValueError:
|
|
mtime_ns = 0
|
|
items.append({
|
|
"name": name,
|
|
"title_id": title_id,
|
|
"region": row.get("region") or "",
|
|
"pkg_url": pkg_url,
|
|
"zrif": row.get("zrif") or "",
|
|
"content_id": row.get("content_id") or "",
|
|
"size": size,
|
|
"sha256": (row.get("sha256") or "").strip().lower(),
|
|
"release_date": date,
|
|
"mtime_ns": mtime_ns,
|
|
"content_type": content_type,
|
|
"system": system,
|
|
"content_kind": row.get("content_kind") or "",
|
|
})
|
|
return items
|
|
|
|
|
|
def fetch_items(content_type: str, url: str = "", timeout: int = 60) -> List[dict]:
|
|
"""Download (and cache) a TSV, returning parsed item dicts."""
|
|
url = url or tsv_url(content_type)
|
|
name = _TSV_NAMES.get(content_type, "PSP_GAMES.tsv")
|
|
cache = _tsv_dir() / name
|
|
if cache.is_file():
|
|
try:
|
|
return parse_tsv(cache.read_text(encoding="utf-8", errors="replace"), content_type)
|
|
except OSError:
|
|
pass
|
|
req = urllib.request.Request(url, headers={"User-Agent": "psp-preserve/0.1"})
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
data = resp.read()
|
|
text = data.decode("utf-8", errors="replace")
|
|
cache.write_text(text, encoding="utf-8")
|
|
return parse_tsv(text, content_type)
|
|
|
|
|
|
def find_pkg2zip(configured_path: str = "") -> Optional[Path]:
|
|
"""Locate the pkg2zip binary (configured path, else PATH)."""
|
|
if configured_path:
|
|
p = Path(configured_path)
|
|
return p if p.is_file() else None
|
|
found = shutil.which("pkg2zip")
|
|
return Path(found) if found else None
|
|
|
|
|
|
def _safe_folder(value: str) -> str:
|
|
import re
|
|
cleaned = re.sub(r"[^A-Za-z0-9]", "", value or "").upper()
|
|
return cleaned or "PKG"
|
|
|
|
|
|
def extract_pkg(pkg_path, title_id: str, dest_dir, pkg2zip: Path,
|
|
progress_cb=None, cancel_event=None) -> Path:
|
|
"""Decrypt/extract a ``.pkg`` with pkg2zip into ``dest_dir/<TITLE_ID>/``.
|
|
|
|
pkg2zip extracts PSP/PSX content into a folder layout we then normalize:
|
|
the game's ``EBOOT.PBP`` (plus siblings) are copied into the destination
|
|
game folder. Returns the destination folder path.
|
|
"""
|
|
from .store import TransferCancelled
|
|
|
|
pkg_path = Path(pkg_path)
|
|
folder_name = _safe_folder(title_id)
|
|
dest = Path(dest_dir) / folder_name
|
|
if (dest / "EBOOT.PBP").exists():
|
|
return dest
|
|
|
|
tmp = Path(dest_dir).with_name(".nps-tmp-" + folder_name)
|
|
if tmp.exists():
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
tmp.mkdir(parents=True, exist_ok=True)
|
|
|
|
if progress_cb is not None:
|
|
progress_cb("Extracting", 0.0, "")
|
|
try:
|
|
proc = subprocess.Popen(
|
|
[str(pkg2zip), "-x", str(pkg_path)],
|
|
cwd=str(tmp), stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
text=True, bufsize=1,
|
|
)
|
|
assert proc.stdout is not None
|
|
for _line in proc.stdout:
|
|
if cancel_event is not None and cancel_event.is_set():
|
|
proc.kill()
|
|
raise TransferCancelled()
|
|
proc.wait()
|
|
if proc.returncode != 0:
|
|
raise RuntimeError(f"pkg2zip failed (exit {proc.returncode}) for {pkg_path.name}")
|
|
finally:
|
|
if cancel_event is not None and cancel_event.is_set():
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
raise TransferCancelled()
|
|
|
|
eboots = list(tmp.rglob("EBOOT.PBP"))
|
|
if not eboots:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
raise RuntimeError(f"pkg2zip produced no EBOOT.PBP from {pkg_path.name}")
|
|
src_folder = eboots[0].parent
|
|
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
for p in sorted(src_folder.rglob("*")):
|
|
if not p.is_file():
|
|
continue
|
|
rel = p.relative_to(src_folder)
|
|
target = dest / rel
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(p, target)
|
|
if progress_cb is not None:
|
|
progress_cb("Extracting", 1.0, "")
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
return dest
|