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 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__)
|
||||
|
||||
|
|
@ -201,9 +202,22 @@ class BaseScraper(ABC):
|
|||
|
||||
|
||||
# ── Запись пачки результатов в 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.
|
||||
|
||||
Дополнительно пишет 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:
|
||||
(inserted, updated) — counters для логов.
|
||||
"""
|
||||
|
|
@ -326,6 +340,28 @@ def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
|
|||
else:
|
||||
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 ─────────────
|
||||
# Idempotent (UPSERT on listing_sources UNIQUE(ext_source, ext_id))
|
||||
# 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 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__)
|
||||
|
||||
|
|
@ -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
|
||||
for change in enrichment.price_changes:
|
||||
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),
|
||||
'cian'
|
||||
)
|
||||
ON CONFLICT DO NOTHING
|
||||
ON CONFLICT ON CONSTRAINT offer_price_history_listing_change_uq DO NOTHING
|
||||
"""),
|
||||
{
|
||||
"lid": listing_id,
|
||||
|
|
@ -317,8 +349,10 @@ def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnric
|
|||
|
||||
db.commit()
|
||||
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,
|
||||
len(enrichment.price_changes),
|
||||
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