backend/app/api/v1/parcels.py

@@ -63,6 +63,7 @@ from app.services.site_finder.competitors import get_competitors
 from app.services.site_finder.custom_pois import (
     get_overlaps_for_scoring as _get_custom_poi_overlaps,
 )
+from app.services.site_finder.developer_attribution import get_developer_attribution
 from app.services.site_finder.gate_verdict import compute_gate_verdict
 from app.services.site_finder.ird_analyze import build_ird_analyze_block
 from app.services.site_finder.poi_score import PoiScoreResponse, compute_poi_weighted_top7
@@ -2856,6 +2857,24 @@ def analyze_parcel(
                 exc_info=True,
             )
 
+    # Атрибуция застройщика (#1088 «GG-форсайт»): топ-1 застройщик участка + track-record +
+    # nearby_developers (fn_developer_for_parcel / developer_registry, миграция 149). БЕЗ флага
+    # (в отличие от ird-блока выше): чистый DB-резолвер по индексам (point-lookup ~0.06ms,
+    # резолвер ~30ms), без live-geoportal в hot-пути — низкий риск. get_developer_attribution
+    # сам graceful (None при пустом матче / нет геометрии / pre-migration), но обёрнут в
+    # try/except как остальные блоки — сбой не меняет успех/форму ответа (extra="allow"). Ключ
+    # добавляется ТОЛЬКО при наличии застройщика (None → ключ отсутствует, не null-шум).
+    try:
+        _dev_attr = get_developer_attribution(db, cad_num)
+        if _dev_attr is not None:
+            result_payload["developer_attribution"] = _dev_attr.model_dump()
+    except Exception:
+        logger.warning(
+            "developer_attribution block failed for cad=%s — analyze response unaffected",
+            cad_num,
+            exc_info=True,
+        )
+
     return result_payload
 
 

backend/app/schemas/parcel.py

@@ -225,6 +225,73 @@ class CompetitorsResponse(BaseModel):
     summary: CompetitorsSummary
 
 
+# ── Developer attribution (#1088 «GG-форсайт») ─────────────────────────────────
+# Детерминированная атрибуция «пятно земли ↔ застройщик»: канон-застройщик участка
+# + его track-record (РНС/РВЭ + домрф ЖК/квартиры) из fn_developer_for_parcel /
+# developer_registry (миграция 149). Источник истины формы — fn_developer_for_parcel
+# SETOF + developer_registry track-record; схемы зеркалят его поля 1:1.
+
+
+class DeveloperAttribution(BaseModel):
+    """Топ-1 застройщик участка + его track-record (#1088).
+
+    developer_inn — нормализованный ИНН (10/12-знач), стабильный кросс-источниковый
+    ключ. match_method — лучшая сработавшая ступень резолвера
+    (exact_cadastral > spatial_150m > quarter). track-record-поля денормализованы
+    из developer_registry по ИНН (могут быть None для застройщика, известного только
+    одной стороне). project/status/obj_class — из конкретного матча (РНС или домрф-ЖК).
+    """
+
+    developer_inn: str
+    developer_name: str | None
+    source: str | None  # 'permits' | 'domrf'
+    match_method: str | None  # 'exact_cadastral' | 'spatial_150m' | 'quarter'
+    distance_m: float | None
+    project: str | None
+    status: str | None
+    obj_class: str | None
+    # track-record (developer_registry, по норм-ИНН)
+    permits_total: int | None
+    domrf_objects_total: int | None
+    domrf_active_objects: int | None
+    in_both_sources: bool | None
+
+
+class NearbyDeveloper(BaseModel):
+    """Соседний застройщик участка (дедупнутый, match_method spatial_150m/quarter).
+
+    Контекст «кто ещё строит рядом / в квартале». Тот же набор полей, что primary,
+    но без exact-кадастра (это всегда вторичный матч). Дедупнут по developer_inn —
+    один застройщик в nearby представлен своим лучшим (первым по ORDER BY) матчем.
+    """
+
+    developer_inn: str
+    developer_name: str | None
+    source: str | None
+    match_method: str | None
+    distance_m: float | None
+    project: str | None
+    status: str | None
+    obj_class: str | None
+    permits_total: int | None
+    domrf_objects_total: int | None
+    domrf_active_objects: int | None
+    in_both_sources: bool | None
+
+
+class DeveloperAttributionResult(BaseModel):
+    """Результат атрибуции застройщика участка (#1088).
+
+    primary — лучший матч (топ-1 по ORDER BY резолвера: exact dist=0 → spatial →
+    quarter). nearby_developers — остальные дедупнутые застройщики (spatial/quarter),
+    «кто ещё строит рядом». Сервис возвращает None при пустом результате (нет матча /
+    нет геометрии / миграция не применена) — analyze получает ключ только при наличии.
+    """
+
+    primary: DeveloperAttribution
+    nearby_developers: list[NearbyDeveloper]
+
+
 # ── Layout analysis (Issue #113) ───────────────────────────────────────────
 
 
@@ -570,6 +637,11 @@ class AnalyzeResponse(BaseModel):
     #   functional_zone/krt (геопортал WFS) + zone_regulation (C8b). Отсутствует при флаге off —
     ird: dict[str, Any] | None = None
 
+    # — Атрибуция застройщика (#1088 «GG-форсайт»): топ-1 застройщик участка + track-record +
+    #   nearby_developers (fn_developer_for_parcel / developer_registry, миграция 149). Additive,
+    #   без флага (чистый DB-lookup, graceful). Отсутствует при пустом матче / pre-migration —
+    developer_attribution: dict[str, Any] | None = None
+
     # — 202 fetch-stub поля (#93): present ТОЛЬКО на 202-ответе, иначе отсутствуют —
     status: str | None = None
     job_id: int | None = None

backend/app/services/site_finder/developer_attribution.py

@@ -0,0 +1,207 @@
+"""Детерминированная атрибуция застройщика для 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"]

backend/app/workers/beat_schedule.py

@@ -264,6 +264,19 @@ def build_beat_schedule() -> dict:
         "options": {"queue": "celery"},
     }
 
+    # Developer registry refresh (#1088 «GG-форсайт») — ежемесячно 1-го числа в 05:30 МСК,
+    # через 30 мин ПОСЛЕ ekburg-permits-monthly (05:00), чтобы permits-сторона реестра
+    # обновлялась на свежих РНС/РВЭ. fn_refresh_developer_registry() → REFRESH MATERIALIZED
+    # VIEW CONCURRENTLY developer_registry (миграция 149; не блокирует читателей analyze).
+    # Celery conf.timezone=Europe/Moscow → crontab трактуется в МСК (#1233). Техническая
+    # infra-задача (не в job_settings, как refresh-quarter-price-index). domrf-сторона
+    # обновляется ежедневно (scrape_kn), но ИНН/track-record меняются редко — monthly хватает.
+    schedule["developer-registry-refresh-monthly"] = {
+        "task": "tasks.developer_registry_refresh.refresh_developer_registry",
+        "schedule": _parse_cron("30 5 1 * *"),  # 05:30 MSK, 1st of month
+        "options": {"queue": "celery"},
+    }
+
     # ПЗЗ территориальных зон ЕКБ из PKK6 ArcGIS — ежемесячно 1-го числа в 03:00 МСК.
     # Celery conf.timezone=Europe/Moscow → crontab трактуется в МСК (#1233).
     # PKK6 нестабилен под нагрузкой, но данные ПЗЗ меняются редко — раз в месяц достаточно.

backend/app/workers/celery_app.py

@@ -78,6 +78,7 @@ celery_app = Celery(
         "app.workers.tasks.okn_objects_sync",
         "app.workers.tasks.pat_subzones_load",
         "app.workers.tasks.izyatie_ocr_ingest",
+        "app.workers.tasks.developer_registry_refresh",
     ],
 )
 celery_app.conf.timezone = "Europe/Moscow"

backend/app/workers/tasks/developer_registry_refresh.py

@@ -0,0 +1,58 @@
+"""Celery task: refresh developer_registry materialized view (#1088 «GG-форсайт»).
+
+Вызывает ``fn_refresh_developer_registry()`` (миграция 149,
+``data/sql/149_developer_attribution.sql``) — REFRESH MATERIALIZED VIEW CONCURRENTLY
+``developer_registry`` (реестр застройщиков по норм-ИНН: канон-имя + track-record
+permits+domrf). CONCURRENTLY не блокирует SELECT-читателей (analyze) во время refresh.
+
+Scheduled via hardcoded beat entry в workers/beat_schedule.py:
+  'developer-registry-refresh-monthly' — ежемесячно 1-го числа в 05:30 МСК,
+  через 30 мин после 'ekburg-permits-monthly' (05:00 МСК), чтобы permits-сторона
+  реестра обновлялась на свежих РНС/РВЭ. domrf-сторона обновляется ежедневно
+  (scrape_kn), но ИНН/track-record застройщиков меняются редко — monthly refresh
+  реестра с запасом достаточно (резолвер всё равно читает domrf-quarter напрямую).
+
+Issue: #1088.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from sqlalchemy import text
+
+from app.core.db import SessionLocal
+from app.workers.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(
+    bind=True,
+    name="tasks.developer_registry_refresh.refresh_developer_registry",
+    max_retries=2,
+)
+def refresh_developer_registry(self: Any) -> dict[str, Any]:
+    """REFRESH MATERIALIZED VIEW CONCURRENTLY developer_registry.
+
+    Лёгкая задача (реестр ~1024 застройщика). CONCURRENTLY — non-blocking для
+    читателей analyze; требует unique-индекс uq_developer_registry_inn (создан
+    миграцией 149). Возвращает row-count реестра после refresh для observability.
+
+    Returns:
+        dict для Celery result store / логирования.
+    """
+    db = SessionLocal()
+    try:
+        db.execute(text("SELECT fn_refresh_developer_registry()"))
+        db.commit()
+        row = db.execute(text("SELECT COUNT(*) FROM developer_registry")).first()
+        count = int(row[0]) if row else 0
+        logger.info("refresh_developer_registry: completed, developers=%d", count)
+        return {"status": "ok", "developer_registry_rows": count}
+    except Exception as e:
+        logger.exception("refresh_developer_registry failed: %s", e)
+        raise
+    finally:
+        db.close()

backend/tests/services/site_finder/test_developer_attribution.py

@@ -0,0 +1,291 @@
+"""Unit-тесты атрибуции застройщика участка (#1088 «GG-форсайт»).
+
+Тестируем get_developer_attribution с MagicMock-сессией (без живой БД, без app.main):
+  • happy-path: rows резолвера → primary (топ-1) + nearby_developers;
+  • дедуп по developer_inn: один застройщик в exact+spatial+quarter → один в выводе,
+    сохраняется ЛУЧШИЙ (первый по ORDER BY резолвера) match_method;
+  • nearby содержит только spatial_150m/quarter (не exact, не дубль primary);
+  • пустой результат резолвера → None;
+  • graceful-degrade при БД-ошибке (OperationalError/ProgrammingError/DataError —
+    incl. ещё не задеплоенная миграция 149) → None, analyze не падает;
+  • SQL использует CAST(:x AS type), НИКОГДА :x::type (psycopg v3).
+
+Детерминированно, без LLM. rows мокаются как dict (RowMapping-совместимо через
+__getitem__, как реальный .mappings().all()).
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+from sqlalchemy.exc import DataError, OperationalError, ProgrammingError
+
+from app.services.site_finder.developer_attribution import (
+    _DEVELOPER_FOR_PARCEL_SQL,
+    get_developer_attribution,
+)
+
+# ── helpers ────────────────────────────────────────────────────────────────────
+
+
+def _mock_db(rows: list[dict[str, Any]]) -> MagicMock:
+    """Сессия с одним execute → .mappings().all() == rows (зеркало test_future_supply)."""
+    db = MagicMock()
+    result = MagicMock()
+    result.mappings.return_value.all.return_value = rows
+    db.execute.return_value = result
+    return db
+
+
+def _row(
+    inn: str,
+    *,
+    name: str | None = "Канон Девелопмент",
+    source: str = "permits",
+    match_method: str = "exact_cadastral",
+    distance_m: float | None = 0.0,
+    project: str | None = "ЖК Тест",
+    status: str | None = "RNS",
+    obj_class: str | None = "Комфорт",
+    permits_total: int | None = 5,
+    domrf_objects_total: int | None = 3,
+    domrf_active_objects: int | None = 2,
+    in_both_sources: bool | None = True,
+) -> dict[str, Any]:
+    """Одна строка резолвера fn_developer_for_parcel (полный набор колонок)."""
+    return {
+        "developer_inn": inn,
+        "developer_name": name,
+        "source": source,
+        "match_method": match_method,
+        "distance_m": distance_m,
+        "project": project,
+        "status": status,
+        "obj_class": obj_class,
+        "permits_total": permits_total,
+        "domrf_objects_total": domrf_objects_total,
+        "domrf_active_objects": domrf_active_objects,
+        "in_both_sources": in_both_sources,
+    }
+
+
+# ── SQL-конвенция (psycopg v3) ──────────────────────────────────────────────────
+
+
+def test_sql_uses_cast_not_double_colon() -> None:
+    """psycopg v3: запрещён :name::type (SQLAlchemy 2.0 молча роняет его → SyntaxError)."""
+    sql = str(_DEVELOPER_FOR_PARCEL_SQL)
+    assert "::" not in sql, f"double-colon cast найден в SQL: {sql!r}"
+    assert "CAST(:radius AS double precision)" in sql
+
+
+# ── happy-path ───────────────────────────────────────────────────────────────
+
+
+class TestPrimaryAttribution:
+    def test_single_developer_becomes_primary(self) -> None:
+        db = _mock_db([_row("6671138299", name="ЕКБ Девелопмент")])
+        result = get_developer_attribution(db, "66:41:0303161:123")
+
+        assert result is not None
+        assert result.primary.developer_inn == "6671138299"
+        assert result.primary.developer_name == "ЕКБ Девелопмент"
+        assert result.primary.match_method == "exact_cadastral"
+        assert result.primary.distance_m == 0.0
+        assert result.primary.permits_total == 5
+        assert result.primary.in_both_sources is True
+        assert result.nearby_developers == []
+
+    def test_radius_passed_to_resolver(self) -> None:
+        db = _mock_db([_row("6671138299")])
+        get_developer_attribution(db, "66:41:0303161:123", radius_m=250.0)
+
+        _args, _kwargs = db.execute.call_args
+        params = _args[1]
+        assert params == {"cad_num": "66:41:0303161:123", "radius": 250.0}
+
+    def test_default_radius_is_150(self) -> None:
+        db = _mock_db([_row("6671138299")])
+        get_developer_attribution(db, "66:41:0303161:123")
+        params = db.execute.call_args[0][1]
+        assert params["radius"] == 150.0
+
+    def test_first_row_is_primary_resolver_order_trusted(self) -> None:
+        # Резолвер уже ORDER BY method (exact > spatial > quarter): первая строка = топ-1.
+        rows = [
+            _row("1111111111", match_method="exact_cadastral", distance_m=0.0),
+            _row("2222222222", match_method="spatial_150m", distance_m=42.0),
+        ]
+        result = get_developer_attribution(db := _mock_db(rows), "66:41:0303161:1")
+        assert db is not None
+        assert result is not None
+        assert result.primary.developer_inn == "1111111111"
+        assert result.primary.match_method == "exact_cadastral"
+
+
+# ── дедуп по developer_inn ───────────────────────────────────────────────────
+
+
+class TestDedup:
+    def test_same_inn_across_stages_dedups_keeping_best_method(self) -> None:
+        # Один застройщик встречается в exact + spatial + quarter (резолвер не
+        # взаимоисключающий). Дедуп оставляет ПЕРВЫЙ (лучший по ORDER BY) — exact.
+        rows = [
+            _row("6671138299", match_method="exact_cadastral", distance_m=0.0, project="ЖК A"),
+            _row("6671138299", match_method="spatial_150m", distance_m=80.0, project="ЖК A-2"),
+            _row("6671138299", match_method="quarter", distance_m=None, project="ЖК A-3"),
+        ]
+        result = get_developer_attribution(_mock_db(rows), "66:41:0303161:1")
+
+        assert result is not None
+        assert result.primary.developer_inn == "6671138299"
+        # ЛУЧШИЙ метод сохранён (первый по ORDER BY), не перезаписан spatial/quarter.
+        assert result.primary.match_method == "exact_cadastral"
+        assert result.primary.project == "ЖК A"
+        # Тот же ИНН не попадает в nearby (дедупнут целиком).
+        assert result.nearby_developers == []
+
+    def test_distinct_developers_split_primary_and_nearby(self) -> None:
+        rows = [
+            _row("1111111111", match_method="exact_cadastral", distance_m=0.0),
+            _row("2222222222", match_method="spatial_150m", distance_m=55.0),
+            _row("3333333333", match_method="quarter", distance_m=None),
+        ]
+        result = get_developer_attribution(_mock_db(rows), "66:41:0303161:1")
+
+        assert result is not None
+        assert result.primary.developer_inn == "1111111111"
+        nearby_inns = [d.developer_inn for d in result.nearby_developers]
+        assert nearby_inns == ["2222222222", "3333333333"]
+
+    def test_nearby_developer_dedups_too(self) -> None:
+        # Сосед, дважды встретившийся (spatial + quarter), в nearby один раз (лучший).
+        rows = [
+            _row("1111111111", match_method="exact_cadastral", distance_m=0.0),
+            _row("2222222222", match_method="spatial_150m", distance_m=55.0),
+            _row("2222222222", match_method="quarter", distance_m=None),
+        ]
+        result = get_developer_attribution(_mock_db(rows), "66:41:0303161:1")
+
+        assert result is not None
+        assert [d.developer_inn for d in result.nearby_developers] == ["2222222222"]
+        assert result.nearby_developers[0].match_method == "spatial_150m"
+
+
+class TestNearbyFiltering:
+    def test_nearby_excludes_secondary_exact_method(self) -> None:
+        # Второй (другой ИНН) с exact (multi-value РНС на тот же ЗУ) — не primary, но
+        # в nearby кладём только spatial/quarter-контекст («строят рядом/в квартале»).
+        rows = [
+            _row("1111111111", match_method="exact_cadastral", distance_m=0.0),
+            _row("2222222222", match_method="exact_cadastral", distance_m=0.0),
+            _row("3333333333", match_method="spatial_150m", distance_m=60.0),
+        ]
+        result = get_developer_attribution(_mock_db(rows), "66:41:0303161:1")
+
+        assert result is not None
+        assert result.primary.developer_inn == "1111111111"
+        # Вторичный exact (2222) отфильтрован; остаётся только spatial-сосед (3333).
+        assert [d.developer_inn for d in result.nearby_developers] == ["3333333333"]
+
+    def test_nearby_includes_both_spatial_and_quarter(self) -> None:
+        rows = [
+            _row("1111111111", match_method="exact_cadastral", distance_m=0.0),
+            _row("2222222222", match_method="spatial_150m", distance_m=30.0),
+            _row("3333333333", match_method="quarter", distance_m=None, source="domrf"),
+        ]
+        result = get_developer_attribution(_mock_db(rows), "66:41:0303161:1")
+
+        assert result is not None
+        methods = [d.match_method for d in result.nearby_developers]
+        assert methods == ["spatial_150m", "quarter"]
+
+
+# ── nullable track-record (FULL OUTER JOIN → None-колонки) ───────────────────
+
+
+class TestNullableFields:
+    def test_developer_known_only_to_one_source(self) -> None:
+        # Застройщик из domrf без РНС: permits_total может быть 0, distance_m None.
+        row = _row(
+            "7777777777",
+            source="domrf",
+            match_method="quarter",
+            distance_m=None,
+            status="Строящиеся",
+            permits_total=0,
+            in_both_sources=False,
+        )
+        result = get_developer_attribution(_mock_db([row]), "66:41:0303161:1")
+
+        assert result is not None
+        assert result.primary.distance_m is None
+        assert result.primary.permits_total == 0
+        assert result.primary.in_both_sources is False
+
+    def test_null_track_record_columns_pass_through(self) -> None:
+        row = _row(
+            "8888888888",
+            name=None,
+            domrf_objects_total=None,
+            domrf_active_objects=None,
+            in_both_sources=None,
+        )
+        result = get_developer_attribution(_mock_db([row]), "66:41:0303161:1")
+
+        assert result is not None
+        assert result.primary.developer_name is None
+        assert result.primary.domrf_objects_total is None
+        assert result.primary.in_both_sources is None
+
+
+# ── пустой результат → None ──────────────────────────────────────────────────
+
+
+def test_empty_resolver_result_returns_none() -> None:
+    result = get_developer_attribution(_mock_db([]), "66:41:0303161:1")
+    assert result is None
+
+
+def test_all_rows_null_inn_returns_none() -> None:
+    # Резолвер гарантирует NOT NULL inn (WHERE c.inn IS NOT NULL), но защищаемся:
+    # все строки с NULL-ИНН → нечего атрибутировать → None.
+    rows = [_row("ignored")]
+    rows[0]["developer_inn"] = None
+    result = get_developer_attribution(_mock_db(rows), "66:41:0303161:1")
+    assert result is None
+
+
+# ── graceful-degrade при БД-ошибке ───────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [
+        OperationalError("stmt", {}, Exception("connection lost")),
+        ProgrammingError("stmt", {}, Exception("function fn_developer_for_parcel does not exist")),
+        DataError("stmt", {}, Exception("invalid input syntax")),
+    ],
+    ids=["operational", "programming_pre_migration", "data"],
+)
+def test_db_error_degrades_to_none(exc: Exception) -> None:
+    # Миграция 149 не задеплоена (ProgrammingError) / коннект упал (OperationalError) /
+    # битый ввод (DataError) → None, analyze не падает.
+    db = MagicMock()
+    db.execute.side_effect = exc
+    result = get_developer_attribution(db, "66:41:0303161:1")
+    assert result is None
+
+
+def test_unexpected_error_also_degrades_to_none() -> None:
+    # Любая иная ошибка (не из перечисленных) тоже не валит analyze — None.
+    db = MagicMock()
+    db.execute.side_effect = RuntimeError("boom")
+    result = get_developer_attribution(db, "66:41:0303161:1")
+    assert result is None

data/sql/149_developer_attribution.sql

@@ -0,0 +1,477 @@
+-- 149_developer_attribution.sql
+-- #1088 (GG-форсайт) — детерминированный geo-join «пятно земли ↔ застройщик» для
+-- Site Finder analyze/competitors. Связывает РНС-сторону (ekburg_construction_permits)
+-- с домрф-стороной (domrf_kn_objects) через НОРМАЛИЗОВАННЫЙ ИНН — единственный
+-- стабильный кросс-источниковый ключ застройщика — и резолвит участок → застройщик
+-- (имя+ИНН), его проект/статус, и track-record застройщика по району/кварталу.
+--
+-- ── ЗАЧЕМ ────────────────────────────────────────────────────────────────────────
+-- До #1088 analyze/competitors «слепы» к РНС: domrf_kn_objects.dev_inn даёт застройщика
+-- для домрф-объектов, но НЕТ связки с РНС-разрешениями и НЕТ канон-имени застройщика
+-- (одно ИНН пишется по-разному: prod — 163 ИНН в permits имеют >1 написания имени;
+-- «ИП Хабибуллин О.В.» vs «ИП Хабибуллин Олег Вахалиевич»). Этот слой даёт:
+--   1. fn_normalize_inn      — детерминированная нормализация ИНН (10/12-знач, иначе NULL)
+--   2. developer_registry    — MATERIALIZED реестр застройщиков по норм-ИНН: канон-имя
+--      + track-record (refresh через fn_refresh_developer_registry, unique-idx по ИНН)
+--   3. developer_attribution — алиас-VIEW с denormalized полями для analyze-блока
+--   4. fn_developer_for_parcel(cad_num) — резолвер участок→застройщик (3-ступенчатый
+--      fallback: точный кадастр → spatial 150 м → кадастровый квартал)
+--
+-- ── PROD-ФАКТЫ (подтверждено EXPLAIN/COUNT, 2026-06-13) ───────────────────────────
+-- ekburg_construction_permits: 4059 строк; developer_inn TEXT not-null=3564 (87.8%),
+--   все pure-digit; 3401×10-знач (юрлица), 154×12-знач (ИП), 9×11-знач (СЛИПШИЙСЯ
+--   мусор → норм. отбрасывает в NULL). cadastral_number not-null=4054 (99.9%); 3439
+--   full-cad, 57 multi-value (запятая/«;» — несколько ЗУ в одной РНС), 2 quarter-only.
+--   geom GEOMETRY(Point,4326) not-null=3994 (98.4%), GIST gist_ekburg_permits_geom.
+--   НЕТ btree на cadastral_number → exact-lookup = Seq Scan 33мс (фикс ниже).
+-- domrf_kn_objects: 10699 строк = 1543 distinct obj_id × снапшоты (2026-04-27..06-10),
+--   ВСЕ region_cd=66. dev_inn TEXT, на latest snapshot not-null=1543 (100%), все
+--   10-знач. ИНН СТАБИЛЕН между снапшотами (0 объектов меняют ИНН). cad_quarter —
+--   формат «66:41:NNNNNNN», заполнен на 1131/1543 объектах (73%), НА LATEST snapshot
+--   часто NULL → берём с ЛЮБОГО снапшота где NOT NULL (тоже стабилен, 0 изменений).
+--   cad_building_num — ВЕЗДЕ NULL (не используем). НЕТ колонки geom (только
+--   latitude/longitude NUMERIC, у всех 1543) → domrf spatial идёт через MakePoint.
+--   district_name заполнен на 1315/1543 (85%), 8 админ-районов ЕКБ.
+-- Кросс-матч норм-ИНН: 888 distinct в permits, 428 в domrf, СОВПАДАЮТ 292 (68.2%
+--   застройщиков domrf имеют РНС в ЕКБ). 1740 permit-строк (42.9%) принадлежат
+--   застройщику из domrf; 1240 domrf-объектов (80.4%) — застройщику из permits.
+--
+-- ── ИНН-НОРМАЛИЗАЦИЯ (детерминированная, edge-cases) ──────────────────────────────
+-- fn_normalize_inn: btrim → если результат matches '^[0-9]{10}$' ИЛИ '^[0-9]{12}$'
+-- → вернуть как есть; ИНАЧЕ NULL. Покрывает edge-cases:
+--   NULL / '' / '0'         → NULL  (495 permit-строк, 0 domrf)
+--   11-знач '66711382990'   → NULL  (9 permit-строк — два слипшихся ИНН/ИП; чинить
+--                                    эвристикой НЕ беремся — детерминированно мусор)
+--   ' 6671138299 ' (пробелы)→ '6671138299'  (btrim)
+--   буквы/дефисы/запятые    → NULL  (не pure-digit)
+-- Ведущие нули НЕ режем: российский ИНН — фикс-длины строка-идентификатор, '00…' —
+-- значимые цифры (срезка сломала бы матч). IMMUTABLE — годен для functional index.
+--
+-- ── КАНОН-ИМЯ (dedupe, детерминированный) ─────────────────────────────────────────
+-- Один ИНН → много написаний. Канон выбираем так (ORDER BY, без LLM/рандома):
+--   1) приоритет источника: domrf-имя > permits-имя (domrf — юрлица из реестра дом.рф,
+--      чище и полнее, чем ручной ввод РНС);
+--   2) внутри источника — самое ЧАСТОЕ написание (по числу строк);
+--   3) tie-break — лексикографически наименьшее (btrim'нутое) → стабильно при ре-apply.
+--
+-- ── РЕЗОЛВЕР УЧАСТОК→ЗАСТРОЙЩИК (3 ступени, детерминированный приоритет) ───────────
+-- fn_developer_for_parcel(cad_num) возвращает атрибуцию с match_method:
+--   'exact_cadastral' — permits.cadastral_number = cad_num (точное, max precision);
+--   'spatial_150m'    — permit.geom в 150 м от geom участка (GIST envelope + geography
+--                        refine); берём ближайший permit;
+--   'quarter'         — по кадастровому кварталу: permits и domrf того же квартала.
+-- Ступени НЕ взаимоисключающи на уровне данных, но метод проставляется по ЛУЧШЕЙ
+-- сработавшей ступени (exact > spatial > quarter). Возвращаем ВСЕХ кандидатов с
+-- match_method и distance_m — caller (backend) решает, брать топ-1 или ранжировать.
+--
+-- ── SPATIAL GIST-TRAP (важно для перфа) ───────────────────────────────────────────
+-- ST_DWithin(geom::geography, …) НЕ использует GIST (индекс на geometry, не geography)
+-- → Seq Scan 91мс. Правильно: ST_DWithin(geom, geom, degrees) для GIST-envelope +
+-- ST_DWithin(::geography, ::geography, meters) для точной метрики. EXPLAIN: 91→12мс,
+-- 1689→132 буфера, Index Scan using gist_ekburg_permits_geom. 0.0025° envelope на
+-- широте ЕКБ (~56.8°N) безопасно перекрывает 150 м (1°lat≈111 км, 1°lon≈61 км).
+--
+-- ── DEPLOY ────────────────────────────────────────────────────────────────────────
+-- Auto-applied deploy.yml через _schema_migrations (ровно один раз, NN=149).
+-- Idempotent: CREATE OR REPLACE FUNCTION/VIEW + CREATE INDEX IF NOT EXISTS +
+-- DROP MATERIALIZED VIEW IF EXISTS ... CASCADE перед CREATE MATERIALIZED VIEW (matview
+-- не поддерживает CREATE OR REPLACE). CREATE MATERIALIZED VIEW наполняется данными
+-- СРАЗУ (без WITH NO DATA) → реестр готов к чтению по завершении миграции; первый
+-- плановый fn_refresh_developer_registry() просто обновит его при след. sync.
+-- Migration order: эта SQL-миграция ПЕРВОЙ; backend-сервис
+-- (app/services/site_finder/developer_attribution.py — рекомендация в отчёте) — следом.
+-- ЧИСТО ADDITIVE: ни одна существующая таблица/вью/колонка не меняется и не дропается.
+
+BEGIN;
+
+-- ──────────────────────────────────────────────────────────────────────────────────
+-- 1. fn_normalize_inn — детерминированная нормализация ИНН
+-- ──────────────────────────────────────────────────────────────────────────────────
+-- btrim + валидация длины/цифр. 10 (юрлицо) или 12 (физлицо/ИП) → как есть, иначе NULL.
+-- IMMUTABLE STRICT: STRICT → NULL-вход даёт NULL-выход без вызова тела; IMMUTABLE →
+-- можно использовать в functional index и планировщик кэширует результат.
+CREATE OR REPLACE FUNCTION fn_normalize_inn(raw text)
+RETURNS text
+LANGUAGE sql
+IMMUTABLE
+STRICT
+AS $$
+    SELECT CASE
+        WHEN btrim(raw) ~ '^[0-9]{10}$' OR btrim(raw) ~ '^[0-9]{12}$'
+        THEN btrim(raw)
+        ELSE NULL
+    END;
+$$;
+
+COMMENT ON FUNCTION fn_normalize_inn(text) IS
+    '#1088: детерминированная нормализация ИНН → 10/12-значная pure-digit строка, '
+    'иначе NULL. NULL/пусто/0/11-знач-мусор/буквы → NULL. Ведущие нули сохраняются '
+    '(фикс-длины идентификатор). IMMUTABLE — годна для functional index.';
+
+-- ──────────────────────────────────────────────────────────────────────────────────
+-- 2. Индексы для горячего пути резолвера (закрывают Seq Scan / ускоряют join по ИНН)
+-- ──────────────────────────────────────────────────────────────────────────────────
+-- 2a. btree на cadastral_number — exact-lookup (= cad_num) делал Seq Scan (4059 строк,
+--     33мс). С индексом — Bitmap Index Scan (0.09мс). Equality collation-safe.
+CREATE INDEX IF NOT EXISTS idx_ekburg_permits_cadastral
+    ON ekburg_construction_permits (cadastral_number);
+
+-- 2a-bis. text_pattern_ops btree — для quarter-ступени LIKE 'QUARTER:%'. БД collation
+--     en_US.utf8 (НЕ C) → обычный btree НЕ обслуживает ни LIKE-prefix, ни range
+--     ['X:','X;') (collation сортирует ':'/';'/цифры не по байтам → range вернул 0
+--     строк, см. dry-run-баг). text_pattern_ops индексирует по C-семантике байтов →
+--     LIKE 'prefix%' даёт Bitmap Index Scan (0.74мс vs Seq Scan 6мс от regex '^').
+CREATE INDEX IF NOT EXISTS idx_ekburg_permits_cad_cpat
+    ON ekburg_construction_permits (cadastral_number text_pattern_ops);
+
+-- 2b. functional index на нормализованный ИНН — ускоряет join permits⋈domrf по
+--     fn_normalize_inn(developer_inn). Partial (WHERE norm IS NOT NULL) — мусорные
+--     ИНН (NULL после норм.) в индекс не попадают, держим его компактным.
+CREATE INDEX IF NOT EXISTS idx_ekburg_permits_inn_norm
+    ON ekburg_construction_permits (fn_normalize_inn(developer_inn))
+    WHERE fn_normalize_inn(developer_inn) IS NOT NULL;
+
+-- 2c. functional index на норм-ИНН для domrf (join вторая сторона). domrf
+--     версионируется по snapshot_date → индекс по (norm_inn) хватает для матча
+--     застройщика (ИНН стабилен между снапшотами).
+CREATE INDEX IF NOT EXISTS idx_domrf_kn_objects_inn_norm
+    ON domrf_kn_objects (fn_normalize_inn(dev_inn))
+    WHERE fn_normalize_inn(dev_inn) IS NOT NULL;
+
+-- ──────────────────────────────────────────────────────────────────────────────────
+-- 3. developer_registry — реестр застройщиков по нормализованному ИНН + track-record
+-- ──────────────────────────────────────────────────────────────────────────────────
+-- Канон-имя застройщика (dedupe вариантов) + агрегаты по ОБЕИМ сторонам:
+--   permits-сторона: число РНС/РВЭ, суммарные площади, диапазон лет, набор кварталов;
+--   domrf-сторона:   число ЖК (latest snapshot per obj_id), суммарно квартир/в продаже,
+--                    набор районов, набор классов.
+--
+-- MATERIALIZED (не plain VIEW): как VIEW point-lookup по ИНН стоил 52мс, а резолвер
+-- 142-184мс — потому что план КАЖДЫЙ РАЗ пересчитывал весь реестр (DISTINCT ON 10699
+-- domrf-строк + два FULL JOIN + dedupe-имя) ради джойна track-record к ~14-460
+-- кандидатам участка. Реестр мал (1024 застройщика) и меняется РЕДКО (permits —
+-- monthly sync #105; domrf — снапшоты ~раз в сутки), поэтому материализуем: резолвер
+-- и прямой lookup читают готовые 1024 строки через unique-индекс за микросекунды.
+-- Refresh — fn_refresh_developer_registry() по расписанию (см. ниже + рекомендация
+-- backend). DROP ... CASCADE для идемпотентности (matview не поддерживает CREATE OR
+-- REPLACE): уронит зависимый developer_attribution VIEW — он пересоздаётся ниже в §4.
+DROP MATERIALIZED VIEW IF EXISTS developer_registry CASCADE;
+CREATE MATERIALIZED VIEW developer_registry AS
+WITH
+-- 3.1 domrf latest snapshot per obj_id (ИНН/имя стабильны между снапшотами).
+--      Только поля, нужные registry-агрегатам (canonical_name + domrf_agg). cad_quarter
+--      здесь НЕ тянем (registry агрегирует по району/классу, не по кварталу — а резолвер
+--      quarter_domrf читает domrf_kn_objects напрямую). Это снимает O(N) коррелированный
+--      subplan (был 93k буферов / 10699 loops в EXPLAIN) — один проход DISTINCT ON.
+domrf_latest AS (
+    SELECT DISTINCT ON (o.obj_id)
+        o.obj_id,
+        fn_normalize_inn(o.dev_inn)                              AS inn,
+        btrim(o.dev_name)                                        AS dev_name,
+        o.obj_class,
+        o.site_status,
+        o.flat_count,
+        o.free_flats,
+        o.district_name
+    FROM domrf_kn_objects o
+    WHERE fn_normalize_inn(o.dev_inn) IS NOT NULL
+    ORDER BY o.obj_id, o.snapshot_date DESC
+),
+-- 3.2 Канон-имя: domrf-имя приоритетнее permits-имени; внутри — самое частое;
+--      tie → min lexical. Собираем кандидаты обеих сторон с приоритетом источника.
+name_candidates AS (
+    SELECT inn, dev_name AS name, 1 AS source_priority   -- domrf > permits
+    FROM domrf_latest
+    WHERE dev_name IS NOT NULL AND dev_name <> ''
+    UNION ALL
+    SELECT fn_normalize_inn(developer_inn) AS inn, btrim(developer_name) AS name, 2 AS source_priority
+    FROM ekburg_construction_permits
+    WHERE fn_normalize_inn(developer_inn) IS NOT NULL
+      AND developer_name IS NOT NULL AND btrim(developer_name) <> ''
+),
+canonical_name AS (
+    SELECT DISTINCT ON (inn)
+        inn,
+        name AS developer_name
+    FROM (
+        SELECT inn, name, source_priority, count(*) AS freq
+        FROM name_candidates
+        GROUP BY inn, name, source_priority
+    ) ranked
+    ORDER BY inn, source_priority ASC, freq DESC, name ASC
+),
+-- 3.3 permits track-record по застройщику (норм-ИНН)
+permits_agg AS (
+    SELECT
+        fn_normalize_inn(developer_inn)                                       AS inn,
+        count(*)                                                              AS permits_total,
+        count(*) FILTER (WHERE permit_type = 'RNS')                           AS permits_rns,
+        count(*) FILTER (WHERE permit_type = 'RVE')                           AS permits_rve,
+        sum(total_area_sqm)                                                   AS permits_total_area_sqm,
+        sum(living_area_sqm)                                                  AS permits_living_area_sqm,
+        min(issue_date)                                                       AS permits_first_issue_date,
+        max(issue_date)                                                       AS permits_last_issue_date,
+        array_agg(DISTINCT substring(cadastral_number from '^(\d{2}:\d{2}:\d{6,7})'))
+            FILTER (WHERE cadastral_number ~ '^\d{2}:\d{2}:\d{6,7}')          AS permits_quarters
+    FROM ekburg_construction_permits
+    WHERE fn_normalize_inn(developer_inn) IS NOT NULL
+    GROUP BY fn_normalize_inn(developer_inn)
+),
+-- 3.4 domrf track-record по застройщику (по latest-snapshot объектам)
+domrf_agg AS (
+    SELECT
+        inn,
+        count(*)                                                  AS domrf_objects_total,
+        sum(flat_count)                                           AS domrf_flats_total,
+        sum(free_flats)                                           AS domrf_free_flats_total,
+        count(*) FILTER (WHERE site_status = 'Строящиеся')        AS domrf_active_objects,
+        array_agg(DISTINCT district_name)
+            FILTER (WHERE district_name IS NOT NULL)              AS domrf_districts,
+        array_agg(DISTINCT obj_class)
+            FILTER (WHERE obj_class IS NOT NULL)                  AS domrf_classes
+    FROM domrf_latest
+    GROUP BY inn
+)
+SELECT
+    COALESCE(p.inn, d.inn)                          AS developer_inn,
+    cn.developer_name                               AS developer_name,
+    -- track-record permits-сторона
+    COALESCE(p.permits_total, 0)                    AS permits_total,
+    COALESCE(p.permits_rns, 0)                      AS permits_rns,
+    COALESCE(p.permits_rve, 0)                      AS permits_rve,
+    p.permits_total_area_sqm,
+    p.permits_living_area_sqm,
+    p.permits_first_issue_date,
+    p.permits_last_issue_date,
+    p.permits_quarters,
+    -- track-record domrf-сторона
+    COALESCE(d.domrf_objects_total, 0)              AS domrf_objects_total,
+    COALESCE(d.domrf_active_objects, 0)             AS domrf_active_objects,
+    d.domrf_flats_total,
+    d.domrf_free_flats_total,
+    d.domrf_districts,
+    d.domrf_classes,
+    -- известен ли застройщик обеим сторонам (кросс-валидация)
+    (p.inn IS NOT NULL AND d.inn IS NOT NULL)       AS in_both_sources
+FROM permits_agg p
+FULL OUTER JOIN domrf_agg d ON p.inn = d.inn
+LEFT JOIN canonical_name cn ON cn.inn = COALESCE(p.inn, d.inn);
+
+COMMENT ON MATERIALIZED VIEW developer_registry IS
+    '#1088: реестр застройщиков по НОРМАЛИЗОВАННОМУ ИНН (fn_normalize_inn) — канон-имя '
+    '(domrf>permits, most-frequent, tie→min-lexical) + track-record обеих сторон '
+    '(permits РНС/РВЭ/площади/кварталы, domrf ЖК/квартиры/районы/классы). '
+    'in_both_sources — кросс-валидация (prod: 292 ИНН). FULL OUTER JOIN — застройщик '
+    'может быть только в одной стороне. MATERIALIZED — refresh через '
+    'fn_refresh_developer_registry() (permits monthly / domrf daily).';
+
+-- 3.5 UNIQUE индекс на developer_inn — обязателен для REFRESH ... CONCURRENTLY
+-- (иначе refresh блокирует читателей). Одновременно обслуживает point-lookup по ИНН
+-- (резолвер джойнит сюда, analyze тянет track-record по конкретному застройщику).
+-- developer_inn здесь уже НОРМАЛИЗОВАН и NOT NULL (COALESCE двух not-null норм-ИНН).
+CREATE UNIQUE INDEX IF NOT EXISTS uq_developer_registry_inn
+    ON developer_registry (developer_inn);
+
+-- 3.6 fn_refresh_developer_registry — обёртка REFRESH CONCURRENTLY (вызывает Celery-
+-- beat / sync-задачи permits+domrf после обновления источников). CONCURRENTLY — не
+-- блокирует SELECT-читателей (analyze) во время refresh; требует unique-индекс выше.
+CREATE OR REPLACE FUNCTION fn_refresh_developer_registry()
+RETURNS void
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    REFRESH MATERIALIZED VIEW CONCURRENTLY developer_registry;
+END;
+$$;
+
+COMMENT ON FUNCTION fn_refresh_developer_registry() IS
+    '#1088: REFRESH MATERIALIZED VIEW CONCURRENTLY developer_registry. Вызывать после '
+    'обновления источников (ekburg_permits_sync monthly / domrf_kn daily snapshot).';
+
+-- ──────────────────────────────────────────────────────────────────────────────────
+-- 4. developer_attribution — алиас-VIEW для analyze-блока (стабильное имя контракта)
+-- ──────────────────────────────────────────────────────────────────────────────────
+-- ТЗ #1088 называет целевую сущность `developer_attribution`. Держим её тонким VIEW
+-- поверх materialized developer_registry: тот же набор строк, имя зафиксировано в
+-- контракте (backend ходит сюда). VIEW поверх matview читается так же дёшево (matview
+-- — физическая таблица с unique-индексом). Пересоздаётся ЗДЕСЬ после DROP ... CASCADE
+-- в §3 (CASCADE роняет этот зависимый VIEW при ре-apply).
+CREATE OR REPLACE VIEW developer_attribution AS
+    SELECT * FROM developer_registry;
+
+COMMENT ON VIEW developer_attribution IS
+    '#1088: стабильное имя контракта для analyze-блока. Алиас materialized '
+    'developer_registry (канон застройщик + track-record по норм-ИНН).';
+
+-- ──────────────────────────────────────────────────────────────────────────────────
+-- 5. fn_developer_for_parcel — резолвер участок → застройщик (3-ступенчатый fallback)
+-- ──────────────────────────────────────────────────────────────────────────────────
+-- Вход: cad_num участка. Геометрию/квартал берём ВНУТРИ из cad_parcels_geom (view над
+-- cad_parcels, GIST cad_parcels_geom_gist) + _quarter_from_cad по строке номера.
+-- Выход: SETOF строк-кандидатов с источником/методом/дистанцией + denormalized
+-- track-record (join к developer_registry по ИНН). Caller (backend) ранжирует/берёт
+-- топ-1. p_radius_m — радиус spatial-ступени (default 150).
+--
+-- STABLE (не IMMUTABLE) — читает таблицы. Параметр geom не принимаем: cad_num —
+-- единственный вход (геометрия резолвится внутри), чтобы контракт был как у
+-- competitors.get_competitors(cad_num). Если участка нет в cad_parcels_geom —
+-- spatial/quarter-domrf по geom не сработают, но exact-cadastral и quarter-permits
+-- (по строке cad_num) всё равно отработают.
+CREATE OR REPLACE FUNCTION fn_developer_for_parcel(
+    p_cad_num  text,
+    p_radius_m double precision DEFAULT 150.0
+)
+RETURNS TABLE (
+    developer_inn   text,
+    developer_name  text,
+    source          text,    -- 'permits' | 'domrf'
+    match_method    text,    -- 'exact_cadastral' | 'spatial_150m' | 'quarter'
+    distance_m      double precision,
+    project         text,    -- permit object_name / domrf comm_name
+    status          text,    -- permit_type / domrf site_status
+    obj_class       text,
+    permits_total       integer,
+    domrf_objects_total integer,
+    domrf_active_objects integer,
+    in_both_sources bool
+)
+LANGUAGE sql
+STABLE
+AS $$
+    WITH
+    quarter AS (
+        -- кадастровый квартал из номера участка: '66:41:0313010:2' → '66:41:0313010'
+        SELECT substring(p_cad_num from '^(\d{2}:\d{2}:\d{6,7})') AS q
+    ),
+    parcel AS (
+        SELECT geom, ST_Centroid(geom) AS centroid
+        FROM cad_parcels_geom
+        WHERE cad_num = p_cad_num AND geom IS NOT NULL
+        LIMIT 1
+    ),
+    -- ── Ступень 1: точный кадастр (permits.cadastral_number = cad_num) ──────────────
+    -- Две ветки, объединённые UNION (каждая sargable / узкая):
+    --   1a. РОВНОЕ равенство = cad_num → Bitmap Index Scan по idx_ekburg_permits_cadastral
+    --       (0.09мс; покрывает single-value строки = 98.6% permits);
+    --   1b. multi-value строки РНС (несколько ЗУ через запятую/«;» — prod: 57 строк):
+    --       cad_num как подстрока на границе. Regex применяется ТОЛЬКО к узкому
+    --       подмножеству (LIKE '%,%' OR '%;%') — не Seq Scan по всем 4059.
+    exact_match AS (
+        SELECT
+            fn_normalize_inn(e.developer_inn) AS inn,
+            'permits'::text  AS source,
+            'exact_cadastral'::text AS match_method,
+            0.0::double precision AS distance_m,
+            e.object_name AS project,
+            e.permit_type AS status,
+            e.object_type AS obj_class
+        FROM ekburg_construction_permits e
+        WHERE e.cadastral_number = p_cad_num          -- 1a: btree equality
+        UNION
+        SELECT
+            fn_normalize_inn(e.developer_inn) AS inn,
+            'permits'::text, 'exact_cadastral'::text, 0.0::double precision,
+            e.object_name, e.permit_type, e.object_type
+        FROM ekburg_construction_permits e
+        WHERE (e.cadastral_number LIKE '%,%' OR e.cadastral_number LIKE '%;%')  -- 1b: multi-value subset
+          AND e.cadastral_number ~ ('(^|[,; ])' || replace(p_cad_num, '.', '\.') || '($|[,; ])')
+    ),
+    -- ── Ступень 2: spatial 150 м (permit.geom рядом с geom участка) ─────────────────
+    -- GIST-friendly: ST_DWithin(geom, geom, deg) даёт index-scan по
+    -- gist_ekburg_permits_geom, затем geography-refine на точные метры. 0.0025° ⊃ 150м.
+    spatial_match AS (
+        SELECT
+            fn_normalize_inn(e.developer_inn) AS inn,
+            'permits'::text AS source,
+            'spatial_150m'::text AS match_method,
+            ST_Distance(e.geom::geography, pa.centroid::geography) AS distance_m,
+            e.object_name AS project,
+            e.permit_type AS status,
+            e.object_type AS obj_class
+        FROM ekburg_construction_permits e
+        CROSS JOIN parcel pa
+        WHERE e.geom IS NOT NULL
+          AND ST_DWithin(e.geom, pa.geom, p_radius_m / 111000.0)
+          AND ST_DWithin(e.geom::geography, pa.centroid::geography, p_radius_m)
+    ),
+    -- ── Ступень 3a: кадастровый квартал — permits ───────────────────────────────────
+    -- LIKE 'QUARTER:%' — sargable через text_pattern_ops индекс (idx_ekburg_permits_
+    -- cad_cpat): все ЗУ квартала имеют вид 'QUARTER:NNN'. → Bitmap Index Scan (0.74мс),
+    -- вместо Seq Scan от regex '^' (6мс). НЕ range ['X:','X;') — он collation-unsafe на
+    -- en_US.utf8 (вернул 0 строк в dry-run); LIKE-prefix на text_pattern_ops надёжен.
+    -- '%' в q.q невозможен (квартал — '\d{2}:\d{2}:\d{6,7}'), escaping не нужен.
+    quarter_permits AS (
+        SELECT
+            fn_normalize_inn(e.developer_inn) AS inn,
+            'permits'::text AS source,
+            'quarter'::text AS match_method,
+            NULL::double precision AS distance_m,
+            e.object_name AS project,
+            e.permit_type AS status,
+            e.object_type AS obj_class
+        FROM ekburg_construction_permits e, quarter q
+        WHERE q.q IS NOT NULL
+          AND e.cadastral_number LIKE q.q || ':%'
+    ),
+    -- ── Ступень 3b: кадастровый квартал — domrf (latest snapshot per obj_id) ─────────
+    quarter_domrf AS (
+        SELECT DISTINCT ON (o.obj_id)
+            fn_normalize_inn(o.dev_inn) AS inn,
+            'domrf'::text AS source,
+            'quarter'::text AS match_method,
+            NULL::double precision AS distance_m,
+            o.comm_name AS project,
+            o.site_status AS status,
+            o.obj_class
+        FROM domrf_kn_objects o, quarter q
+        WHERE q.q IS NOT NULL
+          AND btrim(o.cad_quarter) = q.q
+        ORDER BY o.obj_id, o.snapshot_date DESC
+    ),
+    -- Объединяем кандидатов всех ступеней. Если сработала ступень exact — caller
+    -- получит её первой (distance_m=0); spatial/quarter дополняют контекст соседей.
+    all_candidates AS (
+        SELECT inn, source, match_method, distance_m, project, status, obj_class FROM exact_match
+        UNION ALL
+        SELECT inn, source, match_method, distance_m, project, status, obj_class FROM spatial_match
+        UNION ALL
+        SELECT inn, source, match_method, distance_m, project, status, obj_class FROM quarter_permits
+        UNION ALL
+        SELECT inn, source, match_method, distance_m, project, status, obj_class FROM quarter_domrf
+    )
+    SELECT
+        c.inn AS developer_inn,
+        reg.developer_name,
+        c.source,
+        c.match_method,
+        c.distance_m,
+        c.project,
+        c.status,
+        c.obj_class,
+        reg.permits_total,
+        reg.domrf_objects_total,
+        reg.domrf_active_objects,
+        reg.in_both_sources
+    FROM all_candidates c
+    LEFT JOIN developer_registry reg ON reg.developer_inn = c.inn
+    WHERE c.inn IS NOT NULL
+    -- детерминированный порядок: лучший метод первым, ближе — раньше, затем ИНН
+    ORDER BY
+        CASE c.match_method
+            WHEN 'exact_cadastral' THEN 1
+            WHEN 'spatial_150m'    THEN 2
+            WHEN 'quarter'         THEN 3
+            ELSE 4
+        END,
+        c.distance_m ASC NULLS LAST,
+        c.inn;
+$$;
+
+COMMENT ON FUNCTION fn_developer_for_parcel(text, double precision) IS
+    '#1088: резолвер участок→застройщик. 3 ступени (exact_cadastral > spatial_150m > '
+    'quarter), геометрия/квартал берутся из cad_parcels_geom по cad_num. Возвращает '
+    'SETOF кандидатов с match_method/distance_m + track-record из developer_registry. '
+    'Caller берёт топ-1 (ORDER BY method, distance) или ранжирует. p_radius_m=150 по '
+    'умолчанию. GIST-friendly spatial (ST_DWithin geom degrees + geography refine).';
+
+COMMIT;