Merge pull request 'fix(#231): switch search_by_quarter area layers to grid-walk (Sub-PR B)' (#260) from fix/231-nspd-search-by-quarter-grid into main

backend/app/scrapers/nspd_bulk_client.py

@@ -57,8 +57,12 @@ _CAT_TO_THEMATIC_SEARCH_ID: dict[int, int] = {
     39663: 15,  # ЕНК
 }
 
-# Лимит одновременных запросов (глобальный, shared между всеми instance'ами)
-_SEMAPHORE = asyncio.Semaphore(3)
+# Concurrency cap per HTTP fetch loop. Issue #260 (Sub-PR B re-review):
+# module-level asyncio.Semaphore() bind'ится к event loop первого asyncio.run()
+# вызова, на втором — RuntimeError: bound to a different event loop.
+# search_by_quarter делает N sequential asyncio.run() per layer → semaphore надо
+# создавать per-instance в __aenter__ (под текущим running loop).
+_SEMAPHORE_LIMIT = 3
 
 # Таймаут подключения и чтения
 DEFAULT_TIMEOUT = httpx.Timeout(30.0, connect=10.0)
@@ -131,6 +135,10 @@ class NSPDBulkClient:
         self._headers = headers or DEFAULT_HEADERS
         self._max_retries = max_retries
         self._client: httpx.AsyncClient | None = None
+        # Issue #260: semaphore создаётся в __aenter__ под текущий event loop.
+        # Module-level не работал — sequential asyncio.run() bind'ил его к
+        # закрытому loop'у первого вызова.
+        self._sem: asyncio.Semaphore | None = None
 
     async def __aenter__(self) -> NSPDBulkClient:
         # NSPD prod chain contains a self-signed/internal CA on Beget VPS →
@@ -144,6 +152,8 @@ class NSPDBulkClient:
             follow_redirects=True,
             verify=False,
         )
+        # Создаём semaphore под running loop — safe для sequential asyncio.run().
+        self._sem = asyncio.Semaphore(_SEMAPHORE_LIMIT)
         return self
 
     async def __aexit__(self, *_: Any) -> None:
@@ -161,12 +171,12 @@ class NSPDBulkClient:
             NspdBulkRateLimitError при 429 после исчерпания retries.
             NspdBulkError при прочих 4xx/5xx.
         """
-        if self._client is None:
+        if self._client is None or self._sem is None:
             raise RuntimeError("NSPDBulkClient не инициализирован — используй async with")
 
         attempt = 0
         while True:
-            async with _SEMAPHORE:
+            async with self._sem:
                 # Небольшой jitter чтобы N одновременных запросов не начинались в 0мс
                 await asyncio.sleep(0.05)
                 try:

backend/app/services/scrapers/nspd_client.py

@@ -43,6 +43,7 @@ import urllib.request
 from dataclasses import dataclass
 from typing import Any
 
+from app.services.scrapers.nspd_denorm import classify_engineering_kind
 from app.services.scrapers.nspd_lite import (
     _SSL_CTX,
     HEADERS,
@@ -110,6 +111,37 @@ LAYERS: dict[str, int] = {
 }
 
 
+# Layers where grid-walk (get_features_in_bbox_grid) must be used instead of
+# the legacy single-pixel WMS probe (get_features_in_bbox). These are area/
+# linear layers (territorial zones, red lines, engineering structures, ЗОУИТ,
+# risks) that span large areas and are under-returned by a single GetFeatureInfo
+# call.  Point/polygon EGRN layers (parcels, buildings) stay on legacy for now.
+_GRID_WALK_LAYERS: frozenset[str] = frozenset(
+    {
+        "territorial_zones",
+        "red_lines",
+        "engineering_structures",
+        # ЗОУИТ (TIER 2)
+        "zouit_okn",
+        "zouit_engineering",
+        "zouit_natural",
+        "zouit_protected",
+        "zouit_other",
+        # Risks (TIER 3)
+        "risk_flooding_underground",
+        "risk_flooding",
+        "risk_swampification",
+        "risk_landslide",
+        "risk_abrasion",
+        "risk_erosion_water",
+        "risk_erosion_linear",
+        "risk_erosion_wind",
+        "risk_desertification",
+        "risk_clutter",
+        "risk_burns",
+    }
+)
+
 # Default rate limit (мс между запросами) — баланс между скоростью и WAF
 DEFAULT_RATE_MS = 600
 
@@ -448,7 +480,7 @@ class NSPDClient:
         asyncio.run(). Предназначен для вызова из синхронного кода (Celery
         task, FastAPI sync handler).
 
-        Concurrency: NSPDBulkClient._SEMAPHORE(3) ограничивает параллельные
+        Concurrency: NSPDBulkClient._sem (per-instance, capacity=3) ограничивает параллельные
         запросы. При grid_n=7 (49 ячеек) — все 49 ячеек запускаются одним
         gather; семафор пропускает не более 3 одновременно. Thread-safety:
         каждый вызов get_features_in_bbox_grid создаёт новый event loop
@@ -471,7 +503,7 @@ class NSPDClient:
 
         Note:
             Не делает live HTTP вызовы если вызван с mock NSPDBulkClient.
-            Rate-limit управляется семафором NSPDBulkClient._SEMAPHORE(3) +
+            Rate-limit управляется семафором NSPDBulkClient._sem (per-instance, capacity=3) +
             asyncio.sleep(0.05) jitter — не через self.rate_ms.
         """
         # Импортируем здесь чтобы избежать circular import:
@@ -709,7 +741,17 @@ class NSPDClient:
         layers_fetched: list[str] = ["search"]
 
         def _fetch_layer(name_in_dump: str, layer_key: str) -> list[NSPDFeature]:
-            """Helper: безопасно получить features для одного layer."""
+            """Helper: безопасно получить features для одного layer.
+
+            Dispatch:
+              - area/linear layers (_GRID_WALK_LAYERS) → grid-walk
+                (get_features_in_bbox_grid, grid_n=7, step_m=50)
+              - point/polygon EGRN layers (parcels, buildings, ons, enk)
+                → legacy single-pixel WMS (get_features_in_bbox)
+
+            Engineering structures features дополнительно обогащаются
+            properties["classified_kind"] через classify_engineering_kind.
+            """
             if bbox is None:
                 return []
             layer_id = LAYERS.get(layer_key)
@@ -717,7 +759,29 @@ class NSPDClient:
                 logger.warning("search_by_quarter: unknown layer key %s", layer_key)
                 return []
             layers_fetched.append(name_in_dump)
-            return self.get_features_in_bbox(layer_id, bbox)
+            if layer_key in _GRID_WALK_LAYERS:
+                features = self.get_features_in_bbox_grid(layer_id, bbox, grid_n=7, step_m=50.0)
+                # Обогатить engineering_structures classified_kind
+                if layer_key == "engineering_structures":
+                    for feat in features:
+                        feat.properties["classified_kind"] = classify_engineering_kind(
+                            feat.properties
+                        )
+                logger.info(
+                    "search_by_quarter layer=%s method=grid_walk count=%d quarter=%s",
+                    name_in_dump,
+                    len(features),
+                    quarter_cad,
+                )
+                return features
+            features_legacy = self.get_features_in_bbox(layer_id, bbox)
+            logger.info(
+                "search_by_quarter layer=%s method=legacy count=%d quarter=%s",
+                name_in_dump,
+                len(features_legacy),
+                quarter_cad,
+            )
+            return features_legacy
 
         # 3. Core layers
         parcels = _fetch_layer("parcels", "parcels")

backend/app/workers/tasks/nspd_sync.py

@@ -258,7 +258,9 @@ def _upsert_dump(
     bind=True,
     name="tasks.nspd_sync.harvest_quarter",
     max_retries=3,
-    soft_time_limit=120,  # 2 мин — worst-case 22 HTTP × ~600ms = ~13s + margin
+    soft_time_limit=600,  # 10 мин — grid-walk: 11 layers × 49 cells × ~70ms ≈ 40s + retries
+    time_limit=900,  # 15 мин hard kill — safety net (PR #260 re-review): task может игнорить
+    # SoftTimeLimitExceeded → процесс не освобождается. Hard limit гарантирует worker recovery.
     autoretry_for=(NspdLiteWafError,),
     retry_backoff=True,
     retry_backoff_max=120,

backend/tests/scrapers/test_nspd_bulk_client.py

@@ -316,7 +316,7 @@ async def test_list_objects_in_building_parses(
 
 @pytest.mark.asyncio
 async def test_rate_limit_semaphore_max_3_concurrent() -> None:
-    """Не более 3 одновременных запросов через _SEMAPHORE.
+    """Не более 3 одновременных запросов через per-instance self._sem (PR #260).
 
     Мокируем httpx.AsyncClient.get (нижний уровень) с задержкой, чтобы
     реальный семафор работал. Считаем max in-flight внутри семафора.
@@ -332,7 +332,7 @@ async def test_rate_limit_semaphore_max_3_concurrent() -> None:
 
     async def slow_get(*args: Any, **kwargs: Any) -> httpx.Response:
         nonlocal max_concurrent, current, call_count
-        # Фиксируем вход — уже внутри семафора (httpx.get вызывается после async with _SEMAPHORE)
+        # Фиксируем вход — уже внутри семафора (httpx.get вызывается после async with self._sem)
         async with lock:
             current += 1
             call_count += 1
@@ -353,11 +353,12 @@ async def test_rate_limit_semaphore_max_3_concurrent() -> None:
             tasks = [client.search_by_quarter(f"66:41:{i:07d}") for i in range(6)]
             await asyncio.gather(*tasks)
 
-    # _SEMAPHORE(3) → не более 3 одновременно внутри slow_get
+    # self._sem(3) → не более 3 одновременно внутри slow_get
     assert max_concurrent <= 3, f"Expected ≤3 concurrent, got {max_concurrent}"
     assert call_count == 6  # все 6 вызовов прошли
-    # Убедимся что _SEMAPHORE в модуле имеет нужное значение capacity
-    assert bulk_mod._SEMAPHORE._value >= 0  # семафор сброшен после всех задач
+    # NB: per-instance self._sem cleanup проверяется через max_concurrent <= 3 выше.
+    # Module-level _SEMAPHORE удалён в PR #260 (cross-loop binding fix); smoke на capacity:
+    assert bulk_mod._SEMAPHORE_LIMIT == 3
 
 
 @pytest.mark.asyncio

backend/tests/scrapers/test_nspd_grid_walk.py

@@ -11,7 +11,7 @@ Live integration (помечены @pytest.mark.integration — пропуска
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, ClassVar
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -301,6 +301,139 @@ class TestClassifyEngineeringKind:
         assert classify_engineering_kind({"name": "канализация бытовая"}) == "sewage"
 
 
+# ── Sub-PR B: _fetch_layer dispatch tests ─────────────────────────────────────
+
+
+class TestFetchLayerDispatch:
+    """Проверяем что _fetch_layer внутри search_by_quarter правильно диспатчит
+    area-слои на get_features_in_bbox_grid, а EGRN-слои — на legacy get_features_in_bbox.
+
+    Моки: search_by_cad и get_features_in_bbox_grid / get_features_in_bbox
+    через patch, без live HTTP.
+    """
+
+    # Минимальный bbox квартала (Web Mercator), который возвращает search_by_cad
+    _MOCK_GEOM: ClassVar[dict[str, Any]] = {
+        "type": "Polygon",
+        "coordinates": [
+            [
+                [6_600_000.0, 7_700_000.0],
+                [6_600_700.0, 7_700_000.0],
+                [6_600_700.0, 7_700_700.0],
+                [6_600_000.0, 7_700_700.0],
+                [6_600_000.0, 7_700_000.0],
+            ]
+        ],
+    }
+
+    def _make_search_result(self) -> MagicMock:
+        """Mock NSPDSearchResult с первым feature, у которого есть geometry."""
+        feat = MagicMock(spec=NSPDFeature)
+        feat.geometry = self._MOCK_GEOM
+        feat.properties = {}
+        feat.feature_id = "q-001"
+        result = MagicMock()
+        result.first = feat
+        return result
+
+    def test_fetch_layer_uses_grid_walk_for_area_layers(self) -> None:
+        """territorial_zones (area layer) → вызывает get_features_in_bbox_grid."""
+        client = NSPDClient()
+        mock_search_result = self._make_search_result()
+
+        with (
+            patch.object(client, "search_by_cad", return_value=mock_search_result),
+            patch.object(client, "get_features_in_bbox_grid", return_value=[]) as mock_grid,
+            patch.object(client, "get_features_in_bbox", return_value=[]) as mock_legacy,
+        ):
+            client.search_by_quarter(
+                "66:41:0303161",
+                include_zouit=False,
+                include_risks=False,
+                include_opportunity=False,
+            )
+
+        # territorial_zones, red_lines, engineering_structures — все три должны
+        # быть запрошены через grid-walk
+        called_layer_ids = [call.args[0] for call in mock_grid.call_args_list]
+        from app.services.scrapers.nspd_client import LAYERS
+
+        assert (
+            LAYERS["territorial_zones"] in called_layer_ids
+        ), "territorial_zones должен использовать grid-walk"
+        assert LAYERS["red_lines"] in called_layer_ids, "red_lines должен использовать grid-walk"
+        assert (
+            LAYERS["engineering_structures"] in called_layer_ids
+        ), "engineering_structures должен использовать grid-walk"
+        # parcels и buildings — legacy, не grid
+        called_legacy_ids = [call.args[0] for call in mock_legacy.call_args_list]
+        assert LAYERS["parcels"] in called_legacy_ids, "parcels должен идти через legacy"
+        assert LAYERS["buildings"] in called_legacy_ids, "buildings должен идти через legacy"
+
+    def test_fetch_layer_uses_legacy_for_egrn_layers(self) -> None:
+        """parcels и buildings → вызывают legacy get_features_in_bbox (не grid)."""
+        client = NSPDClient()
+        mock_search_result = self._make_search_result()
+
+        with (
+            patch.object(client, "search_by_cad", return_value=mock_search_result),
+            patch.object(client, "get_features_in_bbox_grid", return_value=[]) as mock_grid,
+            patch.object(client, "get_features_in_bbox", return_value=[]) as mock_legacy,
+        ):
+            client.search_by_quarter(
+                "66:41:0303161",
+                include_zouit=False,
+                include_risks=False,
+                include_opportunity=False,
+            )
+
+        from app.services.scrapers.nspd_client import LAYERS
+
+        called_legacy_ids = [call.args[0] for call in mock_legacy.call_args_list]
+        called_grid_ids = [call.args[0] for call in mock_grid.call_args_list]
+
+        assert LAYERS["parcels"] in called_legacy_ids
+        assert LAYERS["buildings"] in called_legacy_ids
+        # EGRN layers должны НЕ попасть в grid-walk
+        assert LAYERS["parcels"] not in called_grid_ids, "parcels не должен использовать grid"
+        assert LAYERS["buildings"] not in called_grid_ids, "buildings не должен использовать grid"
+
+    def test_engineering_structures_classified_kind_enriched(self) -> None:
+        """engineering_structures features получают classified_kind в properties."""
+        from app.services.scrapers.nspd_client import LAYERS
+
+        # Создаём feature с mutable properties (как реальный NSPDFeature)
+        eng_feat = MagicMock(spec=NSPDFeature)
+        eng_feat.feature_id = "eng-001"
+        eng_feat.geometry = None
+        eng_feat.properties = {"params_name": "Газопровод высокого давления"}
+
+        client = NSPDClient()
+        mock_search_result = self._make_search_result()
+
+        def _grid_side_effect(layer_id: int, bbox: Any, **kwargs: Any) -> list[Any]:
+            if layer_id == LAYERS["engineering_structures"]:
+                return [eng_feat]
+            return []
+
+        with (
+            patch.object(client, "search_by_cad", return_value=mock_search_result),
+            patch.object(client, "get_features_in_bbox_grid", side_effect=_grid_side_effect),
+            patch.object(client, "get_features_in_bbox", return_value=[]),
+        ):
+            dump = client.search_by_quarter(
+                "66:41:0303161",
+                include_zouit=False,
+                include_risks=False,
+                include_opportunity=False,
+            )
+
+        assert len(dump.engineering_structures) == 1
+        enriched = dump.engineering_structures[0]
+        got = enriched.properties.get("classified_kind")
+        assert got == "gas", f"Ожидали classified_kind='gas', получили {got!r}"
+
+
 # ── Integration marker (skip in CI) ──────────────────────────────────────────
 
 

backend/tests/test_nspd_client.py

@@ -507,6 +507,40 @@ def _make_fake_http(
     return fake_http
 
 
+def _make_fake_grid_walk(
+    layer_feature_counts: dict[str, int] | None = None,
+) -> Any:
+    """Возвращает fake `NSPDClient.get_features_in_bbox_grid`.
+
+    После Sub-PR B (#260) area/linear layers (territorial_zones, red_lines,
+    engineering_structures, zouit_*, risk_*) идут через grid-walk вместо
+    legacy `_http_get_json` → тесты обязаны мокать оба пути, иначе грид-walk
+    бьёт по живому NSPD API.
+    """
+    counts = layer_feature_counts or {}
+    from app.services.scrapers.nspd_client import LAYERS as _LAYERS
+    from app.services.scrapers.nspd_client import NSPDFeature
+
+    id_to_name: dict[int, str] = {v: k for k, v in _LAYERS.items()}
+
+    def fake_grid(
+        self: Any,
+        layer_id: int,
+        bbox: tuple[float, float, float, float],
+        *,
+        grid_n: int = 7,
+        step_m: float = 50.0,
+    ) -> list[NSPDFeature]:
+        layer_name = id_to_name.get(layer_id, "unknown")
+        n = counts.get(layer_name, 1)
+        return [
+            NSPDFeature.from_raw({"id": f"{layer_name}-{i}", "geometry": None, "properties": {}})
+            for i in range(n)
+        ]
+
+    return fake_grid
+
+
 def test_search_by_quarter_core_only(monkeypatch: pytest.MonkeyPatch) -> None:
     """core_only (include_zouit=False, include_risks=False): 1 search + 5 bulk."""
     search_calls: list[str] = []
@@ -520,6 +554,11 @@ def test_search_by_quarter_core_only(monkeypatch: pytest.MonkeyPatch) -> None:
         "app.services.scrapers.nspd_client._http_get_json",
         _make_fake_http(_LAYER_FEATURE_COUNTS),
     )
+    # Sub-PR B (#260): area layers идут через grid-walk — мокаем оба пути.
+    monkeypatch.setattr(
+        "app.services.scrapers.nspd_client.NSPDClient.get_features_in_bbox_grid",
+        _make_fake_grid_walk(_LAYER_FEATURE_COUNTS),
+    )
 
     result = NSPDClient().search_by_quarter(
         "66:41:0204016", include_zouit=False, include_risks=False
@@ -566,6 +605,11 @@ def test_search_by_quarter_with_zouit(monkeypatch: pytest.MonkeyPatch) -> None:
         "app.services.scrapers.nspd_client._http_get_json",
         _make_fake_http(),
     )
+    # Sub-PR B (#260): zouit_* и area layers идут через grid-walk.
+    monkeypatch.setattr(
+        "app.services.scrapers.nspd_client.NSPDClient.get_features_in_bbox_grid",
+        _make_fake_grid_walk(),
+    )
 
     result = NSPDClient().search_by_quarter(
         "66:41:0204016", include_zouit=True, include_risks=False
@@ -628,6 +672,11 @@ def test_search_by_quarter_layers_fetched_with_risks(monkeypatch: pytest.MonkeyP
         "app.services.scrapers.nspd_client._http_get_json",
         _make_fake_http(),
     )
+    # Sub-PR B (#260): risk_* и area layers идут через grid-walk.
+    monkeypatch.setattr(
+        "app.services.scrapers.nspd_client.NSPDClient.get_features_in_bbox_grid",
+        _make_fake_grid_walk(),
+    )
 
     result = NSPDClient().search_by_quarter(
         "66:41:0204016", include_zouit=False, include_risks=True