bot-backend
9c52e0b29f
Some checks failed
Deploy / build-worker (push) Blocked by required conditions
Deploy / build-frontend (push) Blocked by required conditions
Deploy / deploy (push) Blocked by required conditions
Deploy / build-backend (push) Blocked by required conditions
Deploy / changes (push) Has been cancelled
housing-class normalization fallback via yandex_realty trigram (#38): migration 169 + ETL + COALESCE consumers. Refs #38
328 lines
13 KiB
Python
328 lines
13 KiB
Python
"""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,
|
||
}
|