fix(tradein): implement listings_snapshots + offer_price_history writers (#567) #624
4 changed files with 635 additions and 3 deletions
|
|
@ -30,6 +30,7 @@ from sqlalchemy.orm import Session
|
||||||
from tenacity import retry, stop_after_attempt, wait_exponential
|
from tenacity import retry, stop_after_attempt, wait_exponential
|
||||||
|
|
||||||
from app.services.matching import match_or_create_house, upsert_listing_source
|
from app.services.matching import match_or_create_house, upsert_listing_source
|
||||||
|
from app.services.scrapers.snapshot_writer import upsert_listing_snapshot
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
@ -201,9 +202,22 @@ class BaseScraper(ABC):
|
||||||
|
|
||||||
|
|
||||||
# ── Запись пачки результатов в Postgres ─────────────────────────────────────
|
# ── Запись пачки результатов в Postgres ─────────────────────────────────────
|
||||||
def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
|
def save_listings(
|
||||||
|
db: Session,
|
||||||
|
lots: list[ScrapedLot],
|
||||||
|
*,
|
||||||
|
run_id: int | None = None,
|
||||||
|
) -> tuple[int, int]:
|
||||||
"""Пишем list[ScrapedLot] в `listings` с upsert по dedup_hash.
|
"""Пишем list[ScrapedLot] в `listings` с upsert по dedup_hash.
|
||||||
|
|
||||||
|
Дополнительно пишет point-in-time snapshot в listings_snapshots (ON CONFLICT DO UPDATE
|
||||||
|
— берётся последний за день). Ошибка записи snapshot НЕ прерывает основной INSERT.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
db: SQLAlchemy session.
|
||||||
|
lots: список распарсенных объявлений.
|
||||||
|
run_id: FK scrape_runs.id текущего run'а (опционально; None при ad-hoc вызовах).
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
(inserted, updated) — counters для логов.
|
(inserted, updated) — counters для логов.
|
||||||
"""
|
"""
|
||||||
|
|
@ -326,6 +340,28 @@ def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
|
||||||
else:
|
else:
|
||||||
updated += 1
|
updated += 1
|
||||||
|
|
||||||
|
# ── Snapshot: point-in-time observation in listings_snapshots ───
|
||||||
|
# Fault-tolerant: failure here MUST NOT abort the listings batch.
|
||||||
|
# ON CONFLICT DO UPDATE — самый последний snapshot за день.
|
||||||
|
if listing_id is not None:
|
||||||
|
try:
|
||||||
|
with db.begin_nested():
|
||||||
|
upsert_listing_snapshot(
|
||||||
|
db,
|
||||||
|
listing_id=listing_id,
|
||||||
|
price_rub=lot.price_rub,
|
||||||
|
price_per_m2=ppm2,
|
||||||
|
run_id=run_id,
|
||||||
|
status="active",
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(
|
||||||
|
"save_listings:snapshot_failed source=%s listing_id=%s: %s",
|
||||||
|
lot.source,
|
||||||
|
listing_id,
|
||||||
|
e,
|
||||||
|
)
|
||||||
|
|
||||||
# ── Hook: link listing → house via matching service ─────────────
|
# ── Hook: link listing → house via matching service ─────────────
|
||||||
# Idempotent (UPSERT on listing_sources UNIQUE(ext_source, ext_id))
|
# Idempotent (UPSERT on listing_sources UNIQUE(ext_source, ext_id))
|
||||||
# and fault-tolerant: failure here MUST NOT abort the listings batch.
|
# and fault-tolerant: failure here MUST NOT abort the listings batch.
|
||||||
|
|
|
||||||
|
|
@ -21,6 +21,7 @@ from sqlalchemy import text
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
from app.services.scrapers.cian_state_parser import extract_all_states, extract_state
|
from app.services.scrapers.cian_state_parser import extract_all_states, extract_state
|
||||||
|
from app.services.scrapers.snapshot_writer import upsert_listing_snapshot
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
@ -248,6 +249,37 @@ def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnric
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Snapshot: point-in-time observation в listings_snapshots.
|
||||||
|
# Берём price_rub / price_per_m2 из listings строки (только что обновлённой).
|
||||||
|
# Используем COALESCE — на случай если listings.price_rub NULL (не должно быть, но safe).
|
||||||
|
_snap_row = db.execute(
|
||||||
|
text("""
|
||||||
|
SELECT price_rub, price_per_m2
|
||||||
|
FROM listings
|
||||||
|
WHERE id = CAST(:lid AS bigint)
|
||||||
|
"""),
|
||||||
|
{"lid": listing_id},
|
||||||
|
).fetchone()
|
||||||
|
_snap_written = False
|
||||||
|
if _snap_row is not None and _snap_row.price_rub is not None:
|
||||||
|
try:
|
||||||
|
with db.begin_nested():
|
||||||
|
upsert_listing_snapshot(
|
||||||
|
db,
|
||||||
|
listing_id=listing_id,
|
||||||
|
price_rub=int(_snap_row.price_rub),
|
||||||
|
price_per_m2=int(_snap_row.price_per_m2) if _snap_row.price_per_m2 else None,
|
||||||
|
run_id=None,
|
||||||
|
status="active",
|
||||||
|
)
|
||||||
|
_snap_written = True
|
||||||
|
except Exception as exc:
|
||||||
|
logger.warning(
|
||||||
|
"save_detail_enrichment: snapshot write failed listing_id=%s: %s",
|
||||||
|
listing_id,
|
||||||
|
exc,
|
||||||
|
)
|
||||||
|
|
||||||
# Price changes — INSERT new rows, de-dup by listing_id + change_time
|
# Price changes — INSERT new rows, de-dup by listing_id + change_time
|
||||||
for change in enrichment.price_changes:
|
for change in enrichment.price_changes:
|
||||||
if not change.get("change_time") or not change.get("price_rub"):
|
if not change.get("change_time") or not change.get("price_rub"):
|
||||||
|
|
@ -263,7 +295,7 @@ def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnric
|
||||||
CAST(:diff AS numeric),
|
CAST(:diff AS numeric),
|
||||||
'cian'
|
'cian'
|
||||||
)
|
)
|
||||||
ON CONFLICT DO NOTHING
|
ON CONFLICT ON CONSTRAINT offer_price_history_listing_change_uq DO NOTHING
|
||||||
"""),
|
"""),
|
||||||
{
|
{
|
||||||
"lid": listing_id,
|
"lid": listing_id,
|
||||||
|
|
@ -317,8 +349,10 @@ def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnric
|
||||||
|
|
||||||
db.commit()
|
db.commit()
|
||||||
logger.info(
|
logger.info(
|
||||||
"Cian detail enrichment saved for listing_id=%s (price_changes=%d, agent=%s)",
|
"Cian detail enrichment saved for listing_id=%s "
|
||||||
|
"(price_changes=%d, agent=%s, snapshot=%s)",
|
||||||
listing_id,
|
listing_id,
|
||||||
len(enrichment.price_changes),
|
len(enrichment.price_changes),
|
||||||
bool(enrichment.agent_profile),
|
bool(enrichment.agent_profile),
|
||||||
|
_snap_written,
|
||||||
)
|
)
|
||||||
|
|
|
||||||
92
tradein-mvp/backend/app/services/scrapers/snapshot_writer.py
Normal file
92
tradein-mvp/backend/app/services/scrapers/snapshot_writer.py
Normal file
|
|
@ -0,0 +1,92 @@
|
||||||
|
"""listings_snapshots writer — point-in-time observation per listing per day.
|
||||||
|
|
||||||
|
Таблица listings_snapshots (016_listings_snapshots.sql):
|
||||||
|
PRIMARY KEY (listing_id, snapshot_date) — максимум 1 snapshot в сутки.
|
||||||
|
ON CONFLICT DO UPDATE — берём последний за день (перезаписываем при повторном run'е).
|
||||||
|
|
||||||
|
Используется двумя путями:
|
||||||
|
1. SERP scrape (save_listings в base.py) — записывает цену + позицию в выдаче
|
||||||
|
каждый раз когда listing появляется в поиске.
|
||||||
|
2. Detail backfill (save_detail_enrichment в cian_detail.py) — записывает цену
|
||||||
|
из detail-страницы для listings которые раньше не имели snapshot'а.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from datetime import date
|
||||||
|
|
||||||
|
from sqlalchemy import text
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def upsert_listing_snapshot(
|
||||||
|
db: Session,
|
||||||
|
*,
|
||||||
|
listing_id: int,
|
||||||
|
price_rub: int,
|
||||||
|
price_per_m2: int | None = None,
|
||||||
|
run_id: int | None = None,
|
||||||
|
snapshot_date: date | None = None,
|
||||||
|
position_in_serp: int | None = None,
|
||||||
|
status: str | None = "active",
|
||||||
|
) -> None:
|
||||||
|
"""Записать / обновить snapshot для listing за текущий (или указанный) день.
|
||||||
|
|
||||||
|
Идемпотентно: повторный вызов с теми же listing_id + snapshot_date перезаписывает
|
||||||
|
строку (берём последние данные за день — ON CONFLICT DO UPDATE).
|
||||||
|
|
||||||
|
Args:
|
||||||
|
db: открытая SQLAlchemy session. Commit делает CALLER.
|
||||||
|
listing_id: PK из таблицы listings.
|
||||||
|
price_rub: текущая цена в рублях.
|
||||||
|
price_per_m2: цена за кв.м (вычисляется в caller'е, None если нет).
|
||||||
|
run_id: FK scrape_runs.id — с каким run'ом связан snapshot (None если backfill
|
||||||
|
запускается вне run'а).
|
||||||
|
snapshot_date: дата наблюдения; если None — используется CURRENT_DATE (по БД).
|
||||||
|
position_in_serp: позиция в SERP (1..60+), только при SERP-scrape.
|
||||||
|
status: 'active' / 'closed' / None. Default 'active' при обычном scrape.
|
||||||
|
"""
|
||||||
|
db.execute(
|
||||||
|
text("""
|
||||||
|
INSERT INTO listings_snapshots
|
||||||
|
(listing_id, snapshot_date, run_id,
|
||||||
|
price_rub, price_per_m2, position_in_serp, status, observed_at)
|
||||||
|
VALUES (
|
||||||
|
CAST(:lid AS bigint),
|
||||||
|
COALESCE(CAST(:snap_date AS date), CURRENT_DATE),
|
||||||
|
CAST(:run_id AS bigint),
|
||||||
|
CAST(:price AS bigint),
|
||||||
|
CAST(:ppm2 AS int),
|
||||||
|
CAST(:pos AS int),
|
||||||
|
:status,
|
||||||
|
NOW()
|
||||||
|
)
|
||||||
|
ON CONFLICT (listing_id, snapshot_date) DO UPDATE SET
|
||||||
|
run_id = COALESCE(EXCLUDED.run_id, listings_snapshots.run_id),
|
||||||
|
price_rub = EXCLUDED.price_rub,
|
||||||
|
price_per_m2 = COALESCE(EXCLUDED.price_per_m2, listings_snapshots.price_per_m2),
|
||||||
|
position_in_serp = COALESCE(
|
||||||
|
EXCLUDED.position_in_serp, listings_snapshots.position_in_serp
|
||||||
|
),
|
||||||
|
status = COALESCE(EXCLUDED.status, listings_snapshots.status),
|
||||||
|
observed_at = EXCLUDED.observed_at
|
||||||
|
"""),
|
||||||
|
{
|
||||||
|
"lid": listing_id,
|
||||||
|
"snap_date": snapshot_date,
|
||||||
|
"run_id": run_id,
|
||||||
|
"price": price_rub,
|
||||||
|
"ppm2": price_per_m2,
|
||||||
|
"pos": position_in_serp,
|
||||||
|
"status": status,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
logger.debug(
|
||||||
|
"upsert_listing_snapshot: listing_id=%s date=%s price=%s run_id=%s",
|
||||||
|
listing_id,
|
||||||
|
snapshot_date or "CURRENT_DATE",
|
||||||
|
price_rub,
|
||||||
|
run_id,
|
||||||
|
)
|
||||||
470
tradein-mvp/backend/tests/test_snapshot_writer.py
Normal file
470
tradein-mvp/backend/tests/test_snapshot_writer.py
Normal file
|
|
@ -0,0 +1,470 @@
|
||||||
|
"""Тесты для snapshot_writer.py и интеграции с save_detail_enrichment / save_listings.
|
||||||
|
|
||||||
|
Цель: доказать что:
|
||||||
|
1. upsert_listing_snapshot строит корректный INSERT INTO listings_snapshots.
|
||||||
|
2. Повторный вызов с теми же listing_id + snapshot_date НЕ дублирует строку
|
||||||
|
(ON CONFLICT DO UPDATE обновляет observed_at).
|
||||||
|
3. save_detail_enrichment вызывает snapshot-write после SELECT price из listings.
|
||||||
|
4. save_listings вызывает snapshot-write для каждого сохранённого listing.
|
||||||
|
5. Ошибка snapshot не прерывает основной INSERT в save_listings.
|
||||||
|
|
||||||
|
Используем unittest.mock — без реальной БД.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from contextlib import contextmanager
|
||||||
|
from datetime import date
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
|
from app.services.scrapers.cian_detail import DetailEnrichment, save_detail_enrichment
|
||||||
|
from app.services.scrapers.snapshot_writer import upsert_listing_snapshot
|
||||||
|
|
||||||
|
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def _mock_db_simple() -> MagicMock:
|
||||||
|
"""Минимальный mock db.execute — scalar None для всех вызовов."""
|
||||||
|
db = MagicMock()
|
||||||
|
db.execute.return_value = MagicMock()
|
||||||
|
db.execute.return_value.fetchone.return_value = None
|
||||||
|
db.execute.return_value.scalar_one_or_none.return_value = None
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def _nested():
|
||||||
|
yield MagicMock()
|
||||||
|
|
||||||
|
db.begin_nested.side_effect = _nested
|
||||||
|
return db
|
||||||
|
|
||||||
|
|
||||||
|
def _get_all_sqls(db: MagicMock) -> list[str]:
|
||||||
|
"""Собрать все SQL-строки из всех db.execute() вызовов."""
|
||||||
|
sqls = []
|
||||||
|
for call_args in db.execute.call_args_list:
|
||||||
|
if call_args.args:
|
||||||
|
sqls.append(str(call_args.args[0]))
|
||||||
|
return sqls
|
||||||
|
|
||||||
|
|
||||||
|
def _get_all_params(db: MagicMock) -> list[dict]:
|
||||||
|
"""Собрать все params dict из db.execute() вызовов."""
|
||||||
|
params_list = []
|
||||||
|
for call_args in db.execute.call_args_list:
|
||||||
|
if len(call_args.args) >= 2:
|
||||||
|
params_list.append(call_args.args[1])
|
||||||
|
return params_list
|
||||||
|
|
||||||
|
|
||||||
|
def _find_snapshot_call(db: MagicMock) -> tuple[str, dict] | None:
|
||||||
|
"""Найти первый db.execute вызов содержащий 'listings_snapshots'."""
|
||||||
|
for call_args in db.execute.call_args_list:
|
||||||
|
if not call_args.args:
|
||||||
|
continue
|
||||||
|
sql = str(call_args.args[0])
|
||||||
|
if "listings_snapshots" in sql:
|
||||||
|
params = call_args.args[1] if len(call_args.args) >= 2 else {}
|
||||||
|
return sql, params
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
# ── upsert_listing_snapshot — SQL shape ──────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_upsert_snapshot_sql_contains_insert_and_on_conflict():
|
||||||
|
"""SQL содержит INSERT INTO listings_snapshots и ON CONFLICT (listing_id, snapshot_date)."""
|
||||||
|
db = _mock_db_simple()
|
||||||
|
upsert_listing_snapshot(db, listing_id=42, price_rub=5_000_000)
|
||||||
|
|
||||||
|
sqls = _get_all_sqls(db)
|
||||||
|
assert len(sqls) == 1, "ровно один db.execute вызов"
|
||||||
|
sql = sqls[0]
|
||||||
|
assert "listings_snapshots" in sql
|
||||||
|
assert "ON CONFLICT" in sql
|
||||||
|
assert "DO UPDATE" in sql
|
||||||
|
|
||||||
|
|
||||||
|
def test_upsert_snapshot_params_minimal():
|
||||||
|
"""Минимальный вызов: только listing_id + price_rub — остальные None."""
|
||||||
|
db = _mock_db_simple()
|
||||||
|
upsert_listing_snapshot(db, listing_id=7, price_rub=3_000_000)
|
||||||
|
|
||||||
|
params = _get_all_params(db)
|
||||||
|
assert len(params) == 1
|
||||||
|
p = params[0]
|
||||||
|
assert p["lid"] == 7
|
||||||
|
assert p["price"] == 3_000_000
|
||||||
|
assert p["run_id"] is None
|
||||||
|
assert p["snap_date"] is None
|
||||||
|
assert p["ppm2"] is None
|
||||||
|
assert p["pos"] is None
|
||||||
|
assert p["status"] == "active"
|
||||||
|
|
||||||
|
|
||||||
|
def test_upsert_snapshot_params_full():
|
||||||
|
"""Полный вызов — все поля правильно попадают в params."""
|
||||||
|
db = _mock_db_simple()
|
||||||
|
snap_date = date(2026, 5, 1)
|
||||||
|
upsert_listing_snapshot(
|
||||||
|
db,
|
||||||
|
listing_id=100,
|
||||||
|
price_rub=7_200_000,
|
||||||
|
price_per_m2=130_000,
|
||||||
|
run_id=99,
|
||||||
|
snapshot_date=snap_date,
|
||||||
|
position_in_serp=3,
|
||||||
|
status="active",
|
||||||
|
)
|
||||||
|
|
||||||
|
params = _get_all_params(db)[0]
|
||||||
|
assert params["lid"] == 100
|
||||||
|
assert params["price"] == 7_200_000
|
||||||
|
assert params["ppm2"] == 130_000
|
||||||
|
assert params["run_id"] == 99
|
||||||
|
assert params["snap_date"] == snap_date
|
||||||
|
assert params["pos"] == 3
|
||||||
|
assert params["status"] == "active"
|
||||||
|
|
||||||
|
|
||||||
|
def test_upsert_snapshot_no_commit():
|
||||||
|
"""upsert_listing_snapshot НЕ вызывает db.commit() — commit делает caller."""
|
||||||
|
db = _mock_db_simple()
|
||||||
|
upsert_listing_snapshot(db, listing_id=1, price_rub=1_000_000)
|
||||||
|
db.commit.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
# ── save_detail_enrichment интеграция ────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def _mock_db_detail(price_rub: int = 5_500_000, price_per_m2: int | None = 110_000) -> MagicMock:
|
||||||
|
"""Mock для save_detail_enrichment: SELECT listings возвращает price_rub/price_per_m2."""
|
||||||
|
db = MagicMock()
|
||||||
|
|
||||||
|
# SELECT price_rub, price_per_m2 FROM listings
|
||||||
|
price_row = MagicMock()
|
||||||
|
price_row.price_rub = price_rub
|
||||||
|
price_row.price_per_m2 = price_per_m2
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def _nested():
|
||||||
|
yield MagicMock()
|
||||||
|
|
||||||
|
db.begin_nested.side_effect = _nested
|
||||||
|
|
||||||
|
# Разные вызовы db.execute() возвращают разное:
|
||||||
|
# 1. UPDATE listings → result.fetchone() = None (не используется)
|
||||||
|
# 2. SELECT price → result.fetchone() = price_row
|
||||||
|
# 3+. INSERT offer_price_history / agents → не важно
|
||||||
|
execute_results = []
|
||||||
|
|
||||||
|
def _execute_side_effect(sql, params=None):
|
||||||
|
sql_str = str(sql)
|
||||||
|
mock_result = MagicMock()
|
||||||
|
mock_result.scalar_one_or_none.return_value = None
|
||||||
|
if "SELECT price_rub" in sql_str:
|
||||||
|
mock_result.fetchone.return_value = price_row
|
||||||
|
else:
|
||||||
|
mock_result.fetchone.return_value = None
|
||||||
|
execute_results.append((sql_str, params, mock_result))
|
||||||
|
return mock_result
|
||||||
|
|
||||||
|
db.execute.side_effect = _execute_side_effect
|
||||||
|
db._execute_log = execute_results
|
||||||
|
return db
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_detail_enrichment_writes_snapshot():
|
||||||
|
"""save_detail_enrichment вызывает upsert_listing_snapshot через SELECT listings."""
|
||||||
|
db = _mock_db_detail(price_rub=6_000_000, price_per_m2=120_000)
|
||||||
|
enrichment = DetailEnrichment(ceiling_height=2.7)
|
||||||
|
|
||||||
|
save_detail_enrichment(db, 55, enrichment)
|
||||||
|
|
||||||
|
snap = _find_snapshot_call(db)
|
||||||
|
assert snap is not None, "должен быть INSERT INTO listings_snapshots"
|
||||||
|
_sql, params = snap
|
||||||
|
assert params["lid"] == 55
|
||||||
|
assert params["price"] == 6_000_000
|
||||||
|
assert params["ppm2"] == 120_000
|
||||||
|
assert params["status"] == "active"
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_detail_enrichment_snapshot_skipped_when_no_price_row():
|
||||||
|
"""Если listings строка не найдена (None) — snapshot не записывается, ошибки нет."""
|
||||||
|
db = MagicMock()
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def _nested():
|
||||||
|
yield MagicMock()
|
||||||
|
|
||||||
|
db.begin_nested.side_effect = _nested
|
||||||
|
|
||||||
|
def _exec(sql, params=None):
|
||||||
|
mock_result = MagicMock()
|
||||||
|
mock_result.fetchone.return_value = None
|
||||||
|
mock_result.scalar_one_or_none.return_value = None
|
||||||
|
return mock_result
|
||||||
|
|
||||||
|
db.execute.side_effect = _exec
|
||||||
|
|
||||||
|
enrichment = DetailEnrichment()
|
||||||
|
# Должно завершиться без исключений
|
||||||
|
save_detail_enrichment(db, 99, enrichment)
|
||||||
|
|
||||||
|
# Убедимся что listings_snapshots не упоминается
|
||||||
|
sqls = _get_all_sqls(db)
|
||||||
|
assert not any("listings_snapshots" in s for s in sqls)
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_detail_enrichment_snapshot_failure_does_not_abort():
|
||||||
|
"""Если upsert_listing_snapshot падает — save_detail_enrichment всё равно commit'ит."""
|
||||||
|
db = _mock_db_detail(price_rub=4_000_000)
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"app.services.scrapers.cian_detail.upsert_listing_snapshot",
|
||||||
|
side_effect=RuntimeError("db snapshot blip"),
|
||||||
|
):
|
||||||
|
enrichment = DetailEnrichment(ceiling_height=2.5)
|
||||||
|
# Должно завершиться без исключений — snapshot error логируется, не пробрасывается
|
||||||
|
save_detail_enrichment(db, 10, enrichment)
|
||||||
|
|
||||||
|
db.commit.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_detail_enrichment_writes_offer_price_history_rows():
|
||||||
|
"""price_changes из enrichment записываются в offer_price_history."""
|
||||||
|
db = _mock_db_detail(price_rub=5_000_000)
|
||||||
|
enrichment = DetailEnrichment(
|
||||||
|
price_changes=[
|
||||||
|
{"change_time": "2026-04-01T00:00:00Z", "price_rub": 5_200_000, "diff_percent": None},
|
||||||
|
{"change_time": "2026-04-10T00:00:00Z", "price_rub": 5_000_000, "diff_percent": -3.8},
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
|
save_detail_enrichment(db, 20, enrichment)
|
||||||
|
|
||||||
|
sqls = _get_all_sqls(db)
|
||||||
|
oph_inserts = [s for s in sqls if "offer_price_history" in s]
|
||||||
|
assert len(oph_inserts) == 2, "по одному INSERT для каждой price_change"
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_detail_enrichment_oph_on_conflict_constraint():
|
||||||
|
"""INSERT offer_price_history использует именованный ON CONFLICT ON CONSTRAINT."""
|
||||||
|
db = _mock_db_detail(price_rub=5_000_000)
|
||||||
|
enrichment = DetailEnrichment(
|
||||||
|
price_changes=[
|
||||||
|
{"change_time": "2026-05-01T12:00:00Z", "price_rub": 5_000_000, "diff_percent": 0},
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
|
save_detail_enrichment(db, 21, enrichment)
|
||||||
|
|
||||||
|
sqls = _get_all_sqls(db)
|
||||||
|
oph_sqls = [s for s in sqls if "offer_price_history" in s]
|
||||||
|
assert oph_sqls, "INSERT offer_price_history не найден"
|
||||||
|
assert "offer_price_history_listing_change_uq" in oph_sqls[0], (
|
||||||
|
"ON CONFLICT должен ссылаться на UNIQUE constraint"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_detail_enrichment_skips_price_change_without_change_time():
|
||||||
|
"""price_changes без change_time пропускаются, не вызывают INSERT."""
|
||||||
|
db = _mock_db_detail(price_rub=5_000_000)
|
||||||
|
enrichment = DetailEnrichment(
|
||||||
|
price_changes=[
|
||||||
|
{"change_time": None, "price_rub": 5_000_000}, # пропустить
|
||||||
|
{"change_time": "2026-04-01T00:00:00Z", "price_rub": None}, # пропустить
|
||||||
|
{"change_time": "2026-04-01T00:00:00Z", "price_rub": 5_000_000}, # записать
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
|
save_detail_enrichment(db, 22, enrichment)
|
||||||
|
|
||||||
|
sqls = _get_all_sqls(db)
|
||||||
|
oph_inserts = [s for s in sqls if "offer_price_history" in s]
|
||||||
|
assert len(oph_inserts) == 1
|
||||||
|
|
||||||
|
|
||||||
|
# ── save_listings интеграция ─────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def _mock_db_for_save_listings(
|
||||||
|
inserted: bool = True,
|
||||||
|
listing_id: int = 42,
|
||||||
|
) -> MagicMock:
|
||||||
|
"""Mock db для save_listings с правильным RETURNING + begin_nested."""
|
||||||
|
row = MagicMock()
|
||||||
|
row.id = listing_id
|
||||||
|
row.inserted = inserted
|
||||||
|
|
||||||
|
ins_result = MagicMock()
|
||||||
|
ins_result.fetchone.return_value = row
|
||||||
|
|
||||||
|
db = MagicMock()
|
||||||
|
db.execute.return_value = ins_result
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def _nested():
|
||||||
|
yield MagicMock()
|
||||||
|
|
||||||
|
db.begin_nested.side_effect = _nested
|
||||||
|
return db
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_listings_writes_snapshot_per_lot():
|
||||||
|
"""save_listings вызывает upsert_listing_snapshot для каждого сохранённого lot."""
|
||||||
|
from app.services.scrapers.base import ScrapedLot, save_listings
|
||||||
|
|
||||||
|
db = _mock_db_for_save_listings(inserted=True, listing_id=77)
|
||||||
|
lots = [
|
||||||
|
ScrapedLot(
|
||||||
|
source="cian",
|
||||||
|
source_url="https://ekb.cian.ru/sale/flat/111/",
|
||||||
|
source_id="111",
|
||||||
|
price_rub=5_000_000,
|
||||||
|
),
|
||||||
|
ScrapedLot(
|
||||||
|
source="cian",
|
||||||
|
source_url="https://ekb.cian.ru/sale/flat/222/",
|
||||||
|
source_id="222",
|
||||||
|
price_rub=6_000_000,
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("app.services.scrapers.base.match_or_create_house", return_value=(1, 1.0, "new")),
|
||||||
|
patch("app.services.scrapers.base.upsert_listing_source"),
|
||||||
|
patch("app.services.scrapers.base.upsert_listing_snapshot") as mock_snap,
|
||||||
|
):
|
||||||
|
save_listings(db, lots, run_id=10)
|
||||||
|
|
||||||
|
assert mock_snap.call_count == 2, "snapshot должен вызваться для каждого lot"
|
||||||
|
# Первый вызов
|
||||||
|
kwargs0 = mock_snap.call_args_list[0].kwargs
|
||||||
|
assert kwargs0["listing_id"] == 77
|
||||||
|
assert kwargs0["price_rub"] == 5_000_000
|
||||||
|
assert kwargs0["run_id"] == 10
|
||||||
|
assert kwargs0["status"] == "active"
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_listings_snapshot_receives_run_id():
|
||||||
|
"""Если передан run_id — он пробрасывается в upsert_listing_snapshot."""
|
||||||
|
from app.services.scrapers.base import ScrapedLot, save_listings
|
||||||
|
|
||||||
|
db = _mock_db_for_save_listings(listing_id=88)
|
||||||
|
lot = ScrapedLot(
|
||||||
|
source="avito",
|
||||||
|
source_url="https://avito.ru/1",
|
||||||
|
source_id="1",
|
||||||
|
price_rub=3_000_000,
|
||||||
|
)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("app.services.scrapers.base.match_or_create_house", return_value=(1, 1.0, "new")),
|
||||||
|
patch("app.services.scrapers.base.upsert_listing_source"),
|
||||||
|
patch("app.services.scrapers.base.upsert_listing_snapshot") as mock_snap,
|
||||||
|
):
|
||||||
|
save_listings(db, [lot], run_id=55)
|
||||||
|
|
||||||
|
kwargs = mock_snap.call_args.kwargs
|
||||||
|
assert kwargs["run_id"] == 55
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_listings_snapshot_none_run_id_by_default():
|
||||||
|
"""Без run_id — upsert_listing_snapshot получает run_id=None."""
|
||||||
|
from app.services.scrapers.base import ScrapedLot, save_listings
|
||||||
|
|
||||||
|
db = _mock_db_for_save_listings(listing_id=9)
|
||||||
|
lot = ScrapedLot(
|
||||||
|
source="cian",
|
||||||
|
source_url="https://ekb.cian.ru/sale/flat/9/",
|
||||||
|
source_id="9",
|
||||||
|
price_rub=2_000_000,
|
||||||
|
)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("app.services.scrapers.base.match_or_create_house", return_value=(1, 1.0, "new")),
|
||||||
|
patch("app.services.scrapers.base.upsert_listing_source"),
|
||||||
|
patch("app.services.scrapers.base.upsert_listing_snapshot") as mock_snap,
|
||||||
|
):
|
||||||
|
save_listings(db, [lot]) # no run_id
|
||||||
|
|
||||||
|
kwargs = mock_snap.call_args.kwargs
|
||||||
|
assert kwargs["run_id"] is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_listings_snapshot_failure_does_not_abort_batch():
|
||||||
|
"""Ошибка snapshot не прерывает основной INSERT; все lots сохраняются."""
|
||||||
|
from app.services.scrapers.base import ScrapedLot, save_listings
|
||||||
|
|
||||||
|
db = _mock_db_for_save_listings(inserted=True, listing_id=5)
|
||||||
|
lots = [
|
||||||
|
ScrapedLot(
|
||||||
|
source="cian",
|
||||||
|
source_url=f"https://ekb.cian.ru/sale/flat/{i}/",
|
||||||
|
source_id=str(i),
|
||||||
|
price_rub=1_000_000 * i,
|
||||||
|
)
|
||||||
|
for i in range(1, 4)
|
||||||
|
]
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("app.services.scrapers.base.match_or_create_house", return_value=(1, 1.0, "new")),
|
||||||
|
patch("app.services.scrapers.base.upsert_listing_source"),
|
||||||
|
patch(
|
||||||
|
"app.services.scrapers.base.upsert_listing_snapshot",
|
||||||
|
side_effect=RuntimeError("snapshot fail"),
|
||||||
|
),
|
||||||
|
):
|
||||||
|
inserted, _updated = save_listings(db, lots)
|
||||||
|
|
||||||
|
# INSERT listings всё равно прошёл для всех 3 lots
|
||||||
|
assert inserted == 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_listings_empty_no_snapshot_call():
|
||||||
|
"""Пустой список → upsert_listing_snapshot не вызывается."""
|
||||||
|
from app.services.scrapers.base import save_listings
|
||||||
|
|
||||||
|
db = _mock_db_for_save_listings()
|
||||||
|
|
||||||
|
with patch("app.services.scrapers.base.upsert_listing_snapshot") as mock_snap:
|
||||||
|
save_listings(db, [])
|
||||||
|
|
||||||
|
mock_snap.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
# ── Идемпотентность: повторный вызов тот же listing_id + дата ────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_upsert_snapshot_idempotent_same_params_no_error():
|
||||||
|
"""Два одинаковых вызова (listing_id + date совпадают) — нет ошибок.
|
||||||
|
|
||||||
|
Реальная де-дупликация — ON CONFLICT DO UPDATE в БД.
|
||||||
|
Здесь проверяем что Python-код не падает при повторном вызове с теми же данными.
|
||||||
|
"""
|
||||||
|
db = _mock_db_simple()
|
||||||
|
snap_date = date(2026, 5, 28)
|
||||||
|
|
||||||
|
upsert_listing_snapshot(db, listing_id=1, price_rub=5_000_000, snapshot_date=snap_date)
|
||||||
|
upsert_listing_snapshot(db, listing_id=1, price_rub=5_000_000, snapshot_date=snap_date)
|
||||||
|
|
||||||
|
# Два вызова db.execute — каждый попытается INSERT ON CONFLICT DO UPDATE
|
||||||
|
assert db.execute.call_count == 2
|
||||||
|
# Убедимся что оба SQL содержат ON CONFLICT DO UPDATE
|
||||||
|
sqls = _get_all_sqls(db)
|
||||||
|
assert all("ON CONFLICT" in s for s in sqls)
|
||||||
|
|
||||||
|
|
||||||
|
def test_upsert_snapshot_different_dates_no_conflict():
|
||||||
|
"""Разные даты для одного listing → два отдельных snapshot (нет конфликта по PK)."""
|
||||||
|
db = _mock_db_simple()
|
||||||
|
|
||||||
|
upsert_listing_snapshot(db, listing_id=1, price_rub=5_000_000, snapshot_date=date(2026, 5, 1))
|
||||||
|
upsert_listing_snapshot(db, listing_id=1, price_rub=4_800_000, snapshot_date=date(2026, 5, 2))
|
||||||
|
|
||||||
|
# Каждый вызов делает INSERT: разные snapshot_date → разные PRIMARY KEY → нет конфликта
|
||||||
|
params_list = _get_all_params(db)
|
||||||
|
assert params_list[0]["snap_date"] == date(2026, 5, 1)
|
||||||
|
assert params_list[1]["snap_date"] == date(2026, 5, 2)
|
||||||
|
assert params_list[0]["price"] == 5_000_000
|
||||||
|
assert params_list[1]["price"] == 4_800_000
|
||||||
Loading…
Add table
Reference in a new issue