"""Унифицированный NSPD API client — foundation для G1/G2/G3/E1/#93. Issue #94: при исследовании НСПД через debug browser найдено что REST `/api/geoportal/v2/search/geoportal?query={cad}` возвращает GeoJSON + полные properties (ВРИ, категория земель, кадастровая стоимость, адрес) за один запрос. Плюс WMS endpoints для тематических слоёв (ПЗЗ, ЗОУИТ, территориальные зоны, risk zones). Этот клиент покрывает 4 базовых метода API: - search_by_cad(cad) → SearchResult — поиск по cad-номеру - get_feature_info(layer_id, lon, lat) → list[Feature] — feature data на точке - get_features_in_bbox(layer_id, bbox) → list[Feature] — bbox bulk через GetFeatureInfo (WFS GetCapabilities → 404, workaround) - list_layers(theme_id) → list[Layer] — каталог слоёв в теме Reuse: `fetch_geoportal` из `nspd_lite.py` (WAF-compatible urllib + rate limit). Расширения сверх nspd_lite: - Typed wrappers (NSPDFeature, NSPDLayer, NSPDSearchResult) - WMS GetFeatureInfo / GetMap helpers (нет в nspd_lite) - Coordinate transform EPSG:3857 ↔ 4326 (NSPD WMS работает в Mercator) WMS endpoints (per #94 issue body, TIER 1-6 каталог слоёв): - TIER 1 критичные: 36048 ЗУ, 36071 кварталы, 36049 здания, 36328 сооружения, 875838 территориальные зоны (G1), 879243 красные линии - TIER 2 ЗОУИТ (G3): 37577 ОКН, 37578 энергетики/связи, 37580 природные, 37579 охраняемых объектов, 37581 иные - TIER 3 риски: 872202 подтопление, 872205 затопление, 872203 заболачивание, 872210 обвально-осыпные, 872211 абразия, etc. """ from __future__ import annotations import json import logging import math import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any from app.services.scrapers.nspd_lite import ( _SSL_CTX, HEADERS, NspdLiteError, NspdLiteWafError, fetch_geoportal, ) logger = logging.getLogger(__name__) # WMS base — отдельный путь от geoportal/v2/search NSPD_WMS_BASE = "https://nspd.gov.ru/api/aeggis/v4" # Layers catalog endpoint NSPD_LAYERS_TREE = "https://nspd.gov.ru/api/geoportal/v1/layers-theme-tree" # Themes list NSPD_THEMES = "https://nspd.gov.ru/api/geoportal/v1/layers-theme" # Стандартные theme ID THEME_PKK = 1 THEME_ARN = 665 # ── Layer ID catalog (TIER 1-6 per #94) ────────────────────────────────────── # Не enum, а module-level dict — для удобства интроспекции и тестов. # Использовать как: LAYERS["territorial_zones"] = 875838. LAYERS: dict[str, int] = { # TIER 1 — критичные для МКД (Gate-факторы) "parcels": 36048, # Земельные участки ЕГРН "quarters": 36071, # Кадастровые кварталы "buildings": 36049, # Здания ЕГРН "engineering_structures": 36328, # Сооружения ЕГРН "territorial_zones": 875838, # ПЗЗ (G1 #28) "red_lines": 879243, # Красные линии # TIER 2 — ЗОУИТ (G3 #30) "zouit_okn": 37577, # ОКН "zouit_engineering": 37578, # энергетики/связи/транспорта "zouit_natural": 37580, # природные (водоохранные, СЗО) "zouit_protected": 37579, # охраняемых объектов (СЗЗ предприятий) "zouit_other": 37581, # иные (публичный сервитут, ОЭЗ) # TIER 3 — Risk / Negative "risk_flooding_underground": 872202, # подтопление "risk_flooding": 872205, # затопление "risk_swampification": 872203, # заболачивание "risk_landslide": 872210, # обвально-осыпные "risk_abrasion": 872211, # абразия "risk_erosion_water": 872153, # водная эрозия "risk_erosion_linear": 872155, "risk_erosion_wind": 872164, "risk_desertification": 872182, "risk_clutter": 872206, # захламление "risk_burns": 872222, # гари # TIER 4 — Opportunity / nature "protected_areas": 875845, # ООПТ "coast_lines_river": 875832, "coast_lines_sea": 875835, "forestries": 875866, "auction_parcels": 37299, # ЗУ на аукционе — opportunity "scheme_parcels": 37294, # ЗУ по схеме расположения "free_parcels": 37298, # ЗУ свободные от прав "future_parcels": 36473, # ЗУ по проекту межевания "okn_territory": 875840, "cadastral_cost_heatmap": 37236, "cadastral_cost_per_m2_heatmap": 37758, } # Default rate limit (мс между запросами) — баланс между скоростью и WAF DEFAULT_RATE_MS = 600 # SSL context — reuse `_SSL_CTX` из nspd_lite (one source of truth для WAF- # compatible TLS settings; нет дублирования если что-то поменяется). # ── Typed responses ────────────────────────────────────────────────────────── @dataclass(frozen=True, slots=True) class NSPDFeature: """Парсенный GeoJSON Feature из NSPD API. `geometry` — raw GeoJSON dict (Polygon/LineString/Point в EPSG:3857). `properties` — все returned attributes (cad_num, ВРИ, кост и т.д.). """ feature_id: str | None geometry: dict[str, Any] | None properties: dict[str, Any] crs: str = "EPSG:3857" # NSPD default; для search /v2 фактически 3857 @classmethod def from_raw(cls, raw: dict[str, Any]) -> NSPDFeature: """Парсим GeoJSON Feature dict в typed NSPDFeature.""" props = raw.get("properties") or {} return cls( feature_id=str(raw.get("id")) if raw.get("id") is not None else None, geometry=raw.get("geometry"), properties=props, ) @dataclass(frozen=True, slots=True) class NSPDSearchResult: """Aggregated результат поиска по cad — typed wrapper над feature list.""" cad_num: str features: list[NSPDFeature] raw: dict[str, Any] # original response для debugging / future-proofing @property def first(self) -> NSPDFeature | None: """Удобный shortcut: features[0] или None.""" return self.features[0] if self.features else None @property def is_empty(self) -> bool: return not self.features @dataclass(frozen=True, slots=True) class NSPDLayer: """Метаданные layer'а из layers-theme-tree response.""" layer_id: int title: str layer_type: str | None # 'wms' | 'wfs' | etc metadata: dict[str, Any] # ── HTTP helper ────────────────────────────────────────────────────────────── def _http_get_json(url: str, *, timeout: int = 15, rate_ms: int = 0) -> Any: """Внутренний GET-JSON helper с WAF-handling, paralleling nspd_lite. Используется для endpoints которые не покрыты `fetch_geoportal` (WMS, layers-tree). Reuse HEADERS + SSL context из nspd_lite — общая защита от WAF блокировки по TLS fingerprint. """ if rate_ms > 0: time.sleep(rate_ms / 1000.0) req = urllib.request.Request(url, headers=HEADERS) try: with urllib.request.urlopen(req, context=_SSL_CTX, timeout=timeout) as r: body = r.read().decode("utf-8", "ignore") return json.loads(body) except urllib.error.HTTPError as e: body = e.read().decode("utf-8", "ignore")[:300] if e.fp else "" if e.code in (403, 429): raise NspdLiteWafError(f"HTTP {e.code} (WAF/rate-limit): {body}") from e raise NspdLiteError(f"HTTP {e.code}: {body}") from e except urllib.error.URLError as e: raise NspdLiteError(f"Network error: {e}") from e # ── Coordinate transforms ──────────────────────────────────────────────────── def lonlat_to_3857(lon: float, lat: float) -> tuple[float, float]: """WGS84 lon/lat → EPSG:3857 Web Mercator (метры). Стандартная формула; стандартная для всех Web mapping. NSPD WMS принимает bbox в 3857. """ x = lon * 20037508.34 / 180.0 y = math.log(math.tan((90.0 + lat) * math.pi / 360.0)) / (math.pi / 180.0) y = y * 20037508.34 / 180.0 return (x, y) def bbox_around_point_m( lon: float, lat: float, buffer_m: int, ) -> tuple[float, float, float, float]: """Микро-bbox вокруг точки в EPSG:3857 — для WMS GetFeatureInfo. Returns (xmin, ymin, xmax, ymax). buffer_m по m в каждую сторону. """ x, y = lonlat_to_3857(lon, lat) return (x - buffer_m, y - buffer_m, x + buffer_m, y + buffer_m) # ── Public client API ──────────────────────────────────────────────────────── class NSPDClient: """Foundation-level NSPD client. 4 core methods. Stateless — все методы могут вызываться независимо. Используй один instance на process (rate-limit shared между вызовами через time.sleep). """ def __init__(self, rate_ms: int = DEFAULT_RATE_MS, timeout: int = 15) -> None: self.rate_ms = rate_ms self.timeout = timeout # ── 1. search_by_cad ──────────────────────────────────────────────────── def search_by_cad(self, cad_num: str, thematic_id: int = 1) -> NSPDSearchResult: """REST поиск /v2/search/geoportal — 1 запрос = geometry + ВРИ + кост. Args: cad_num: канонический формат — `NN:NN:NNNNNN[:NN[:NN]]`, цифры через `:`. Примеры: - 3-сегментный квартал: `'66:41:0204016'` - 4-сегментный участок: `'66:41:0204016:10'` - 5-сегментное здание: `'66:41:0204016:10:1'` Спецсимволы / пробелы корректно URL-encode'нутся (нет SQL/URL injection), но НСПД молча вернёт empty result. Caller обязан сам валидировать формат если нужна defensive обработка (см. `app.services.site_finder.cadastre_fetch.validate_cad_format`). thematic_id: 1 ЗУ / 2 квартал / 5 здание / 7 территориальная зона. Default 1 — соответствует use-case G2 ВРИ. Returns: NSPDSearchResult с typed features. `result.is_empty` → cad не найден. Raises: NspdLiteWafError при 403/429 — caller должен сделать backoff. NspdLiteError при прочих 4xx/5xx. Закрывает: G2 #29 ВРИ, поддержка on-demand #93/#51. """ raw = fetch_geoportal( cad_num, thematic_id=thematic_id, timeout=self.timeout, rate_ms=self.rate_ms, ) # Response shape: {"data": {"type": "FeatureCollection", "features": [...]}} data = raw.get("data") or {} feats = data.get("features") or [] features = [NSPDFeature.from_raw(f) for f in feats] return NSPDSearchResult(cad_num=cad_num, features=features, raw=raw) # ── 2. get_feature_info ───────────────────────────────────────────────── def get_feature_info( self, layer_id: int, lon: float, lat: float, buffer_m: int = 100, ) -> list[NSPDFeature]: """WMS GetFeatureInfo на точке — какие feature'ы layer'а покрывают (lon, lat). Использует микро-bbox вокруг точки в EPSG:3857 (WMS требует bbox + I/J пиксель). Tile 256x256, I=128, J=128 — центр. Returns features (может быть 0+). Use cases: - ВРИ конкретного участка (layer_id=875838 territorial_zones) - ЗОУИТ на участке (layer_id=37577..37581) - Risk zones overlap (872xxx) """ xmin, ymin, xmax, ymax = bbox_around_point_m(lon, lat, buffer_m) params = { "SERVICE": "WMS", "VERSION": "1.3.0", "REQUEST": "GetFeatureInfo", "FORMAT": "image/png", "TRANSPARENT": "true", "QUERY_LAYERS": str(layer_id), "LAYERS": str(layer_id), "INFO_FORMAT": "application/json", # STYLES обязателен (пустой), иначе INFO_FORMAT игнорируется per NSPD WAF "STYLES": "", "I": "128", "J": "128", "WIDTH": "256", "HEIGHT": "256", "CRS": "EPSG:3857", "BBOX": f"{xmin},{ymin},{xmax},{ymax}", } url = f"{NSPD_WMS_BASE}/{layer_id}/wms?{urllib.parse.urlencode(params)}" data = _http_get_json(url, timeout=self.timeout, rate_ms=self.rate_ms) feats = (data or {}).get("features") or [] return [NSPDFeature.from_raw(f) for f in feats] # ── 3. get_features_in_bbox ───────────────────────────────────────────── def get_features_in_bbox( self, layer_id: int, bbox_3857: tuple[float, float, float, float], *, width: int = 4096, height: int = 4096, ) -> list[NSPDFeature]: """Bulk fetch features в bbox через GetFeatureInfo с большим bbox. Workaround: WFS GetCapabilities → 404 на nspd.gov.ru, нет WFS GetFeature endpoint. Решение: использовать GetFeatureInfo с large bbox и точкой в центре (I=W/2, J=H/2) — возвращает все features пересекающиеся с bbox. Args: bbox_3857: (xmin, ymin, xmax, ymax) в EPSG:3857 метрах. width/height: размер виртуального tile. Большой → большой bbox. Returns: list[NSPDFeature]; пусто если ничего не найдено. Use cases (per #94 acceptance): - sync_territorial_zones_bbox → закрывает G1 #28 ПЗЗ - sync_zouit_*_bbox → G3 #30 - sync_risk_zones_bbox → новый risk overlay """ xmin, ymin, xmax, ymax = bbox_3857 params = { "SERVICE": "WMS", "VERSION": "1.3.0", "REQUEST": "GetFeatureInfo", "FORMAT": "image/png", "TRANSPARENT": "true", "QUERY_LAYERS": str(layer_id), "LAYERS": str(layer_id), "INFO_FORMAT": "application/json", "STYLES": "", "I": str(width // 2), "J": str(height // 2), "WIDTH": str(width), "HEIGHT": str(height), "CRS": "EPSG:3857", "BBOX": f"{xmin},{ymin},{xmax},{ymax}", } url = f"{NSPD_WMS_BASE}/{layer_id}/wms?{urllib.parse.urlencode(params)}" data = _http_get_json(url, timeout=self.timeout, rate_ms=self.rate_ms) feats = (data or {}).get("features") or [] return [NSPDFeature.from_raw(f) for f in feats] # ── 4. list_layers ────────────────────────────────────────────────────── def list_layers(self, theme_id: int = THEME_PKK) -> list[NSPDLayer]: """Полный каталог слоёв в теме (PKK / ARN). Use case: admin UI «выбрать layer для bulk-импорта», discovery новых слоёв при изменении NSPD каталога. """ url = f"{NSPD_LAYERS_TREE}?themeId={theme_id}" data = _http_get_json(url, timeout=self.timeout, rate_ms=self.rate_ms) # Shape defense: NSPD иногда оборачивает в {"data": [...]} — паттерн уже # ловили в Bug_Nspd_Geo_Str_Object_No_Get_Fixed. Без проверки walker # молча вернул бы [], а caller увидел бы пустой catalog. if isinstance(data, dict) and "data" in data and not data.get("children"): data = data.get("data") or [] if not isinstance(data, list | dict): logger.warning( "list_layers: unexpected NSPD response type %s for theme_id=%s", type(data).__name__, theme_id, ) return [] layers: list[NSPDLayer] = [] # Tree response: nested {"children": [...], "id": N, "title": "..."} # Используем walker по subtree чтобы вытащить все leaf layers for entry in _walk_layer_tree(data): layers.append( NSPDLayer( layer_id=int(entry["id"]), title=str(entry.get("title", "")), layer_type=entry.get("layerType") or entry.get("layer_type"), metadata=entry, ) ) return layers def _walk_layer_tree(node: Any) -> list[dict[str, Any]]: """Рекурсивный walker для NSPD layers-theme-tree. Yields leaf nodes.""" out: list[dict[str, Any]] = [] if isinstance(node, dict): # Leaf если есть id + нет children (или children пустой) children = node.get("children") or [] if "id" in node and not children: out.append(node) elif "id" in node and children: # Branch с id — может быть и сам layer, и группа. NSPD сейчас # treats group nodes как "не имеющий wms-endpoint" — но включаем # на случай если они оба leaves (хитрая branch). out.append(node) for ch in children: out.extend(_walk_layer_tree(ch)) elif isinstance(node, list): for item in node: out.extend(_walk_layer_tree(item)) return out __all__ = [ "LAYERS", "NSPD_THEMES", "NSPD_WMS_BASE", "THEME_ARN", "THEME_PKK", "NSPDClient", "NSPDFeature", "NSPDLayer", "NSPDSearchResult", "NspdLiteError", "NspdLiteWafError", "bbox_around_point_m", "lonlat_to_3857", ]