psp-preservation-tool/app/nopaystation.py

"""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/<TITLE_ID>/``-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": "psppt/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/<TITLE_ID>/``.

    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