feat(etl): housing-class normalization fallback via yandex_realty trigram match (#38)
kn-API не отдаёт obj_class для ~3032 ЖК Свердл, что ломает per-class velocity/absorption/weight. Добавлен fallback: - 169_obj_class_fallback.sql: obj_class_fallback + obj_class_source колонки, pg_trgm + GIN trgm индексы на LOWER(comm_name) и LOWER(yandex_realty_zk.name). - app/services/etl/obj_class_backfill.py: trigram-match comm_name ↔ yandex_realty_zk.name (similarity >= 0.6, CROSS JOIN LATERAL) → obj_class_source='yandex_match'; price-inference по yandex_realty_class_prices → 'price_inference'. Idempotent, SAVEPOINT per row. - Консьюмеры (analytics_queries / velocity / competitors): класс-фильтр и dominant_class читают COALESCE(obj_class, obj_class_fallback) — реальный класс в приоритете, поведение для размеченных ЖК не меняется. Внутренний backfill — response-схема (parcel.py / OpenAPI) не меняется. Refs #38
This commit is contained in:
parent
67d41df880
commit
95b2fb34e2
6 changed files with 640 additions and 12 deletions
|
|
@ -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,
|
||||
|
|
|
|||
328
backend/app/services/etl/obj_class_backfill.py
Normal file
328
backend/app/services/etl/obj_class_backfill.py
Normal file
|
|
@ -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,
|
||||
}
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
227
backend/tests/services/test_obj_class_backfill.py
Normal file
227
backend/tests/services/test_obj_class_backfill.py
Normal file
|
|
@ -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
|
||||
50
data/sql/169_obj_class_fallback.sql
Normal file
50
data/sql/169_obj_class_fallback.sql
Normal file
|
|
@ -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;
|
||||
Loading…
Add table
Reference in a new issue