From dbf31cc9118998829da058d51f14f6f96df48a14 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sat, 16 May 2026 12:47:09 +0000 Subject: [PATCH] feat(etl): objective_complex_mapping backfill via pg_trgm (#203) (#214) --- backend/app/api/v1/admin_etl.py | 70 ++++++ backend/app/main.py | 6 + backend/app/services/etl/__init__.py | 0 .../app/services/etl/objective_backfill.py | 218 ++++++++++++++++++ .../tests/services/test_objective_backfill.py | 176 ++++++++++++++ data/sql/97_objective_backfill_indexes.sql | 31 +++ 6 files changed, 501 insertions(+) create mode 100644 backend/app/api/v1/admin_etl.py create mode 100644 backend/app/services/etl/__init__.py create mode 100644 backend/app/services/etl/objective_backfill.py create mode 100644 backend/tests/services/test_objective_backfill.py create mode 100644 data/sql/97_objective_backfill_indexes.sql diff --git a/backend/app/api/v1/admin_etl.py b/backend/app/api/v1/admin_etl.py new file mode 100644 index 00000000..96b2d51a --- /dev/null +++ b/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: +""" + +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 diff --git a/backend/app/main.py b/backend/app/main.py index 93db3a29..4197a6b4 100644 --- a/backend/app/main.py +++ b/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") diff --git a/backend/app/services/etl/__init__.py b/backend/app/services/etl/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/app/services/etl/objective_backfill.py b/backend/app/services/etl/objective_backfill.py new file mode 100644 index 00000000..dc566af9 --- /dev/null +++ b/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) diff --git a/backend/tests/services/test_objective_backfill.py b/backend/tests/services/test_objective_backfill.py new file mode 100644 index 00000000..3da13880 --- /dev/null +++ b/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 diff --git a/data/sql/97_objective_backfill_indexes.sql b/data/sql/97_objective_backfill_indexes.sql new file mode 100644 index 00000000..13a0b968 --- /dev/null +++ b/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;