gendesign/backend/app/schemas/nspd_bulk.py

"""Pydantic schemas для NSPD bulk cadastre ingest (PR 2/5, issue #168).

Отдельный модуль от legacy NSPDFeature dataclass (nspd_client.py) —
там frozen dataclass для on-demand flow, тут Pydantic для bulk pipeline
с extra='allow' чтобы не потерять неизвестные поля → raw_props JSONB.
"""

from __future__ import annotations

from typing import Any

from pydantic import AliasChoices, BaseModel, ConfigDict, Field


class NSPDOptions(BaseModel):
    """Опции кадастрового объекта из NSPD.

    Все поля Optional — разные categoryId возвращают разные subsets.
    extra='allow' → любые незнакомые поля сохраняются (forward-compatible).
    """

    model_config = ConfigDict(extra="allow", populate_by_name=True)

    cad_num: str | None = None
    cad_number: str | None = None  # некоторые категории используют cad_number
    quarter_cad_number: str | None = None
    parent_cad_number: str | None = None
    area: float | None = None
    cost_value: float | None = None
    cost_index: float | None = None
    purpose: str | None = None
    floors: str | int | None = None
    year_built: str | int | None = None
    readable_address: str | None = None
    ownership_type: str | None = None
    right_type: str | None = None
    status: str | None = None
    land_record_area: float | None = None
    specified_area: float | None = None
    declared_area: float | None = None
    permitted_use_established_by_document: str | None = None
    category_type: str | None = None
    build_record_type: str | None = None
    build_record_type_value: str | None = None
    # NSPD geoportal search отдаёт objdoc/registers в camelCase (objdocId /
    # registersId), а bulk grid-walk иногда в snake_case. AliasChoices ловит оба
    # (verified live на 66:41:0106036:183 — search вернул objdocId=40995027,
    # snake-only парсинг давал None и ломал list_objects_in_building). objdoc_id —
    # ключ к tab-group objectsList (помещения/машино-места).
    objdoc_id: int | None = Field(
        default=None,
        validation_alias=AliasChoices("objdoc_id", "objdocId"),
    )
    registers_id: int | None = Field(
        default=None,
        validation_alias=AliasChoices("registers_id", "registersId"),
    )
    okato: str | None = None
    kladr: str | None = None

    @property
    def effective_cad_num(self) -> str | None:
        """Нормализованный кад. номер: cad_num или cad_number (разные категории)."""
        return self.cad_num or self.cad_number


class NSPDBulkFeature(BaseModel):
    """GeoJSON Feature из NSPD bulk endpoints.

    Pydantic-версия (не dataclass) — для удобной сериализации + extra='allow'.
    geometry — GeoJSON-dict в EPSG:3857 (transform на стороне Celery-таска).
    """

    model_config = ConfigDict(extra="allow", populate_by_name=True)

    id: int | str | None = None
    type: str = "Feature"
    geometry: dict[str, Any] | None = None
    properties: dict[str, Any] = {}

    @property
    def category_id(self) -> int | None:
        """category из properties (ключ для routing к target table).

        NSPD response shape: `properties.category` (НЕ `categoryId`).
        Field name был неверно interpretirovan в initial PR2 implementation
        based on HAR audit; в #177 поправили оба пути.

        Использует `is not None` (НЕ `or`) — иначе `category=0` truthy-trapped
        бы skipped (хотя observed NSPD IDs все positive, paranoia не вредна).
        int() wrapped в try/except — если NSPD когда-то вернёт string ID
        ("ЗУ"), мы не падаем посреди upsert_features loop.
        """
        val = self.properties.get("category")
        if val is None:
            val = self.properties.get("categoryId")
        if val is None:
            return None
        try:
            return int(val)
        except (TypeError, ValueError):
            return None

    @property
    def options(self) -> NSPDOptions:
        """Распарсить options из properties.options если есть, иначе пустой."""
        opts = self.properties.get("options") or {}
        return NSPDOptions.model_validate(opts)

    @property
    def cad_num(self) -> str | None:
        """Shortcut: cad_num из options (часто нужен в caller)."""
        return self.options.effective_cad_num


class QuarterSnapshot(BaseModel):
    """Результат search_by_quarter — snapshot данных для одного квартала.

    Первые 20 объектов на категорию из NSPD search endpoint.
    overflow_categories — категории требующие grid-walk (totalCount > 20).
    """

    model_config = ConfigDict(extra="allow")

    quarter: str
    fetched_at: str  # ISO datetime string
    features: list[NSPDBulkFeature] = []
    meta_counts: dict[int, int] = {}  # {categoryId: totalCount}

    @property
    def overflow_categories(self) -> list[int]:
        """Категории где totalCount > 20 — нужен grid-walk для полноты."""
        return [cat_id for cat_id, count in self.meta_counts.items() if count > 20]

    @property
    def total_meta_count(self) -> int:
        """Суммарное количество объектов по meta (до cap)."""
        return sum(self.meta_counts.values())


class ObjectsListing(BaseModel):
    """Результат list_objects_in_building — список помещений и машино-мест здания.

    Получается через /api/geoportal/v1/tab-group-data?tabClass=objectsList.
    Q3 deferred — код готов, но в bulk_harvest_quarter MVP не используется.
    """

    model_config = ConfigDict(extra="allow")

    objdoc_id: int
    flats_count: int = 0
    parking_count: int = 0
    flats_cad_nums: list[str] = []
    parking_cad_nums: list[str] = []

    @property
    def parking_ratio(self) -> float | None:
        """Обеспеченность паркингом: машино-места / помещения (#96).

        Сигнал класса/дефицита для recommend: бизнес обычно ≥1.0, эконом <0.5.
        None если помещений 0 (нельзя делить) — caller трактует как «нет данных»,
        НЕ как 0 (graceful, не выдумываем дефицит из пустоты). Округление до 0.01
        — детерминированно, без накопления float-шума при агрегации по соседям.
        """
        if self.flats_count <= 0:
            return None
        return round(self.parking_count / self.flats_count, 2)


__all__ = [
    "NSPDBulkFeature",
    "NSPDOptions",
    "ObjectsListing",
    "QuarterSnapshot",
]