diff --git a/backend/app/scrapers/nspd_bulk_client.py b/backend/app/scrapers/nspd_bulk_client.py index 53d93c8e..7475abc9 100644 --- a/backend/app/scrapers/nspd_bulk_client.py +++ b/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: diff --git a/backend/app/services/scrapers/nspd_client.py b/backend/app/services/scrapers/nspd_client.py index 639ba3bb..670b66e2 100644 --- a/backend/app/services/scrapers/nspd_client.py +++ b/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") diff --git a/backend/app/workers/tasks/nspd_sync.py b/backend/app/workers/tasks/nspd_sync.py index 3c377919..a7d4989a 100644 --- a/backend/app/workers/tasks/nspd_sync.py +++ b/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, diff --git a/backend/tests/scrapers/test_nspd_bulk_client.py b/backend/tests/scrapers/test_nspd_bulk_client.py index 48e482fb..7add40f5 100644 --- a/backend/tests/scrapers/test_nspd_bulk_client.py +++ b/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 diff --git a/backend/tests/scrapers/test_nspd_grid_walk.py b/backend/tests/scrapers/test_nspd_grid_walk.py index 25bdfec6..377ba20c 100644 --- a/backend/tests/scrapers/test_nspd_grid_walk.py +++ b/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) ────────────────────────────────────────── diff --git a/backend/tests/test_nspd_client.py b/backend/tests/test_nspd_client.py index b7b7479b..9fa82d81 100644 --- a/backend/tests/test_nspd_client.py +++ b/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