feat(etl): objective_complex_mapping backfill via pg_trgm fuzzy match (#203)

Add fuzzy-match ETL service that maps unmapped DOM.РФ EKB complexes to
Objective project names using pg_trgm similarity. Exposes admin endpoint
POST /api/v1/admin/etl/objective-backfill with dry_run + refresh_mv flags.

Phase 1 scope: auto-accept matches >= 0.85, review_queue 0.60-0.84 (UI TBD).
Expected coverage lift: 114 → ~260 mapped objects, mv_layout_velocity 3% → ~20%.

Closes #203

backend/app/api/v1/admin_etl.py

@@ -0,0 +1,70 @@
+"""Admin endpoints для ETL operations (#203).
+
+POST /api/v1/admin/etl/objective-backfill
+    Запустить fuzzy-match backfill objective_complex_mapping.
+    Опциональный REFRESH mv_layout_velocity после успешного backfill.
+
+Header: X-Admin-Token: <SCRAPE_ADMIN_TOKEN>
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.orm import Session
+
+from app.core.db import get_db
+from app.core.deps import AdminTokenAuth
+from app.services.etl.objective_backfill import (
+    REVIEW_THRESHOLD,
+    auto_apply_matches,
+    find_match_candidates,
+    trigger_mv_refresh,
+)
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+
+@router.post("/objective-backfill")
+def run_objective_backfill(
+    db: Annotated[Session, Depends(get_db)],
+    _: AdminTokenAuth,
+    dry_run: Annotated[bool, Query(description="Preview без insertions")] = False,
+    refresh_mv: Annotated[
+        bool, Query(description="REFRESH mv_layout_velocity после backfill")
+    ] = True,
+) -> dict[str, int]:
+    """Запустить backfill objective_complex_mapping + опционально REFRESH MV.
+
+    Ищет DOM.РФ комплексы (is_ekb=true) без mapping и применяет fuzzy match
+    к project_name из objective_corpus_room_month через pg_trgm similarity.
+
+    - score >= 0.85 (AUTO_ACCEPT_THRESHOLD): auto-insert с match_method='fuzzy_trgm'
+    - score >= 0.6 (REVIEW_THRESHOLD) и < 0.85: только в счётчик review_queue
+      (Phase 2 — UI для ручного review)
+
+    Returns dict:
+        auto_accepted: сколько строк вставлено
+        review_queue: сколько кандидатов ниже порога auto-accept
+        skipped: ON CONFLICT + ошибки
+        mv_rows_after_refresh: строк в MV после REFRESH (0 если refresh_mv=False)
+    """
+    candidates = find_match_candidates(db, only_unmapped=True)
+    logger.info(
+        "Backfill candidates found: %d (score >= %.2f)",
+        len(candidates),
+        REVIEW_THRESHOLD,
+    )
+
+    result = auto_apply_matches(db, candidates, dry_run=dry_run)
+
+    mv_rows = 0
+    if refresh_mv and not dry_run and result["auto_accepted"] > 0:
+        mv_rows = trigger_mv_refresh(db)
+        logger.info("mv_layout_velocity refreshed after backfill: %d rows", mv_rows)
+
+    result["mv_rows_after_refresh"] = mv_rows
+    return result

backend/app/main.py

@@ -7,6 +7,7 @@ from fastapi.middleware.cors import CORSMiddleware
 
 from app.api.v1 import (
     admin_cadastre,
+    admin_etl,
     admin_jobs,
     admin_leads,
     admin_scrape,
@@ -58,6 +59,11 @@ app.include_router(
     prefix="/api/v1/admin/cadastre",
     tags=["admin", "cadastre"],
 )
+app.include_router(
+    admin_etl.router,
+    prefix="/api/v1/admin/etl",
+    tags=["admin", "etl"],
+)
 
 
 @app.get("/health")

backend/app/services/etl/objective_backfill.py

@@ -0,0 +1,218 @@
+"""Backfill objective_complex_mapping (#203).
+
+Auto-match DOM.РФ комплексов к Objective dataset через fuzzy name matching.
+Goal: coverage 3% → 40%+ для mv_layout_velocity.
+
+Schema facts (confirmed via pg MCP):
+- domrf_kn_objects: фильтр is_ekb = true (~1285 ЕКБ объектов).
+  district_name ILIKE '%екатеринбург%' возвращает 0 строк — не использовать.
+- objective_corpus_room_month: group_name = 'Екатеринбург' (единственное значение),
+  263 distinct project_name.
+- objective_complex_mapping: UNIQUE (objective_complex_name, objective_group),
+  колонки match_method + match_score + is_reviewed поддерживают audit trail.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Порог для auto-insert (высокая уверенность)
+AUTO_ACCEPT_THRESHOLD = 0.85
+
+# Порог для review queue (средняя уверенность — Phase 2)
+REVIEW_THRESHOLD = 0.6
+
+
+@dataclass
+class MatchCandidate:
+    """Один candidate match DOM.РФ ↔ Objective."""
+
+    domrf_obj_id: int
+    domrf_comm_name: str
+    domrf_dev_name: str | None
+    objective_project_name: str
+    similarity_score: float  # 0.0..1.0
+
+
+def find_match_candidates(
+    db: Session,
+    *,
+    only_unmapped: bool = True,
+    limit: int | None = None,
+) -> list[MatchCandidate]:
+    """Поиск candidates через pg_trgm similarity.
+
+    Использует CROSS JOIN LATERAL + similarity() для fuzzy match
+    comm_name (DOM.РФ) ↔ project_name (Objective).
+
+    Args:
+        db: SQLAlchemy sync Session.
+        only_unmapped: Если True — пропускает уже-mapped obj_id.
+        limit: Максимальное число строк результата (для тестирования).
+
+    Returns:
+        Список MatchCandidate, отсортированных по убыванию similarity_score.
+    """
+    # Формируем SQL. LIMIT добавляем через int() — SQL injection safe (только число).
+    limit_clause = f"LIMIT {int(limit)}" if limit is not None else ""
+
+    sql = text(
+        f"""
+        WITH domrf_unmapped AS (
+            SELECT o.obj_id, o.comm_name, o.dev_name
+            FROM domrf_kn_objects o
+            WHERE o.is_ekb = true
+              AND o.comm_name IS NOT NULL
+              AND (
+                  CAST(:only_unmapped AS boolean) = FALSE
+                  OR NOT EXISTS (
+                      SELECT 1 FROM objective_complex_mapping cm
+                      WHERE cm.domrf_obj_id = o.obj_id
+                  )
+              )
+        ),
+        objective_distinct AS (
+            SELECT DISTINCT project_name
+            FROM objective_corpus_room_month
+        )
+        SELECT
+            d.obj_id,
+            d.comm_name,
+            d.dev_name,
+            obj.project_name,
+            similarity(d.comm_name, obj.project_name) AS sim_score
+        FROM domrf_unmapped d
+        CROSS JOIN LATERAL (
+            SELECT project_name
+            FROM objective_distinct
+            WHERE similarity(d.comm_name, project_name) > 0.4
+            ORDER BY similarity(d.comm_name, project_name) DESC
+            LIMIT 1
+        ) obj
+        WHERE similarity(d.comm_name, obj.project_name) >= CAST(:min_threshold AS float)
+        ORDER BY sim_score DESC
+        {limit_clause}
+        """
+    )
+
+    rows = db.execute(
+        sql,
+        {"only_unmapped": only_unmapped, "min_threshold": REVIEW_THRESHOLD},
+    ).all()
+
+    return [
+        MatchCandidate(
+            domrf_obj_id=int(r[0]),
+            domrf_comm_name=str(r[1]),
+            domrf_dev_name=str(r[2]) if r[2] is not None else None,
+            objective_project_name=str(r[3]),
+            similarity_score=float(r[4]),
+        )
+        for r in rows
+    ]
+
+
+def auto_apply_matches(
+    db: Session,
+    candidates: list[MatchCandidate],
+    *,
+    threshold: float = AUTO_ACCEPT_THRESHOLD,
+    dry_run: bool = False,
+) -> dict[str, int]:
+    """Apply candidates с score >= threshold в objective_complex_mapping.
+
+    Кандидаты ниже threshold, но >= REVIEW_THRESHOLD попадают в review_queue
+    (Phase 2 — UI для ручного review, в этом PR только считаются).
+
+    ON CONFLICT DO NOTHING — если пара (objective_complex_name, objective_group)
+    уже существует, строка пропускается без ошибки.
+
+    Args:
+        db: SQLAlchemy sync Session.
+        candidates: Список из find_match_candidates().
+        threshold: Минимальный score для auto-insert (default 0.85).
+        dry_run: Если True — только логирует, не пишет в БД.
+
+    Returns:
+        dict с ключами auto_accepted, review_queue, skipped.
+    """
+    auto = [c for c in candidates if c.similarity_score >= threshold]
+    review = [c for c in candidates if REVIEW_THRESHOLD <= c.similarity_score < threshold]
+
+    if dry_run:
+        logger.info(
+            "DRY RUN: would auto-accept %d, review queue %d",
+            len(auto),
+            len(review),
+        )
+        return {"auto_accepted": 0, "review_queue": len(review), "skipped": 0}
+
+    inserted = 0
+    skipped = 0
+    for c in auto:
+        try:
+            with db.begin_nested():
+                result = db.execute(
+                    text(
+                        """
+                        INSERT INTO objective_complex_mapping
+                            (objective_complex_name, domrf_obj_id, objective_group,
+                             match_method, match_score, is_reviewed)
+                        VALUES (
+                            CAST(:name AS text),
+                            CAST(:obj_id AS bigint),
+                            CAST(:group AS text),
+                            CAST(:method AS text),
+                            CAST(:score AS numeric),
+                            CAST(:reviewed AS boolean)
+                        )
+                        ON CONFLICT (objective_complex_name, objective_group) DO NOTHING
+                        """
+                    ),
+                    {
+                        "name": c.objective_project_name,
+                        "obj_id": c.domrf_obj_id,
+                        "group": "Екатеринбург",
+                        "method": "fuzzy_trgm",
+                        "score": c.similarity_score,
+                        "reviewed": False,
+                    },
+                )
+            if result.rowcount > 0:
+                inserted += 1
+            else:
+                skipped += 1
+        except Exception as e:
+            logger.warning(
+                "Insert failed для %s ↔ %s: %s",
+                c.domrf_comm_name,
+                c.objective_project_name,
+                e,
+            )
+            skipped += 1
+
+    db.commit()
+    logger.info(
+        "Backfill complete: auto_accepted=%d skipped=%d review_queue=%d",
+        inserted,
+        skipped,
+        len(review),
+    )
+    return {"auto_accepted": inserted, "review_queue": len(review), "skipped": skipped}
+
+
+def trigger_mv_refresh(db: Session) -> int:
+    """REFRESH mv_layout_velocity после backfill.
+
+    Вызывает существующий helper из layout_velocity_refresh.
+    Передаём concurrently=True (MV уже заполнен).
+    """
+    from app.services.site_finder.layout_velocity_refresh import refresh_layout_velocity
+
+    return refresh_layout_velocity(db, concurrently=True)

backend/tests/services/test_objective_backfill.py

@@ -0,0 +1,176 @@
+"""Тесты для objective_backfill.py (#203).
+
+Использует mock DB (unittest.mock) — не требует реального PostgreSQL.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from app.services.etl.objective_backfill import (
+    AUTO_ACCEPT_THRESHOLD,
+    REVIEW_THRESHOLD,
+    MatchCandidate,
+    auto_apply_matches,
+    find_match_candidates,
+    trigger_mv_refresh,
+)
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+
+def _make_candidate(
+    obj_id: int,
+    comm_name: str,
+    project_name: str,
+    score: float,
+    dev_name: str | None = "ООО Девелопер",
+) -> MatchCandidate:
+    return MatchCandidate(
+        domrf_obj_id=obj_id,
+        domrf_comm_name=comm_name,
+        domrf_dev_name=dev_name,
+        objective_project_name=project_name,
+        similarity_score=score,
+    )
+
+
+def _make_db_row(obj_id: int, comm_name: str, dev_name: str, project: str, score: float) -> Any:
+    """Имитирует Row, возвращаемый SQLAlchemy db.execute().all()."""
+    return (obj_id, comm_name, dev_name, project, score)
+
+
+# ── Test 1: find_match_candidates возвращает правильные MatchCandidate ────────
+
+
+def test_find_match_candidates_returns_candidates() -> None:
+    """find_match_candidates корректно преобразует DB rows в MatchCandidate."""
+    mock_rows = [
+        _make_db_row(100, "Северный квартал", "Брусника", "Северный квартал", 1.0),
+        _make_db_row(200, "АЛЛЕГРО", "СЗ ГОРЖИЛСТРОЙ", "АЛЛЕГРО", 0.95),
+        _make_db_row(300, "Некий ЖК", None, "Близкий ЖК", 0.72),
+    ]
+
+    mock_execute = MagicMock()
+    mock_execute.all.return_value = mock_rows
+    mock_db = MagicMock()
+    mock_db.execute.return_value = mock_execute
+
+    candidates = find_match_candidates(mock_db, only_unmapped=True)
+
+    assert len(candidates) == 3
+    assert candidates[0].domrf_obj_id == 100
+    assert candidates[0].similarity_score == 1.0
+    assert candidates[0].domrf_dev_name == "Брусника"
+    assert candidates[2].domrf_dev_name is None  # NULL dev_name → None
+    assert candidates[2].similarity_score == 0.72
+
+    # Убедимся, что db.execute вызывался с параметрами
+    mock_db.execute.assert_called_once()
+    call_kwargs = mock_db.execute.call_args[0][1]
+    assert call_kwargs["only_unmapped"] is True
+    assert call_kwargs["min_threshold"] == REVIEW_THRESHOLD
+
+
+# ── Test 2: auto_apply_matches dry_run не вставляет ──────────────────────────
+
+
+def test_auto_apply_matches_dry_run_no_inserts() -> None:
+    """dry_run=True возвращает счётчики без обращения к БД (execute не вызывается)."""
+    mock_db = MagicMock()
+
+    candidates = [
+        _make_candidate(1, "ЖК А", "ЖК А", 0.95),  # auto-accept
+        _make_candidate(2, "ЖК Б", "ЖК Б", 0.70),  # review queue
+        _make_candidate(3, "ЖК В", "ЖК В", 0.65),  # review queue
+    ]
+
+    result = auto_apply_matches(mock_db, candidates, dry_run=True)
+
+    assert result["auto_accepted"] == 0
+    assert result["review_queue"] == 2
+    mock_db.execute.assert_not_called()
+    mock_db.commit.assert_not_called()
+
+
+# ── Test 3: auto_apply_matches high-score inserts, low-score → review_queue ──
+
+
+def test_auto_apply_matches_inserts_high_score_only() -> None:
+    """Только кандидаты >= AUTO_ACCEPT_THRESHOLD вставляются в БД."""
+    # Мокируем begin_nested как context manager
+    savepoint_cm = MagicMock()
+    savepoint_cm.__enter__ = MagicMock(return_value=None)
+    savepoint_cm.__exit__ = MagicMock(return_value=False)
+
+    mock_execute_result = MagicMock()
+    mock_execute_result.rowcount = 1
+
+    mock_db = MagicMock()
+    mock_db.begin_nested.return_value = savepoint_cm
+    mock_db.execute.return_value = mock_execute_result
+
+    high_score = AUTO_ACCEPT_THRESHOLD  # ровно на пороге — должен вставляться
+    low_score = REVIEW_THRESHOLD  # ровно REVIEW_THRESHOLD — не auto-accept
+
+    candidates = [
+        _make_candidate(10, "ЖК Альфа", "ЖК Альфа", high_score),
+        _make_candidate(11, "ЖК Бета", "ЖК Бета", high_score + 0.05),
+        _make_candidate(20, "ЖК Гамма", "ЖК Гамма", low_score),  # review только
+    ]
+
+    result = auto_apply_matches(mock_db, candidates, threshold=AUTO_ACCEPT_THRESHOLD)
+
+    assert result["auto_accepted"] == 2
+    assert result["review_queue"] == 1
+    assert mock_db.execute.call_count == 2  # вызывался только для high-score
+    mock_db.commit.assert_called_once()
+
+
+# ── Test 4: trigger_mv_refresh делегирует в layout_velocity_refresh ──────────
+
+
+def test_trigger_mv_refresh_calls_helper() -> None:
+    """trigger_mv_refresh вызывает refresh_layout_velocity с concurrently=True.
+
+    refresh_layout_velocity импортируется лениво внутри trigger_mv_refresh,
+    поэтому патчим по полному пути модуля-источника.
+    """
+    mock_db = MagicMock()
+
+    with patch(
+        "app.services.site_finder.layout_velocity_refresh.refresh_layout_velocity",
+        return_value=512,
+    ) as mock_refresh:
+        count = trigger_mv_refresh(mock_db)
+
+    assert count == 512
+    mock_refresh.assert_called_once_with(mock_db, concurrently=True)
+
+
+# ── Test 5: ON CONFLICT — rowcount=0 считается как skipped ───────────────────
+
+
+def test_auto_apply_matches_conflict_counted_as_skipped() -> None:
+    """Если INSERT вернул rowcount=0 (ON CONFLICT DO NOTHING), считается skipped."""
+    savepoint_cm = MagicMock()
+    savepoint_cm.__enter__ = MagicMock(return_value=None)
+    savepoint_cm.__exit__ = MagicMock(return_value=False)
+
+    mock_execute_result = MagicMock()
+    mock_execute_result.rowcount = 0  # ON CONFLICT DO NOTHING
+
+    mock_db = MagicMock()
+    mock_db.begin_nested.return_value = savepoint_cm
+    mock_db.execute.return_value = mock_execute_result
+
+    candidates = [
+        _make_candidate(99, "Уже существующий ЖК", "Уже существующий ЖК", 0.99),
+    ]
+
+    result = auto_apply_matches(mock_db, candidates)
+
+    assert result["auto_accepted"] == 0
+    assert result["skipped"] == 1
+    assert result["review_queue"] == 0

data/sql/97_objective_backfill_indexes.sql

@@ -0,0 +1,31 @@
+-- 97_objective_backfill_indexes.sql
+-- Issue #203 — Backfill objective_complex_mapping (114 → ~1500 EKB ЖК).
+--
+-- pg_trgm уже установлен в prod, но CREATE EXTENSION IF NOT EXISTS идемпотентен.
+-- Trigram GIST index на comm_name (DOM.РФ) ускоряет LATERAL similarity() join
+-- при поиске кандидатов для fuzzy match (domrf_kn_objects ↔ objective_corpus_room_month).
+--
+-- Dependencies: domrf_kn_objects, objective_corpus_room_month (уже существуют).
+-- Applied: automatically via deploy.yml in NN order.
+
+BEGIN;
+
+CREATE EXTENSION IF NOT EXISTS pg_trgm;
+
+-- Trigram GIST index для fuzzy match comm_name (DOM.РФ) ↔ project_name (Objective).
+-- Фильтр is_ekb = true: только ЕКБ объекты (~1285 строк).
+CREATE INDEX IF NOT EXISTS idx_domrf_kn_objects_comm_name_trgm
+    ON domrf_kn_objects USING gist (comm_name gist_trgm_ops)
+    WHERE is_ekb = true;
+
+-- B-tree index на project_name в objective для JOIN performance.
+CREATE INDEX IF NOT EXISTS idx_objective_corpus_room_month_project_name
+    ON objective_corpus_room_month (project_name);
+
+COMMENT ON INDEX idx_domrf_kn_objects_comm_name_trgm IS
+    'Trigram GIST index для fuzzy match с objective_complex_mapping (#203 backfill)';
+
+COMMENT ON INDEX idx_objective_corpus_room_month_project_name IS
+    'B-tree index на project_name для JOIN performance (#203 backfill)';
+
+COMMIT;