gendesign/data/sql/58_fetch_cad_quarters_nspd.py

"""Fetch кадастровые квартал-полигоны из NSPD и загрузить в PostGIS.

Usage:
    cd backend && uv run python ../data/sql/58_fetch_cad_quarters_nspd.py [options]

Options:
    --scope ekb|region   ekb (default): только cad-кварталы с ДДУ-сделками
                         в ЕКБ (~674 кварталов). region: вся Свердл (~10 853).
    --batch-size N       commit каждые N кварталов (default 25).
    --rate-limit-ms N    пауза между запросами в мс (default 700).
    --resume             пропустить cad-кварталы уже в cad_quarters_geom.
    --max N              ограничить N запросами (для smoke-теста).
    --dry-run            fetch+parse only, skip DB upsert.

ВАЖНО: NSPD блокирует запросы с не-российских IP (WAF). Скрипт нужно запускать
с машины в РФ. Если получаешь HTTP 403 + 'Client IP: ...' — это блок гео.

Endpoint: GET https://nspd.gov.ru/api/geoportal/v2/search/geoportal
          ?thematicSearchId=1&query={cad_number}
Returns: GeoJSON FeatureCollection. Геометрия MultiPolygon в EPSG:3857 (web
         mercator). Скрипт переводит в EPSG:4326 через ST_Transform на стороне
         БД при INSERT.
"""

from __future__ import annotations

import argparse
import json
import logging
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

# Add backend to path so we can import app.* in standalone CLI mode.
ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(ROOT / "backend"))

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger(__name__)

NSPD_URL = "https://nspd.gov.ru/api/geoportal/v2/search/geoportal"
HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    ),
    "Accept": "application/json",
    "Accept-Language": "ru-RU,ru;q=0.9",
    "Referer": "https://nspd.gov.ru/map",
}


def fetch_cad_quarter(cad_number: str, *, retries: int = 3, timeout: int = 30) -> dict | None:
    """Запрос NSPD API для одного cad-номера.

    Returns: GeoJSON FeatureCollection or None если квартал не найден / ошибка.
    """
    qs = urllib.parse.urlencode(
        {
            "thematicSearchId": 1,  # 1 = кадастровые квартал
            "query": cad_number,
        }
    )
    url = f"{NSPD_URL}?{qs}"
    req = urllib.request.Request(url, headers=HEADERS)
    last_err: Exception | None = None
    for attempt in range(retries):
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            if e.code == 403:
                # WAF: гео-блок не лечится retry. Бросаем выше — оператор решит.
                body = e.read().decode("utf-8", errors="ignore")[:300]
                log.error("HTTP 403 from NSPD (likely geo-WAF). Body: %s", body)
                raise
            if e.code == 404:
                return None  # cad-квартал не найден — норма для глухих номеров.
            last_err = e
            log.warning("HTTP %d for %s (attempt %d/%d)", e.code, cad_number, attempt + 1, retries)
        except (urllib.error.URLError, TimeoutError, OSError) as e:
            last_err = e
            log.warning("Network err for %s: %s (attempt %d/%d)", cad_number, e, attempt + 1, retries)
        time.sleep(2 ** attempt)
    log.error("Failed %s after %d retries: %s", cad_number, retries, last_err)
    return None


def extract_geometry(geojson: dict | None, cad_number: str) -> tuple[dict | None, dict | None]:
    """Из GeoJSON выбрать первый MultiPolygon (или Polygon) под точный
    cad-номер. Returns (geometry_dict, raw_properties)."""
    if not geojson or "features" not in geojson:
        return None, None
    for feat in geojson["features"]:
        props = feat.get("properties") or {}
        # NSPD возвращает кадастровый номер в одном из этих полей.
        cn = (
            props.get("cn")
            or props.get("cadastralNumber")
            or props.get("cad_num")
            or props.get("CAD_NUMBER")
        )
        if cn and str(cn).strip() == cad_number:
            geom = feat.get("geometry")
            if not geom:
                continue
            # Принимаем Polygon → MultiPolygon (для unification).
            if geom.get("type") == "Polygon":
                geom = {"type": "MultiPolygon", "coordinates": [geom["coordinates"]]}
            return geom, props
    # Fallback: если features=1 и cn совпадает по prefix — берём первую.
    if len(geojson["features"]) == 1:
        feat = geojson["features"][0]
        geom = feat.get("geometry")
        if geom:
            if geom.get("type") == "Polygon":
                geom = {"type": "MultiPolygon", "coordinates": [geom["coordinates"]]}
            return geom, feat.get("properties") or {}
    return None, None


def candidates_to_fetch(scope: str, resume: bool) -> list[str]:
    """Возвращает список cad-номеров для скрейпа из rosreestr_deals."""
    from sqlalchemy import text

    from app.core.db import SessionLocal

    db = SessionLocal()
    try:
        if scope == "ekb":
            rows = db.execute(
                text(
                    """
                    SELECT DISTINCT quarter_cad_number
                      FROM rosreestr_deals
                     WHERE region_code = 66
                       AND doc_type = 'ДДУ'
                       AND realestate_type_code = '002001003000'
                       AND (district ILIKE '%Екатеринбург%' OR city ILIKE '%Екатеринбург%')
                       AND quarter_cad_number IS NOT NULL
                       AND quarter_cad_number <> ''
                    """
                )
            ).all()
        else:  # region
            rows = db.execute(
                text(
                    """
                    SELECT DISTINCT quarter_cad_number
                      FROM rosreestr_deals
                     WHERE region_code = 66
                       AND quarter_cad_number IS NOT NULL
                       AND quarter_cad_number <> ''
                    """
                )
            ).all()
        cads = sorted({r[0] for r in rows if r[0]})

        if resume:
            done = {
                r[0]
                for r in db.execute(
                    text("SELECT cad_number FROM cad_quarters_geom")
                ).all()
            }
            cads = [c for c in cads if c not in done]
            log.info("Resume: skipping %d already-loaded quarters", len(done))

        return cads
    finally:
        db.close()


def upsert_quarter(db, cad_number: str, geom_dict: dict, props: dict) -> None:
    """INSERT или UPDATE в cad_quarters_geom. Полигон приходит в EPSG:3857
    (web mercator); конвертируем в 4326 на стороне БД через ST_Transform."""
    from sqlalchemy import text

    db.execute(
        text(
            """
            INSERT INTO cad_quarters_geom (cad_number, geom, raw_props, fetched_at, source)
            VALUES (
                :cn,
                ST_Multi(
                    ST_Transform(
                        ST_SetSRID(ST_GeomFromGeoJSON(:gj), 3857),
                        4326
                    )
                )::geometry(MultiPolygon, 4326),
                CAST(:props AS jsonb),
                NOW(),
                'nspd'
            )
            ON CONFLICT (cad_number) DO UPDATE SET
                geom = EXCLUDED.geom,
                raw_props = EXCLUDED.raw_props,
                fetched_at = NOW()
            """
        ),
        {
            "cn": cad_number,
            "gj": json.dumps(geom_dict),
            "props": json.dumps(props, ensure_ascii=False),
        },
    )


def main() -> int:
    p = argparse.ArgumentParser(description="Fetch+load NSPD cad-quarter polygons")
    p.add_argument("--scope", choices=["ekb", "region"], default="ekb")
    p.add_argument("--batch-size", type=int, default=25)
    p.add_argument("--rate-limit-ms", type=int, default=700)
    p.add_argument("--resume", action="store_true", help="Skip already-loaded cad-numbers")
    p.add_argument("--max", type=int, default=None, help="Limit N requests (smoke test)")
    p.add_argument("--dry-run", action="store_true")
    args = p.parse_args()

    cads = candidates_to_fetch(args.scope, resume=args.resume)
    if args.max:
        cads = cads[: args.max]
    log.info("Plan to fetch %d cad-quarters (scope=%s)", len(cads), args.scope)
    if not cads:
        log.info("Nothing to do.")
        return 0

    if args.dry_run:
        log.info("Dry-run: would fetch %s", cads[:5])
        return 0

    from sqlalchemy import text  # noqa: F401

    from app.core.db import SessionLocal

    db = SessionLocal()
    ok = 0
    skipped = 0
    failed = 0
    try:
        for i, cn in enumerate(cads, 1):
            try:
                gj = fetch_cad_quarter(cn)
                geom, props = extract_geometry(gj, cn)
                if geom is None:
                    log.warning("[%d/%d] %s — no geometry returned", i, len(cads), cn)
                    skipped += 1
                else:
                    upsert_quarter(db, cn, geom, props or {})
                    ok += 1
                    if i % 10 == 0 or i == len(cads):
                        log.info("[%d/%d] %s ✓ (ok=%d skip=%d fail=%d)",
                                 i, len(cads), cn, ok, skipped, failed)
            except urllib.error.HTTPError as e:
                if e.code == 403:
                    db.commit()
                    log.error(
                        "Aborting: NSPD WAF блокирует запросы (HTTP 403). "
                        "Скрипт нужно запускать с российского IP. Saved %d so far.",
                        ok,
                    )
                    return 2
                failed += 1
            except Exception as e:  # noqa: BLE001
                failed += 1
                log.exception("[%d/%d] %s ✗ %s", i, len(cads), cn, e)
            if i % args.batch_size == 0:
                db.commit()
                log.info("  → committed batch (i=%d, ok=%d)", i, ok)
            time.sleep(args.rate_limit_ms / 1000.0)
        db.commit()
        log.info("Done: ok=%d, skipped=%d, failed=%d", ok, skipped, failed)
        return 0
    finally:
        db.close()


if __name__ == "__main__":
    sys.exit(main())