"""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 is decrypted by the bundled pure-Python decryptor (``app/pkg.py``) into a ``PSP/GAME//``-style folder — no Wine or ``pkg2zip`` required. """ from __future__ import annotations import logging import shutil import urllib.parse import urllib.request from datetime import datetime from pathlib import Path from typing import List from . import config log = logging.getLogger("nopaystation") _TSV_NAMES = { "psp": "PSP_GAMES.tsv", "psx": "PSX_GAMES.tsv", "theme": "PSP_THEMES.tsv", "dlc": "PSP_DLCS.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`` / ``theme`` / ``dlc``).""" 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.USER_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 = {"psp": "games", "psx": "ps1", "theme": "theme", "dlc": "dlc"}.get(content_type, "games") 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 "" # NoPayStation marks delisted packages with the literal "MISSING" in the # URL column — skip them so they can't be queued for download. if not name or not pkg_url or not pkg_url.lower().startswith(("http://", "https://")): 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, force: bool = False) -> List[dict]: """Download (and cache) a TSV, returning parsed item dicts. ``force`` re-downloads the TSV (used by the Store's Refresh button); the local cache is only used as a fallback when the network fails. """ url = url or tsv_url(content_type) name = _TSV_NAMES.get(content_type, "PSP_GAMES.tsv") cache = _tsv_dir() / name def read_cache(): try: return parse_tsv(cache.read_text(encoding="utf-8", errors="replace"), content_type) except OSError: return None if not force and cache.is_file(): items = read_cache() if items is not None: return items try: 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) except Exception: # noqa: BLE001 - fall back to the cached list on failure items = read_cache() if items is not None: log.warning("could not refresh %s TSV; using cached copy", content_type) return items raise def _safe_folder(value: str) -> str: import re cleaned = re.sub(r"[^A-Za-z0-9]", "", value or "").upper() return cleaned or "PKG" def _ensure_keys_bin(folder, kind): """PS1 Classics need KEYS.BIN next to EBOOT.PBP to boot on CFW (npdrm).""" if kind != "psx": return src = config.DATA_DIR / "loader" / "KEYS.BIN" dest = Path(folder) / "KEYS.BIN" if src.is_file() and not dest.exists(): try: shutil.copy2(src, dest) except OSError: pass def extract_pkg(pkg_path, title_id: str, dest_dir, kind_hint: str = None, progress_cb=None, cancel_event=None) -> Path: """Decrypt/extract a ``.pkg`` into ``dest_dir//``. Uses the bundled pure-Python decryptor (no Wine or pkg2zip). Returns the destination game folder path. """ from . import pkg as pkgmod from .store import TransferCancelled pkg_path = Path(pkg_path) dest = Path(dest_dir) / _safe_folder(title_id) if dest.is_dir() and any(dest.iterdir()): return dest try: _kind, _tid, folder = pkgmod.extract(pkg_path, dest_dir, kind_hint, progress_cb, cancel_event) _ensure_keys_bin(folder, _kind) return folder except TransferCancelled: raise except Exception as exc: # noqa: BLE001 raise RuntimeError(f"could not decrypt {pkg_path.name}: {exc}") from exc