"""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//`` 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"} # TSV columns (tab-separated), in order. _COLUMNS = ( "title_id", "region", "name", "pkg_url", "zrif", "content_id", "date", "size", "sha256", ) 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 _cache_dir() -> Path: d = config.STATE_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. 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] = [] lines = text.splitlines() for line in lines: line = line.rstrip("\r\n") if not line.strip(): continue cols = line.split("\t") # Skip the header row. if cols and cols[0].strip().lower() == "title id": continue if len(cols) < 5: continue row = dict(zip(_COLUMNS, cols)) name = (row.get("name") or "").strip() pkg_url = (row.get("pkg_url") or "").strip() title_id = (row.get("title_id") or "").strip() if not name or not pkg_url: continue 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 "").strip(), "pkg_url": pkg_url, "zrif": (row.get("zrif") or "").strip(), "content_id": (row.get("content_id") or "").strip(), "size": size, "sha256": (row.get("sha256") or "").strip().lower(), "release_date": date, "mtime_ns": mtime_ns, "content_type": content_type, "system": system, }) 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 = _cache_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//``. 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