feat(#126): Sub-PR A — NSPDClient.get_features_in_bbox_grid + engineering classifier #247

Merged
lekss361 merged 1 commit from fix/126-nspd-grid-walk-scraper into main 2026-05-17 05:54:31 +00:00
3 changed files with 548 additions and 11 deletions

View file

@ -31,6 +31,7 @@ WMS endpoints (per #94 issue body, TIER 1-6 каталог слоёв):
from __future__ import annotations
import asyncio
import datetime as _dt
import json
import logging
@ -384,24 +385,22 @@ class NSPDClient:
width: int = 4096,
height: int = 4096,
) -> list[NSPDFeature]:
"""Bulk fetch features в bbox через GetFeatureInfo с большим bbox.
"""WMS GetFeatureInfo на одном центральном пикселе bbox.
Workaround: WFS GetCapabilities 404 на nspd.gov.ru, нет WFS
GetFeature endpoint. Решение: использовать GetFeatureInfo с large
bbox и точкой в центре (I=W/2, J=H/2) возвращает все features
пересекающиеся с bbox.
DEPRECATED: возвращает 0-3 features под одним пикселем (I=W/2, J=H/2).
НЕ является bulk fetch несмотря на исходный docstring WMS GetFeatureInfo
по стандарту OGC возвращает объекты строго под одной pixel-точкой, а не
во всём bbox. Для получения всех объектов в bbox используй
`get_features_in_bbox_grid`.
See: fixes/Bug_NSPD_WMS_NotBulk_2026_May14.md
Args:
bbox_3857: (xmin, ymin, xmax, ymax) в EPSG:3857 метрах.
width/height: размер виртуального tile. Большой большой bbox.
width/height: размер виртуального tile.
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 = {
@ -426,6 +425,149 @@ class NSPDClient:
feats = (data or {}).get("features") or []
return [NSPDFeature.from_raw(f) for f in feats]
# ── 3b. get_features_in_bbox_grid ───────────────────────────────────────
def get_features_in_bbox_grid(
self,
layer_id: int,
bbox: tuple[float, float, float, float],
*,
grid_n: int = 7,
step_m: float = 50.0,
tile_size: int = 512,
) -> list[NSPDFeature]:
"""Bulk-аппроксимация bbox через grid-walk WMS GetFeatureInfo.
Разбивает bbox на grid_n × grid_n равных ячеек. В каждой ячейке
вызывает WMS GetFeatureInfo в центральном пикселе. Дедуплицирует
результаты по feature_id / cad_num / reg_numb_border возвращает
список уникальных NSPDFeature.
Делегирует HTTP через NSPDBulkClient.wms_feature_info (async httpx
с semaphore и retry), запуская asyncio event loop синхронно через
asyncio.run(). Предназначен для вызова из синхронного кода (Celery
task, FastAPI sync handler).
Concurrency: NSPDBulkClient._SEMAPHORE(3) ограничивает параллельные
запросы. При grid_n=7 (49 ячеек) все 49 ячеек запускаются одним
gather; семафор пропускает не более 3 одновременно. Thread-safety:
каждый вызов get_features_in_bbox_grid создаёт новый event loop
через asyncio.run() безопасно из разных Celery workers (process-
уровень изоляции).
Args:
layer_id: NSPD layer ID (например 36328 сооружения, 37578 ЗОУИТ).
bbox: (xmin, ymin, xmax, ymax) в EPSG:3857 (метры).
grid_n: размер сетки по каждой оси. 7 49 запросов (~coarse),
15 225 запросов (~fine). По умолчанию 7 для первичного scan.
step_m: минимальный шаг ячейки в метрах. Если bbox меньше
grid_n*step_m grid_n уменьшается автоматически чтобы
ячейки не становились меньше step_m.
tile_size: размер виртуального WMS тайла (пиксели).
Returns:
Дедуплицированный list[NSPDFeature]. Может быть пуст если в bbox
нет объектов данного layer'а.
Note:
Не делает live HTTP вызовы если вызван с mock NSPDBulkClient.
Rate-limit управляется семафором NSPDBulkClient._SEMAPHORE(3) +
asyncio.sleep(0.05) jitter не через self.rate_ms.
"""
# Импортируем здесь чтобы избежать circular import:
# nspd_client ← nspd_bulk_client (оба top-level scrapers, не cross-domain)
from app.scrapers.nspd_bulk_client import NSPDBulkClient
xmin, ymin, xmax, ymax = bbox
width_m = xmax - xmin
height_m = ymax - ymin
# Авто-коррекция grid_n если bbox мал для шага step_m
effective_n = min(
grid_n,
max(1, int(width_m / step_m)),
max(1, int(height_m / step_m)),
)
if effective_n < grid_n:
logger.info(
"get_features_in_bbox_grid layer=%d: bbox %.0fx%.0fm < grid_n=%d×step_m=%.0f"
" — уменьшаем grid до %d×%d",
layer_id,
width_m,
height_m,
grid_n,
step_m,
effective_n,
effective_n,
)
x_step = width_m / effective_n
y_step = height_m / effective_n
# Генерируем список (sub_bbox, click_xy) ячеек
cells: list[tuple[tuple[float, float, float, float], tuple[int, int]]] = []
click_px = tile_size // 2
for i in range(effective_n):
for j in range(effective_n):
cell_xmin = xmin + i * x_step
cell_ymin = ymin + j * y_step
cell_xmax = cell_xmin + x_step
cell_ymax = cell_ymin + y_step
cells.append(((cell_xmin, cell_ymin, cell_xmax, cell_ymax), (click_px, click_px)))
async def _run_grid() -> list[NSPDFeature]:
async with NSPDBulkClient() as client:
tasks = [
client.wms_feature_info(layer_id, sub_bbox, click_xy, tile_size, tile_size)
for sub_bbox, click_xy in cells
]
results = await asyncio.gather(*tasks, return_exceptions=True)
features: list[NSPDFeature] = []
for r in results:
if isinstance(r, Exception):
logger.warning("get_features_in_bbox_grid layer=%d cell error: %s", layer_id, r)
continue
for bulk_feat in r:
raw = {
"id": bulk_feat.id,
"geometry": bulk_feat.geometry,
"properties": bulk_feat.properties,
}
features.append(NSPDFeature.from_raw(raw))
return features
raw_features = asyncio.run(_run_grid())
# Дедупликация — приоритет ключей: feature_id > cad_num > reg_numb_border
seen: set[str] = set()
deduped: list[NSPDFeature] = []
for feat in raw_features:
props = feat.properties
dedup_key = (
feat.feature_id
or props.get("cad_num")
or props.get("cad_number")
or props.get("reg_numb_border")
or props.get("label")
)
if dedup_key is not None:
if dedup_key in seen:
continue
seen.add(dedup_key)
deduped.append(feat)
logger.info(
"get_features_in_bbox_grid layer=%d grid=%dx%d cells=%d raw=%d deduped=%d",
layer_id,
effective_n,
effective_n,
len(cells),
len(raw_features),
len(deduped),
)
return deduped
# ── 4. list_layers ──────────────────────────────────────────────────────
def list_layers(self, theme_id: int = THEME_PKK) -> list[NSPDLayer]:

View file

@ -32,6 +32,7 @@ from __future__ import annotations
import datetime
import json
import logging
import re
from typing import Any
from sqlalchemy import text
@ -39,6 +40,84 @@ from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
# ── Engineering classifier ─────────────────────────────────────────────────────
# Паттерны для классификации инженерных сооружений (layer 36328) по текстовым
# свойствам. Источник: bug-post-mortem fixes/Bug_NSPD_WMS_NotBulk_2026_May14.md
# + live test данные 7×7 grid на 1км² центра ЕКБ.
# Порядок проверки важен: более специфичные паттерны идут первыми.
_ENGINEERING_PATTERNS: list[tuple[re.Pattern[str], str]] = [
# gas — газопроводы, ГРП, ГК (газовые колодцы)
(re.compile(r"газопровод|газоснабж|ГРП\b|ГК\b", re.IGNORECASE), "gas"),
# sewage — канализация, сток, ливневые сети
(re.compile(r"канализ|сточ|ливнев|самотёч|самотеч", re.IGNORECASE), "sewage"),
# heat — тепловые сети, котельные, ТЭЦ
(
re.compile(r"теплов(ая|ой|ые)|теплосеть|теплоснабж|котельн|ТЭЦ\b", re.IGNORECASE),
"heat",
),
# electric — электросети, ВЛ, КЛ, ЛЭП, подстанции, трансформаторы, ТП
(
re.compile(
r"электроэнерг|ВЛ[\s\-]|ВЛ-?\d|КЛ[\s\-]|КЛ-?\d|ЛЭП\b"
r"|подстанц|трансформ|ТП[\s\-]?\d",
re.IGNORECASE,
),
"electric",
),
# water — водопровод, водоснабжение, хозбытовые водосети
(re.compile(r"водопровод|водоснабж|хозбытов|водовод", re.IGNORECASE), "water"),
]
# Поля из NSPD properties в которых ищем паттерны (по приоритету)
_ENGINEERING_TEXT_FIELDS = ("params_name", "name", "params_purpose", "purpose", "label")
def classify_engineering_kind(properties: dict[str, Any]) -> str:
"""Классифицировать инженерное сооружение (layer 36328) по его properties.
Проверяет поля `params_name`, `name`, `params_purpose`, `purpose`, `label`
против regex-паттернов. Возвращает первое совпадение.
Args:
properties: dict свойств NSPDFeature.properties из WMS GetFeatureInfo
или search/geoportal response.
Returns:
Одно из: ``"water"`` | ``"sewage"`` | ``"gas"`` | ``"heat"``
| ``"electric"`` | ``"other"``.
Examples:
>>> classify_engineering_kind({"params_name": "Газопровод высокого давления"})
'gas'
>>> classify_engineering_kind({"name": "КЛ 10 кВ ТП 64102"})
'electric'
>>> classify_engineering_kind({"params_purpose": "Водопровод хозбытовой"})
'water'
>>> classify_engineering_kind({"params_name": "Тепловая сеть"})
'heat'
>>> classify_engineering_kind({"name": "Канализация"})
'sewage'
"""
# Собираем текст для проверки из всех релевантных полей
text_parts: list[str] = []
for field in _ENGINEERING_TEXT_FIELDS:
val = properties.get(field)
if val and isinstance(val, str):
text_parts.append(val)
combined = " ".join(text_parts)
if not combined:
return "other"
for pattern, kind in _ENGINEERING_PATTERNS:
if pattern.search(combined):
return kind
return "other"
# ── Type coercions ─────────────────────────────────────────────────────────────

View file

@ -0,0 +1,316 @@
"""Unit-тесты для NSPDClient.get_features_in_bbox_grid + classify_engineering_kind.
Sub-PR A foundation (#126): grid-walk — НЕ live HTTP (all mocked).
Запуск:
cd backend && uv run pytest tests/scrapers/test_nspd_grid_walk.py -v
Live integration (помечены @pytest.mark.integration пропускаются в CI):
cd backend && uv run pytest tests/scrapers/test_nspd_grid_walk.py -m integration -s
"""
from __future__ import annotations
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from app.services.scrapers.nspd_client import NSPDClient, NSPDFeature
from app.services.scrapers.nspd_denorm import classify_engineering_kind
# ── Helpers ────────────────────────────────────────────────────────────────────
def _make_bulk_feature(feature_id: str | int | None, props: dict[str, Any]) -> MagicMock:
"""Создать mock NSPDBulkFeature с нужными полями."""
feat = MagicMock()
feat.id = feature_id
feat.geometry = None
feat.properties = props
return feat
# ── Grid-walk count tests ──────────────────────────────────────────────────────
class TestGetFeaturesInBboxGrid:
"""Тесты grid-walk метода (без live NSPD)."""
# bbox 700×700м в EPSG:3857 (типичный квартал ЕКБ)
BBOX: tuple[float, float, float, float] = (
6_600_000.0,
7_700_000.0,
6_600_700.0,
7_700_700.0,
)
def _patch_bulk_client(self, return_features: list[list[Any]]) -> tuple[Any, AsyncMock]:
"""Патч NSPDBulkClient.wms_feature_info.
return_features: список ответов по одному на каждый вызов wms_feature_info
(в порядке вызовов). Если список короче числа calls последний элемент
повторяется.
"""
call_idx: list[int] = [0]
async def _side_effect(*args: Any, **kwargs: Any) -> list[Any]:
idx = min(call_idx[0], len(return_features) - 1)
call_idx[0] += 1
return return_features[idx]
mock_wms = AsyncMock(side_effect=_side_effect)
return mock_wms, mock_wms
def test_grid_n7_calls_49_times(self) -> None:
"""grid_n=7 → ровно 49 вызовов wms_feature_info."""
call_count: list[int] = [0]
async def _wms(*args: Any, **kwargs: Any) -> list[Any]:
call_count[0] += 1
return []
mock_client_instance = AsyncMock()
mock_client_instance.wms_feature_info = AsyncMock(side_effect=_wms)
mock_client_instance.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_client_instance.__aexit__ = AsyncMock(return_value=None)
with patch(
"app.scrapers.nspd_bulk_client.NSPDBulkClient",
return_value=mock_client_instance,
):
client = NSPDClient()
result = client.get_features_in_bbox_grid(36328, self.BBOX, grid_n=7, step_m=1.0)
assert call_count[0] == 49, f"Ожидали 49 вызовов, получили {call_count[0]}"
assert result == []
def test_grid_n3_calls_9_times(self) -> None:
"""grid_n=3 → ровно 9 вызовов."""
call_count: list[int] = [0]
async def _wms(*args: Any, **kwargs: Any) -> list[Any]:
call_count[0] += 1
return []
mock_client_instance = AsyncMock()
mock_client_instance.wms_feature_info = AsyncMock(side_effect=_wms)
mock_client_instance.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_client_instance.__aexit__ = AsyncMock(return_value=None)
with patch(
"app.scrapers.nspd_bulk_client.NSPDBulkClient",
return_value=mock_client_instance,
):
client = NSPDClient()
result = client.get_features_in_bbox_grid(36328, self.BBOX, grid_n=3, step_m=1.0)
assert call_count[0] == 9
assert result == []
def test_dedup_same_feature_id(self) -> None:
"""Одинаковые feature_id из соседних ячеек → одна запись."""
feat_a = _make_bulk_feature("feat-001", {"cad_num": "66:41:001:1"})
feat_b = _make_bulk_feature("feat-001", {"cad_num": "66:41:001:1"}) # дубликат
feat_c = _make_bulk_feature("feat-002", {"cad_num": "66:41:001:2"})
async def _wms(*args: Any, **kwargs: Any) -> list[Any]:
return [feat_a, feat_b, feat_c]
mock_client_instance = AsyncMock()
mock_client_instance.wms_feature_info = AsyncMock(side_effect=_wms)
mock_client_instance.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_client_instance.__aexit__ = AsyncMock(return_value=None)
with patch(
"app.scrapers.nspd_bulk_client.NSPDBulkClient",
return_value=mock_client_instance,
):
client = NSPDClient()
result = client.get_features_in_bbox_grid(36328, self.BBOX, grid_n=2, step_m=1.0)
# 4 cells × 3 features = 12 raw, но feature_id уникальных 2
feature_ids = [f.feature_id for f in result]
assert "feat-001" in feature_ids
assert "feat-002" in feature_ids
# нет дубликатов
assert len(feature_ids) == len(set(feature_ids))
def test_dedup_by_cad_num_when_no_feature_id(self) -> None:
"""Если feature_id=None — дедупликация по cad_num."""
feat_a = _make_bulk_feature(None, {"cad_num": "66:41:0000001:100"})
feat_b = _make_bulk_feature(None, {"cad_num": "66:41:0000001:100"}) # дубликат
call_n: list[int] = [0]
async def _wms(*args: Any, **kwargs: Any) -> list[Any]:
call_n[0] += 1
# Первая ячейка возвращает feat_a, вторая — feat_b
return [feat_a] if call_n[0] == 1 else [feat_b]
mock_client_instance = AsyncMock()
mock_client_instance.wms_feature_info = AsyncMock(side_effect=_wms)
mock_client_instance.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_client_instance.__aexit__ = AsyncMock(return_value=None)
with patch(
"app.scrapers.nspd_bulk_client.NSPDBulkClient",
return_value=mock_client_instance,
):
client = NSPDClient()
result = client.get_features_in_bbox_grid(36328, self.BBOX, grid_n=2, step_m=1.0)
cad_nums = [f.properties.get("cad_num") for f in result]
assert cad_nums.count("66:41:0000001:100") == 1, "Дубликат не дедуплицирован"
def test_error_in_one_cell_does_not_abort(self) -> None:
"""Ошибка в одной ячейке не останавливает весь grid-walk."""
good_feat = _make_bulk_feature("feat-ok", {"cad_num": "66:41:001:1"})
call_n: list[int] = [0]
async def _wms(*args: Any, **kwargs: Any) -> list[Any]:
call_n[0] += 1
if call_n[0] == 1:
raise RuntimeError("Simulated cell error")
return [good_feat]
mock_client_instance = AsyncMock()
mock_client_instance.wms_feature_info = AsyncMock(side_effect=_wms)
mock_client_instance.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_client_instance.__aexit__ = AsyncMock(return_value=None)
with patch(
"app.scrapers.nspd_bulk_client.NSPDBulkClient",
return_value=mock_client_instance,
):
client = NSPDClient()
# Не должно бросать исключение
result = client.get_features_in_bbox_grid(36328, self.BBOX, grid_n=2, step_m=1.0)
# 4 cells: 1 error + 3 good_feat → 1 unique feature
assert any(f.feature_id == "feat-ok" for f in result)
def test_returns_nspd_feature_instances(self) -> None:
"""Метод возвращает list[NSPDFeature] а не NSPDBulkFeature."""
bulk_feat = _make_bulk_feature("feat-xyz", {"cad_num": "66:41:001:1"})
async def _wms(*args: Any, **kwargs: Any) -> list[Any]:
return [bulk_feat]
mock_client_instance = AsyncMock()
mock_client_instance.wms_feature_info = AsyncMock(side_effect=_wms)
mock_client_instance.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_client_instance.__aexit__ = AsyncMock(return_value=None)
with patch(
"app.scrapers.nspd_bulk_client.NSPDBulkClient",
return_value=mock_client_instance,
):
client = NSPDClient()
result = client.get_features_in_bbox_grid(36328, self.BBOX, grid_n=1, step_m=1.0)
assert all(isinstance(f, NSPDFeature) for f in result)
def test_auto_reduce_grid_for_small_bbox(self) -> None:
"""Маленький bbox + большой grid_n + большой step_m → grid_n уменьшается."""
# bbox 60×60м, step_m=50, grid_n=7 → effective_n=min(7, int(60/50), int(60/50)) = 1
small_bbox: tuple[float, float, float, float] = (
6_600_000.0,
7_700_000.0,
6_600_060.0,
7_700_060.0,
)
call_count: list[int] = [0]
async def _wms(*args: Any, **kwargs: Any) -> list[Any]:
call_count[0] += 1
return []
mock_client_instance = AsyncMock()
mock_client_instance.wms_feature_info = AsyncMock(side_effect=_wms)
mock_client_instance.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_client_instance.__aexit__ = AsyncMock(return_value=None)
with patch(
"app.scrapers.nspd_bulk_client.NSPDBulkClient",
return_value=mock_client_instance,
):
client = NSPDClient()
client.get_features_in_bbox_grid(36328, small_bbox, grid_n=7, step_m=50.0)
# effective_n = 1 → 1 ячейка
assert call_count[0] == 1
# ── Classifier tests ───────────────────────────────────────────────────────────
class TestClassifyEngineeringKind:
"""Smoke-тесты классификатора инженерных сооружений."""
@pytest.mark.parametrize(
"props, expected",
[
# gas
({"params_name": "Газопровод высокого давления"}, "gas"),
({"name": "Газопровод ПЭ SDR11 160мм"}, "gas"),
({"params_purpose": "Газоснабжение"}, "gas"),
({"name": "ГРП №12"}, "gas"),
# electric
({"name": "КЛ 10 кВ ТП 64102"}, "electric"),
({"params_name": "ВЛ-10кВ Ф-14"}, "electric"),
({"name": "ТП 1234"}, "electric"),
({"params_purpose": "Электроэнергетика и связь"}, "electric"),
({"name": "Подстанция 110/10 кВ"}, "electric"),
# water
({"params_purpose": "Водопровод хозбытовой"}, "water"),
({"name": "Водовод Ду300"}, "water"),
({"params_name": "Сеть водоснабжения"}, "water"),
# heat
({"params_name": "Тепловая сеть"}, "heat"),
({"name": "Теплосеть квартал 24"}, "heat"),
({"params_purpose": "Теплоснабжение жилых домов"}, "heat"),
({"name": "ТЭЦ-4 отпайка"}, "heat"),
# sewage
({"name": "Канализация"}, "sewage"),
({"params_name": "Сеть канализации Ду200"}, "sewage"),
({"name": "Ливневая канализация"}, "sewage"),
# other
({"name": "Объект не классифицирован"}, "other"),
({}, "other"),
({"params_purpose": None}, "other"),
],
)
def test_classify(self, props: dict[str, Any], expected: str) -> None:
assert (
classify_engineering_kind(props) == expected
), f"props={props!r} → expected {expected!r}"
def test_field_priority_params_name_over_purpose(self) -> None:
"""params_name проверяется раньше purpose."""
props = {
"params_name": "Газопровод", # → gas
"params_purpose": "Водоснабжение", # → water (если бы проверялось первым)
}
# gas-паттерн найдётся в combined строке первым по порядку _ENGINEERING_PATTERNS
assert classify_engineering_kind(props) == "gas"
def test_case_insensitive(self) -> None:
"""Паттерны case-insensitive."""
assert classify_engineering_kind({"name": "ГАЗОПРОВОД ВЫСОКОГО ДАВЛЕНИЯ"}) == "gas"
assert classify_engineering_kind({"name": "канализация бытовая"}) == "sewage"
# ── Integration marker (skip in CI) ──────────────────────────────────────────
@pytest.mark.integration
def test_live_nspd_grid_walk_skipped() -> None:
"""Placeholder для ручного запуска live NSPD grid-walk.
Запуск:
cd backend && uv run pytest tests/scrapers/test_nspd_grid_walk.py -m integration -s
При запуске делает реальные HTTP-запросы к nspd.gov.ru.
"""
pytest.skip("Live NSPD integration test — запускать вручную с -m integration")