backend/app/api/v1/parcels.py

@@ -1084,7 +1084,7 @@ def _compute_confidence(
     poi_rows: list[dict[str, Any]],
     district_row: dict[str, Any] | None,
     competitor_rows: list[dict[str, Any]],
-    noise_sources_count: int,
+    noise_map_rows_nearby: int,
     air_q: dict[str, Any] | None,
     weather: dict[str, Any] | None,
     market_trend: dict[str, Any] | None,
@@ -1164,9 +1164,14 @@ def _compute_confidence(
         caveats.append("Нет конкурентов-ЖК в 3км — низкая урбанизация / окраина")
 
     # 6) Environmental data freshness
-    env_ok = sum([bool(noise_sources_count > 0), bool(air_q), bool(weather)])
+    # #2464-G: считаем строки шумовой КАРТЫ в радиусе (любого типа, включая
+    # water/utility), а не отфильтрованные источники для скоринга. Вопрос здесь —
+    # «есть ли у нас данные по этой точке», и ноль означает непокрытие карты.
+    # Отфильтрованный список дал бы 0 у трети участков, где рядом просто тихо, и
+    # оговорка ниже утверждала бы неправду.
+    env_ok = sum([bool(noise_map_rows_nearby > 0), bool(air_q), bool(weather)])
     subscores["environment"] = env_ok / 3.0
-    if noise_sources_count == 0:
+    if noise_map_rows_nearby == 0:
         caveats.append("Шумовая карта не загружена — noise score = stub")
     if not air_q:
         caveats.append("Air Quality API недоступен — exposure unknown")
@@ -2531,7 +2536,24 @@ def analyze_parcel(
             }
         )
 
-    # 7) Noise score — шумовые источники в радиусе 2 км
+    # 7) Noise score — шумовые источники в радиусе 2 км.
+    #
+    # #2464-G: фильтр по source_type обязателен. Таблица osm_noise_sources_ekb
+    # держит и НЕшумовые слои — 'water' (870 строк) и 'utility' (1 487), их
+    # отдельно читает гидрология в 9c ниже. Для скорера они мусор: ключа в
+    # NOISE_L_BASE у них нет, поэтому `.get(key, 50.0)` выдавал им ровно 50 дБ —
+    # значение, совпадающее с порогом попадания в список источников.
+    #
+    # Главное — `LIMIT 30` берётся ПО БЛИЗОСТИ, поэтому вода вытесняла настоящие
+    # источники. Замер 19.08 на 1 000 участков (детерминированная выборка по
+    # cad_num): 19 755 занятых слотов, из них 8 797 (44.5%) — вода и коммуникации;
+    # 562 участка теряли хотя бы один настоящий источник.
+    #
+    # Честно про эффект: сегодня пользователь этого не видит — все вытесненные
+    # источники оказались тише порога 50 дБ (участков, теряющих ВИДИМЫЙ источник:
+    # 0 из 729), и максимум дБ не меняется ни у одного. Правка убирает не видимую
+    # поломку, а скрытый потолок: почти половина бюджета LIMIT уходила на строки,
+    # которые скорер не умеет оценивать.
     noise_rows = (
         db.execute(
             text("""
@@ -2541,7 +2563,8 @@ def analyze_parcel(
                        ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
                    ) AS distance_m
               FROM osm_noise_sources_ekb n
-             WHERE ST_DWithin(
+             WHERE n.source_type IN ('highway', 'railway', 'industrial', 'aerodrome')
+               AND ST_DWithin(
                        n.geom::geography,
                        ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
                        2000
@@ -2555,6 +2578,30 @@ def analyze_parcel(
         .all()
     )
 
+    # Покрытие шумовой карты — ОТДЕЛЬНО от списка источников, и это не педантизм.
+    # _compute_confidence спрашивает «загружена ли шумовая карта», а не «шумно ли
+    # тут»: при нуле она пишет «Шумовая карта не загружена — noise score = stub».
+    # У 345 участков из 1 000 в радиусе 2 км нет НИ ОДНОГО шумового источника, но
+    # вода/коммуникации есть. Передай туда len(noise_rows) после фильтра — и треть
+    # участков получит утверждение о незагруженной карте, которое неверно: карта
+    # загружена, просто рядом тихо. До этой правки верный ответ получался
+    # случайно — ровно потому, что в счёт шли и нешумовые строки.
+    noise_map_rows_nearby: int = (
+        db.execute(
+            text("""
+            SELECT COUNT(*)
+              FROM osm_noise_sources_ekb n
+             WHERE ST_DWithin(
+                       n.geom::geography,
+                       ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
+                       2000
+                   )
+        """),
+            {"wkt": geom_wkt},
+        ).scalar()
+        or 0
+    )
+
     noise_db_max = 0.0
     nearby_noise_sources: list[dict[str, Any]] = []
     for nr in noise_rows:
@@ -3802,7 +3849,7 @@ def analyze_parcel(
         poi_rows=[dict(p) for p in poi_rows],
         district_row=dict(district_row) if district_row else None,
         competitor_rows=[dict(c) for c in competitor_rows],
-        noise_sources_count=len(noise_rows),
+        noise_map_rows_nearby=noise_map_rows_nearby,
         air_q=air_q,
         weather=weather,
         market_trend=market_trend,

backend/tests/api/v1/test_2464g_noise_source_filter.py

@@ -0,0 +1,156 @@
+"""#2464-G: запрос шумовых источников обязан отсеивать нешумовые слои.
+
+`osm_noise_sources_ekb` держит не только источники шума. Замер на проде 19.08:
+
+    highway     5 616   есть в NOISE_L_BASE
+    railway     1 994   есть в NOISE_L_BASE
+    industrial    585   есть в NOISE_L_BASE
+    utility     1 487   -> дефолт 50 дБ
+    water         870   -> дефолт 50 дБ
+
+`water`/`utility` — предмет отдельного блока гидрологии (9c), для шумового скорера
+это мусор: ключа в NOISE_L_BASE у них нет, и `.get(key, 50.0)` выдаёт им ровно 50 дБ,
+что совпадает с порогом попадания в список источников. При этом `LIMIT 30` берётся
+ПО БЛИЗОСТИ, поэтому вода вытесняла настоящие источники: 8 797 занятых слотов из
+19 755 (44.5%) на выборке в 1 000 участков.
+
+Ловушка, ради которой написан второй тест
+─────────────────────────────────────────
+Наивная правка — просто добавить фильтр и оставить `noise_sources_count=len(noise_rows)` —
+ломает треть участков. `_compute_confidence` спрашивает «загружена ли шумовая карта»
+(при нуле пишет «Шумовая карта не загружена — noise score = stub»), а у 345 участков
+из 1 000 в радиусе 2 км нет НИ ОДНОГО шумового источника при наличии воды. После
+наивной правки они получили бы утверждение о незагруженной карте — неверное.
+
+До правки верный ответ там выходил СЛУЧАЙНО: в счёт шли и нешумовые строки.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+from fastapi.testclient import TestClient
+
+from app.main import app
+from tests.api.v1.test_analyze_market_price import (
+    _GEOJSON,
+    _WKT,
+    _make_mapping,
+    _override_db,
+    _start_patches,
+    _stop_patches,
+)
+
+_CAD = "66:41:0204016:10"
+
+# Строки «шумовой» таблицы вокруг участка: настоящих источников нет, есть вода
+# вплотную (10 м → base 50 дБ − 20·log10(1) = ровно 50, порог списка) и коммуникации.
+_TABLE_ROWS = [
+    {"source_type": "water", "road_class": None, "name": "Пруд", "distance_m": 10.0},
+    {"source_type": "utility", "road_class": None, "name": "ЛЭП", "distance_m": 900.0},
+]
+
+
+def _make_db(rows: list[dict[str, Any]]) -> MagicMock:
+    """Mock Session, ведущий себя как БД: применяет предикат source_type запроса.
+
+    Мок здесь стоит вместо Postgres, поэтому обязан УВАЖАТЬ WHERE — иначе тест
+    проверял бы не фильтрацию, а то, что мы отдали в мок.
+    """
+    db = MagicMock()
+
+    def _execute(*args: Any, **kwargs: Any) -> MagicMock:
+        sql = " ".join(str(args[0]).split()) if args else ""
+        first_val: Any = None
+        all_val: list[Any] = []
+        scalar_val: Any = 0
+
+        if "AS geom_geojson" in sql:
+            first_val = _make_mapping(
+                {"geom_geojson": _GEOJSON, "geom_wkb": None, "source": "cad_quarter"}
+            )
+        elif "AS wkt" in sql:
+            first_val = _make_mapping({"wkt": _WKT})
+        elif "AS ekb_reference_median" in sql and "district_name" in sql:
+            first_val = _make_mapping(
+                {
+                    "district_name": "Октябрьский",
+                    "ekb_reference_median": 120000,
+                    "quarter_median_12m": None,
+                    "quarter_deals_count": 0,
+                    "dist_to_center": 1500.0,
+                }
+            )
+        elif "AS lon" in sql and "AS lat" in sql:
+            first_val = _make_mapping({"lat": 56.84, "lon": 60.605})
+        elif "osm_noise_sources_ekb" in sql:
+            visible = rows
+            # Уважаем предикат так же, как это сделал бы Postgres.
+            if "source_type IN (" in sql:
+                visible = [r for r in rows if r["source_type"] not in ("water", "utility")]
+            elif "source_type = 'water'" in sql:
+                visible = [r for r in rows if r["source_type"] == "water"]
+            elif "source_type = 'industrial'" in sql:
+                visible = [r for r in rows if r["source_type"] == "industrial"]
+            if "COUNT(*)" in sql:
+                scalar_val = len(visible)
+            else:
+                all_val = [_make_mapping(r) for r in visible]
+
+        r = MagicMock()
+        r.mappings.return_value.first.return_value = first_val
+        r.mappings.return_value.all.return_value = all_val
+        r.scalar.return_value = scalar_val
+        return r
+
+    db.execute.side_effect = _execute
+    ctx = MagicMock()
+    ctx.__enter__ = MagicMock(return_value=ctx)
+    ctx.__exit__ = MagicMock(return_value=False)
+    db.begin_nested.return_value = ctx
+    return db
+
+
+def _analyze(rows: list[dict[str, Any]]) -> dict[str, Any]:
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(_make_db(rows))
+    _start_patches()
+    try:
+        resp = TestClient(app).post(f"/api/v1/parcels/{_CAD}/analyze")
+        assert resp.status_code == 200, resp.text
+        return resp.json()
+    finally:
+        app.dependency_overrides.clear()
+        _stop_patches()
+
+
+def test_water_not_reported_as_noise_source() -> None:
+    """Пруд в 10 м не должен числиться источником шума.
+
+    На main фильтра нет, водоём получает дефолтные 50 дБ (= порог) и попадает в
+    список источников с именем «Пруд».
+    """
+    body = _analyze(_TABLE_ROWS)
+    noise = body.get("noise") or {}
+    sources = noise.get("nearby_sources") or noise.get("sources") or []
+    types = {s.get("source_type") for s in sources}
+    assert (
+        "water" not in types and "utility" not in types
+    ), f"нешумовой слой попал в источники шума: {sources}"
+
+
+def test_no_false_map_not_loaded_caveat_when_only_water_nearby() -> None:
+    """Ловушка наивной правки: карта ЗАГРУЖЕНА, рядом просто тихо.
+
+    Зелёный и на main, и с правкой — но красный, если считать покрытие карты по
+    отфильтрованному списку источников. Так себя вёл бы «очевидный» вариант фикса,
+    и он соврал бы 345 участкам из 1 000.
+    """
+    body = _analyze(_TABLE_ROWS)
+    caveats = " ".join(body.get("confidence_caveats") or [])
+    assert "Шумовая карта не загружена" not in caveats, (
+        "оговорка о незагруженной карте при загруженной карте: "
+        "покрытие посчитано по отфильтрованному списку, а не по строкам карты"
+    )

backend/tests/api/v1/test_analyze_parcel_meta.py

@@ -7,26 +7,14 @@
 Стратегия mock: аналогична test_analyze_market_price.py — DB mock через
 dependency_overrides, тяжёлые сервисы патчим через unittest.mock.patch.
 
-Порядок db.execute calls в analyze_parcel (с #29 G2, #2464 cluster B):
-  0.  UNION ALL geom + source → .mappings().first()
-  1.  WKT query → .mappings().first()
-  2.  District → .mappings().first()
-  3.  POI rows → .mappings().all()
-  4.  Competitor rows → .mappings().all()
-  5.  competitors_total honest COUNT(*) → .scalar()   ← NEW #2464 cluster B
-  6.  Pipeline rows → .mappings().all()
-  7.  Centroid lat/lon → .mappings().first()
-  8.  Noise rows → .mappings().all()
-  9.  Hydrology → .mappings().all()
-  10. Utilities → .mappings().all()
-  11. parcel_meta (cad_parcels) → .mappings().first()   ← NEW #29 G2
-  12. Market trend → .mappings().first()
-  13. Zoning (begin_nested) → .mappings().first()
-  14. Success recommendation (begin_nested) → .mappings().all()
-  15. Market price (begin_nested) → .mappings().first()
-  16. Recent permits (begin_nested) → .mappings().all()
-  17. _geotech_risk (industrial count) → .scalar()
-  18. _neighbors_summary (single statement, neighbors+overlap+total) → .mappings().first()
+Диспетчеризация мока — ПО СИГНАТУРЕ SQL, а не по порядковому номеру вызова.
+Позиционный вариант ломался при каждом добавлении запроса в analyze_parcel
+(следы в истории: #29 G2, #2464 cluster B, затем #2464-G с запросом покрытия
+шумовой карты). Такое падение говорит не о parcel_meta, ради которого написан
+файл, а о том, что кто-то добавил блок выше по коду — красный CI без отношения
+к предмету теста. Матчинг по сигнатуре к порядку нечувствителен; всё, что не
+распознано, отдаёт пустой результат (обработчик это переживает — прочие блоки
+обёрнуты в try/except SAVEPOINT).
 """
 
 from __future__ import annotations
@@ -75,46 +63,32 @@ def _make_db_for_analyze(
 
     pm_mock = _make_mapping(parcel_meta_row) if parcel_meta_row is not None else None
 
-    call_idx = [0]
-    responses: list[Any] = [
-        ("first", geom_row),  # 0: geom UNION ALL
-        ("first", wkt_row),  # 1: WKT
-        ("first", district_row),  # 2: district
-        ("all", []),  # 3: POI rows
-        ("all", []),  # 4: competitor rows
-        ("scalar", 0),  # 5: competitors_total honest COUNT(*) ← NEW #2464 cluster B
-        ("all", []),  # 6: pipeline rows
-        ("first", centroid_row),  # 7: centroid
-        ("all", []),  # 8: noise rows
-        ("all", []),  # 9: hydrology rows
-        ("all", []),  # 10: utilities rows
-        ("first", pm_mock),  # 11: parcel_meta ← #29 G2
-        ("first", None),  # 12: market trend
-        ("first", None),  # 13: zoning (begin_nested)
-        ("all", []),  # 14: success recommendation (begin_nested)
-        ("first", None),  # 15: market price (begin_nested)
-        ("all", []),  # 16: recent permits (begin_nested)
-        ("scalar", 0),  # 17: geotech_risk
-        # 18: _neighbors_summary — ОДИН statement (neighbors + overlap + total через
-        # json_agg + neighbors_total CTE, PR #1130 / #2464 cluster B) → .mappings().first()
-        # возвращает ОДНУ строку с этими тремя ключами.
-        ("first", {"neighbors": [], "overlap_rows": [], "neighbors_total_count": 0}),
-    ]
-
     def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:
-        idx = call_idx[0]
-        call_idx[0] += 1
-        if idx >= len(responses):
-            r = MagicMock()
-            r.mappings.return_value.first.return_value = None
-            r.mappings.return_value.all.return_value = []
-            r.scalar.return_value = 0
-            return r
-        kind, data = responses[idx]
+        sql = " ".join(str(args[0]).split()) if args else ""
+
+        first_val: Any = None
+        all_val: list[Any] = []
+
+        if "AS geom_geojson" in sql:
+            first_val = geom_row
+        elif "AS wkt" in sql:
+            first_val = wkt_row
+        elif "AS ekb_reference_median" in sql and "district_name" in sql:
+            first_val = district_row
+        elif "AS lon" in sql and "AS lat" in sql:
+            first_val = centroid_row
+        # Предмет файла: parcel_meta из cad_parcels (#29 G2).
+        elif "AS permitted_use" in sql and "FROM cad_parcels" in sql:
+            first_val = pm_mock
+        # _neighbors_summary — ОДИН statement (neighbors + overlap + total через
+        # json_agg + neighbors_total CTE, PR #1130 / #2464 cluster B).
+        elif "neighbors_total_count" in sql or "json_agg" in sql:
+            first_val = {"neighbors": [], "overlap_rows": [], "neighbors_total_count": 0}
+
         r = MagicMock()
-        r.mappings.return_value.first.return_value = data
-        r.mappings.return_value.all.return_value = data if isinstance(data, list) else []
-        r.scalar.return_value = data if kind == "scalar" else 0
+        r.mappings.return_value.first.return_value = first_val
+        r.mappings.return_value.all.return_value = all_val
+        r.scalar.return_value = 0
         return r
 
     db.execute.side_effect = _execute_side_effect