From b16935e6792a239da56f6e8f923bfcf22c75f325 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Thu, 28 May 2026 18:26:52 +0500 Subject: [PATCH] =?UTF-8?q?fix(tradein):=20=D1=80=D0=B5=D0=B0=D0=BB=D0=B8?= =?UTF-8?q?=D0=B7=D0=BE=D0=B2=D0=B0=D1=82=D1=8C=20writers=20listings=5Fsna?= =?UTF-8?q?pshots/offer=5Fprice=5Fhistory=20(#567)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit listings_snapshots не имел ни одного writer'а — таблица оставалась пустой. offer_price_history уже писался в save_detail_enrichment, но ON CONFLICT DO NOTHING не указывал имя constraint'а (добавлено в 047_cian_history_sanitize.sql), что делало де-дупликацию неэффективной. Добавлено: - snapshot_writer.py: upsert_listing_snapshot() — атомарный INSERT ... ON CONFLICT DO UPDATE на (listing_id, snapshot_date), без commit (commit у caller'а). - base.py save_listings(): вызывает upsert_listing_snapshot после каждого INSERT/UPDATE listings; принимает опциональный run_id; ошибка snapshot не прерывает батч. - cian_detail.py save_detail_enrichment(): SELECT price_rub/price_per_m2 из listings, затем upsert_listing_snapshot; ON CONFLICT для offer_price_history теперь ссылается на offer_price_history_listing_change_uq (именованный constraint из мигр. 047). - test_snapshot_writer.py: 17 unit-тестов покрывают SQL-форму, params, идемпотентность, интеграцию с save_listings и save_detail_enrichment, fault-tolerance. --- .../backend/app/services/scrapers/base.py | 38 +- .../app/services/scrapers/cian_detail.py | 35 +- .../app/services/scrapers/snapshot_writer.py | 92 ++++ .../backend/tests/test_snapshot_writer.py | 470 ++++++++++++++++++ 4 files changed, 632 insertions(+), 3 deletions(-) create mode 100644 tradein-mvp/backend/app/services/scrapers/snapshot_writer.py create mode 100644 tradein-mvp/backend/tests/test_snapshot_writer.py diff --git a/tradein-mvp/backend/app/services/scrapers/base.py b/tradein-mvp/backend/app/services/scrapers/base.py index 29603f6c..914c50a1 100644 --- a/tradein-mvp/backend/app/services/scrapers/base.py +++ b/tradein-mvp/backend/app/services/scrapers/base.py @@ -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. diff --git a/tradein-mvp/backend/app/services/scrapers/cian_detail.py b/tradein-mvp/backend/app/services/scrapers/cian_detail.py index 59d281cf..fb7f824f 100644 --- a/tradein-mvp/backend/app/services/scrapers/cian_detail.py +++ b/tradein-mvp/backend/app/services/scrapers/cian_detail.py @@ -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,34 @@ 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() + if _snap_row is not None and _snap_row.price_rub is not None: + try: + 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", + ) + 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 +292,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 +346,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_row is not None and _snap_row.price_rub is not None, ) diff --git a/tradein-mvp/backend/app/services/scrapers/snapshot_writer.py b/tradein-mvp/backend/app/services/scrapers/snapshot_writer.py new file mode 100644 index 00000000..624637bd --- /dev/null +++ b/tradein-mvp/backend/app/services/scrapers/snapshot_writer.py @@ -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, + ) diff --git a/tradein-mvp/backend/tests/test_snapshot_writer.py b/tradein-mvp/backend/tests/test_snapshot_writer.py new file mode 100644 index 00000000..b259a78d --- /dev/null +++ b/tradein-mvp/backend/tests/test_snapshot_writer.py @@ -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