rf4-spotter/scripts/fetch-waterbodies.py

#!/usr/bin/env python3
"""Manually fetch RF4DB waterbody snapshots through the guarded research CLI."""

from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from dataclasses import asdict
from pathlib import Path
from urllib.parse import urlsplit

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from rf4_research.community_cli import _validate_url_before_io
from rf4_research.community_sources import parse_rf4db_waterbodies, parse_rf4db_waterbody_detail


DEFAULT_STATE_FILE = ROOT / ".cache" / "community-fetch-state.json"
DEFAULT_OUTPUT_DIR = ROOT / ".cache" / "waterbodies"
CATALOG_SOURCE = "rf4db-waterbodies"
DETAIL_SOURCE = "rf4db-waterbody"


def _timestamp() -> str:
    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")


def _safe_detail_name(url: str) -> str:
    name = Path(urlsplit(url).path.rstrip("/")).name or "detail"
    return re.sub(r"[^A-Za-z0-9_.-]+", "-", name)


def _default_output(mode: str, url: str | None) -> Path:
    stem = "rf4db-catalog" if mode == "catalog" else f"rf4db-{_safe_detail_name(url or '')}"
    return DEFAULT_OUTPUT_DIR / f"{stem}-{_timestamp()}.json"


def _write_json_atomically(path: Path, payload: object) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    handle = tempfile.NamedTemporaryFile(
        mode="w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.",
        suffix=".tmp", delete=False,
    )
    temporary = Path(handle.name)
    try:
        with handle:
            json.dump(payload, handle, ensure_ascii=False, indent=2)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    except Exception:
        temporary.unlink(missing_ok=True)
        raise


def fetch_snapshot(
    mode: str, *, url: str | None, html: Path | None = None, output: Path,
    state_file: Path, limit: int,
) -> int:
    if html is not None:
        source_url = url or "https://rf4db.com/ru/maps"
        try:
            _validate_url_before_io(source_url)
            document = html.read_text(encoding="utf-8")
            parsed = (
                parse_rf4db_waterbodies(document)
                if mode == "catalog"
                else parse_rf4db_waterbody_detail(document, source_url=source_url)
            )
        except Exception as exc:
            print(f"local HTML parse failed: {exc}", file=sys.stderr)
            return 1
        payload = [asdict(item) for item in parsed] if mode == "catalog" else asdict(parsed)
        _write_json_atomically(output, payload)
        count = len(payload) if isinstance(payload, list) else 1
        print(f"parsed and saved {count} waterbody snapshot(s) to {output}")
        return 0

    source = CATALOG_SOURCE if mode == "catalog" else DETAIL_SOURCE
    command = [
        sys.executable, "-m", "rf4_research.community_cli", source,
        "--state-file", str(state_file), "--limit", str(limit),
    ]
    if url:
        command.extend(["--url", url])
    result = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, check=False)
    if result.returncode != 0:
        if result.stderr:
            print(result.stderr, file=sys.stderr, end="")
        return result.returncode
    try:
        payload = json.loads(result.stdout)
    except json.JSONDecodeError as exc:
        print(f"fetch succeeded but returned invalid JSON: {exc}", file=sys.stderr)
        return 1
    if mode == "catalog" and not isinstance(payload, list):
        print("fetch succeeded but catalog payload is not a JSON array", file=sys.stderr)
        return 1
    if mode == "detail" and not isinstance(payload, dict):
        print("fetch succeeded but detail payload is not a JSON object", file=sys.stderr)
        return 1
    _write_json_atomically(output, payload)
    count = len(payload) if isinstance(payload, list) else 1
    print(f"saved {count} waterbody snapshot(s) to {output}")
    print(f"next import: python -m app.cli import-waterbody-{'catalog' if mode == 'catalog' else 'detail'} --input {output}")
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Manually fetch RF4DB waterbody data with the shared cooldown guard"
    )
    subparsers = parser.add_subparsers(dest="mode", required=True)

    catalog = subparsers.add_parser("catalog", help="fetch the public waterbody catalog")
    catalog.add_argument("--output", type=Path, help="snapshot path; default: .cache/waterbodies/")
    catalog.add_argument("--html", type=Path, help="parse an HTML file saved from an authorized browser session; no network")
    catalog.add_argument("--limit", type=int, default=500, choices=range(1, 501), metavar="1..500")

    detail = subparsers.add_parser("detail", help="fetch one waterbody detail page")
    detail.add_argument("--url", required=True, help="authorized RF4DB detail URL")
    detail.add_argument("--output", type=Path, help="snapshot path; default: .cache/waterbodies/")
    detail.add_argument("--html", type=Path, help="parse an HTML file saved from an authorized browser session; no network")
    detail.add_argument("--limit", type=int, default=100, choices=range(1, 501), metavar="1..500")

    for command in (catalog, detail):
        command.add_argument(
            "--state-file", type=Path, default=DEFAULT_STATE_FILE,
            help=f"cooldown state file (default: {DEFAULT_STATE_FILE})",
        )
    args = parser.parse_args(argv)
    output = args.output or _default_output(args.mode, getattr(args, "url", None))
    return fetch_snapshot(
        args.mode,
        url=getattr(args, "url", None),
        html=args.html,
        output=output,
        state_file=args.state_file,
        limit=args.limit,
    )


if __name__ == "__main__":
    raise SystemExit(main())