Bulk-downloader for Slay the Spire 2 runs from spire-codex.com (concurrent listing + incremental top-up). Standalone, single-file. Deps: httpx, typer, tenacity.

download_spire_codex.py

"""Bulk-download every run from spire-codex.com.

Lists all runs from ``/api/runs/list`` then fetches ``/api/runs/shared/{hash}``
for every run not already on disk, writing the JSON to
``<data>/files-spire-codex/<character>/<YYYY-MM>/<hash>.run``.

The listing endpoint is the bottleneck (~10s/page, capped at 50 rows/page), so
pages are fetched concurrently rather than one at a time — sequential paging of
the full corpus took *hours*. Detail fetches are fast (~0.3s) and also run
concurrently. Runs already on disk, and run_hashes recorded as permanently
unretrievable in ``_errors.log``, are skipped without any network call so
re-runs (top-ups) are cheap.

For a cheap incremental top-up (e.g. from a cron), pass ``--since-newest``: the
listing is newest-first, so it walks from page 1 and stops once it reaches runs
already on disk instead of enumerating all ~2,300 pages.

Output directory and User-Agent are read from the environment:
    STS2_DATA_DIR   output root (default: ./data)  -> writes under <root>/files-spire-codex
    STS2_USER_AGENT HTTP User-Agent header         (default: a Chrome UA string)

Dependencies (PyPI): httpx, typer, tenacity. With uv you can run it inline:
    uv run --with httpx --with typer --with tenacity python download_spire_codex.py

Usage:
    python download_spire_codex.py [--concurrency 32] [--list-concurrency 32]
    python download_spire_codex.py --since-newest   # incremental top-up
"""

from __future__ import annotations

import asyncio
import json
import os
import re
import sys
from pathlib import Path

import httpx
import typer
from tenacity import (
    retry,
    retry_if_exception,
    stop_after_attempt,
    wait_exponential,
)

DEFAULT_UA = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)


def data_root() -> Path:
    """Output root for downloaded runs: ``<STS2_DATA_DIR>/files-spire-codex``."""
    data_dir = Path(os.environ.get("STS2_DATA_DIR", "data")).resolve()
    return data_dir / "files-spire-codex"


def user_agent() -> str:
    return os.environ.get("STS2_USER_AGENT", DEFAULT_UA)


BASE = "https://spire-codex.com/api"
# The listing endpoint reads `limit` (NOT `per_page`, which it silently ignores)
# and caps it server-side at 50 rows per page.
SERVER_MAX_PAGE_SIZE = 50
_SLUG = re.compile(r"[^a-z0-9]+")
_HASH_RE = re.compile(r"/shared/([0-9a-f]{16})")


def slug(s: str) -> str:
    return _SLUG.sub("-", s.lower()).strip("-") or "unknown"


def target_path(root: Path, row: dict) -> Path:
    char = slug(row.get("character") or "unknown")
    submitted = row.get("submitted_at") or ""
    bucket = submitted[:7] if len(submitted) >= 7 else "unknown"
    return root / char / bucket / f"{row['run_hash']}.run"


def _is_transient(exc: BaseException) -> bool:
    """Retry only on server/network hiccups — never on 4xx (e.g. 404s)."""
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code >= 500 or exc.response.status_code == 429
    return isinstance(exc, httpx.HTTPError)


@retry(
    reraise=True,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    retry=retry_if_exception(_is_transient),
)
async def get_json(client: httpx.AsyncClient, url: str) -> dict:
    r = await client.get(url)
    r.raise_for_status()
    return r.json()


async def list_page(client: httpx.AsyncClient, page: int) -> dict:
    return await get_json(client, f"{BASE}/runs/list?page={page}&limit={SERVER_MAX_PAGE_SIZE}")


def load_known_bad(root: Path) -> set[str]:
    """run_hashes previously logged as unretrievable (404) — skip them outright."""
    log = root / "_errors.log"
    if not log.exists():
        return set()
    bad: set[str] = set()
    for line in log.read_text().splitlines():
        try:
            obj = json.loads(line)
        except ValueError:
            continue
        h = obj.get("run_hash")
        if not h or h == "?":
            m = _HASH_RE.search(obj.get("error", ""))
            h = m.group(1) if m else None
        if h:
            bad.add(h)
    return bad


async def probe_first_page(client: httpx.AsyncClient, min_total: int) -> dict | None:
    """Fetch page 1, retrying through the API's stale-total cache flapping."""
    for attempt in range(20):
        probe = await list_page(client, 1)
        if probe.get("total", 0) >= min_total:
            return probe
        typer.echo(
            f"  stale total {probe.get('total')}/{probe.get('total_pages')} "
            f"on probe {attempt}; waiting for cache to refresh..."
        )
        await asyncio.sleep(min(30, 5 + attempt * 2))
    return None


async def fetch_page_rows(client: httpx.AsyncClient, page: int) -> list[dict]:
    """One list page, retrying transient empties / errors before giving up."""
    for attempt in range(4):
        try:
            payload = await list_page(client, page)
        except httpx.HTTPError as exc:
            typer.echo(f"  page {page} list failed (attempt {attempt}): {exc!r}", err=True)
            await asyncio.sleep(1 + attempt)
            continue
        rows = payload.get("runs") or []
        if rows:
            return rows
        await asyncio.sleep(1 + attempt)  # empty: API may be flaking
    return []


async def list_all_rows(
    client: httpx.AsyncClient, first: dict, total_pages: int, concurrency: int
) -> list[dict]:
    """Fetch every list page concurrently; page 1 is reused from the probe."""
    sem = asyncio.Semaphore(concurrency)
    done = 0

    async def one(page: int) -> list[dict]:
        nonlocal done
        async with sem:
            if page == 1:
                rows = list(first.get("runs") or [])
            else:
                rows = await fetch_page_rows(client, page)
        done += 1
        if done % 200 == 0:
            typer.echo(f"  listed {done}/{total_pages} pages")
        return rows

    pages = await asyncio.gather(*[one(p) for p in range(1, total_pages + 1)])
    return [row for page_rows in pages for row in page_rows]


def _take_new(rows: list[dict], skip: set[str], seen: set[str], todo: list[dict]) -> int:
    """Append rows not already seen/skipped to ``todo``; return how many were new."""
    new = 0
    for row in rows:
        h = row.get("run_hash")
        if not h or h in seen:
            continue
        seen.add(h)
        if h not in skip:
            todo.append(row)
            new += 1
    return new


def select_new_rows(rows: list[dict], skip: set[str]) -> tuple[list[dict], int]:
    """Dedupe by hash and drop runs already on disk or known-unretrievable."""
    seen: set[str] = set()
    todo: list[dict] = []
    _take_new(rows, skip, seen, todo)
    return todo, len(seen)


async def list_until_known(
    client: httpx.AsyncClient, first: dict, skip: set[str], total_pages: int, concurrency: int
) -> tuple[list[dict], int]:
    """Incremental top-up: the list is newest-first, so new runs cluster at the top.

    Walk pages in concurrent batches and stop after the first batch that yields no
    new runs — everything older is already on disk. Avoids enumerating all ~2,300
    pages when only a handful of runs are new.
    """
    sem = asyncio.Semaphore(concurrency)

    async def page_rows(page: int) -> list[dict]:
        async with sem:
            if page == 1:
                return list(first.get("runs") or [])
            return await fetch_page_rows(client, page)

    seen: set[str] = set()
    todo: list[dict] = []
    page = 1
    while page <= total_pages:
        batch = list(range(page, min(page + concurrency, total_pages + 1)))
        results = await asyncio.gather(*[page_rows(p) for p in batch])
        new_in_batch = sum(_take_new(rows, skip, seen, todo) for rows in results)
        typer.echo(
            f"  scanned pages {batch[0]}-{batch[-1]}: {new_in_batch} new (total {len(todo)})"
        )
        if new_in_batch == 0:
            break
        page = batch[-1] + 1
    return todo, len(seen)


async def download_one(
    client: httpx.AsyncClient, row: dict, root: Path, sem: asyncio.Semaphore
) -> None:
    async with sem:
        data = await get_json(client, f"{BASE}/runs/shared/{row['run_hash']}")
    path = target_path(root, row)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".run.tmp")
    tmp.write_text(json.dumps(data, indent=2))
    tmp.replace(path)


async def download_all(
    client: httpx.AsyncClient, todo: list[dict], root: Path, concurrency: int
) -> tuple[int, int, list[dict]]:
    sem = asyncio.Semaphore(concurrency)
    downloaded = errors = 0
    failures: list[dict] = []

    async def one(row: dict) -> None:
        nonlocal downloaded, errors
        try:
            await download_one(client, row, root, sem)
            downloaded += 1
        except httpx.HTTPError as exc:
            errors += 1
            failures.append({"run_hash": row.get("run_hash", "?"), "error": repr(exc)})

    tasks = [asyncio.create_task(one(r)) for r in todo]
    for done, coro in enumerate(asyncio.as_completed(tasks), 1):
        await coro
        if done % 200 == 0:
            typer.echo(f"  fetched {done}/{len(todo)} (new={downloaded} err={errors})")
    return downloaded, errors, failures


def write_failures(root: Path, failures: list[dict]) -> None:
    log = root / "_errors.log"
    with log.open("a") as fh:
        for f in failures:
            fh.write(json.dumps(f) + "\n")
    typer.echo(f"  wrote {len(failures)} error entries to {log}")


async def main_async(
    concurrency: int, list_concurrency: int, max_runs: int | None, since_newest: bool
) -> None:
    root = data_root()
    root.mkdir(parents=True, exist_ok=True)

    max_conn = max(concurrency, list_concurrency) * 2
    timeout = httpx.Timeout(90.0, connect=10.0)
    limits = httpx.Limits(max_connections=max_conn, max_keepalive_connections=max_conn)
    headers = {"User-Agent": user_agent(), "Accept": "application/json"}

    async with httpx.AsyncClient(timeout=timeout, limits=limits, headers=headers) as client:
        on_disk = {p.stem for p in root.rglob("*.run")}
        known_bad = load_known_bad(root)
        # The listing endpoint intermittently reports a stale total (e.g. 359
        # instead of ~114k) from a cache replica; insist on a fresh-looking total.
        first = await probe_first_page(client, max(9000, len(on_disk)))
        if first is None:
            typer.echo("gave up waiting for fresh listing", err=True)
            return
        total, total_pages = first["total"], first["total_pages"]
        typer.echo(
            f"spire-codex: {total} runs across {total_pages} pages; "
            f"{len(on_disk)} on disk, {len(known_bad)} known-unretrievable"
        )

        skip = on_disk | known_bad
        if since_newest:
            todo, unique = await list_until_known(
                client, first, skip, total_pages, list_concurrency
            )
        else:
            rows = await list_all_rows(client, first, total_pages, list_concurrency)
            todo, unique = select_new_rows(rows, skip)
        if max_runs is not None:
            todo = todo[:max_runs]
        typer.echo(f"listed {unique} unique runs; {len(todo)} new to download")

        downloaded, errors, failures = await download_all(client, todo, root, concurrency)

    typer.echo(f"done: downloaded={downloaded} skipped={len(on_disk)} errors={errors} root={root}")
    if failures:
        write_failures(root, failures)


def main(
    concurrency: int = typer.Option(32, help="Concurrent shared-run detail fetches."),
    list_concurrency: int = typer.Option(32, help="Concurrent list-page fetches."),
    max_runs: int | None = typer.Option(None, help="Stop after N new downloads."),
    since_newest: bool = typer.Option(
        False,
        "--since-newest",
        help="Incremental top-up: walk newest-first and stop once caught up "
        "(cheap when only a few runs are new; skips full enumeration).",
    ),
) -> None:
    try:
        asyncio.run(main_async(concurrency, list_concurrency, max_runs, since_newest))
    except KeyboardInterrupt:
        typer.echo("interrupted", err=True)
        sys.exit(130)


if __name__ == "__main__":
    typer.run(main)