gendesign/data/sql/62_scrape_nspd_full.py

"""Полный NSPD scrape для всех cad-кварталов ЕКБ с ДДУ-сделками.

Запускать с машины пользователя (РФ-IP, иначе WAF).
SSH-tunnel к prod БД на localhost:15432.

Что делает:
1. Берёт список cad-номеров из rosreestr_deals (region=66, ЕКБ, ДДУ, квартиры).
2. Пропускает уже скрейпнутые (есть в cad_quarters_geom).
3. Для каждого cad-квартала:
   a. fetch /api/geoportal/v2/search/geoportal?thematicSearchId=2 → polygon
   b. fetch /api/geoportal/v2/search/geoportal?thematicSearchId=1 → 20 зданий + земля
4. INSERT в cad_quarters_geom + cad_buildings (фильтр cad_num LIKE '66:41:%').
5. После каждого квартала commit + лог. Можно прервать и перезапустить.

Usage:
    cd backend && uv run python ../data/sql/62_scrape_nspd_full.py [--limit N] [--rate-ms 600]
"""

from __future__ import annotations

import argparse
import json
import os
import ssl
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any

import psycopg

PG_DSN = os.environ.get(
    "TUNNEL_DSN",
    "postgresql://gendesign:2J2SBPMKuS998fiwhtQqDhMI@localhost:15432/gendesign",
)

NSPD_BASE = "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",
}
SSL_CTX = ssl._create_unverified_context()


def nspd_fetch(thematic_search_id: int, query: str, *, retries: int = 5, timeout: int = 30) -> dict | None:
    """Fetch NSPD with backoff. WAF banит burst-запросы — на 403 ждём долго."""
    qs = urllib.parse.urlencode({"thematicSearchId": thematic_search_id, "query": query})
    url = f"{NSPD_BASE}?{qs}"
    req = urllib.request.Request(url, headers=HEADERS)
    last_err = None
    for attempt in range(retries):
        try:
            with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as r:
                return json.loads(r.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            if e.code == 404:
                return None
            last_err = e
            if e.code == 403:
                # WAF rate-limit — ждём дольше (60 сек)
                wait = 60 + 30 * attempt
                print(f"  WAF 403 for {query} (attempt {attempt+1}/{retries}) — sleeping {wait}s", flush=True)
                time.sleep(wait)
                continue
            print(f"  HTTP {e.code} for {query} (attempt {attempt+1})", flush=True)
        except (urllib.error.URLError, TimeoutError, OSError) as e:
            last_err = e
            print(f"  Network err for {query}: {e} (attempt {attempt+1})", flush=True)
        time.sleep(min(2 ** attempt, 30))
    print(f"  FAILED {query}: {last_err}", flush=True)
    return None


def poly_to_wkt(geom: dict | None) -> str | None:
    if not geom:
        return None
    def ring(r):
        return "(" + ",".join(f"{p[0]} {p[1]}" for p in r) + ")"
    t = geom.get("type")
    if t == "Polygon":
        return "POLYGON(" + ",".join(ring(r) for r in geom["coordinates"]) + ")"
    if t == "MultiPolygon":
        return "MULTIPOLYGON(" + ",".join("(" + ",".join(ring(r) for r in p) + ")" for p in geom["coordinates"]) + ")"
    if t == "Point":
        return f"POINT({geom['coordinates'][0]} {geom['coordinates'][1]})"
    return None


def get_pending_cads(cur) -> list[str]:
    """Все cad-номера с ДДУ-сделками в ЕКБ — минус уже scraped."""
    cur.execute("""
        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 <> ''
           AND quarter_cad_number <> '66:41:0000000'
    """)
    all_cads = {r[0] for r in cur.fetchall()}
    cur.execute("SELECT cad_number FROM cad_quarters_geom")
    done = {r[0] for r in cur.fetchall()}
    pending = sorted(all_cads - done)
    return pending


def insert_quarter(cur, cad: str, wkt: str) -> None:
    cur.execute(
        """
        INSERT INTO cad_quarters_geom (cad_number, geom, source)
        VALUES (
            %s,
            ST_Multi(ST_Transform(ST_SetSRID(ST_GeomFromText(%s), 3857), 4326))::geometry(MultiPolygon, 4326),
            'nspd'
        )
        ON CONFLICT (cad_number) DO UPDATE SET geom = EXCLUDED.geom, fetched_at = NOW()
        """,
        (cad, wkt),
    )


def insert_buildings(cur, qcad: str, features: list[dict]) -> int:
    """Insert 'Здания' filtered by cad LIKE '66:41:%'."""
    n = 0
    for f in features:
        if f.get("properties", {}).get("categoryName") != "Здания":
            continue
        opts = f["properties"].get("options", {}) or {}
        cn = opts.get("cad_num")
        if not cn or not cn.startswith("66:41:"):
            continue
        wkt = poly_to_wkt(f.get("geometry"))
        if not wkt:
            continue
        cur.execute(
            """
            INSERT INTO cad_buildings (
                cad_num, quarter_cad_num, geom, purpose, building_name,
                readable_address, area, floors, year_built, year_commisioning,
                cost_value, registration_date
            )
            VALUES (
                %s, %s, ST_Transform(ST_SetSRID(ST_GeomFromText(%s), 3857), 4326),
                %s, %s, %s, %s, %s, %s, %s, %s, %s
            )
            ON CONFLICT (cad_num) DO UPDATE SET
                cost_value = EXCLUDED.cost_value,
                area = EXCLUDED.area,
                fetched_at = NOW()
            """,
            (
                cn, qcad, wkt,
                opts.get("purpose"), opts.get("building_name"),
                opts.get("readable_address"), opts.get("area"), opts.get("floors"),
                int(opts["year_built"]) if opts.get("year_built") else None,
                int(opts["year_commisioning"]) if opts.get("year_commisioning") else None,
                opts.get("cost_value"),
                opts.get("registration_date", "").split(" ")[0] if opts.get("registration_date") else None,
            ),
        )
        n += 1
    return n


def main() -> int:
    p = argparse.ArgumentParser()
    p.add_argument("--limit", type=int, default=None, help="Limit N quarters (smoke)")
    p.add_argument("--rate-ms", type=int, default=600, help="Pause between cads (ms)")
    p.add_argument("--commit-every", type=int, default=10, help="Commit every N quarters")
    args = p.parse_args()

    conn = psycopg.connect(PG_DSN, connect_timeout=10)
    cur = conn.cursor()
    pending = get_pending_cads(cur)
    if args.limit:
        pending = pending[: args.limit]
    print(f"Pending: {len(pending)} cad-quarters")
    if not pending:
        print("Nothing to do.")
        return 0

    started = time.time()
    ok = 0
    n_buildings = 0
    failed = 0
    for i, cn in enumerate(pending, 1):
        try:
            j2 = nspd_fetch(2, cn)
            qf = None
            if j2:
                features = j2.get("data", {}).get("features") or []
                for f in features:
                    if f.get("properties", {}).get("label") == cn:
                        qf = f
                        break
            wkt = poly_to_wkt(qf.get("geometry") if qf else None) if qf else None
            if wkt:
                insert_quarter(cur, cn, wkt)

            j1 = nspd_fetch(1, cn)
            if j1:
                features = j1.get("data", {}).get("features") or []
                n_buildings += insert_buildings(cur, cn, features)

            if wkt:
                ok += 1
            else:
                failed += 1

            if i % args.commit_every == 0:
                conn.commit()
                elapsed = time.time() - started
                rate = ok / elapsed * 60 if elapsed > 0 else 0
                eta = (len(pending) - i) / max(rate, 1) if rate > 0 else 0
                print(
                    f"[{i}/{len(pending)}] {cn} "
                    f"ok={ok} fail={failed} buildings={n_buildings} "
                    f"rate={rate:.1f}/min eta={eta:.0f}min",
                    flush=True,
                )
        except Exception as e:  # noqa: BLE001
            failed += 1
            conn.rollback()
            print(f"  ERR {cn}: {e}", flush=True)

        time.sleep(args.rate_ms / 1000.0)

    conn.commit()
    elapsed = time.time() - started
    cur.execute("SELECT COUNT(*) FROM cad_quarters_geom")
    db_q = cur.fetchone()[0]
    cur.execute("SELECT COUNT(*) FROM cad_buildings")
    db_b = cur.fetchone()[0]
    print(f"\nDONE in {elapsed:.0f}s. ok={ok} failed={failed} buildings_inserted={n_buildings}")
    print(f"DB totals: cad_quarters_geom={db_q}, cad_buildings={db_b}")
    conn.close()
    return 0


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