fix(tradein/data): external_valuations.house_id — резолв в write-path + идемпотентный бэкфилл (#2236)
This commit is contained in:
bot-backend
parent
14d76fb136
commit
291717fa2f
4 changed files with 753 additions and 3 deletions
|
|
@ -509,11 +509,17 @@ async def _get_or_fetch_yandex_valuation_cached(
|
|||
address: str,
|
||||
offer_category: str = YANDEX_VALUATION_DEFAULT_CATEGORY,
|
||||
offer_type: str = YANDEX_VALUATION_DEFAULT_TYPE,
|
||||
house_id: int | None = None,
|
||||
) -> YandexValuationResult | None:
|
||||
"""Cached Yandex Valuation lookup. TTL 24h via external_valuations table.
|
||||
|
||||
Returns None on any error / cache miss + fetch failure — caller continues
|
||||
without Yandex enrichment (graceful degradation).
|
||||
|
||||
house_id: канонический дом, уже разрезолвленный estimate-путём через
|
||||
match_house_readonly. Пишется в external_valuations.house_id при fresh-fetch
|
||||
(best-effort; None → колонка остаётся NULL). ON CONFLICT сохраняет уже
|
||||
проставленный house_id через COALESCE.
|
||||
"""
|
||||
cache_key = _yandex_valuation_cache_key(address, offer_category, offer_type)
|
||||
|
||||
|
|
@ -580,15 +586,20 @@ async def _get_or_fetch_yandex_valuation_cached(
|
|||
"""
|
||||
INSERT INTO external_valuations (
|
||||
source, cache_key, address,
|
||||
house_id,
|
||||
raw_payload,
|
||||
fetched_at, expires_at
|
||||
) VALUES (
|
||||
'yandex_valuation', :ck, :addr,
|
||||
CAST(:hid AS bigint),
|
||||
CAST(:payload AS jsonb),
|
||||
NOW(), NOW() + (:ttl_hours || ' hours')::interval
|
||||
)
|
||||
ON CONFLICT (source, cache_key) DO UPDATE
|
||||
SET raw_payload = EXCLUDED.raw_payload,
|
||||
house_id = COALESCE(
|
||||
EXCLUDED.house_id, external_valuations.house_id
|
||||
),
|
||||
fetched_at = NOW(),
|
||||
expires_at = NOW() + (:ttl_hours || ' hours')::interval
|
||||
"""
|
||||
|
|
@ -596,6 +607,7 @@ async def _get_or_fetch_yandex_valuation_cached(
|
|||
{
|
||||
"ck": cache_key,
|
||||
"addr": address,
|
||||
"hid": house_id,
|
||||
"payload": json.dumps(result.model_dump(mode="json"), ensure_ascii=False),
|
||||
"ttl_hours": YANDEX_VALUATION_CACHE_TTL_HOURS,
|
||||
},
|
||||
|
|
@ -2907,17 +2919,19 @@ async def estimate_quality(
|
|||
yandex_val: YandexValuationResult | None = None
|
||||
if geo is not None and geo.full_address:
|
||||
yandex_val = await _with_budget(
|
||||
_get_or_fetch_yandex_valuation_cached(db, address=geo.full_address),
|
||||
_get_or_fetch_yandex_valuation_cached(
|
||||
db, address=geo.full_address, house_id=target_house_id
|
||||
),
|
||||
settings.estimate_yandex_valuation_timeout_s,
|
||||
label="yandex_valuation",
|
||||
)
|
||||
if yandex_val is not None:
|
||||
saved_hist = await asyncio.to_thread(_save_yandex_history_items, db, yandex_val)
|
||||
logger.info(
|
||||
"yandex_valuation: history items processed=%d saved=%d"
|
||||
" (house_id=NULL — matching deferred)",
|
||||
"yandex_valuation: history items processed=%d saved=%d (ext_val house_id=%s)",
|
||||
len(yandex_val.history_items),
|
||||
saved_hist,
|
||||
target_house_id,
|
||||
)
|
||||
|
||||
# ── Stage 9: Cian Valuation as 7th source (on-demand, 24h cached) ──────
|
||||
|
|
@ -2942,6 +2956,7 @@ async def estimate_quality(
|
|||
repair_type="cosmetic",
|
||||
deal_type="sale",
|
||||
use_cache=True,
|
||||
house_id=target_house_id,
|
||||
),
|
||||
settings.estimate_cian_valuation_timeout_s,
|
||||
label="cian_valuation",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,366 @@
|
|||
"""Backfill external_valuations.house_id for existing rows (issue #2236).
|
||||
|
||||
Внешние оценки (Cian Valuation Calculator, Yandex Оценка квартиры) исторически
|
||||
писались в `external_valuations` без привязки к канонической таблице `houses`:
|
||||
на проде 1579/1579 строк имели `house_id IS NULL`. Write-path теперь резолвит
|
||||
house_id тем же матчером, что и estimate (`match_house_readonly`), но уже
|
||||
накопленные строки остаются без ссылки.
|
||||
|
||||
Эта one-shot джоба идёт по всем `external_valuations WHERE house_id IS NULL`
|
||||
и резолвит канонический дом ТЕМ ЖЕ путём, что estimate-запрос — через
|
||||
`match_house_readonly(db, address=...)`. В external_valuations нет lat/lon и
|
||||
кадастра, поэтому резолв опирается на адрес (Tier 1 fingerprint / — при пустых
|
||||
координатах — по нормализованному адресу дома). Best-effort: дом не нашёлся →
|
||||
house_id остаётся NULL, строка не трогается.
|
||||
|
||||
Per-row SAVEPOINT (`db.begin_nested()`) per `.claude/rules/backend.md`: один
|
||||
битый матч не должен ронять весь батч.
|
||||
|
||||
Идемпотентность / резюмируемость:
|
||||
- Source-query: `WHERE house_id IS NULL` — уже проставленные строки
|
||||
выпадают из выборки. Повторный прогон резолвит только оставшиеся NULL,
|
||||
результат тот же.
|
||||
- UPDATE guard `AND house_id IS NULL` — гонок с write-path не перетирает
|
||||
уже проставленный house_id.
|
||||
- Курсор `id > :after_id` двигается и по нерезолвленным строкам, поэтому
|
||||
один прогон не зацикливается на строке, которую матчер вернул None.
|
||||
|
||||
Никаких сетевых вызовов — матчер работает целиком в БД.
|
||||
|
||||
Usage:
|
||||
DATABASE_URL=postgresql+psycopg://... \\
|
||||
python -m scripts.backfill_external_valuations_house_id --batch-size 500
|
||||
|
||||
# Канарейка сначала (без записи)
|
||||
python -m scripts.backfill_external_valuations_house_id --limit 100 --dry-run
|
||||
|
||||
# Ограничить одним источником
|
||||
python -m scripts.backfill_external_valuations_house_id --source cian_valuation
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# Allow running both as `python -m scripts.backfill_external_valuations_house_id`
|
||||
# (preferred) and as a stand-alone file (fallback for adhoc invocation).
|
||||
try:
|
||||
from app.core.db import SessionLocal # type: ignore[import-not-found]
|
||||
from app.services.matching.houses import ( # type: ignore[import-not-found]
|
||||
match_house_readonly,
|
||||
)
|
||||
except ImportError: # pragma: no cover — fallback for adhoc invocation
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
from app.core.db import SessionLocal
|
||||
from app.services.matching.houses import match_house_readonly
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(name)s %(message)s",
|
||||
)
|
||||
logger = logging.getLogger("backfill_external_valuations_house_id")
|
||||
|
||||
# Резолвер: (db, address) -> house_id | None. Инъектируемый для тестов, по
|
||||
# умолчанию — тот же match_house_readonly, что зовёт estimate-путь.
|
||||
Matcher = Callable[[Session, str], "int | None"]
|
||||
|
||||
|
||||
def _default_matcher(db: Session, address: str) -> int | None:
|
||||
"""estimate-совместимый резолв дома по адресу (read-only, без создания)."""
|
||||
match = match_house_readonly(db, address=address)
|
||||
return match[0] if match is not None else None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Domain types
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExtValRow:
|
||||
"""Одна строка external_valuations, которую резолвим."""
|
||||
|
||||
id: int
|
||||
source: str
|
||||
address: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class Stats:
|
||||
"""Агрегатные счётчики — печатаются по батчу и в конце прогона."""
|
||||
|
||||
processed: int = 0
|
||||
resolved: int = 0
|
||||
unresolved: int = 0
|
||||
errors: int = 0
|
||||
by_source: dict[str, dict[str, int]] = field(default_factory=dict)
|
||||
|
||||
def bump(self, source: str, key: str) -> None:
|
||||
bucket = self.by_source.setdefault(source, defaultdict(int))
|
||||
bucket[key] += 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Source query — stream rows without a house_id
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _build_select_sql(*, source: str | None) -> str:
|
||||
"""Streaming SELECT — только строки с house_id IS NULL и непустым адресом."""
|
||||
where_source = " AND source = :source " if source is not None else ""
|
||||
return (
|
||||
"SELECT id, source, address "
|
||||
"FROM external_valuations "
|
||||
"WHERE house_id IS NULL "
|
||||
" AND address IS NOT NULL "
|
||||
" AND id > :after_id "
|
||||
f" {where_source} "
|
||||
"ORDER BY id "
|
||||
"LIMIT CAST(:limit AS int)"
|
||||
)
|
||||
|
||||
|
||||
def _fetch_batch(
|
||||
db: Session, *, after_id: int, batch_size: int, source: str | None
|
||||
) -> list[ExtValRow]:
|
||||
"""Следующий батч нерезолвленных строк по возрастанию id."""
|
||||
sql = _build_select_sql(source=source)
|
||||
params: dict[str, Any] = {"after_id": after_id, "limit": batch_size}
|
||||
if source is not None:
|
||||
params["source"] = source
|
||||
|
||||
rows = db.execute(text(sql), params).mappings().all()
|
||||
return [ExtValRow(id=int(r["id"]), source=r["source"], address=r["address"]) for r in rows]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Per-row work
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _process_row(
|
||||
db: Session,
|
||||
row: ExtValRow,
|
||||
*,
|
||||
dry_run: bool,
|
||||
stats: Stats,
|
||||
matcher: Matcher,
|
||||
) -> None:
|
||||
"""Резолвит дом и (если найден) проставляет house_id одной строке.
|
||||
|
||||
Обёрнуто в per-row SAVEPOINT: битый матч/UPDATE не ломает батч. Best-effort —
|
||||
matcher вернул None → строка не трогается, house_id остаётся NULL.
|
||||
"""
|
||||
stats.processed += 1
|
||||
stats.bump(row.source, "processed")
|
||||
|
||||
try:
|
||||
if dry_run:
|
||||
house_id = matcher(db, row.address)
|
||||
else:
|
||||
with db.begin_nested():
|
||||
house_id = matcher(db, row.address)
|
||||
if house_id is not None:
|
||||
db.execute(
|
||||
text(
|
||||
"UPDATE external_valuations "
|
||||
" SET house_id = CAST(:hid AS bigint) "
|
||||
" WHERE id = CAST(:eid AS bigint) "
|
||||
" AND house_id IS NULL"
|
||||
),
|
||||
{"hid": house_id, "eid": row.id},
|
||||
)
|
||||
|
||||
if house_id is not None:
|
||||
stats.resolved += 1
|
||||
stats.bump(row.source, "resolved")
|
||||
else:
|
||||
stats.unresolved += 1
|
||||
stats.bump(row.source, "unresolved")
|
||||
|
||||
logger.debug(
|
||||
"backfill ext_val id=%s source=%s house_id=%s",
|
||||
row.id,
|
||||
row.source,
|
||||
house_id,
|
||||
)
|
||||
except Exception as e:
|
||||
stats.errors += 1
|
||||
stats.bump(row.source, "errors")
|
||||
logger.warning(
|
||||
"backfill failed ext_val id=%s source=%s: %s",
|
||||
row.id,
|
||||
row.source,
|
||||
e,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Driver
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def run_backfill(
|
||||
db: Session,
|
||||
*,
|
||||
batch_size: int,
|
||||
limit: int | None,
|
||||
source: str | None,
|
||||
dry_run: bool,
|
||||
matcher: Matcher = _default_matcher,
|
||||
) -> Stats:
|
||||
"""Главный драйвер — стримит батчи и проставляет house_id.
|
||||
|
||||
Args:
|
||||
db: SQLAlchemy Session.
|
||||
batch_size: строк за один SELECT (каждый батч коммитится).
|
||||
limit: остановиться после стольких строк (--limit). None = до конца.
|
||||
source: фильтр по source ('cian_valuation'/'yandex_valuation'). None = все.
|
||||
dry_run: не писать в БД, только считать.
|
||||
matcher: (db, address) -> house_id|None; по умолчанию match_house_readonly.
|
||||
"""
|
||||
stats = Stats()
|
||||
after_id = 0
|
||||
batch_idx = 0
|
||||
|
||||
while True:
|
||||
if limit is not None and stats.processed >= limit:
|
||||
logger.info(
|
||||
"limit reached: stopping (processed=%d, limit=%d)",
|
||||
stats.processed,
|
||||
limit,
|
||||
)
|
||||
break
|
||||
|
||||
effective_size = batch_size
|
||||
if limit is not None:
|
||||
effective_size = min(batch_size, limit - stats.processed)
|
||||
if effective_size <= 0:
|
||||
break
|
||||
|
||||
batch = _fetch_batch(db, after_id=after_id, batch_size=effective_size, source=source)
|
||||
if not batch:
|
||||
logger.info("no more rows — done")
|
||||
break
|
||||
|
||||
batch_idx += 1
|
||||
for row in batch:
|
||||
_process_row(db, row, dry_run=dry_run, stats=stats, matcher=matcher)
|
||||
after_id = max(after_id, row.id)
|
||||
|
||||
if not dry_run:
|
||||
db.commit()
|
||||
|
||||
logger.info(
|
||||
"batch %d done: size=%d total processed=%d resolved=%d unresolved=%d errors=%d",
|
||||
batch_idx,
|
||||
len(batch),
|
||||
stats.processed,
|
||||
stats.resolved,
|
||||
stats.unresolved,
|
||||
stats.errors,
|
||||
)
|
||||
|
||||
return stats
|
||||
|
||||
|
||||
def _log_summary(stats: Stats, *, dry_run: bool) -> None:
|
||||
"""Финальная разбивка по источникам + итог."""
|
||||
logger.info("=" * 72)
|
||||
logger.info(
|
||||
"backfill done (dry_run=%s): processed=%d resolved=%d unresolved=%d errors=%d",
|
||||
dry_run,
|
||||
stats.processed,
|
||||
stats.resolved,
|
||||
stats.unresolved,
|
||||
stats.errors,
|
||||
)
|
||||
for src, bucket in sorted(stats.by_source.items()):
|
||||
logger.info(
|
||||
" %-18s processed=%d resolved=%d unresolved=%d errors=%d",
|
||||
src,
|
||||
bucket.get("processed", 0),
|
||||
bucket.get("resolved", 0),
|
||||
bucket.get("unresolved", 0),
|
||||
bucket.get("errors", 0),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
"""argparse setup, вынесено ради тестируемости."""
|
||||
p = argparse.ArgumentParser(
|
||||
description=(
|
||||
"Backfill external_valuations.house_id — резолвит канонический дом "
|
||||
"по адресу тем же match_house_readonly, что и estimate-путь. "
|
||||
"Идемпотентно на re-run."
|
||||
),
|
||||
)
|
||||
p.add_argument(
|
||||
"--batch-size",
|
||||
type=int,
|
||||
default=500,
|
||||
help="Строк за SELECT (default: 500). Каждый батч коммитится.",
|
||||
)
|
||||
p.add_argument(
|
||||
"--limit",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Потолок на общее число строк за прогон (для канарейки).",
|
||||
)
|
||||
p.add_argument(
|
||||
"--source",
|
||||
choices=("cian_valuation", "yandex_valuation"),
|
||||
default=None,
|
||||
help="Ограничить одним source. Default: все.",
|
||||
)
|
||||
p.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Логировать что было бы сделано; без записи в БД.",
|
||||
)
|
||||
return p.parse_args(argv)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
"""CLI entry point. Возвращает число обработанных строк за прогон."""
|
||||
args = _parse_args(argv)
|
||||
logger.info(
|
||||
"starting backfill: batch_size=%d limit=%s source=%s dry_run=%s",
|
||||
args.batch_size,
|
||||
args.limit if args.limit is not None else "all",
|
||||
args.source or "all",
|
||||
args.dry_run,
|
||||
)
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
stats = run_backfill(
|
||||
db,
|
||||
batch_size=args.batch_size,
|
||||
limit=args.limit,
|
||||
source=args.source,
|
||||
dry_run=args.dry_run,
|
||||
)
|
||||
_log_summary(stats, dry_run=args.dry_run)
|
||||
return stats.processed
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
sys.exit(0 if main() >= 0 else 1)
|
||||
|
|
@ -0,0 +1,214 @@
|
|||
"""Tests for the external_valuations.house_id backfill job (issue #2236).
|
||||
|
||||
Coverage:
|
||||
- happy path: resolvable rows get UPDATE-ed with house_id, counts add up.
|
||||
- unresolved rows (matcher → None) stay untouched, no UPDATE, counted.
|
||||
- per-row error isolation: one raising matcher call doesn't kill the batch.
|
||||
- batching: batch_size < total → multiple batches, one commit each.
|
||||
- idempotency: after a run marks rows resolved, a re-run finds nothing.
|
||||
- dry-run: no UPDATE / no commit.
|
||||
|
||||
No real Postgres — DB is a stateful MagicMock modelling the
|
||||
`WHERE house_id IS NULL AND id > :after_id` source query and the guarded
|
||||
UPDATE (same convention as test_backfill_house_coords).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
# Settings требует DATABASE_URL на import.
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from scripts.backfill_external_valuations_house_id import (
|
||||
ExtValRow,
|
||||
_fetch_batch,
|
||||
run_backfill,
|
||||
)
|
||||
|
||||
|
||||
def _make_db(pool: list[dict]):
|
||||
"""Stateful MagicMock DB.
|
||||
|
||||
`pool` = list of {id, source, address, house_id}. SELECT returns rows with
|
||||
house_id IS NULL AND id > after_id (ordered, capped by limit). UPDATE sets
|
||||
house_id on the matching pool row (guarded on house_id IS NULL).
|
||||
Records commits.
|
||||
"""
|
||||
db = MagicMock()
|
||||
db.begin_nested.return_value.__enter__ = lambda self: self
|
||||
db.begin_nested.return_value.__exit__ = lambda self, *a: False
|
||||
commits: list[int] = []
|
||||
|
||||
def execute_side_effect(sql, params=None):
|
||||
sql_str = str(sql)
|
||||
result = MagicMock()
|
||||
if "SELECT id, source, address" in sql_str:
|
||||
after_id = params["after_id"]
|
||||
limit = params["limit"]
|
||||
src = params.get("source")
|
||||
rows = [
|
||||
{"id": r["id"], "source": r["source"], "address": r["address"]}
|
||||
for r in sorted(pool, key=lambda x: x["id"])
|
||||
if r["house_id"] is None
|
||||
and r["id"] > after_id
|
||||
and r["address"] is not None
|
||||
and (src is None or r["source"] == src)
|
||||
][:limit]
|
||||
result.mappings.return_value.all.return_value = rows
|
||||
elif "UPDATE external_valuations" in sql_str:
|
||||
for r in pool:
|
||||
if r["id"] == params["eid"] and r["house_id"] is None:
|
||||
r["house_id"] = params["hid"]
|
||||
return result
|
||||
|
||||
db.execute.side_effect = execute_side_effect
|
||||
db.commit.side_effect = lambda: commits.append(1)
|
||||
db.close = MagicMock()
|
||||
return db, commits
|
||||
|
||||
|
||||
def _pool(n: int) -> list[dict]:
|
||||
return [
|
||||
{
|
||||
"id": i,
|
||||
"source": "cian_valuation",
|
||||
"address": f"Екатеринбург, ул Тест, {i}",
|
||||
"house_id": None,
|
||||
}
|
||||
for i in range(1, n + 1)
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _fetch_batch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_fetch_batch_maps_rows():
|
||||
db, _ = _make_db(_pool(3))
|
||||
rows = _fetch_batch(db, after_id=0, batch_size=10, source=None)
|
||||
assert [r.id for r in rows] == [1, 2, 3]
|
||||
assert all(isinstance(r, ExtValRow) for r in rows)
|
||||
|
||||
|
||||
def test_fetch_batch_respects_after_id_and_limit():
|
||||
db, _ = _make_db(_pool(5))
|
||||
rows = _fetch_batch(db, after_id=2, batch_size=2, source=None)
|
||||
assert [r.id for r in rows] == [3, 4]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Happy path — resolvable rows updated
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_run_resolves_and_updates():
|
||||
pool = _pool(3)
|
||||
db, commits = _make_db(pool)
|
||||
matcher = MagicMock(return_value=555)
|
||||
|
||||
stats = run_backfill(db, batch_size=10, limit=None, source=None, dry_run=False, matcher=matcher)
|
||||
|
||||
assert stats.processed == 3
|
||||
assert stats.resolved == 3
|
||||
assert stats.unresolved == 0
|
||||
assert stats.errors == 0
|
||||
assert all(r["house_id"] == 555 for r in pool)
|
||||
assert len(commits) == 1 # one batch → one commit
|
||||
|
||||
|
||||
def test_run_unresolved_rows_untouched():
|
||||
"""matcher → None: house_id остаётся NULL, UPDATE не вызывается, счётчик растёт."""
|
||||
pool = _pool(2)
|
||||
db, _ = _make_db(pool)
|
||||
matcher = MagicMock(return_value=None)
|
||||
|
||||
stats = run_backfill(db, batch_size=10, limit=None, source=None, dry_run=False, matcher=matcher)
|
||||
|
||||
assert stats.processed == 2
|
||||
assert stats.resolved == 0
|
||||
assert stats.unresolved == 2
|
||||
assert all(r["house_id"] is None for r in pool)
|
||||
|
||||
|
||||
def test_run_row_error_isolated():
|
||||
"""Матчер кидает на одной строке — батч продолжается, ошибка посчитана."""
|
||||
pool = _pool(3)
|
||||
db, _ = _make_db(pool)
|
||||
|
||||
def matcher(_db, addr):
|
||||
if "2" in addr:
|
||||
raise RuntimeError("boom")
|
||||
return 999
|
||||
|
||||
stats = run_backfill(db, batch_size=10, limit=None, source=None, dry_run=False, matcher=matcher)
|
||||
assert stats.processed == 3
|
||||
assert stats.resolved == 2
|
||||
assert stats.errors == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Batching
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_run_multiple_batches_commit_each():
|
||||
pool = _pool(5)
|
||||
db, commits = _make_db(pool)
|
||||
matcher = MagicMock(return_value=7)
|
||||
|
||||
stats = run_backfill(db, batch_size=2, limit=None, source=None, dry_run=False, matcher=matcher)
|
||||
# 5 rows / batch 2 → batches of 2, 2, 1 = 3 commits
|
||||
assert stats.processed == 5
|
||||
assert stats.resolved == 5
|
||||
assert len(commits) == 3
|
||||
|
||||
|
||||
def test_run_limit_caps_total():
|
||||
pool = _pool(10)
|
||||
db, _ = _make_db(pool)
|
||||
matcher = MagicMock(return_value=1)
|
||||
|
||||
stats = run_backfill(db, batch_size=3, limit=4, source=None, dry_run=False, matcher=matcher)
|
||||
assert stats.processed == 4
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Idempotency
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_run_idempotent_second_pass_noop():
|
||||
pool = _pool(4)
|
||||
db, _ = _make_db(pool)
|
||||
matcher = MagicMock(return_value=42)
|
||||
|
||||
first = run_backfill(db, batch_size=10, limit=None, source=None, dry_run=False, matcher=matcher)
|
||||
assert first.processed == 4 and first.resolved == 4
|
||||
|
||||
# Все строки теперь house_id != NULL → source-query пуст на re-run.
|
||||
second = run_backfill(
|
||||
db, batch_size=10, limit=None, source=None, dry_run=False, matcher=matcher
|
||||
)
|
||||
assert second.processed == 0
|
||||
assert second.resolved == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dry-run
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_run_dry_run_no_writes():
|
||||
pool = _pool(3)
|
||||
db, commits = _make_db(pool)
|
||||
matcher = MagicMock(return_value=123)
|
||||
|
||||
stats = run_backfill(db, batch_size=10, limit=None, source=None, dry_run=True, matcher=matcher)
|
||||
assert stats.processed == 3
|
||||
assert stats.resolved == 3 # matcher нашёл дом
|
||||
assert all(r["house_id"] is None for r in pool) # но НЕ записали
|
||||
assert commits == []
|
||||
155
tradein-mvp/backend/tests/test_extval_house_id_write_path.py
Normal file
155
tradein-mvp/backend/tests/test_extval_house_id_write_path.py
Normal file
|
|
@ -0,0 +1,155 @@
|
|||
"""Write-path tests: external_valuations.house_id резолвится при сохранении (#2236).
|
||||
|
||||
Cian и Yandex valuation-кэши теперь принимают house_id (уже разрезолвленный
|
||||
estimate-путём через match_house_readonly) и пишут его в external_valuations.
|
||||
Best-effort: house_id=None → колонка NULL, сохранение НЕ падает.
|
||||
|
||||
DB — MagicMock, записываем параметры db.execute (никакого реального Postgres,
|
||||
как в test_yandex_valuation_save / test_backfill_house_coords).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
# Settings требует DATABASE_URL на import. Ставим фиктивный DSN заранее.
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from app.services.scrapers.cian_valuation import CianValuationResult, _save_to_cache
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Cian valuation — _save_to_cache прокидывает house_id в INSERT params
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _cian_result() -> CianValuationResult:
|
||||
return CianValuationResult(
|
||||
sale_price_rub=10_000_000,
|
||||
sale_accuracy=85.0,
|
||||
sale_price_from=9_000_000,
|
||||
sale_price_to=11_000_000,
|
||||
rent_price_rub=40_000,
|
||||
rent_accuracy=70.0,
|
||||
chart=[],
|
||||
chart_change_pct=None,
|
||||
chart_change_direction=None,
|
||||
external_house_id=54016,
|
||||
filters_hash="abc",
|
||||
raw_state=None,
|
||||
)
|
||||
|
||||
|
||||
def _cian_save(db: MagicMock, house_id: int | None) -> dict:
|
||||
_save_to_cache(
|
||||
db,
|
||||
cache_key="ck",
|
||||
address="Екатеринбург, улица Малышева, 51",
|
||||
total_area=50.0,
|
||||
rooms_count=2,
|
||||
floor=3,
|
||||
total_floors=9,
|
||||
repair_type="cosmetic",
|
||||
deal_type="sale",
|
||||
result=_cian_result(),
|
||||
house_id=house_id,
|
||||
)
|
||||
# _save_to_cache делает один db.execute(INSERT ...) + commit
|
||||
return db.execute.call_args_list[0].args[1]
|
||||
|
||||
|
||||
def test_cian_save_persists_house_id():
|
||||
"""Резолвленный house_id попадает в INSERT-параметры."""
|
||||
db = MagicMock()
|
||||
params = _cian_save(db, house_id=777)
|
||||
assert params["hid"] == 777
|
||||
db.commit.assert_called_once()
|
||||
|
||||
|
||||
def test_cian_save_null_house_id_no_exception():
|
||||
"""house_id=None → hid=None в params, сохранение не падает."""
|
||||
db = MagicMock()
|
||||
params = _cian_save(db, house_id=None)
|
||||
assert params["hid"] is None
|
||||
db.commit.assert_called_once()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Yandex valuation — _get_or_fetch_yandex_valuation_cached пишет house_id
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_db_cache_miss() -> MagicMock:
|
||||
"""MagicMock db, у которого cache-lookup даёт miss (first() → None)."""
|
||||
db = MagicMock()
|
||||
db.execute.return_value.mappings.return_value.first.return_value = None
|
||||
return db
|
||||
|
||||
|
||||
def _yandex_result():
|
||||
from app.services.scrapers.yandex_valuation import (
|
||||
ValuationHouseMeta,
|
||||
YandexValuationResult,
|
||||
)
|
||||
|
||||
return YandexValuationResult(
|
||||
address="Екатеринбург, улица Куйбышева, 106",
|
||||
offer_category="APARTMENT",
|
||||
offer_type="SELL",
|
||||
page=1,
|
||||
source_url="https://realty.yandex.ru/otsenka/?address=test&page=1",
|
||||
house=ValuationHouseMeta(year_built=2005, total_floors=25),
|
||||
history_items=[],
|
||||
)
|
||||
|
||||
|
||||
def _insert_params(db: MagicMock) -> dict:
|
||||
"""Найти params INSERT INTO external_valuations среди вызовов db.execute."""
|
||||
for call in db.execute.call_args_list:
|
||||
args = call.args
|
||||
if len(args) >= 2 and "INSERT INTO external_valuations" in str(args[0]):
|
||||
return args[1]
|
||||
raise AssertionError("INSERT INTO external_valuations not found in db.execute calls")
|
||||
|
||||
|
||||
async def test_yandex_save_persists_house_id():
|
||||
"""Fresh fetch пишет house_id в INSERT external_valuations."""
|
||||
from app.services import estimator
|
||||
|
||||
db = _make_db_cache_miss()
|
||||
scraper = AsyncMock()
|
||||
scraper.fetch_house_history = AsyncMock(return_value=_yandex_result())
|
||||
cm = MagicMock()
|
||||
cm.__aenter__ = AsyncMock(return_value=scraper)
|
||||
cm.__aexit__ = AsyncMock(return_value=False)
|
||||
|
||||
with patch.object(estimator, "YandexValuationScraper", return_value=cm):
|
||||
res = await estimator._get_or_fetch_yandex_valuation_cached(
|
||||
db, address="Екатеринбург, улица Куйбышева, 106", house_id=4242
|
||||
)
|
||||
|
||||
assert res is not None
|
||||
params = _insert_params(db)
|
||||
assert params["hid"] == 4242
|
||||
|
||||
|
||||
async def test_yandex_save_null_house_id_no_exception():
|
||||
"""house_id=None → hid=None, INSERT проходит без исключения."""
|
||||
from app.services import estimator
|
||||
|
||||
db = _make_db_cache_miss()
|
||||
scraper = AsyncMock()
|
||||
scraper.fetch_house_history = AsyncMock(return_value=_yandex_result())
|
||||
cm = MagicMock()
|
||||
cm.__aenter__ = AsyncMock(return_value=scraper)
|
||||
cm.__aexit__ = AsyncMock(return_value=False)
|
||||
|
||||
with patch.object(estimator, "YandexValuationScraper", return_value=cm):
|
||||
res = await estimator._get_or_fetch_yandex_valuation_cached(
|
||||
db, address="Екатеринбург, улица Куйбышева, 106"
|
||||
)
|
||||
|
||||
assert res is not None
|
||||
params = _insert_params(db)
|
||||
assert params["hid"] is None
|
||||
Loading…
Add table
Reference in a new issue