gendesign/backend/app/services/chat/retrieval.py

"""Read-only достаёт персистентный §22-форсайт участка для чата (#957, Step 1).

Тонкая обёртка над `latest_run_for(..., schema_version="1.0")` — ровно тот же seam,
что читает `GET /{cad_num}/forecast` (parcels.py): последний ран ИМЕННО §22-схемы
('1.0', НЕ analyze-1.0 inline-dict). `run.result` — это `SiteFinderReport.as_dict()`
(стабильный 8-секционный контракт), `run.id` — run_id для grounding.

Q2-РЕШЕНИЕ = READ-ONLY: рана нет → возвращаем (None, None). Эндпоинт по этому отдаёт
report_status='pending' + детерминированный RU-ответ «отчёт ещё не готов, запустите
анализ участка». НИЧЕГО не считаем, НЕ enqueue'им, НЕ пере-вычисляем (чат не триггерит
форсайт — это делает POST /analyze). Сбой БД пробрасываем НЕ из этого модуля —
гасит/логирует его вызывающий эндпоинт (как get_parcel_forecast → pending).
"""

from __future__ import annotations

import logging
from typing import Any

from sqlalchemy.orm import Session

from app.services.analysis_runs.repository import ANALYZE_SCHEMA_VERSION, latest_run_for

logger = logging.getLogger(__name__)

# Версия схемы §22-форсайта (зеркало _FORECAST_SCHEMA_VERSION в parcels.py). Читаем
# именно её, а не analyze-1.0: view вернул бы §22-ран поверх analyze, но фильтр по
# схеме явный и устойчивый.
_FORECAST_SCHEMA_VERSION = "1.0"

# Максимум названий ЗОУИТ-зон в курируемом срезе (§19: ограничиваем объём, чтобы наружу
# не ушёл длинный «хвост» — сводки хватает списка из первых нескольких зон).
_MAX_ZOUIT_NAMES = 15


def get_report_for_chat(
    db: Session,
    cad_num: str,
) -> tuple[dict[str, Any] | None, int | None]:
    """Достать последний §22-отчёт участка как (report_dict, run_id) или (None, None).

    READ-ONLY (Q2): рана нужной схемы нет → (None, None), БЕЗ enqueue/пере-расчёта.
    `run.result` — `SiteFinderReport.as_dict()` (8 секций). Если result почему-то не
    dict (мусор/частичная запись) — логируем и отдаём (None, None) как «нет отчёта»,
    чтобы чат не упал (graceful).
    """
    run = latest_run_for(db, cad_num, schema_version=_FORECAST_SCHEMA_VERSION)
    if run is None:
        return None, None

    result = run.result
    if not isinstance(result, dict):
        # Ран есть, но result не похож на отчёт (старая/битая запись) — не валим чат,
        # трактуем как «отчёта ещё нет» (pending). Логируем для диагностики.
        logger.warning(
            "chat: run %s for cad=%s has non-dict result (%s) — treating as pending",
            run.id,
            cad_num,
            type(result).__name__,
        )
        return None, None

    return result, run.id


def _as_dict(value: Any) -> dict[str, Any]:
    """Секция payload как dict или {} (graceful — как _as_dict в full_report_html)."""
    return value if isinstance(value, dict) else {}


def _put(target: dict[str, Any], key: str, value: Any) -> None:
    """Положить ТОЛЬКО скаляр в курируемый срез (§19 data-residency).

    Гейт двойной: и по КЛЮЧУ (вызывающий передаёт фиксированный whitelist), и по ТИПУ
    значения. Пропускаем лишь str|int|float|bool — dict/list/любая вложенная структура
    ОТБРАСЫВАЕТСЯ (иначе сырой под-словарь ЕГРН/зонирования с PII мог бы утечь наружу;
    redaction-слой ловит regex-PII в строках, но НЕ структуры). Пустую строку опускаем.

    UP038: union-форма isinstance (не кортеж) — pre-commit пинит старый ruff.
    """
    if not isinstance(value, str | int | float | bool):
        return
    if value == "":
        return
    target[key] = value


def _zouit_names(overlaps: list[Any]) -> list[str]:
    """Собрать dedup-список названий ЗОУИТ-зон (type_zone|name), обрезанный до лимита.

    §19: ТОЛЬКО строки-названия — никакой геометрии/reg_numb/coverage сырых пересечений.
    Не-строковые значения (dict/list/число) ОТБРАСЫВАЕМ — не приводим str()'ом, иначе
    вложенная структура утекла бы как её repr. dict.fromkeys держит порядок и dedup
    (зеркало _build_zouit в §1-рендере).
    """
    # Фильтруем до str ДО dict.fromkeys: не-str (dict/list) не только небезопасны для
    # §19, но и unhashable → уронили бы dict.fromkeys TypeError'ом.
    raw = (ov.get("type_zone") or ov.get("name") for ov in overlaps if isinstance(ov, dict))
    names = list(dict.fromkeys(n for n in raw if isinstance(n, str) and n))
    return names[:_MAX_ZOUIT_NAMES]


def get_parcel_context_for_chat(db: Session, cad_num: str) -> dict[str, Any] | None:
    """Курируемый паспорт участка + градрегламент для чата (§19 data-residency).

    READ-ONLY: последний analyze-ран (schema_version='analyze-1.0', НЕ §22-форсайт) →
    ПЛОСКИЙ whitelist-дикт из ФИКСИРОВАННОГО набора публичных градо-полей. Сырой
    analyze-blob во внешний LLM ЗАПРЕЩЁН (см. safe_payload.py) — поэтому собираем только
    скаляры и список строк по явным ключам, НИКОГДА не протаскивая под-словари/геометрию.

    Ключи payload — те же, что читает §1-рендер отчёта (full_report_html:_build_parcel_facts
    / _build_zoning / _build_zouit): egrn, nspd_zoning, encumbrance, nspd_zouit_overlaps.

    None если analyze-рана/результата нет (graceful, как get_report_for_chat) — вызывающий
    работает как раньше (только §22-отчёт, без паспорта участка).
    """
    run = latest_run_for(db, cad_num, schema_version=ANALYZE_SCHEMA_VERSION)
    if run is None:
        return None

    result = run.result
    if not isinstance(result, dict):
        logger.warning(
            "chat: analyze run %s for cad=%s has non-dict result (%s) — no parcel context",
            run.id,
            cad_num,
            type(result).__name__,
        )
        return None

    egrn = _as_dict(result.get("egrn"))
    nspd_zoning = _as_dict(result.get("nspd_zoning"))
    encumbrance = _as_dict(result.get("encumbrance"))
    overlaps = [ov for ov in (result.get("nspd_zouit_overlaps") or []) if isinstance(ov, dict)]

    context: dict[str, Any] = {}

    # ── Кадастровые факты (ЕГРН) — скаляры, без геометрии ────────────────────────
    _put(context, "address", egrn.get("address"))
    _put(context, "area_m2", egrn.get("area_m2"))
    _put(context, "land_category", egrn.get("land_category"))
    _put(context, "permitted_use_text", egrn.get("permitted_use_text"))
    _put(context, "cadastral_value_rub", egrn.get("cadastral_value_rub"))
    _put(context, "parcel_status", egrn.get("parcel_status"))
    _put(context, "ownership_type", egrn.get("ownership_type"))

    # ── Территориальная зона ПЗЗ + лимиты регламента (те же ключи, что §1-рендер) ─
    _put(
        context,
        "zone_code",
        nspd_zoning.get("zone_code") or nspd_zoning.get("regulation_zone_index"),
    )
    _put(context, "zone_name", nspd_zoning.get("zone_name"))
    _put(context, "max_far", nspd_zoning.get("max_far"))
    _put(context, "max_floors", nspd_zoning.get("max_floors"))
    _put(context, "max_height_m", nspd_zoning.get("max_height_m"))
    _put(context, "max_building_pct", nspd_zoning.get("max_building_pct"))
    _put(context, "min_parcel_area_m2", nspd_zoning.get("min_parcel_area_m2"))
    _put(context, "regulation_source", nspd_zoning.get("regulation_source"))

    # ── ЗОУИТ-обременения: сводка + список названий (dedup, capped) ───────────────
    has_zouit = encumbrance.get("has_zouit")
    zouit_count = encumbrance.get("zouit_count")
    zouit_names = _zouit_names(overlaps)
    # Противоречие «encumbrance=нет, но пересечения ЕСТЬ» — доверяем фактам НСПД (тот же
    # приём, что _build_zouit): свежий геослой перекрывает устаревшую сводку.
    if not has_zouit and zouit_names:
        has_zouit = True
        if zouit_count in (None, 0):
            zouit_count = len(overlaps)
    _put(context, "has_zouit", has_zouit)
    _put(context, "zouit_count", zouit_count)
    if zouit_names:
        context["zouit_zone_names"] = zouit_names

    return context or None