b7889d113b
Add async NSPD client (nspd_bulk_client.py) with 3 new methods: - search_by_quarter: REST /api/geoportal/v2/search/geoportal — snapshot 20 per cat + meta totals - wms_feature_info: WMS GetFeatureInfo для grid-walk discovery - list_objects_in_building: tab-group-data (Q3 deferred — код ready) Rate limit: asyncio.Semaphore(3) + 0.05s jitter. 429 → exponential backoff 2→4→8s, max 3 retries. 403 (WAF) → raise immediately, no retry. Pydantic schemas (nspd_bulk.py): NSPDBulkFeature, QuarterSnapshot, ObjectsListing with extra='allow' to preserve unknown fields → raw_props JSONB downstream. 15 unit tests + 4 e2e (real NSPD, skipped in CI by default). Co-authored-by: lekss361 <claudestars@proton.me>
407 lines
16 KiB
Python
407 lines
16 KiB
Python
"""NSPD bulk cadastre client — async httpx-based, rate-limited.
|
||
|
||
PR 2/5 of GH issue #168: Bulk Cadastre Ingest (ЕКБ 2408 кварталов).
|
||
|
||
Три новых метода:
|
||
- search_by_quarter(quarter) → QuarterSnapshot (phase 1 snapshot)
|
||
- wms_feature_info(layer_id, bbox, click_xy) → list[NSPDBulkFeature] (grid-walk)
|
||
- list_objects_in_building(objdoc_id) → ObjectsListing (Q3 deferred)
|
||
|
||
Ключевые отличия от nspd_client.py (sync urllib):
|
||
- async httpx.AsyncClient (connection pooling, timeout objects)
|
||
- asyncio.Semaphore(3) — не более 3 одновременных запросов
|
||
- asyncio.sleep(0.05) jitter между запросами
|
||
- 429 → exponential backoff (2 → 4 → 8 с, max 3 retries)
|
||
- 403 WAF → raise немедленно (не retry, нужен manual investigate)
|
||
|
||
Geometry: NSPD возвращает geom в EPSG:3857. ST_Transform(geom, 4326)
|
||
делается на стороне Celery-таска (PR 3) при upsert в DB.
|
||
|
||
Rate limit: 3 req/s sustained через Semaphore(3) + 0.05s jitter.
|
||
При 429 — exponential backoff: 2s → 4s → 8s, max 3 retries.
|
||
При 403 (WAF) — raise NspdBulkWafError, не retry.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import datetime as _dt
|
||
import logging
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
from app.schemas.nspd_bulk import NSPDBulkFeature, ObjectsListing, QuarterSnapshot
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Константы ──────────────────────────────────────────────────────────────────
|
||
|
||
NSPD_BASE = "https://nspd.gov.ru/api"
|
||
NSPD_SEARCH_URL = f"{NSPD_BASE}/geoportal/v2/search/geoportal"
|
||
NSPD_WMS_BASE = f"{NSPD_BASE}/aeggis/v4"
|
||
NSPD_TAB_GROUP_URL = f"{NSPD_BASE}/geoportal/v1/tab-group-data"
|
||
|
||
# Лимит одновременных запросов (глобальный, shared между всеми instance'ами)
|
||
_SEMAPHORE = asyncio.Semaphore(3)
|
||
|
||
# Таймаут подключения и чтения
|
||
DEFAULT_TIMEOUT = httpx.Timeout(30.0, connect=10.0)
|
||
|
||
# Headers из HAR audit 2026-05-15
|
||
DEFAULT_HEADERS: dict[str, str] = {
|
||
"User-Agent": (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36"
|
||
),
|
||
"Referer": "https://nspd.gov.ru/map",
|
||
"Accept": "*/*",
|
||
"Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
|
||
"x-timezone": "Europe/Moscow",
|
||
# Origin только для POST endpoints — не включаем для GET (лишний header)
|
||
}
|
||
|
||
# Retry settings
|
||
_MAX_RETRIES = 3
|
||
_BACKOFF_BASE_S = 2.0 # 2 → 4 → 8 секунд
|
||
|
||
|
||
# ── Exceptions ─────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class NspdBulkError(RuntimeError):
|
||
"""Неуспешный ответ от NSPD bulk endpoint."""
|
||
|
||
|
||
class NspdBulkWafError(NspdBulkError):
|
||
"""HTTP 403 WAF блокировка — не retry, нужно manual investigate."""
|
||
|
||
|
||
class NspdBulkRateLimitError(NspdBulkError):
|
||
"""HTTP 429 Rate limit — исчерпаны retries после backoff."""
|
||
|
||
|
||
# ── NSPDBulkClient ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
class NSPDBulkClient:
|
||
"""Async NSPD client для bulk cadastre ingest.
|
||
|
||
Использовать как async context manager:
|
||
|
||
async with NSPDBulkClient() as client:
|
||
snapshot = await client.search_by_quarter("66:41:0303161")
|
||
|
||
Или с explicit lifecycle (для Celery tasks):
|
||
|
||
client = NSPDBulkClient()
|
||
await client.__aenter__()
|
||
try:
|
||
...
|
||
finally:
|
||
await client.__aexit__(None, None, None)
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
timeout: httpx.Timeout = DEFAULT_TIMEOUT,
|
||
headers: dict[str, str] | None = None,
|
||
max_retries: int = _MAX_RETRIES,
|
||
) -> None:
|
||
self._timeout = timeout
|
||
self._headers = headers or DEFAULT_HEADERS
|
||
self._max_retries = max_retries
|
||
self._client: httpx.AsyncClient | None = None
|
||
|
||
async def __aenter__(self) -> NSPDBulkClient:
|
||
self._client = httpx.AsyncClient(
|
||
headers=self._headers,
|
||
timeout=self._timeout,
|
||
follow_redirects=True,
|
||
)
|
||
return self
|
||
|
||
async def __aexit__(self, *_: Any) -> None:
|
||
if self._client is not None:
|
||
await self._client.aclose()
|
||
self._client = None
|
||
|
||
# ── Internal HTTP helper ───────────────────────────────────────────────────
|
||
|
||
async def _get_json(self, url: str, params: dict[str, Any] | None = None) -> Any:
|
||
"""GET → JSON с rate-limit semaphore, jitter, retry при 429.
|
||
|
||
Raises:
|
||
NspdBulkWafError при 403.
|
||
NspdBulkRateLimitError при 429 после исчерпания retries.
|
||
NspdBulkError при прочих 4xx/5xx.
|
||
"""
|
||
if self._client is None:
|
||
raise RuntimeError("NSPDBulkClient не инициализирован — используй async with")
|
||
|
||
attempt = 0
|
||
while True:
|
||
async with _SEMAPHORE:
|
||
# Небольшой jitter чтобы N одновременных запросов не начинались в 0мс
|
||
await asyncio.sleep(0.05)
|
||
try:
|
||
resp = await self._client.get(url, params=params)
|
||
except httpx.TransportError as e:
|
||
raise NspdBulkError(f"Network error: {e}") from e
|
||
|
||
# Обработка статусов — вне семафора чтобы не держать слот при sleep
|
||
if resp.status_code == 403:
|
||
body_preview = resp.text[:200]
|
||
logger.warning("NSPD WAF 403 blocked: url=%s preview=%s", url, body_preview)
|
||
raise NspdBulkWafError(f"HTTP 403 WAF block (manual investigate required): {url}")
|
||
|
||
if resp.status_code == 429:
|
||
attempt += 1
|
||
if attempt > self._max_retries:
|
||
raise NspdBulkRateLimitError(
|
||
f"HTTP 429 rate-limit: exhausted {self._max_retries} retries for {url}"
|
||
)
|
||
backoff = _BACKOFF_BASE_S * (2 ** (attempt - 1))
|
||
logger.warning(
|
||
"NSPD 429 rate-limit: retry %d/%d after %.0fs backoff, url=%s",
|
||
attempt,
|
||
self._max_retries,
|
||
backoff,
|
||
url,
|
||
)
|
||
await asyncio.sleep(backoff)
|
||
continue
|
||
|
||
if resp.status_code >= 400:
|
||
body_preview = resp.text[:300]
|
||
raise NspdBulkError(f"HTTP {resp.status_code}: {url} — {body_preview}")
|
||
|
||
return resp.json()
|
||
|
||
# ── 1. search_by_quarter ──────────────────────────────────────────────────
|
||
|
||
async def search_by_quarter(self, quarter: str) -> QuarterSnapshot:
|
||
"""Поиск объектов в квартале через NSPD search endpoint.
|
||
|
||
GET /api/geoportal/v2/search/geoportal?thematicSearchId=1&query={quarter}
|
||
|
||
Возвращает первые 20 объектов на категорию + meta с totalCount.
|
||
Категории с totalCount > 20 попадают в overflow_categories
|
||
и требуют grid-walk (wms_feature_info) для полноты.
|
||
|
||
Args:
|
||
quarter: кадастровый номер квартала в формате `66:41:NNNNNN`.
|
||
|
||
Returns:
|
||
QuarterSnapshot с features + meta_counts.
|
||
При 404 или пустом ответе — snapshot с features=[].
|
||
|
||
Raises:
|
||
NspdBulkWafError при 403.
|
||
NspdBulkError при сетевых / прочих ошибках.
|
||
"""
|
||
fetched_at = _dt.datetime.now(_dt.UTC).isoformat()
|
||
|
||
try:
|
||
data = await self._get_json(
|
||
NSPD_SEARCH_URL,
|
||
params={"thematicSearchId": 1, "query": quarter},
|
||
)
|
||
except NspdBulkError as e:
|
||
# 404 или пустой ответ → возвращаем пустой snapshot
|
||
err_str = str(e)
|
||
if "HTTP 404" in err_str or "HTTP 400" in err_str:
|
||
logger.info(
|
||
"search_by_quarter: quarter %s not found (404/400), returning empty",
|
||
quarter,
|
||
)
|
||
return QuarterSnapshot(quarter=quarter, fetched_at=fetched_at)
|
||
raise
|
||
|
||
# Парсим response
|
||
# Shape: {"data": {"type": "FeatureCollection", "features": [...], "meta": [...]}, ...}
|
||
# или: {"data": {"features": [...], "meta": [...]}}
|
||
payload = data if isinstance(data, dict) else {}
|
||
data_block = payload.get("data") or {}
|
||
if isinstance(data_block, str):
|
||
# Иногда NSPD оборачивает data в строку (Bug_Nspd_Geo_Str_Object_No_Get_Fixed)
|
||
logger.warning(
|
||
"search_by_quarter: data block is string for quarter %s, returning empty",
|
||
quarter,
|
||
)
|
||
return QuarterSnapshot(quarter=quarter, fetched_at=fetched_at)
|
||
|
||
raw_features: list[dict[str, Any]] = data_block.get("features") or []
|
||
raw_meta: list[dict[str, Any]] = data_block.get("meta") or []
|
||
|
||
features = [NSPDBulkFeature.model_validate(f) for f in raw_features]
|
||
|
||
# meta: [{categoryId: N, totalCount: M}, ...]
|
||
meta_counts: dict[int, int] = {}
|
||
for m in raw_meta:
|
||
cat_id = m.get("categoryId")
|
||
total = m.get("totalCount")
|
||
if cat_id is not None and total is not None:
|
||
meta_counts[int(cat_id)] = int(total)
|
||
|
||
logger.info(
|
||
"search_by_quarter: quarter=%s features=%d meta_cats=%d overflow=%d",
|
||
quarter,
|
||
len(features),
|
||
len(meta_counts),
|
||
len([c for c, cnt in meta_counts.items() if cnt > 20]),
|
||
)
|
||
|
||
return QuarterSnapshot(
|
||
quarter=quarter,
|
||
fetched_at=fetched_at,
|
||
features=features,
|
||
meta_counts=meta_counts,
|
||
)
|
||
|
||
# ── 2. wms_feature_info ───────────────────────────────────────────────────
|
||
|
||
async def wms_feature_info(
|
||
self,
|
||
layer_id: int,
|
||
bbox: tuple[float, float, float, float],
|
||
click_xy: tuple[int, int],
|
||
width: int = 512,
|
||
height: int = 512,
|
||
) -> list[NSPDBulkFeature]:
|
||
"""WMS GetFeatureInfo — объекты layer'а в bbox на pixel-клике.
|
||
|
||
GET /api/aeggis/v4/{layer_id}/wms?REQUEST=GetFeatureInfo&...
|
||
|
||
Используется для grid-walk фазы: каждая ячейка bbox квартала
|
||
кликается в центре (click_xy = (width//2, height//2)).
|
||
|
||
Args:
|
||
layer_id: NSPD layer ID (например 36368 для ЗУ ЕГРН).
|
||
bbox: (x_min, y_min, x_max, y_max) в EPSG:3857 (метры).
|
||
click_xy: (I, J) пиксель-координаты клика в [0..width-1, 0..height-1].
|
||
width: ширина виртуального tile.
|
||
height: высота виртуального tile.
|
||
|
||
Returns:
|
||
Список NSPDBulkFeature (может быть пуст).
|
||
|
||
Raises:
|
||
NspdBulkWafError при 403.
|
||
NspdBulkError при прочих ошибках.
|
||
"""
|
||
xmin, ymin, xmax, ymax = bbox
|
||
click_i, click_j = click_xy
|
||
|
||
url = f"{NSPD_WMS_BASE}/{layer_id}/wms"
|
||
params: dict[str, Any] = {
|
||
"REQUEST": "GetFeatureInfo",
|
||
"SERVICE": "WMS",
|
||
"VERSION": "1.3.0",
|
||
"INFO_FORMAT": "application/json",
|
||
"LAYERS": str(layer_id),
|
||
"QUERY_LAYERS": str(layer_id),
|
||
"CRS": "EPSG:3857",
|
||
"BBOX": f"{xmin},{ymin},{xmax},{ymax}",
|
||
"WIDTH": str(width),
|
||
"HEIGHT": str(height),
|
||
"I": str(click_i),
|
||
"J": str(click_j),
|
||
"STYLES": "",
|
||
"TRANSPARENT": "true",
|
||
"FORMAT": "image/png",
|
||
"FEATURE_COUNT": "10",
|
||
}
|
||
|
||
data = await self._get_json(url, params=params)
|
||
raw_features: list[dict[str, Any]] = (data or {}).get("features") or []
|
||
return [NSPDBulkFeature.model_validate(f) for f in raw_features]
|
||
|
||
# ── 3. list_objects_in_building ───────────────────────────────────────────
|
||
# Q3 deferred — метод реализован, но не вызывается в bulk_harvest_quarter MVP.
|
||
# Готов для per-building помещения/парковка фазы.
|
||
|
||
async def list_objects_in_building(
|
||
self,
|
||
objdoc_id: int,
|
||
registers_id: int = 36441,
|
||
) -> ObjectsListing:
|
||
"""Список помещений и машино-мест в здании.
|
||
|
||
GET /api/geoportal/v1/tab-group-data?tabClass=objectsList
|
||
&objdocId={objdoc_id}®istersId={registers_id}
|
||
|
||
Response shape:
|
||
{title: "Список объектов", object: [
|
||
{title: "Помещения (количество)", value: ["264"]},
|
||
{title: "Машино-места (количество)", value: ["4"]},
|
||
{title: "Помещения (список)", value: [...cad_nums]},
|
||
{title: "Машино-места (список)", value: [...cad_nums]},
|
||
]}
|
||
|
||
NOTE: Q3 deferred — метод готов, но НЕ вызывается в PR 2/3 MVP.
|
||
Использовать при реализации bulk_harvest_flats (отдельный task).
|
||
|
||
Args:
|
||
objdoc_id: NSPD internal object document ID (из NSPDOptions.objdoc_id).
|
||
registers_id: NSPD registers ID (36441 — здания ЕГРН по умолчанию).
|
||
|
||
Returns:
|
||
ObjectsListing с counts и cad_nums списками.
|
||
|
||
Raises:
|
||
NspdBulkWafError при 403.
|
||
NspdBulkError при прочих ошибках.
|
||
"""
|
||
data = await self._get_json(
|
||
NSPD_TAB_GROUP_URL,
|
||
params={
|
||
"tabClass": "objectsList",
|
||
"objdocId": str(objdoc_id),
|
||
"registersId": str(registers_id),
|
||
},
|
||
)
|
||
|
||
flats_count = 0
|
||
parking_count = 0
|
||
flats_cad_nums: list[str] = []
|
||
parking_cad_nums: list[str] = []
|
||
|
||
obj_list: list[dict[str, Any]] = (data or {}).get("object") or []
|
||
for entry in obj_list:
|
||
title: str = (entry.get("title") or "").lower()
|
||
values: list[Any] = entry.get("value") or []
|
||
|
||
if "помещения" in title and "количество" in title:
|
||
try:
|
||
flats_count = int(values[0]) if values else 0
|
||
except (ValueError, TypeError):
|
||
flats_count = 0
|
||
|
||
elif "машино-места" in title and "количество" in title:
|
||
try:
|
||
parking_count = int(values[0]) if values else 0
|
||
except (ValueError, TypeError):
|
||
parking_count = 0
|
||
|
||
elif "помещения" in title and "список" in title:
|
||
flats_cad_nums = [str(v) for v in values if v]
|
||
|
||
elif "машино-места" in title and "список" in title:
|
||
parking_cad_nums = [str(v) for v in values if v]
|
||
|
||
return ObjectsListing(
|
||
objdoc_id=objdoc_id,
|
||
flats_count=flats_count,
|
||
parking_count=parking_count,
|
||
flats_cad_nums=flats_cad_nums,
|
||
parking_cad_nums=parking_cad_nums,
|
||
)
|
||
|
||
|
||
__all__ = [
|
||
"NSPDBulkClient",
|
||
"NspdBulkError",
|
||
"NspdBulkRateLimitError",
|
||
"NspdBulkWafError",
|
||
]
|