gendesign/backend/app/services/site_finder/developer_attribution.py

"""Детерминированная атрибуция застройщика для analyze_parcel (#1088 «GG-форсайт»).

Вызывает резолвер ``fn_developer_for_parcel(cad_num, radius_m)`` (миграция 149,
``data/sql/149_developer_attribution.sql``) и собирает атрибуцию «пятно земли ↔
застройщик»: топ-1 канон-застройщик участка (имя + норм-ИНН) с его проектом/статусом
и track-record (РНС/РВЭ + домрф ЖК/квартиры из ``developer_registry``), плюс список
``nearby_developers`` — «кто ещё строит рядом / в квартале».

Резолвер — 3-ступенчатый fallback (exact_cadastral > spatial_150m > quarter), уже
отсортированный лучшим матчем вперёд (ORDER BY method, distance). Один застройщик
может встретиться в нескольких ступенях (exact+spatial+quarter) — дедупим по
норм-ИНН, оставляя ЛУЧШИЙ match_method (первый по ORDER BY резолвера). Топ-1
дедупнутый застройщик → primary, остальные → nearby_developers.

Связка с ``competitors.py``: иная ответственность. competitors отдаёт ЖК-конкурентов
в радиусе (domrf, velocity), здесь — кто ЗАСТРОЙЩИК конкретного участка и его послужной
список по РНС+домрф. Отдельный сервис (НЕ метод в competitors).

Graceful: пустой результат / нет геометрии участка / БД-ошибка (incl. миграция 149 ещё
не задеплоена) → None — analyze не падает. Зеркалит стиль ``ird_overlay_lookup.py`` /
``okn_lookup.py``.
"""

from __future__ import annotations

import logging
from typing import Any

from sqlalchemy import text
from sqlalchemy.exc import DataError, OperationalError, ProgrammingError
from sqlalchemy.orm import Session

from app.schemas.parcel import (
    DeveloperAttribution,
    DeveloperAttributionResult,
    NearbyDeveloper,
)

logger = logging.getLogger(__name__)

# Резолвер участок→застройщик (миграция 149). Возвращает SETOF кандидатов, УЖЕ
# отсортированный лучшим матчем вперёд: exact_cadastral (dist=0) → spatial_150m →
# quarter, внутри метода — по distance_m ASC, затем ИНН. Caller (этот сервис) берёт
# топ-1 дедупнутый и дополняет nearby. CAST psycopg v3 — НИКОГДА :param::type
# (vault Pattern_CAST_AS_Type / backend.md).
_DEVELOPER_FOR_PARCEL_SQL = text(
    """
    SELECT
        developer_inn,
        developer_name,
        source,
        match_method,
        distance_m,
        project,
        status,
        obj_class,
        permits_total,
        domrf_objects_total,
        domrf_active_objects,
        in_both_sources
    FROM fn_developer_for_parcel(:cad_num, CAST(:radius AS double precision))
    """
)

# match_method, считающиеся «соседними» (вторичными) для nearby_developers. exact —
# это сам застройщик участка (primary), он в nearby не дублируется.
_NEARBY_METHODS = frozenset({"spatial_150m", "quarter"})


def _row_get(row: Any, key: str) -> Any:
    """Безопасно прочитать ключ из row-mapping → None если ключа нет.

    Только __getitem__: и SQLAlchemy RowMapping, и тестовые MagicMock-строки
    реализуют его корректно и кидают KeyError на отсутствующий ключ. (Не .get():
    на MagicMock это авто-атрибут-заглушка, возвращающая MagicMock вместо значения.)
    Зеркалит competitors._row_get — backward-safe чтение колонок из моков.
    """
    try:
        return row[key]
    except (KeyError, TypeError, IndexError):
        return None


def _as_int(value: Any) -> int | None:
    """None-safe int() — track-record-колонки могут быть NULL (FULL OUTER JOIN)."""
    return int(value) if value is not None else None


def _as_float(value: Any) -> float | None:
    """None-safe float() — distance_m NULL для quarter-ступени."""
    return float(value) if value is not None else None


def _attribution_fields(row: Any) -> dict[str, Any]:
    """Извлечь общий набор полей атрибуции из row резолвера (через _row_get).

    Используется и для primary (DeveloperAttribution), и для nearby (NearbyDeveloper)
    — у них идентичный набор полей. developer_inn — единственное обязательное (NOT NULL
    в резолвере, гарантирован WHERE c.inn IS NOT NULL); остальное nullable.
    """
    return {
        "developer_inn": str(_row_get(row, "developer_inn")),
        "developer_name": _row_get(row, "developer_name"),
        "source": _row_get(row, "source"),
        "match_method": _row_get(row, "match_method"),
        "distance_m": _as_float(_row_get(row, "distance_m")),
        "project": _row_get(row, "project"),
        "status": _row_get(row, "status"),
        "obj_class": _row_get(row, "obj_class"),
        "permits_total": _as_int(_row_get(row, "permits_total")),
        "domrf_objects_total": _as_int(_row_get(row, "domrf_objects_total")),
        "domrf_active_objects": _as_int(_row_get(row, "domrf_active_objects")),
        "in_both_sources": _row_get(row, "in_both_sources"),
    }


def get_developer_attribution(
    db: Session,
    cad_num: str,
    radius_m: float = 150.0,
) -> DeveloperAttributionResult | None:
    """Атрибуция застройщика участка: топ-1 + track-record + соседи (#1088).

    Шаги:
    1. Вызвать ``fn_developer_for_parcel(cad_num, radius_m)`` — SETOF кандидатов,
       уже отсортированный лучшим матчем вперёд (резолвер: exact > spatial > quarter).
    2. Дедупнуть по нормализованному developer_inn, сохраняя ЛУЧШИЙ (первый по
       ORDER BY) match_method каждого застройщика.
    3. Топ-1 дедупнутый → primary; остальные (match_method spatial/quarter) →
       nearby_developers.

    Args:
        db: SQLAlchemy-сессия.
        cad_num: кадастровый номер участка.
        radius_m: радиус spatial-ступени резолвера (default 150 м).

    Returns:
        ``DeveloperAttributionResult`` (primary + nearby_developers) либо None при:
          - пустом результате резолвера (нет матча застройщика);
          - отсутствии геометрии участка (spatial/quarter-domrf не сработали, а
            exact/quarter-permits ничего не нашли);
          - БД-ошибке, включая ещё не задеплоенную миграцию 149 (analyze не падает).
    """
    try:
        rows = (
            db.execute(
                _DEVELOPER_FOR_PARCEL_SQL,
                {"cad_num": cad_num, "radius": radius_m},
            )
            .mappings()
            .all()
        )
    except (OperationalError, ProgrammingError, DataError) as exc:
        # Миграция 149 ещё не задеплоена / БД-ошибка — graceful degrade.
        logger.warning(
            "get_developer_attribution: fn_developer_for_parcel недоступна для cad=%s, skip: %s",
            cad_num,
            exc,
        )
        return None
    except Exception as exc:
        logger.warning(
            "get_developer_attribution: неожиданная ошибка для cad=%s, skip: %s",
            cad_num,
            exc,
        )
        return None

    if not rows:
        return None

    # ── Дедуп по норм-ИНН, сохраняя ПЕРВЫЙ матч (он же лучший — резолвер ORDER BY) ──
    # rows УЖЕ отсортированы (exact > spatial > quarter, distance ASC), поэтому первое
    # вхождение ИНН — его лучший match_method. dict сохраняет порядок вставки (3.7+),
    # значит первый ключ — топ-1 застройщик участка.
    deduped: dict[str, Any] = {}
    for row in rows:
        inn = _row_get(row, "developer_inn")
        if inn is None:
            # Резолвер гарантирует NOT NULL (WHERE c.inn IS NOT NULL), но защищаемся.
            continue
        inn_key = str(inn)
        if inn_key not in deduped:
            deduped[inn_key] = row

    if not deduped:
        return None

    deduped_rows = list(deduped.values())
    primary_row = deduped_rows[0]
    primary = DeveloperAttribution(**_attribution_fields(primary_row))

    # ── nearby: остальные дедупнутые застройщики (вторичные spatial/quarter матчи) ──
    # primary уже исключён (срез [1:]). Доп. фильтр по _NEARBY_METHODS: если у соседа
    # как-то оказался exact (другой ИНН, multi-value РНС на тот же ЗУ) — он всё равно
    # «другой застройщик участка», но в nearby кладём только spatial/quarter-контекст,
    # чтобы семантика nearby = «строят рядом/в квартале» оставалась чистой.
    nearby: list[NearbyDeveloper] = []
    for row in deduped_rows[1:]:
        if _row_get(row, "match_method") not in _NEARBY_METHODS:
            continue
        nearby.append(NearbyDeveloper(**_attribution_fields(row)))

    return DeveloperAttributionResult(primary=primary, nearby_developers=nearby)


__all__ = ["get_developer_attribution"]