From 9c52e0b29fef72737d22cfb23c53dcffbf01577e Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 26 Jun 2026 07:44:25 +0000 Subject: [PATCH] feat(etl): housing-class normalization fallback via yandex_realty trigram match (#38) (#1911) housing-class normalization fallback via yandex_realty trigram (#38): migration 169 + ETL + COALESCE consumers. Refs #38 --- backend/app/services/analytics_queries.py | 31 +- .../app/services/etl/obj_class_backfill.py | 328 ++++++++++++++++++ .../app/services/site_finder/competitors.py | 5 +- backend/app/services/site_finder/velocity.py | 11 +- .../tests/services/test_obj_class_backfill.py | 227 ++++++++++++ data/sql/169_obj_class_fallback.sql | 50 +++ 6 files changed, 640 insertions(+), 12 deletions(-) create mode 100644 backend/app/services/etl/obj_class_backfill.py create mode 100644 backend/tests/services/test_obj_class_backfill.py create mode 100644 data/sql/169_obj_class_fallback.sql diff --git a/backend/app/services/analytics_queries.py b/backend/app/services/analytics_queries.py index cd368a89..0c87d6e3 100644 --- a/backend/app/services/analytics_queries.py +++ b/backend/app/services/analytics_queries.py @@ -1811,9 +1811,11 @@ def _active_competitors_count( return int(n or 0) # Tier 1: район + класс (через PostGIS-полигоны district_name) + # #38: реальный obj_class в приоритете, иначе obj_class_fallback. if target_class: n = _q( - "AND district_name = :dn AND obj_class = :cls", + "AND district_name = :dn" + " AND COALESCE(obj_class, obj_class_fallback) = :cls", {"rc": region_code, "dn": district_name, "cls": target_class}, ) if n >= 2: @@ -2218,7 +2220,10 @@ def _competitors_two_dim( ) return 0, n, float(n), scope - class_filter = "AND obj_class = :cls" if target_class else "" + # #38: реальный obj_class в приоритете, иначе obj_class_fallback. + class_filter = ( + "AND COALESCE(obj_class, obj_class_fallback) = :cls" if target_class else "" + ) params: dict[str, Any] = { "rc": region_code, "dn": district_name, @@ -2548,8 +2553,12 @@ def recommend_mix( SELECT 1 FROM domrf_kn_objects WHERE region_cd = :rc AND district_name = :dn - AND obj_class IS NOT NULL - AND (CAST(:cls AS TEXT) IS NULL OR obj_class = :cls) + -- #38: реальный класс ИЛИ fallback (yandex_match/price_inf) + AND COALESCE(obj_class, obj_class_fallback) IS NOT NULL + AND ( + CAST(:cls AS TEXT) IS NULL + OR COALESCE(obj_class, obj_class_fallback) = :cls + ) LIMIT 1 """ ), @@ -2964,15 +2973,21 @@ def recommend_mix( ), latest_obj AS ( -- domrf_kn_objects содержит ~3 snapshot'а на obj_id; - -- берём только самый свежий, иначе comparables дублируются - SELECT DISTINCT ON (obj_id) * + -- берём только самый свежий, иначе comparables дублируются. + -- #38: класс-фильтр по COALESCE(obj_class, obj_class_fallback); + -- eff_class отдаём наружу как эффективный класс. + SELECT DISTINCT ON (obj_id) *, + COALESCE(obj_class, obj_class_fallback) AS eff_class FROM domrf_kn_objects WHERE region_cd = :rc AND district_name = :dn - AND (CAST(:cls AS TEXT) IS NULL OR obj_class = :cls) + AND ( + CAST(:cls AS TEXT) IS NULL + OR COALESCE(obj_class, obj_class_fallback) = :cls + ) ORDER BY obj_id, snapshot_date DESC NULLS LAST ) - SELECT o.obj_id, o.comm_name, o.dev_name, o.obj_class, o.flat_count, + SELECT o.obj_id, o.comm_name, o.dev_name, o.eff_class AS obj_class, o.flat_count, a.perc AS sold_perc, c.cad_quarter, c.latitude AS lat, diff --git a/backend/app/services/etl/obj_class_backfill.py b/backend/app/services/etl/obj_class_backfill.py new file mode 100644 index 00000000..ab71fe78 --- /dev/null +++ b/backend/app/services/etl/obj_class_backfill.py @@ -0,0 +1,328 @@ +"""Backfill obj_class_fallback для domrf_kn_objects (issue #38, Demand D6). + +kn-API не отдаёт obj_class для ~3032 ЖК Свердл +(Bug_Kn_API_Obj_Class_Always_Null) → per-class velocity/absorption/weight +ломаются. Workaround: производим класс из yandex_realty. + +Двухступенчатый fallback (записывается только когда реальный obj_class = NULL): + 1. yandex_match — trigram-match LOWER(comm_name) ↔ LOWER(yandex_realty_zk.name) + через similarity(); auto-accept >= AUTO_ACCEPT_THRESHOLD (0.6). + 2. price_inference — для оставшихся NULL с известным price_per_m2: класс по + вхождению цены в диапазон yandex_realty_class_prices. + +Результат: obj_class_fallback + obj_class_source ('yandex_match'|'price_inference'). + +Schema facts (см. 169_obj_class_fallback.sql + 43_anton_import.sql): +- domrf_kn_objects: obj_class, obj_class_fallback, obj_class_source, + comm_name, price_per_m2_min, price_per_m2_max, region_cd (66 = Свердл). +- yandex_realty_zk: name, obj_class (uppercase EN: 'COMFORT'/'BUSINESS'/…), + price_from. ~12 строк сейчас, scrape расширяет до >500 (отдельный day). +- yandex_realty_class_prices: obj_class (PK), price_per_m2_min, price_per_m2_max. + +Idempotent: пишет только WHERE obj_class IS NULL AND obj_class_fallback IS NULL +(yandex_match) либо AND obj_class_source IS NULL (price_inference). Batched, +SAVEPOINT per row (begin_nested) — падение одной строки не рушит outer tx. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass + +from sqlalchemy import text +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + +# Порог auto-accept для trigram-match (issue #38 указывает similarity > 0.6). +AUTO_ACCEPT_THRESHOLD = 0.6 + +# Регион Свердловской обл. (ЕКБ) в domrf_kn_objects.region_cd. +SVERDL_REGION_CD = 66 + + +@dataclass +class ClassMatchCandidate: + """Один candidate производного класса DOM.РФ ↔ yandex_realty_zk.""" + + domrf_obj_id: int + domrf_comm_name: str + yandex_name: str + yandex_obj_class: str + similarity_score: float # 0.0..1.0 + + +def find_class_candidates( + db: Session, + *, + region_cd: int = SVERDL_REGION_CD, + min_threshold: float = AUTO_ACCEPT_THRESHOLD, + limit: int | None = None, +) -> list[ClassMatchCandidate]: + """Поиск candidates класса через pg_trgm similarity. + + CROSS JOIN LATERAL + similarity() для fuzzy match comm_name (DOM.РФ) ↔ + name (yandex_realty_zk). Для каждого ЖК без класса берётся ЛУЧШИЙ + (max similarity) yandex-кандидат с непустым obj_class. + + Берётся latest snapshot на obj_id (DISTINCT ON), чтобы не плодить дубли + по версионированной domrf_kn_objects. + + Args: + db: SQLAlchemy sync Session. + region_cd: Регион-фильтр (по умолчанию 66 — Свердл). + min_threshold: Нижняя граница similarity для кандидата (0.6 по issue). + limit: Максимум строк результата (для тестирования / батчинга). + + Returns: + Список ClassMatchCandidate, отсортированных по убыванию similarity. + """ + # LIMIT добавляем через int() — SQL injection safe (только число). + limit_clause = f"LIMIT {int(limit)}" if limit is not None else "" + + sql = text( + f""" + WITH domrf_unclassed AS ( + SELECT DISTINCT ON (o.obj_id) + o.obj_id, o.comm_name + FROM domrf_kn_objects o + WHERE o.region_cd = CAST(:region_cd AS integer) + AND o.obj_class IS NULL + AND o.obj_class_fallback IS NULL + AND o.comm_name IS NOT NULL + AND btrim(o.comm_name) <> '' + ORDER BY o.obj_id, o.snapshot_date DESC NULLS LAST + ) + SELECT + d.obj_id, + d.comm_name, + y.name AS yandex_name, + y.obj_class AS yandex_obj_class, + similarity(LOWER(d.comm_name), LOWER(y.name)) AS sim_score + FROM domrf_unclassed d + CROSS JOIN LATERAL ( + SELECT yz.name, yz.obj_class + FROM yandex_realty_zk yz + WHERE yz.obj_class IS NOT NULL + AND yz.name IS NOT NULL + AND similarity(LOWER(d.comm_name), LOWER(yz.name)) + >= CAST(:min_threshold AS float) + ORDER BY similarity(LOWER(d.comm_name), LOWER(yz.name)) DESC + LIMIT 1 + ) y + ORDER BY sim_score DESC + {limit_clause} + """ + ) + + rows = db.execute( + sql, + {"region_cd": region_cd, "min_threshold": min_threshold}, + ).all() + + return [ + ClassMatchCandidate( + domrf_obj_id=int(r[0]), + domrf_comm_name=str(r[1]), + yandex_name=str(r[2]), + yandex_obj_class=str(r[3]), + similarity_score=float(r[4]), + ) + for r in rows + ] + + +def apply_class_matches( + db: Session, + candidates: list[ClassMatchCandidate], + *, + threshold: float = AUTO_ACCEPT_THRESHOLD, + dry_run: bool = False, +) -> dict[str, int]: + """Записать obj_class_fallback + obj_class_source='yandex_match'. + + Пишет ТОЛЬКО строки latest-snapshot, у которых obj_class IS NULL и + obj_class_fallback ещё не проставлен (idempotent — повторный прогон не + перезаписывает уже найденный класс). SAVEPOINT per row. + + Args: + db: SQLAlchemy sync Session. + candidates: Список из find_class_candidates(). + threshold: Минимальный score для записи (default 0.6). + dry_run: Если True — только логирует, не пишет в БД. + + Returns: + dict с ключами updated, skipped. + """ + accepted = [c for c in candidates if c.similarity_score >= threshold] + + if dry_run: + logger.info("DRY RUN: would set obj_class_fallback (yandex_match) для %d", len(accepted)) + return {"updated": len(accepted), "skipped": 0} + + updated = 0 + skipped = 0 + for c in accepted: + try: + with db.begin_nested(): + result = db.execute( + text( + """ + UPDATE domrf_kn_objects + SET obj_class_fallback = CAST(:cls AS text), + obj_class_source = 'yandex_match' + WHERE obj_id = CAST(:obj_id AS bigint) + AND obj_class IS NULL + AND obj_class_fallback IS NULL + """ + ), + {"cls": c.yandex_obj_class, "obj_id": c.domrf_obj_id}, + ) + if result.rowcount > 0: + updated += 1 + else: + skipped += 1 + except Exception as e: + # Не глотаем молча: логируем строку и продолжаем (SAVEPOINT откатил + # только её, outer tx чист). Re-raise не нужен — backfill best-effort. + logger.warning( + "obj_class yandex_match failed для obj_id=%s (%s ↔ %s): %s", + c.domrf_obj_id, + c.domrf_comm_name, + c.yandex_name, + e, + ) + skipped += 1 + + db.commit() + logger.info("obj_class yandex_match backfill: updated=%d skipped=%d", updated, skipped) + return {"updated": updated, "skipped": skipped} + + +def apply_price_inference( + db: Session, + *, + region_cd: int = SVERDL_REGION_CD, + dry_run: bool = False, +) -> dict[str, int]: + """Price-inference fallback: класс по price_per_m2 vs class_prices. + + Для ЖК, оставшихся без obj_class И без obj_class_fallback после yandex_match, + но с известной price_per_m2: класс = тот yandex_realty_class_prices, в чей + диапазон [price_per_m2_min, price_per_m2_max] попадает средняя цена ЖК. + + Средняя цена ЖК — midpoint доступных price_per_m2_min/max (COALESCE, чтобы + учесть строки, где задана только одна граница). При пересечении нескольких + классовых диапазонов берётся диапазон с наименьшей серединой (детерминизм). + + Только set-based UPDATE одним запросом — нет per-row цикла (диапазоны + непротиворечивы, конфликтов нет; idempotency через WHERE obj_class_source + IS NULL). Возвращает счётчик обновлённых строк. + + Args: + db: SQLAlchemy sync Session. + region_cd: Регион-фильтр (66 — Свердл). + dry_run: Если True — считает кандидатов SELECT'ом, не пишет. + + Returns: + dict с ключом updated. + """ + # Подзапрос: для каждого obj_id (latest snapshot) без класса и с ценой — + # лучший классовый диапазон по price_per_m2. + select_candidates = text( + """ + WITH priced AS ( + SELECT DISTINCT ON (o.obj_id) + o.obj_id, + o.snapshot_date, + (COALESCE(o.price_per_m2_min, o.price_per_m2_max) + + COALESCE(o.price_per_m2_max, o.price_per_m2_min)) / 2.0 AS ppm2 + FROM domrf_kn_objects o + WHERE o.region_cd = CAST(:region_cd AS integer) + AND o.obj_class IS NULL + AND o.obj_class_fallback IS NULL + AND COALESCE(o.price_per_m2_min, o.price_per_m2_max) IS NOT NULL + ORDER BY o.obj_id, o.snapshot_date DESC NULLS LAST + ), + matched AS ( + SELECT DISTINCT ON (p.obj_id) + p.obj_id, + p.snapshot_date, + cp.obj_class + FROM priced p + JOIN yandex_realty_class_prices cp + ON p.ppm2 >= cp.price_per_m2_min + AND p.ppm2 <= cp.price_per_m2_max + ORDER BY p.obj_id, + (cp.price_per_m2_min + cp.price_per_m2_max) / 2.0 ASC + ) + SELECT obj_id, snapshot_date, obj_class FROM matched + """ + ) + + if dry_run: + n = len(db.execute(select_candidates, {"region_cd": region_cd}).all()) + logger.info("DRY RUN: would set obj_class_fallback (price_inference) для %d", n) + return {"updated": n} + + rows = db.execute(select_candidates, {"region_cd": region_cd}).all() + updated = 0 + for r in rows: + try: + with db.begin_nested(): + result = db.execute( + text( + """ + UPDATE domrf_kn_objects + SET obj_class_fallback = CAST(:cls AS text), + obj_class_source = 'price_inference' + WHERE obj_id = CAST(:obj_id AS bigint) + AND snapshot_date = CAST(:snap AS date) + AND obj_class IS NULL + AND obj_class_fallback IS NULL + """ + ), + {"cls": str(r[2]), "obj_id": int(r[0]), "snap": r[1]}, + ) + updated += result.rowcount + except Exception as e: + logger.warning("obj_class price_inference failed для obj_id=%s: %s", r[0], e) + + db.commit() + logger.info("obj_class price_inference backfill: updated=%d", updated) + return {"updated": updated} + + +def run_backfill( + db: Session, + *, + region_cd: int = SVERDL_REGION_CD, + with_price_inference: bool = True, + batch_limit: int | None = None, + dry_run: bool = False, +) -> dict[str, int]: + """Полный backfill: yandex_match → (опц.) price_inference. + + Args: + db: SQLAlchemy sync Session. + region_cd: Регион (66 — Свердл). + with_price_inference: Запускать ли вторую ступень price-inference. + batch_limit: Лимит кандидатов trigram-match (None — все). + dry_run: Прогон без записи. + + Returns: + dict: yandex_updated, yandex_skipped, price_updated. + """ + candidates = find_class_candidates(db, region_cd=region_cd, limit=batch_limit) + ym = apply_class_matches(db, candidates, dry_run=dry_run) + + price_updated = 0 + if with_price_inference: + pi = apply_price_inference(db, region_cd=region_cd, dry_run=dry_run) + price_updated = pi["updated"] + + return { + "yandex_updated": ym["updated"], + "yandex_skipped": ym["skipped"], + "price_updated": price_updated, + } diff --git a/backend/app/services/site_finder/competitors.py b/backend/app/services/site_finder/competitors.py index 5d436a0a..2ee36d41 100644 --- a/backend/app/services/site_finder/competitors.py +++ b/backend/app/services/site_finder/competitors.py @@ -321,7 +321,10 @@ _COMPETITORS_SQL = text(""" obj_id, comm_name, dev_name, - obj_class, + -- #38: реальный obj_class в приоритете, иначе obj_class_fallback + -- (yandex_match / price_inference). Класс-фильтр и dominant_class + -- в Python-слое работают по этому эффективному значению. + COALESCE(obj_class, obj_class_fallback) AS obj_class, latitude, longitude, flat_count, diff --git a/backend/app/services/site_finder/velocity.py b/backend/app/services/site_finder/velocity.py index 31c1acf1..cc5aa6df 100644 --- a/backend/app/services/site_finder/velocity.py +++ b/backend/app/services/site_finder/velocity.py @@ -155,8 +155,12 @@ def compute_velocity( # ── Step 1: конкуренты по lat/lon в радиусе ────────────────────────────── # DISTINCT ON (obj_id) ORDER BY snapshot_date DESC — latest snapshot only. # obj_class в domrf_kn_objects заполнен слабо (много NULL); фильтруем - # только если явно передан. - class_filter = "AND o.obj_class = :obj_class" if obj_class else "" + # только если явно передан. #38: при NULL реального класса используем + # obj_class_fallback (yandex_match / price_inference) — реальный obj_class + # в приоритете (COALESCE), поведение для размеченных ЖК не меняется. + class_filter = ( + "AND COALESCE(o.obj_class, o.obj_class_fallback) = :obj_class" if obj_class else "" + ) # SAVEPOINT per query: failure rollbacks ТОЛЬКО savepoint, не outer tx. # db.rollback() здесь НЕЛЬЗЯ — он orphan'ит outer SessionTransaction # (см. PR #155 bot review — SQLAlchemy 2.0 begin_nested context cleanup). @@ -171,7 +175,8 @@ def compute_velocity( obj_id, comm_name, dev_name, - obj_class, + -- #38: эффективный класс — реальный, иначе fallback + COALESCE(obj_class, obj_class_fallback) AS obj_class, latitude, longitude, district_name diff --git a/backend/tests/services/test_obj_class_backfill.py b/backend/tests/services/test_obj_class_backfill.py new file mode 100644 index 00000000..0c55872e --- /dev/null +++ b/backend/tests/services/test_obj_class_backfill.py @@ -0,0 +1,227 @@ +"""Тесты для obj_class_backfill.py (#38, Demand D6). + +Mock DB (unittest.mock) — без реального PostgreSQL. Проверяют преобразование +DB-rows в кандидаты, порог auto-accept, idempotent-skip и price-inference. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +from app.services.etl.obj_class_backfill import ( + AUTO_ACCEPT_THRESHOLD, + ClassMatchCandidate, + apply_class_matches, + apply_price_inference, + find_class_candidates, + run_backfill, +) + +# ── Helpers ─────────────────────────────────────────────────────────────────── + + +def _make_candidate( + obj_id: int, + comm_name: str, + yandex_name: str, + yandex_class: str, + score: float, +) -> ClassMatchCandidate: + return ClassMatchCandidate( + domrf_obj_id=obj_id, + domrf_comm_name=comm_name, + yandex_name=yandex_name, + yandex_obj_class=yandex_class, + similarity_score=score, + ) + + +def _make_db_row( + obj_id: int, comm_name: str, yandex_name: str, yandex_class: str, score: float +) -> Any: + """Имитирует Row из find_class_candidates SELECT.""" + return (obj_id, comm_name, yandex_name, yandex_class, score) + + +def _savepoint_cm() -> MagicMock: + cm = MagicMock() + cm.__enter__ = MagicMock(return_value=None) + cm.__exit__ = MagicMock(return_value=False) + return cm + + +# ── Test 1: find_class_candidates маппит rows в ClassMatchCandidate ────────── + + +def test_find_class_candidates_maps_rows() -> None: + """find_class_candidates корректно преобразует DB rows в кандидаты.""" + mock_rows = [ + _make_db_row(100, "Северный квартал", "ЖК Северный квартал", "COMFORT", 0.91), + _make_db_row(200, "АЛЛЕГРО", "Аллегро", "BUSINESS", 0.74), + ] + mock_execute = MagicMock() + mock_execute.all.return_value = mock_rows + mock_db = MagicMock() + mock_db.execute.return_value = mock_execute + + candidates = find_class_candidates(mock_db) + + assert len(candidates) == 2 + assert candidates[0].domrf_obj_id == 100 + assert candidates[0].yandex_obj_class == "COMFORT" + assert candidates[0].similarity_score == 0.91 + assert candidates[1].yandex_obj_class == "BUSINESS" + + mock_db.execute.assert_called_once() + params = mock_db.execute.call_args[0][1] + assert params["region_cd"] == 66 + assert params["min_threshold"] == AUTO_ACCEPT_THRESHOLD + + +# ── Test 2: dry_run не пишет в БД ──────────────────────────────────────────── + + +def test_apply_class_matches_dry_run_no_writes() -> None: + """dry_run=True возвращает projected-счётчик без обращения к БД.""" + mock_db = MagicMock() + candidates = [ + _make_candidate(1, "ЖК А", "ЖК А", "COMFORT", 0.95), + _make_candidate(2, "ЖК Б", "ЖК Б", "BUSINESS", 0.40), # ниже порога + ] + + result = apply_class_matches(mock_db, candidates, dry_run=True) + + assert result["updated"] == 1 # только >= 0.6 + assert result["skipped"] == 0 + mock_db.execute.assert_not_called() + mock_db.commit.assert_not_called() + + +# ── Test 3: пишутся только кандидаты >= threshold ──────────────────────────── + + +def test_apply_class_matches_threshold_gate() -> None: + """Только кандидаты >= AUTO_ACCEPT_THRESHOLD проходят в UPDATE.""" + mock_result = MagicMock() + mock_result.rowcount = 1 + mock_db = MagicMock() + mock_db.begin_nested.return_value = _savepoint_cm() + mock_db.execute.return_value = mock_result + + candidates = [ + _make_candidate(10, "ЖК Альфа", "Альфа", "COMFORT", AUTO_ACCEPT_THRESHOLD), # ровно порог + _make_candidate(11, "ЖК Бета", "Бета", "BUSINESS", 0.80), + _make_candidate(12, "ЖК Гамма", "Гамма", "ELITE", 0.59), # ниже порога + ] + + result = apply_class_matches(mock_db, candidates) + + assert result["updated"] == 2 + assert result["skipped"] == 0 + assert mock_db.execute.call_count == 2 # только два >= порога + mock_db.commit.assert_called_once() + + +# ── Test 4: idempotent skip (rowcount=0) ───────────────────────────────────── + + +def test_apply_class_matches_idempotent_skip() -> None: + """UPDATE rowcount=0 (класс уже проставлен) считается skipped.""" + mock_result = MagicMock() + mock_result.rowcount = 0 # WHERE obj_class IS NULL не сматчил → уже заполнено + mock_db = MagicMock() + mock_db.begin_nested.return_value = _savepoint_cm() + mock_db.execute.return_value = mock_result + + candidates = [_make_candidate(99, "ЖК уже размечен", "уже", "COMFORT", 0.99)] + + result = apply_class_matches(mock_db, candidates) + + assert result["updated"] == 0 + assert result["skipped"] == 1 + mock_db.commit.assert_called_once() + + +# ── Test 5: failure одной строки не рушит цикл (SAVEPOINT) ─────────────────── + + +def test_apply_class_matches_row_failure_isolated() -> None: + """Исключение на одной строке логируется, цикл продолжается (skipped++).""" + mock_ok = MagicMock() + mock_ok.rowcount = 1 + mock_db = MagicMock() + mock_db.begin_nested.return_value = _savepoint_cm() + # Первая строка кидает, вторая — успех. + mock_db.execute.side_effect = [RuntimeError("boom"), mock_ok] + + candidates = [ + _make_candidate(1, "ЖК падающий", "пад", "COMFORT", 0.9), + _make_candidate(2, "ЖК норм", "норм", "BUSINESS", 0.9), + ] + + result = apply_class_matches(mock_db, candidates) + + assert result["updated"] == 1 + assert result["skipped"] == 1 + mock_db.commit.assert_called_once() + + +# ── Test 6: price_inference dry_run считает кандидатов ─────────────────────── + + +def test_apply_price_inference_dry_run_counts() -> None: + """dry_run=True считает кандидатов SELECT'ом, ничего не пишет.""" + mock_execute = MagicMock() + mock_execute.all.return_value = [(1, "2026-01-01", "COMFORT"), (2, "2026-01-01", "BUSINESS")] + mock_db = MagicMock() + mock_db.execute.return_value = mock_execute + + result = apply_price_inference(mock_db, dry_run=True) + + assert result["updated"] == 2 + mock_db.commit.assert_not_called() + + +# ── Test 7: price_inference пишет per-row с SAVEPOINT ──────────────────────── + + +def test_apply_price_inference_writes_rows() -> None: + """price_inference UPDATE'ит каждого кандидата, суммирует rowcount.""" + select_result = MagicMock() + select_result.all.return_value = [(1, "2026-01-01", "COMFORT"), (2, "2026-01-01", "ELITE")] + update_result = MagicMock() + update_result.rowcount = 1 + mock_db = MagicMock() + mock_db.begin_nested.return_value = _savepoint_cm() + # Первый execute — SELECT кандидатов, далее — UPDATE per row. + mock_db.execute.side_effect = [select_result, update_result, update_result] + + result = apply_price_inference(mock_db) + + assert result["updated"] == 2 + mock_db.commit.assert_called_once() + + +# ── Test 8: run_backfill оркестрирует обе ступени ──────────────────────────── + + +def test_run_backfill_orchestrates_both_stages() -> None: + """run_backfill: trigram-match → price_inference, агрегирует счётчики.""" + # SELECT кандидатов (find) → пусто, чтобы apply_class_matches ничего не писал; + # затем price_inference SELECT вернёт 1 кандидата + UPDATE. + find_result = MagicMock() + find_result.all.return_value = [] # нет trigram-кандидатов + price_select = MagicMock() + price_select.all.return_value = [(5, "2026-01-01", "COMFORT")] + update_result = MagicMock() + update_result.rowcount = 1 + mock_db = MagicMock() + mock_db.begin_nested.return_value = _savepoint_cm() + mock_db.execute.side_effect = [find_result, price_select, update_result] + + result = run_backfill(mock_db) + + assert result["yandex_updated"] == 0 + assert result["yandex_skipped"] == 0 + assert result["price_updated"] == 1 diff --git a/data/sql/169_obj_class_fallback.sql b/data/sql/169_obj_class_fallback.sql new file mode 100644 index 00000000..6d5ee377 --- /dev/null +++ b/data/sql/169_obj_class_fallback.sql @@ -0,0 +1,50 @@ +-- 169_obj_class_fallback.sql +-- +-- Контекст (issue #38 — Demand D6): kn-API не отдаёт obj_class для ~3032 ЖК +-- Свердл (Bug_Kn_API_Obj_Class_Always_Null). Это ломает per-class velocity (D2), +-- per-class absorption (D3) и per-class weight profile (I2). +-- +-- Workaround: fuzzy name-match domrf_kn_objects.comm_name ↔ yandex_realty_zk.name +-- через pg_trgm similarity(); опционально — price-inference по +-- yandex_realty_class_prices. Результат пишется в obj_class_fallback с пометкой +-- источника obj_class_source ('domrf' | 'yandex_match' | 'price_inference'). +-- Backfill выполняет ETL-модуль app.services.etl.obj_class_backfill. +-- +-- Порядок применения: схема первой (этот файл), затем backend-код, +-- читающий COALESCE(obj_class, obj_class_fallback). +-- +-- Idempotent: ADD COLUMN IF NOT EXISTS + CREATE INDEX IF NOT EXISTS + CREATE +-- EXTENSION IF NOT EXISTS. Безопасно ре-применять. + +BEGIN; + +-- ── Колонки fallback ──────────────────────────────────────────────────────── +-- obj_class_fallback — производный класс (когда реальный obj_class = NULL). +-- obj_class_source — провенанс класса: +-- 'domrf' — реальный obj_class из kn-API/AI-описания (этим файлом +-- не проставляется; для аудита, проставляет backfill). +-- 'yandex_match' — trigram-match comm_name ↔ yandex_realty_zk.name. +-- 'price_inference' — выведен по price_per_m2 vs yandex_realty_class_prices. +ALTER TABLE domrf_kn_objects + ADD COLUMN IF NOT EXISTS obj_class_fallback TEXT, + ADD COLUMN IF NOT EXISTS obj_class_source TEXT; + +-- ── pg_trgm + GIN trgm индексы для similarity() perf ───────────────────────── +-- similarity(LOWER(a), LOWER(b)) без GIN trgm деградирует в seq-scan на каждой +-- паре (CROSS JOIN LATERAL). Индексы по выражению LOWER(...) ускоряют % / <-> . +CREATE EXTENSION IF NOT EXISTS pg_trgm; + +CREATE INDEX IF NOT EXISTS idx_domrf_kn_objects_comm_name_trgm + ON domrf_kn_objects USING gin (LOWER(comm_name) gin_trgm_ops); + +CREATE INDEX IF NOT EXISTS idx_yandex_realty_zk_name_trgm + ON yandex_realty_zk USING gin (LOWER(name) gin_trgm_ops); + +-- ── Документация ──────────────────────────────────────────────────────────── +COMMENT ON COLUMN domrf_kn_objects.obj_class_fallback IS + 'Производный класс жилья, когда реальный obj_class = NULL (issue #38). ' + 'Источник в obj_class_source. Backfill: app.services.etl.obj_class_backfill.'; +COMMENT ON COLUMN domrf_kn_objects.obj_class_source IS + 'Провенанс класса: domrf | yandex_match | price_inference (issue #38).'; + +COMMIT;