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 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"}

# 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 _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.

    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 = _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