backend/app/services/site_finder/ird_analyze.py

@@ -28,6 +28,7 @@ from sqlalchemy.orm import Session
 
 from app.services.scrapers.ekb_geoportal_client import EKBGeoportalClient
 from app.services.site_finder.ird_overlay_lookup import parcel_ird_overlaps
+from app.services.site_finder.krt_lookup import parcel_krt_requisites
 from app.services.site_finder.planning_lookup import parcel_planning_overlaps
 from app.services.site_finder.reservation_lookup import parcel_reservations
 from app.services.site_finder.zone_regulation import get_or_fetch_zone_regulation
@@ -96,6 +97,7 @@ def build_ird_analyze_block(
         ``opportunity_overlaps`` (подмножество layer_kind='opportunity_*'),
         ``planning_projects`` (ППТ/ПМТ overlap, DB, м.134 #1085),
         ``reservations`` (list, изъятие/резервирование из land_reservation по cad_num, #1091/#1062),
+        ``krt_requisites`` (list, КРТ-реквизиты из ekb_krt_sites + planning_projects, #1060),
         ``functional_zone`` (генплан, dict|None), ``krt`` (list), ``zone_regulation`` (dict|None).
         Любой источник деградирует независимо — блок всегда well-formed.
     """
@@ -108,6 +110,9 @@ def build_ird_analyze_block(
         "opportunity_overlaps": opportunity,
         "planning_projects": parcel_planning_overlaps(db, geom_wkt),
         "reservations": parcel_reservations(db, cad_num),
+        # КРТ-реквизиты (DB, дёшево, graceful): planning_projects ⋈ ekb_krt_sites (#1060).
+        # Идёт рядом с planning_projects — оба DB-источника, не в geoportal-конкурентной части.
+        "krt_requisites": parcel_krt_requisites(db, geom_wkt),
         "functional_zone": None,
         "krt": [],
         "zone_regulation": None,

backend/app/services/site_finder/krt_lookup.py

@@ -0,0 +1,189 @@
+"""КРТ-реквизиты для analyze_parcel (#1060, эпик #1067).
+
+Связывает геометрию КРТ-площадок из ``planning_projects`` (ППТ/ПМТ, м.134) с реестром
+реквизитов ``ekb_krt_sites`` (застройщик / аукционная цена / градпотенциал).
+
+Ключ связки: номер постановления ПАГЕ из ``ekb_krt_sites.paga_decisions`` + год совпадает с
+заголовком ``planning_projects.full_name``. Даёт ``confidence='exact'`` для 3 из 15 площадок
+(по состоянию на 2026-06). Для остальных КРТ-геом-пересечений (opportunity_krt в ird_overlays —
+это ККР, не КРТ-площадки застройки) надёжного ключа нет → ``matched=False, confidence='none'``.
+
+Graceful: нет таблицы / пустой WKT / исключение БД → [].
+Зеркалит стиль ``ird_overlay_lookup.py`` + ``planning_lookup.py``.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.exc import OperationalError, ProgrammingError
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Участок → planning_projects, которые пересекают его И имеют в full_name «ПАГЕ № N от»
+# (первичный документ, не «Внесение изменений»). Geom — 4326. CAST psycopg v3.
+_PLANNING_KRT_SQL = text(
+    """
+    SELECT
+        pp.id           AS pp_id,
+        pp.full_name    AS pp_full_name,
+        pp.dmd_actual_year,
+        pp.doc_status_name,
+        k.id            AS krt_id,
+        k.site_name,
+        k.developer_name,
+        k.contract_number,
+        k.auction_final_price,
+        k.gradpotential_sqm,
+        k.area_ha,
+        k.krt_kind
+    FROM planning_projects pp
+    -- только первичные документы (не «Внесение изменений»)
+    CROSS JOIN ekb_krt_sites k
+    -- пересечение участка с PP-геометрией
+    WHERE ST_Intersects(
+            pp.geom,
+            ST_GeomFromText(CAST(:parcel_wkt AS text), 4326)
+          )
+    -- матч по ПАГЕ-номеру: в full_name есть «ПАГЕ № {paga_num} от», год совпадает
+    AND EXISTS (
+        SELECT 1
+        FROM jsonb_array_elements(k.paga_decisions) d
+        WHERE pp.full_name ~ ('^[[:space:]]*ПАГЕ № ' || (d->>'number') || ' от')
+          AND pp.dmd_actual_year::text = split_part(d->>'date', '.', 3)
+          AND (d->>'title' ILIKE '%планировки%' OR d->>'title' ILIKE '%межевания%')
+    )
+    ORDER BY pp.dmd_actual_year DESC, k.id
+    """
+)
+
+# Вторичный запрос: planning_projects без матча с ekb_krt_sites — geom-only без реквизитов.
+# Запускается только если первый не дал результатов, чтобы не путать пользователя
+# двумя overlapping-источниками.
+_PLANNING_NO_MATCH_SQL = text(
+    """
+    SELECT
+        pp.id          AS pp_id,
+        pp.full_name   AS pp_full_name,
+        pp.dmd_actual_year,
+        pp.doc_status_name
+    FROM planning_projects pp
+    WHERE ST_Intersects(
+            pp.geom,
+            ST_GeomFromText(CAST(:parcel_wkt AS text), 4326)
+          )
+      AND pp.full_name ~ '[кК][рР][тТ]|комплексн.*развити|Краснофлотцев|Донбасской
+        |Шефской|Азина|Черкасской|Кварцевой|Летчиков|Вилонова|Лумумбы
+        |Студенческой|Васильева|Гурзуфск|Бакинских|Шадринский|Бабушкина'
+    ORDER BY pp.dmd_actual_year DESC
+    """
+)
+
+
+def _extract_paga_num(full_name: str) -> str | None:
+    """Извлечь номер ПАГЕ из заголовка planning_project.full_name."""
+    m = re.search(r"ПАГЕ\s*№\s*(\d+)\s+от", full_name)
+    return m.group(1) if m else None
+
+
+def parcel_krt_requisites(db: Session, parcel_wkt: str | None) -> list[dict[str, Any]]:
+    """КРТ-реквизиты для участка: застройщик / цена / градпотенциал.
+
+    Алгоритм:
+      1. ST_Intersects участка с ``planning_projects`` + матч ПАГЕ-номер+год с
+         ``ekb_krt_sites.paga_decisions`` → ``confidence='exact'``.
+      2. Если матчей нет, ищем ППТ/ПМТ в ``planning_projects`` с КРТ-топонимами
+         → ``matched=False, confidence='none'`` (геом-факт без реквизитов).
+      3. Никакой ложной атрибуции: при отсутствии exact-матча — NOT передаём
+         developer_name/auction_final_price (остаются None).
+
+    Args:
+        db: сессия.
+        parcel_wkt: геометрия участка в WKT (EPSG:4326). None/пусто → [].
+
+    Returns:
+        list[dict]: каждый элемент содержит поля ниже; список пуст при graceful-деградации.
+        Поля: ``matched`` (bool), ``confidence`` ('exact'|'none'),
+        ``krt_geom_ref`` (pp_id или None), ``pp_full_name`` (str),
+        ``site_name``, ``developer_name``, ``auction_final_price``,
+        ``gradpotential_sqm``, ``area_ha``, ``contract_number`` — None при matched=False.
+    """
+    if not parcel_wkt:
+        return []
+
+    try:
+        rows = db.execute(_PLANNING_KRT_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
+    except (OperationalError, ProgrammingError) as exc:
+        logger.warning("parcel_krt_requisites: таблица недоступна, skip: %s", exc)
+        return []
+    except Exception as exc:
+        logger.warning("parcel_krt_requisites: неожиданная ошибка, skip: %s", exc)
+        return []
+
+    if rows:
+        # Есть exact-матч — возвращаем с реквизитами
+        seen: set[tuple[int, int]] = set()
+        result: list[dict[str, Any]] = []
+        for r in rows:
+            key = (r["pp_id"], r["krt_id"])
+            if key in seen:
+                continue
+            seen.add(key)
+            result.append(
+                {
+                    "matched": True,
+                    "confidence": "exact",
+                    "krt_geom_ref": r["pp_id"],
+                    "pp_full_name": r["pp_full_name"],
+                    "site_name": r["site_name"],
+                    "developer_name": r["developer_name"],
+                    "contract_number": r["contract_number"],
+                    "auction_final_price": (
+                        float(r["auction_final_price"])
+                        if r["auction_final_price"] is not None
+                        else None
+                    ),
+                    "gradpotential_sqm": (
+                        float(r["gradpotential_sqm"])
+                        if r["gradpotential_sqm"] is not None
+                        else None
+                    ),
+                    "area_ha": float(r["area_ha"]) if r["area_ha"] is not None else None,
+                }
+            )
+        return result
+
+    # Нет exact-матча — geom-only поиск по КРТ-топонимам
+    try:
+        no_match_rows = (
+            db.execute(_PLANNING_NO_MATCH_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
+        )
+    except (OperationalError, ProgrammingError) as exc:
+        logger.warning("parcel_krt_requisites: no-match fallback недоступен: %s", exc)
+        return []
+    except Exception as exc:
+        logger.warning("parcel_krt_requisites: no-match fallback ошибка: %s", exc)
+        return []
+
+    return [
+        {
+            "matched": False,
+            "confidence": "none",
+            "krt_geom_ref": r["pp_id"],
+            "pp_full_name": r["pp_full_name"],
+            "site_name": None,
+            "developer_name": None,
+            "contract_number": None,
+            "auction_final_price": None,
+            "gradpotential_sqm": None,
+            "area_ha": None,
+        }
+        for r in no_match_rows
+    ]
+
+
+__all__ = ["parcel_krt_requisites"]

backend/tests/services/test_ird_analyze.py

@@ -45,10 +45,27 @@ _OVERLAY = {
 _PLANNING = [{"project_type": "ppt", "doc_status_name": "действующий", "full_name": "ПАГЕ №1"}]
 
 
+_KRT_REQUISITES: list[dict[str, Any]] = [
+    {
+        "matched": True,
+        "confidence": "exact",
+        "krt_geom_ref": 1166,
+        "pp_full_name": "ПАГЕ № 1970",
+        "site_name": "КРТ Летчиков",
+        "developer_name": "ООО СЗ Летчиков",
+        "contract_number": "2-7/21.7-14",
+        "auction_final_price": 5_000_000.0,
+        "gradpotential_sqm": 25_000.0,
+        "area_ha": 1.5,
+    }
+]
+
+
 def _patch_overlay(monkeypatch: Any, overlay: dict[str, Any]) -> None:
     monkeypatch.setattr(ird_analyze, "parcel_ird_overlaps", lambda db, wkt: overlay)
     monkeypatch.setattr(ird_analyze, "parcel_planning_overlaps", lambda db, wkt: list(_PLANNING))
     monkeypatch.setattr(ird_analyze, "parcel_reservations", lambda db, cad_num: [])
+    monkeypatch.setattr(ird_analyze, "parcel_krt_requisites", lambda db, wkt: list(_KRT_REQUISITES))
 
 
 def test_full_block_assembly(monkeypatch: Any) -> None:
@@ -239,3 +256,37 @@ def test_reservations_none_cad_num_returns_empty(monkeypatch: Any) -> None:
 
     assert "reservations" in block
     assert block["reservations"] == []
+
+
+# ---------------------------------------------------------------------------
+# KRT-реквизиты (#1060)
+# ---------------------------------------------------------------------------
+
+
+def test_krt_requisites_wired_to_block(monkeypatch: Any) -> None:
+    """parcel_krt_requisites возвращает реквизит → ключ krt_requisites присутствует в блоке."""
+    _patch_overlay(monkeypatch, _OVERLAY)
+    monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
+    client = _FakeClient(fz=[], krt=[])
+
+    block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", None, None, client=client)
+
+    assert "krt_requisites" in block
+    assert len(block["krt_requisites"]) == 1
+    r = block["krt_requisites"][0]
+    assert r["matched"] is True
+    assert r["confidence"] == "exact"
+    assert r["developer_name"] == "ООО СЗ Летчиков"
+
+
+def test_krt_requisites_empty_when_no_overlap(monkeypatch: Any) -> None:
+    """parcel_krt_requisites пуст → krt_requisites == [] в блоке (ключ присутствует)."""
+    _patch_overlay(monkeypatch, {"ird_overlaps": [], "ird_by_kind": {}})
+    monkeypatch.setattr(ird_analyze, "parcel_krt_requisites", lambda db, wkt: [])
+    monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
+    client = _FakeClient(fz=[], krt=[])
+
+    block = ird_analyze.build_ird_analyze_block(object(), None, None, None, client=client)
+
+    assert "krt_requisites" in block
+    assert block["krt_requisites"] == []

backend/tests/services/test_krt_lookup.py

@@ -0,0 +1,176 @@
+"""Тесты parcel_krt_requisites (#1060) — КРТ-реквизиты из ekb_krt_sites + planning_projects.
+
+DB/сеть не дёргаются: мокаем db.execute через FakeDB.
+Проверяем:
+ - exact-матч → matched=True, confidence='exact', реквизиты заполнены;
+ - planning overlap без матча → matched=False, confidence='none', без застройщика;
+ - пустой WKT → [];
+ - OperationalError (таблица не задеплоена) → graceful [];
+ - дубликаты строк (один pp_id × krt_id) → дедублируются.
+"""
+
+from __future__ import annotations
+
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy.exc import OperationalError
+
+from app.services.site_finder.krt_lookup import parcel_krt_requisites
+
+
+class _MappingRow(dict):  # type: ignore[type-arg]
+    """Имитирует sqlalchemy MappingResult row (dict-like)."""
+
+
+class _Result:
+    def __init__(self, rows: list[dict[str, Any]]) -> None:
+        self._rows = [_MappingRow(r) for r in rows]
+
+    def mappings(self) -> _Result:
+        return self
+
+    def all(self) -> list[_MappingRow]:
+        return self._rows
+
+
+class _FakeDB:
+    """Мок SQLAlchemy Session: первый вызов execute → first_rows; второй → second_rows."""
+
+    def __init__(
+        self,
+        first_rows: list[dict[str, Any]] | None = None,
+        second_rows: list[dict[str, Any]] | None = None,
+        raise_first: Exception | None = None,
+        raise_second: Exception | None = None,
+    ) -> None:
+        self._calls = 0
+        self._first = first_rows or []
+        self._second = second_rows or []
+        self._raise_first = raise_first
+        self._raise_second = raise_second
+
+    def execute(self, sql: Any, params: dict[str, Any] | None = None) -> _Result:
+        self._calls += 1
+        if self._calls == 1:
+            if self._raise_first:
+                raise self._raise_first
+            return _Result(self._first)
+        if self._raise_second:
+            raise self._raise_second
+        return _Result(self._second)
+
+
+_WKT = "POLYGON((60.5 56.8, 60.6 56.8, 60.6 56.9, 60.5 56.9, 60.5 56.8))"
+
+# Типичная строка exact-матча (planning_projects ⋈ ekb_krt_sites)
+_EXACT_ROW = {
+    "pp_id": 1166,
+    "pp_full_name": (
+        " ПАГЕ № 1970 от 16.08.2024\nПроект межевания территории в районе улицы Летчиков"
+    ),
+    "dmd_actual_year": 2024,
+    "doc_status_name": "Действующий",
+    "krt_id": 7,
+    "site_name": "2. КРТ жилой застройки в районе ул. Летчиков",
+    "developer_name": "ООО СЗ Летчиков",
+    "contract_number": "2-7/21.7-14",
+    "auction_final_price": Decimal("5000000"),
+    "gradpotential_sqm": Decimal("25000"),
+    "area_ha": Decimal("1.50"),
+    "krt_kind": "жилой застройки",
+}
+
+
+def test_exact_match_returns_requisites() -> None:
+    """Exact-матч по ПАГЕ-номер+год → matched=True, реквизиты застройщика заполнены."""
+    db = _FakeDB(first_rows=[_EXACT_ROW])
+    result = parcel_krt_requisites(db, _WKT)
+
+    assert len(result) == 1
+    r = result[0]
+    assert r["matched"] is True
+    assert r["confidence"] == "exact"
+    assert r["krt_geom_ref"] == 1166
+    assert r["site_name"] == "2. КРТ жилой застройки в районе ул. Летчиков"
+    assert r["developer_name"] == "ООО СЗ Летчиков"
+    assert r["contract_number"] == "2-7/21.7-14"
+    assert r["auction_final_price"] == 5_000_000.0
+    assert r["gradpotential_sqm"] == 25_000.0
+    assert r["area_ha"] == 1.50
+
+
+def test_exact_match_deduplicates_pp_krt_pair() -> None:
+    """Дублирующиеся строки (pp_id × krt_id) дедублируются → один результат."""
+    db = _FakeDB(first_rows=[_EXACT_ROW, _EXACT_ROW])
+    result = parcel_krt_requisites(db, _WKT)
+    assert len(result) == 1
+
+
+def test_no_match_returns_geom_only() -> None:
+    """Первый запрос пуст (нет exact-матча) → второй запрос → matched=False без реквизитов."""
+    no_match_row = {
+        "pp_id": 999,
+        "pp_full_name": " ПАГЕ № 4033 от 27.12.2022\n"
+        "Проект планировки территории в районе улицы Краснофлотцев",
+        "dmd_actual_year": 2022,
+        "doc_status_name": "Действующий",
+    }
+    db = _FakeDB(first_rows=[], second_rows=[no_match_row])
+    result = parcel_krt_requisites(db, _WKT)
+
+    assert len(result) == 1
+    r = result[0]
+    assert r["matched"] is False
+    assert r["confidence"] == "none"
+    assert r["krt_geom_ref"] == 999
+    assert r["developer_name"] is None
+    assert r["auction_final_price"] is None
+    assert r["gradpotential_sqm"] is None
+    assert r["site_name"] is None
+
+
+def test_empty_wkt_returns_empty_list() -> None:
+    """parcel_wkt=None → пустой список без вызова DB."""
+    db = _FakeDB(first_rows=[_EXACT_ROW])
+    assert parcel_krt_requisites(db, None) == []
+    assert db._calls == 0  # DB не вызывалась
+
+
+def test_empty_string_wkt_returns_empty_list() -> None:
+    """parcel_wkt='' → пустой список."""
+    assert parcel_krt_requisites(_FakeDB(), "") == []
+
+
+def test_operational_error_first_query_returns_empty() -> None:
+    """OperationalError на первом запросе (таблица не задеплоена) → []."""
+    db = _FakeDB(raise_first=OperationalError("stmt", {}, Exception("relation does not exist")))
+    result = parcel_krt_requisites(db, _WKT)
+    assert result == []
+
+
+def test_operational_error_second_query_returns_empty() -> None:
+    """OperationalError на втором запросе (no-match fallback) → []."""
+    db = _FakeDB(
+        first_rows=[],
+        raise_second=OperationalError("stmt", {}, Exception("relation does not exist")),
+    )
+    result = parcel_krt_requisites(db, _WKT)
+    assert result == []
+
+
+def test_none_decimal_fields_become_none() -> None:
+    """NULL Decimal-поля → None в результате, не падают при float()."""
+    row = {**_EXACT_ROW, "auction_final_price": None, "gradpotential_sqm": None, "area_ha": None}
+    db = _FakeDB(first_rows=[row])
+    result = parcel_krt_requisites(db, _WKT)
+    assert result[0]["auction_final_price"] is None
+    assert result[0]["gradpotential_sqm"] is None
+    assert result[0]["area_ha"] is None
+
+
+def test_no_overlap_at_all_returns_empty() -> None:
+    """Оба запроса возвращают пустые списки → []."""
+    db = _FakeDB(first_rows=[], second_rows=[])
+    result = parcel_krt_requisites(db, _WKT)
+    assert result == []