gendesign/backend/app/services/scrapers/nspd_kn.py

"""NSPD scraper — кадастровые кварталы и здания через nspd.gov.ru API.

Endpoints:
    GET /api/geoportal/v2/search/geoportal?thematicSearchId=2&query={cad}
        → polygon кадастрового квартала (1 feature с label = cad-номер)
    GET /api/geoportal/v2/search/geoportal?thematicSearchId=1&query={cad}
        → объекты внутри квартала (здания, земля, линейные).
        Фильтруем categoryName='Здания' и cad_num LIKE '<region>:<district>:%'.

WAF nspd.gov.ru банит burst-запросы и нероссийские IP. Стратегия:
- rate-limit 600мс между запросами (≤100 req/min — наблюдаемый предел)
- exponential backoff на 403 (60s + 30s × attempt)
- запускать с RU IP (production worker должен быть в RU/CIS-zone)

Этот модуль — рефакторинг data/sql/62_scrape_nspd_full.py из stand-alone скрипта
в библиотечную функцию для Celery. Логика идентична, но:
- работает через SQLAlchemy Session (FK к cad_quarters_geom/cad_buildings)
- пишет heartbeat и progress в nspd_scrape_runs
- логирует структурированно в nspd_scrape_log
"""

from __future__ import annotations

import json
import logging
import ssl
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.db import SessionLocal

logger = logging.getLogger(__name__)

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()

DEFAULT_RATE_MS = 600
# Heartbeat каждый cad — даёт UI видимый прогресс и быстрый zombie-detect
# на прод-VM, где worker может тихо стопать без error в логе.
DEFAULT_HEARTBEAT_EVERY = 1  # quarters
DEFAULT_COMMIT_EVERY = 10
# Retries × timeout × WAF backoff формирует worst-case на один cad.
# Старые значения (5×30s + 60+90+120+150+180s = 600+150 = ~12.5 мин) дают
# инвиз-зависание worker'а если NSPD WAF режет nonbrowser fingerprint.
# Новые: до ~3 мин на cad, чтобы при 'тихом' WAF run падал быстрее.
DEFAULT_RETRIES = 3
DEFAULT_TIMEOUT_S = 12

# region_code (rosreestr) → cad-prefix фильтр для cad_buildings.
# 66 = Свердловская обл., 66:41 = ЕКБ. Для других регионов добавлять mapping.
REGION_CAD_PREFIX: dict[int, str] = {
    66: "66:41:",  # Екатеринбург
}


# ── HTTP layer ────────────────────────────────────────────────────────────────


class WafBlockedError(RuntimeError):
    """403 после всех retry — WAF банит, прогон не продолжаем."""


def nspd_fetch(
    thematic_search_id: int,
    query: str,
    *,
    retries: int = DEFAULT_RETRIES,
    timeout: int = DEFAULT_TIMEOUT_S,
    on_403: Any = None,
) -> dict | None:
    """Fetch NSPD с retry/backoff. on_403 callable(attempt) — для трекинга WAF."""
    qs = urllib.parse.urlencode({"thematicSearchId": thematic_search_id, "query": query})
    url = f"{NSPD_BASE}?{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, context=SSL_CTX) as r:
                return json.loads(r.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            last_err = e
            if e.code == 404:
                return None
            if e.code == 403:
                if on_403:
                    on_403(attempt)
                wait = 60 + 30 * attempt
                logger.warning(
                    "WAF 403 для %s (попытка %d/%d), пауза %ds", query, attempt + 1, retries, wait
                )
                time.sleep(wait)
                continue
            logger.warning("HTTP %s для %s (попытка %d)", e.code, query, attempt + 1)
        except (urllib.error.URLError, TimeoutError, OSError) as e:
            last_err = e
            logger.warning("Сетевая ошибка для %s: %s (попытка %d)", query, e, attempt + 1)
        time.sleep(min(2**attempt, 30))
    if isinstance(last_err, urllib.error.HTTPError) and last_err.code == 403:
        raise WafBlockedError(f"WAF banned after {retries} retries on {query}")
    logger.error("FAILED %s после %d попыток: %s", query, retries, last_err)
    return None


# ── Geometry helpers ──────────────────────────────────────────────────────────


def poly_to_wkt(geom: dict | None) -> str | None:
    if not geom:
        return None

    def ring(r: list) -> str:
        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


# ── SQL helpers ───────────────────────────────────────────────────────────────


def _to_int(x: Any) -> int | None:
    if x is None or x == "":
        return None
    try:
        return int(x)
    except (TypeError, ValueError):
        return None


def _to_num(x: Any) -> float | None:
    if x is None or x == "":
        return None
    try:
        return float(x)
    except (TypeError, ValueError):
        return None


def _to_date(x: Any) -> str | None:
    if not x:
        return None
    s = str(x).split(" ")[0]
    return s or None


def get_pending_cads(db: Session, region_code: int) -> list[str]:
    """Cad-кварталы с ДДУ-сделками в регионе минус уже scraped.

    NOTE: regex с двоеточиями ('!~ ":0000000$"') ломают SQLAlchemy text() —
    двоеточие парсится как named-parameter prefix. Используем NOT LIKE через
    bindparam (быстрее regex и безопасно).
    """
    rows = db.execute(
        text(
            """
            SELECT DISTINCT quarter_cad_number
              FROM rosreestr_deals
             WHERE region_code = :rc
               AND doc_type = 'ДДУ'
               AND realestate_type_code = '002001003000'
               AND quarter_cad_number IS NOT NULL
               AND quarter_cad_number <> ''
               AND quarter_cad_number NOT LIKE :bad_prefix
               AND quarter_cad_number NOT LIKE :bad_suffix
            """
        ),
        {"rc": region_code, "bad_prefix": "00:00:%", "bad_suffix": "%:0000000"},
    ).all()
    pending = {r[0] for r in rows}
    done = {r[0] for r in db.execute(text("SELECT cad_number FROM cad_quarters_geom")).all()}
    return sorted(pending - done)


def insert_quarter(db: Session, cad: str, wkt: str, raw_props: dict | None) -> None:
    db.execute(
        text(
            """
            INSERT INTO cad_quarters_geom (cad_number, geom, raw_props, source)
            VALUES (
                :cad,
                ST_Multi(ST_Transform(ST_SetSRID(ST_GeomFromText(:wkt), 3857), 4326))
                    ::geometry(MultiPolygon, 4326),
                CAST(:props AS jsonb),
                'nspd'
            )
            ON CONFLICT (cad_number) DO UPDATE SET
                geom = EXCLUDED.geom,
                raw_props = COALESCE(EXCLUDED.raw_props, cad_quarters_geom.raw_props),
                fetched_at = NOW()
            """
        ),
        {
            "cad": cad,
            "wkt": wkt,
            "props": json.dumps(raw_props, ensure_ascii=False) if raw_props else None,
        },
    )


def insert_buildings(db: Session, qcad: str, features: list[dict], cad_prefix: str) -> int:
    n = 0
    for f in features:
        props_outer = f.get("properties") or {}
        if props_outer.get("categoryName") != "Здания":
            continue
        opts = props_outer.get("options") or {}
        cn = opts.get("cad_num")
        if not cn or not cn.startswith(cad_prefix):
            continue
        wkt = poly_to_wkt(f.get("geometry"))
        if not wkt:
            continue
        db.execute(
            text(
                """
                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,
                    status, ownership_type, cultural_heritage, underground_floors,
                    build_record_area, build_record_type, common_data_status, obj_type,
                    raw_props
                )
                VALUES (
                    :cad_num, :qcad,
                    ST_Transform(ST_SetSRID(ST_GeomFromText(:wkt), 3857), 4326),
                    :purpose, :name, :addr, :area, :floors, :yb, :yc,
                    :cost, :reg_date,
                    :status, :ownership, :cultural, :underground,
                    :build_rec_area, :build_rec_type, :common_status, :obj_type,
                    CAST(:raw_props AS jsonb)
                )
                ON CONFLICT (cad_num) DO UPDATE SET
                    quarter_cad_num = EXCLUDED.quarter_cad_num,
                    geom = EXCLUDED.geom,
                    purpose = EXCLUDED.purpose,
                    building_name = EXCLUDED.building_name,
                    readable_address = EXCLUDED.readable_address,
                    area = EXCLUDED.area,
                    floors = EXCLUDED.floors,
                    year_built = EXCLUDED.year_built,
                    year_commisioning = EXCLUDED.year_commisioning,
                    cost_value = EXCLUDED.cost_value,
                    registration_date = EXCLUDED.registration_date,
                    status = EXCLUDED.status,
                    ownership_type = EXCLUDED.ownership_type,
                    cultural_heritage = EXCLUDED.cultural_heritage,
                    underground_floors = EXCLUDED.underground_floors,
                    build_record_area = EXCLUDED.build_record_area,
                    build_record_type = EXCLUDED.build_record_type,
                    common_data_status = EXCLUDED.common_data_status,
                    obj_type = EXCLUDED.obj_type,
                    raw_props = COALESCE(EXCLUDED.raw_props, cad_buildings.raw_props),
                    fetched_at = NOW()
                """
            ),
            {
                "cad_num": cn,
                "qcad": qcad,
                "wkt": wkt,
                "purpose": opts.get("purpose"),
                "name": opts.get("building_name"),
                "addr": opts.get("readable_address"),
                "area": _to_num(opts.get("area")),
                # floors / underground_floors хранятся TEXT — NSPD отдаёт
                # диапазоны '1-2', '2-3'. _to_int съел бы их в NULL и потерял
                # данные. Аналитика делает CAST через regex '^[0-9]+$'.
                "floors": (
                    str(opts["floors"]).strip() if opts.get("floors") not in (None, "") else None
                ),
                "yb": _to_int(opts.get("year_built")),
                "yc": _to_int(opts.get("year_commisioning")),
                "cost": _to_num(opts.get("cost_value")),
                "reg_date": _to_date(opts.get("registration_date")),
                "status": opts.get("status"),
                "ownership": opts.get("ownership_type"),
                "cultural": opts.get("cultural_heritage"),
                "underground": (
                    str(opts["underground_floors"]).strip()
                    if opts.get("underground_floors") not in (None, "")
                    else None
                ),
                "build_rec_area": _to_num(opts.get("build_record_area")),
                "build_rec_type": opts.get("build_record_type"),
                "common_status": opts.get("common_data_status"),
                "obj_type": opts.get("obj_type"),
                "raw_props": json.dumps(opts, ensure_ascii=False) if opts else None,
            },
        )
        n += 1
    return n


# ── Run lifecycle ─────────────────────────────────────────────────────────────


def _start_run(db: Session, region_code: int, triggered_by: str, pending_count: int) -> int:
    row = db.execute(
        text(
            """
            INSERT INTO nspd_scrape_runs (region_code, triggered_by, pending_count, status)
            VALUES (:rc, :tb, :pc, 'running')
            RETURNING run_id
            """
        ),
        {"rc": region_code, "tb": triggered_by, "pc": pending_count},
    ).scalar_one()
    db.commit()
    return int(row)


def _heartbeat(db: Session, run_id: int, **counts: int) -> None:
    sets = ["heartbeat_at = NOW()"]
    params: dict[str, Any] = {"rid": run_id}
    for k, v in counts.items():
        sets.append(f"{k} = :{k}")
        params[k] = v
    db.execute(
        text(f"UPDATE nspd_scrape_runs SET {', '.join(sets)} WHERE run_id = :rid"),
        params,
    )
    db.commit()


def _finish_run(
    db: Session,
    run_id: int,
    *,
    status: str,
    error: str | None = None,
    **counts: int,
) -> None:
    sets = ["finished_at = NOW()", "status = :status"]
    params: dict[str, Any] = {"rid": run_id, "status": status, "error": error}
    sets.append("error = :error")
    for k, v in counts.items():
        sets.append(f"{k} = :{k}")
        params[k] = v
    db.execute(
        text(f"UPDATE nspd_scrape_runs SET {', '.join(sets)} WHERE run_id = :rid"),
        params,
    )
    db.commit()


def _log(
    db: Session,
    run_id: int | None,
    *,
    level: str,
    stage: str,
    message: str,
    cad: str | None = None,
) -> None:
    try:
        db.execute(
            text(
                """
                INSERT INTO nspd_scrape_log (run_id, level, stage, cad_number, message)
                VALUES (:rid, :lvl, :st, :cad, :msg)
                """
            ),
            {"rid": run_id, "lvl": level, "st": stage, "cad": cad, "msg": message[:1000]},
        )
        db.commit()
    except Exception as e:
        logger.warning("nspd_scrape_log insert failed: %s", e)


# ── Main entrypoint ───────────────────────────────────────────────────────────


def run_region_scrape(
    *,
    region_code: int,
    triggered_by: str = "beat",
    limit: int | None = None,
    rate_ms: int = DEFAULT_RATE_MS,
    commit_every: int = DEFAULT_COMMIT_EVERY,
    heartbeat_every: int = DEFAULT_HEARTBEAT_EVERY,
) -> dict[str, Any]:
    """Полный sweep одного региона. Безопасен для повторных запусков —
    pending = (rosreestr cads) − (cad_quarters_geom cads)."""
    cad_prefix = REGION_CAD_PREFIX.get(region_code)
    if not cad_prefix:
        raise ValueError(f"region_code={region_code} unknown — добавь в REGION_CAD_PREFIX")

    db = SessionLocal()
    run_id: int | None = None
    try:
        pending = get_pending_cads(db, region_code)
        if limit:
            pending = pending[:limit]
        run_id = _start_run(db, region_code, triggered_by, len(pending))
        _log(
            db,
            run_id,
            level="info",
            stage="discover",
            message=f"region={region_code} pending={len(pending)} cads",
        )

        if not pending:
            _finish_run(db, run_id, status="done", quarters_ok=0, quarters_failed=0, buildings_ok=0)
            return {"run_id": run_id, "pending": 0, "ok": 0, "failed": 0, "buildings": 0}

        ok = 0
        failed = 0
        n_buildings = 0
        n_requests = 0
        n_waf = 0
        started = time.time()

        for i, cn in enumerate(pending, 1):
            try:
                # Pre-fetch лог в DB — чтобы видеть «застрял на cad X» если
                # urllib зависнет в read(). Heartbeat не пишется во время
                # nspd_fetch, и без этого лога run выглядит как «discover, тишина».
                logger.info("nspd[%d/%d] %s — fetching polygon", i, len(pending), cn)
                if i == 1 or i % 10 == 0:
                    _log(
                        db,
                        run_id,
                        level="info",
                        stage="quarter_fetch",
                        cad=cn,
                        message=f"fetching polygon ({i}/{len(pending)})",
                    )
                j2 = nspd_fetch(2, cn, on_403=lambda _a: None)
                n_requests += 1
                qf = None
                q_props: dict | None = None
                if j2:
                    for f in j2.get("data", {}).get("features") or []:
                        if (f.get("properties") or {}).get("label") == cn:
                            qf = f
                            q_props = (f.get("properties") or {}).get("options") or {}
                            break
                wkt = poly_to_wkt(qf.get("geometry") if qf else None) if qf else None
                if wkt:
                    insert_quarter(db, cn, wkt, q_props)
                    ok += 1
                else:
                    failed += 1
                    _log(
                        db,
                        run_id,
                        level="warn",
                        stage="quarter_fetch",
                        cad=cn,
                        message="no polygon returned",
                    )

                j1 = nspd_fetch(1, cn, on_403=lambda _a: None)
                n_requests += 1
                if j1:
                    features = j1.get("data", {}).get("features") or []
                    n_buildings += insert_buildings(db, cn, features, cad_prefix)

                if i % commit_every == 0:
                    db.commit()
                if i % heartbeat_every == 0:
                    _heartbeat(
                        db,
                        run_id,
                        quarters_ok=ok,
                        quarters_failed=failed,
                        buildings_ok=n_buildings,
                        requests_count=n_requests,
                        waf_429_count=n_waf,
                    )

            except WafBlockedError as e:
                _log(
                    db,
                    run_id,
                    level="error",
                    stage="quarter_fetch",
                    cad=cn,
                    message=f"WAF blocked: {e}",
                )
                _finish_run(
                    db,
                    run_id,
                    status="failed",
                    error=str(e),
                    quarters_ok=ok,
                    quarters_failed=failed,
                    buildings_ok=n_buildings,
                    requests_count=n_requests,
                    waf_429_count=n_waf + 1,
                )
                raise
            except Exception as e:
                failed += 1
                db.rollback()
                _log(
                    db,
                    run_id,
                    level="error",
                    stage="quarter_fetch",
                    cad=cn,
                    message=f"{type(e).__name__}: {e}",
                )

            time.sleep(rate_ms / 1000.0)

        db.commit()
        elapsed = time.time() - started
        _finish_run(
            db,
            run_id,
            status="done",
            quarters_ok=ok,
            quarters_failed=failed,
            buildings_ok=n_buildings,
            requests_count=n_requests,
            waf_429_count=n_waf,
        )
        _log(
            db,
            run_id,
            level="info",
            stage="done",
            message=(
                f"elapsed={elapsed:.0f}s ok={ok} failed={failed} "
                f"buildings={n_buildings} reqs={n_requests}"
            ),
        )
        return {
            "run_id": run_id,
            "pending": len(pending),
            "ok": ok,
            "failed": failed,
            "buildings": n_buildings,
            "requests": n_requests,
            "elapsed_s": int(elapsed),
            "started_at": datetime.utcfromtimestamp(started).isoformat() + "Z",
        }
    except WafBlockedError:
        raise
    except Exception as e:
        if run_id:
            try:
                _finish_run(db, run_id, status="failed", error=f"{type(e).__name__}: {e}")
            except Exception:
                pass
        raise
    finally:
        db.close()