"""ЕКБ муниципальный геопортал (ИСОГД / Geometa) — WFS-клиент (#1067 C7). Геопортал Екатеринбурга (`геопортал.екатеринбург.рф`) — открытый GeoServer на ~1149 слоёв, без авторизации, нормальный TLS. Единый источник «что построят / землепользование» для ЕКБ: ПЗЗ-регламент, функц.зоны генплана, КРТ, ППТ/ПМТ, ЗОУИТ — то, чего НСПД (только геометрия + reg-номер) и городские документы по-отдельности не дают структурно. Primary интерфейс — **стандартный OGC WFS GetFeature** (`/api/gis/ows/3/portal/wfs`): OGC-стандарт стабильнее reverse-engineered `urbanCard`-API и покрывает все слои одним вызовом. ⚠️ WMS GetFeatureInfo на этом GeoServer багует на reproject («too close to a pole») — использовать ТОЛЬКО WFS GetFeature. CRS: WFS отдаёт `geoloc` (геометрия) в EPSG:4326 (lon/lat). В CQL `POINT(lon lat)` порядок X=lon, Y=lat. `geoloc_orig` (МСК-66) НЕ запрашиваем. Источник разведки (live-verified 2026-06-06): vault research/EKB_Geoportal_WFS_Construction_Intel_Jun06. Рич ПЗЗ-карточка (3 списка ВРИ + предельные параметры) — отдельным дополнением через urbanCard (см. research/EKB_Geoportal_PZZ_Content_API_Jun06), здесь НЕ реализуется (follow-up). """ from __future__ import annotations import logging from dataclasses import dataclass from typing import Any import httpx logger = logging.getLogger(__name__) # Punycode-хост геопортала (= геопортал.екатеринбург.рф). # В коде только punycode — кириллица в URL ломает часть HTTP-стеков/логов. EKB_GEOPORTAL_BASE = "https://xn--80afgznagjs.xn--80acgfbsl1azdqr.xn--p1ai" EKB_WFS_URL = f"{EKB_GEOPORTAL_BASE}/api/gis/ows/3/portal/wfs" # Геометрическая колонка GeoServer-слоёв портала (таргет CQL-фильтра). GEOM_COLUMN = "geoloc" # Читаемый ключ → GeoServer typeName. Проверены live WFS GetFeature (vault recon 2026-06-06). # Полный каталог — 1149 typeNames; тут только релевантные ИРД/supply-слои. WFS_LAYERS: dict[str, str] = { # Зонирование / регламент (ИРД п.2 ПЗЗ, capacity) "territorial_zone": "portal:portal_geo_urban_ter_zone_exist", # ПЗЗ терзоны (urban_index) "subzone": "portal:portal_geo_subzone_exist", # ПЗЗ подзоны "functional_zone": "portal:portal_geo_urban_10_functional_zone", # функц.зоны ГП (#1058) "genplan": "portal:portal_geo_dmd_genplan", # генплан (документ-слой) # Future supply / что построят "krt": "portal:portal_geo_oks_krt", # КРТ объекты/границы (ИРД #1060) "ppt": "portal:portal_geo_dmd_prj_plan", # проекты планировки территории "pmt": "portal:portal_geo_dmd_prj_mej", # проекты межевания территории "investment_zone": "portal:portal_geo_urban_10_investment_zone_polygon", # инвест-зоны "draft_parcel": "portal:portal_geo_art_zu_draft", # проектируемые/черновые ЗУ # Ограничения "zouit": "portal:portal_geo_gknspecial_zone", # ЗОУИТ } DEFAULT_TIMEOUT = 20 # ── HTTP helper (module-level — мокается в тестах) ─────────────────────────── def _http_get_json(url: str, params: dict[str, str], *, timeout: int = DEFAULT_TIMEOUT) -> Any: """GET WFS-эндпоинта → JSON. Вынесен на module-level для monkeypatch в тестах. Геопортал ЕКБ — нормальный TLS, без auth: обычный httpx без спец-обвязки (в отличие от НСПД с нац-CA / WAF). Raises httpx.HTTPStatusError на 4xx/5xx (caller решает). """ resp = httpx.get(url, params=params, timeout=timeout) resp.raise_for_status() return resp.json() # ── Typed response ─────────────────────────────────────────────────────────── @dataclass(frozen=True, slots=True) class EKBFeature: """Парсенный WFS GeoJSON Feature геопортала ЕКБ. `geometry` — raw GeoJSON dict в EPSG:4326 (lon/lat). `properties` — все атрибуты слоя (для терзоны: urban_index/full_name/urban_kind_obj; для КРТ: num_oks_krt/status/type/area; для ППТ: doc_full_name/doc_status/dmd_actual_year). """ feature_id: str | None geometry: dict[str, Any] | None properties: dict[str, Any] crs: str = "EPSG:4326" @classmethod def from_raw(cls, raw: dict[str, Any]) -> EKBFeature: return cls( feature_id=str(raw.get("id")) if raw.get("id") is not None else None, geometry=raw.get("geometry"), properties=raw.get("properties") or {}, ) # ── Client ─────────────────────────────────────────────────────────────────── class EKBGeoportalClient: """WFS-клиент геопортала ЕКБ. Stateless; один instance на процесс. Методы возвращают list[EKBFeature]; пусто если по фильтру ничего не найдено. """ def __init__(self, timeout: int = DEFAULT_TIMEOUT) -> None: self.timeout = timeout def _resolve_type_name(self, layer: str) -> str: """Читаемый ключ WFS_LAYERS → typeName; либо сырой typeName (`portal:...`) as-is.""" return WFS_LAYERS.get(layer, layer) def _get_feature(self, type_name: str, cql_filter: str) -> list[EKBFeature]: """WFS 2.0.0 GetFeature с CQL_FILTER → typed features (EPSG:4326).""" params = { "service": "WFS", "version": "2.0.0", "request": "GetFeature", "typeNames": type_name, "outputFormat": "application/json", "srsName": "EPSG:4326", "CQL_FILTER": cql_filter, } data = _http_get_json(EKB_WFS_URL, params, timeout=self.timeout) feats = (data or {}).get("features") or [] return [EKBFeature.from_raw(f) for f in feats] def features_at_point(self, layer: str, lon: float, lat: float) -> list[EKBFeature]: """Фичи слоя, покрывающие точку (lon, lat) — CQL INTERSECTS. Порядок координат в CQL POINT — X=lon, Y=lat (EPSG:4326). Args: layer: ключ WFS_LAYERS ('territorial_zone', 'krt', ...) или сырой typeName. """ type_name = self._resolve_type_name(layer) cql = f"INTERSECTS({GEOM_COLUMN}, POINT({lon} {lat}))" return self._get_feature(type_name, cql) def features_in_bbox( self, layer: str, bbox_4326: tuple[float, float, float, float] ) -> list[EKBFeature]: """Фичи слоя в bbox (minlon, minlat, maxlon, maxlat) EPSG:4326 — CQL BBOX.""" minx, miny, maxx, maxy = bbox_4326 type_name = self._resolve_type_name(layer) cql = f"BBOX({GEOM_COLUMN}, {minx}, {miny}, {maxx}, {maxy}, 'EPSG:4326')" return self._get_feature(type_name, cql) def zone_index_at(self, lon: float, lat: float) -> str | None: """Удобный shortcut: индекс ПЗЗ-терзоны ('Ц-1', 'Ж-2') в точке, либо None. Закрывает crosswalk «координата → индекс зоны ЕКБ» для резолва ИРД-content (#1059): НСПД даёт геометрию зоны без индекса, геопортал — индекс по точке. """ feats = self.features_at_point("territorial_zone", lon, lat) if not feats: return None idx = feats[0].properties.get("urban_index") return str(idx) if idx is not None else None __all__ = [ "EKB_GEOPORTAL_BASE", "EKB_WFS_URL", "WFS_LAYERS", "EKBFeature", "EKBGeoportalClient", ]