From 8436532bf4ecc3f615fdf3c23e9985691240b81c Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 18 Jun 2026 22:01:54 +0300 Subject: [PATCH] feat(listings): card_hash column + snapshot-dedup on unchanged cards Adds listings.card_hash (SHA-256 of volatile SERP-card fields) computed and stored on every scrape. Skips the redundant per-day listings_snapshots write when a listing's card is unchanged. listings.last_seen_at/is_active still bumped on every sighting (delisting TTL). Lays groundwork for detail-refresh gating (C2). --- .../backend/app/services/scrapers/base.py | 68 +++- .../data/sql/128_listings_card_hash.sql | 22 ++ .../backend/tests/test_base_save_listings.py | 315 +++++++++++++++++- 3 files changed, 394 insertions(+), 11 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/128_listings_card_hash.sql diff --git a/tradein-mvp/backend/app/services/scrapers/base.py b/tradein-mvp/backend/app/services/scrapers/base.py index 6c8a2d74..30980a7e 100644 --- a/tradein-mvp/backend/app/services/scrapers/base.py +++ b/tradein-mvp/backend/app/services/scrapers/base.py @@ -156,6 +156,28 @@ class ScrapedLot(BaseModel): h.update((key or "").encode()) return h.hexdigest() + def compute_card_hash(self) -> str: + """SHA-256 over the volatile SERP-card fields. Used to detect whether a + listing's card content changed between scrapes (snapshot-dedup, detail-refresh gate). + Excludes last_seen/scrape timestamps and detail-only fields. + + Хэшируем детерминированную нормализованную репрезентацию полей карточки: + price_rub, area_m2, floor, total_floors, rooms, address (strip+lower, None→""), + len(photo_urls). Порядок частей фиксирован (нет зависимости от порядка + итерации dict) → стабильный hash при неизменной карточке. + """ + address = (self.address or "").strip().lower() + parts = [ + str(self.price_rub), + str(self.area_m2), + str(self.floor), + str(self.total_floors), + str(self.rooms), + address, + str(len(self.photo_urls)), + ] + return hashlib.sha256("|".join(parts).encode()).hexdigest() + def compute_price_per_m2(self) -> int | None: """price_per_m2 = price_rub / area_m2 если area задана.""" if self.area_m2 and self.area_m2 > 0: @@ -256,6 +278,18 @@ def save_listings( for lot in lots: ppm2 = lot.price_per_m2 or lot.compute_price_per_m2() dedup = lot.compute_dedup_hash() + card_hash = lot.compute_card_hash() + + # Pre-read the existing row's card_hash (keyed by dedup_hash) BEFORE the + # upsert — needed to know the *prior* card content. The upsert's + # RETURNING can only expose the NEW hash (EXCLUDED.card_hash), never the + # old one, so a lightweight SELECT is the clean way to detect "unchanged". + # None → row did not exist yet (fresh insert) → snapshot always written. + prior_row = db.execute( + text("SELECT card_hash FROM listings WHERE dedup_hash = :dedup"), + {"dedup": dedup}, + ).fetchone() + prior_card_hash = prior_row.card_hash if prior_row is not None else None result = db.execute( text( @@ -277,7 +311,7 @@ def save_listings( yandex_offer_id, predicted_price_rub, predicted_price_min, predicted_price_max, price_trend, price_previous_rub, - geo_precision, + geo_precision, card_hash, scraped_at, last_seen_at ) VALUES ( :source, :source_url, :source_id, :dedup, @@ -297,7 +331,7 @@ def save_listings( :yandex_offer_id, :predicted_price_rub, :predicted_price_min, :predicted_price_max, :price_trend, :price_previous_rub, - :geo_precision, + :geo_precision, :card_hash, NOW(), NOW() ) ON CONFLICT (dedup_hash) DO UPDATE @@ -352,7 +386,8 @@ def save_listings( ), newbuilding_url = COALESCE( EXCLUDED.newbuilding_url, listings.newbuilding_url - ) + ), + card_hash = EXCLUDED.card_hash RETURNING id, (xmax = 0) AS inserted """ ), @@ -408,6 +443,7 @@ def save_listings( "price_trend": lot.price_trend, "price_previous_rub": lot.price_previous_rub, "geo_precision": lot.geo_precision, + "card_hash": card_hash, }, ).fetchone() @@ -422,7 +458,33 @@ def save_listings( # ── Snapshot: point-in-time observation in listings_snapshots ─── # Fault-tolerant: failure here MUST NOT abort the listings batch. # ON CONFLICT DO UPDATE — самый последний snapshot за день. + # + # Snapshot-dedup: пропускаем запись только когда МЫ УВЕРЕНЫ что карточка + # не изменилась (prior_card_hash == card_hash, row существовал) И snapshot + # за сегодня уже есть. Принцип "when in doubt — write": лишний snapshot + # неизменной карточки безвреден, ПРОПУЩЕННЫЙ snapshot реального изменения — + # недопустим. Поэтому equality-проверки мало (первый скрейп за день мог + # ещё не записать сегодняшний snapshot) — дополнительно подтверждаем + # наличие сегодняшней строки лёгким EXISTS. if listing_id is not None: + skip_snapshot = False + if prior_card_hash is not None and prior_card_hash == card_hash: + today_row = db.execute( + text( + "SELECT 1 FROM listings_snapshots " + "WHERE listing_id = CAST(:lid AS bigint) " + " AND snapshot_date = CURRENT_DATE" + ), + {"lid": listing_id}, + ).fetchone() + skip_snapshot = today_row is not None + if skip_snapshot: + logger.debug( + "save_listings:snapshot_skipped (card unchanged) " "source=%s listing_id=%s", + lot.source, + listing_id, + ) + if listing_id is not None and not skip_snapshot: try: with db.begin_nested(): upsert_listing_snapshot( diff --git a/tradein-mvp/backend/data/sql/128_listings_card_hash.sql b/tradein-mvp/backend/data/sql/128_listings_card_hash.sql new file mode 100644 index 00000000..446ee548 --- /dev/null +++ b/tradein-mvp/backend/data/sql/128_listings_card_hash.sql @@ -0,0 +1,22 @@ +-- 128_listings_card_hash.sql +-- Добавляет listings.card_hash — SHA-256 над волатильными полями SERP-карточки +-- (price_rub, area_m2, floor, total_floors, rooms, address, len(photo_urls)). +-- +-- Назначение: +-- 1. Snapshot-dedup: если карточка объявления не изменилась между скрейпами и +-- snapshot за сегодня уже есть — пропускаем повторную запись в +-- listings_snapshots (единственная реальная экономия записи; саму строку +-- listings всё равно трогаем — last_seen_at/is_active бампятся каждый раз, +-- от этого зависит delisting TTL). +-- 2. Future: detail-refresh gating (C2) — re-fetch detail-страницы только когда +-- card_hash сменился. +-- +-- Колонка сравнивается per-row во время upsert'а (save_listings), НЕ запрашивается +-- по ней — индекс не нужен. +-- +-- ЗАВИСИМОСТИ: таблица listings (создаётся ранними миграциями 0xx). +-- Idempotent: ADD COLUMN IF NOT EXISTS безопасен при повторном запуске. + +BEGIN; +ALTER TABLE listings ADD COLUMN IF NOT EXISTS card_hash text; +COMMIT; diff --git a/tradein-mvp/backend/tests/test_base_save_listings.py b/tradein-mvp/backend/tests/test_base_save_listings.py index 9aa01f9c..65ee349d 100644 --- a/tradein-mvp/backend/tests/test_base_save_listings.py +++ b/tradein-mvp/backend/tests/test_base_save_listings.py @@ -109,19 +109,31 @@ def _cian_lot(**overrides) -> ScrapedLot: # ── Utility: extract params from mock db call ──────────────────────────────── -def _get_execute_params(db: MagicMock) -> dict: - """Получить params dict из первого db.execute() вызова (INSERT listings).""" +def _find_listings_insert_call(db: MagicMock): + """Найти (sql, params) вызова INSERT INTO listings. + + save_listings теперь делает лёгкий pre-read `SELECT card_hash FROM listings` + ПЕРЕД upsert'ом, поэтому INSERT больше не обязательно call_args_list[0]. + Ищем по SQL-тексту, а не по фиксированному индексу. + """ assert db.execute.called, "db.execute was not called" - # First call is INSERT INTO listings ... RETURNING id, inserted. - # Subsequent calls are matching hook (match_or_create_house, listing_sources). - _sql, params = db.execute.call_args_list[0].args + for call in db.execute.call_args_list: + sql = str(call.args[0]) + if "INSERT INTO listings (" in sql: + return sql, call.args[1] + raise AssertionError("INSERT INTO listings call not found") + + +def _get_execute_params(db: MagicMock) -> dict: + """Получить params dict из вызова INSERT listings.""" + _sql, params = _find_listings_insert_call(db) return params def _get_listings_insert_sql(db: MagicMock) -> str: - """Получить SQL-текст INSERT INTO listings (первый db.execute() вызов).""" - assert db.execute.called, "db.execute was not called" - return str(db.execute.call_args_list[0].args[0]) + """Получить SQL-текст INSERT INTO listings.""" + sql, _params = _find_listings_insert_call(db) + return sql # ── Test 1: Cian lot → все новые колонки попадают в params ────────────────── @@ -805,3 +817,290 @@ def test_save_listings_newbuilding_fields_default_none(): params = _get_execute_params(db) assert params["newbuilding_id"] is None assert params["newbuilding_url"] is None + + +# ── card_hash: ScrapedLot.compute_card_hash determinism + sensitivity ──────── + + +def test_compute_card_hash_deterministic(): + """Тот же lot → тот же hash (вызов дважды и два эквивалентных lot).""" + lot = _cian_lot() + assert lot.compute_card_hash() == lot.compute_card_hash() + + lot_b = _cian_lot() + assert lot.compute_card_hash() == lot_b.compute_card_hash() + # sha256 hex + assert len(lot.compute_card_hash()) == 64 + + +def test_compute_card_hash_changes_on_price(): + """Смена price_rub → другой hash.""" + a = _base_lot(price_rub=4_500_000) + b = _base_lot(price_rub=4_300_000) + assert a.compute_card_hash() != b.compute_card_hash() + + +def test_compute_card_hash_changes_on_area_floor(): + """Смена area_m2, floor, total_floors, rooms → другой hash.""" + base = _cian_lot() + assert base.compute_card_hash() != _cian_lot(area_m2=53.0).compute_card_hash() + assert base.compute_card_hash() != _cian_lot(floor=8).compute_card_hash() + assert base.compute_card_hash() != _cian_lot(total_floors=17).compute_card_hash() + assert base.compute_card_hash() != _cian_lot(rooms=3).compute_card_hash() + + +def test_compute_card_hash_changes_on_address_and_photos(): + """Смена address (после нормализации) и числа фото → другой hash.""" + base = _base_lot(address="Екатеринбург, Ленина 1", photo_urls=["a", "b"]) + assert ( + base.compute_card_hash() + != _base_lot(address="Екатеринбург, Ленина 2", photo_urls=["a", "b"]).compute_card_hash() + ) + assert ( + base.compute_card_hash() + != _base_lot( + address="Екатеринбург, Ленина 1", photo_urls=["a", "b", "c"] + ).compute_card_hash() + ) + + +def test_compute_card_hash_address_normalized(): + """address нормализуется (strip + lower) — регистр/пробелы не меняют hash.""" + a = _base_lot(address=" Екатеринбург, Ленина 1 ") + b = _base_lot(address="екатеринбург, ленина 1") + assert a.compute_card_hash() == b.compute_card_hash() + + +def test_compute_card_hash_none_address_stable(): + """address=None → "" — два lot без адреса дают одинаковый hash.""" + a = _base_lot(address=None) + b = _base_lot(address=None) + assert a.compute_card_hash() == b.compute_card_hash() + + +def test_compute_card_hash_stable_when_only_timestamps_or_detail_change(): + """Изменение только timestamp/detail-only полей (не входящих в карточку) → + hash НЕ меняется.""" + import datetime + + base = _cian_lot() + # detail-only / metadata поля — НЕ часть card_hash + variant = _cian_lot( + listing_date=datetime.date(2026, 1, 1), + publish_date=datetime.date(2026, 1, 2), + days_on_market=42, + description="совсем другое описание", + cadastral_number="66:41:9999999:9999", + is_homeowner=False, + agency_name="Другое агентство", + raw_payload={"foo": "bar"}, + ) + assert base.compute_card_hash() == variant.compute_card_hash() + + +# ── save_listings: card_hash persisted on insert + update ──────────────────── + + +def test_save_listings_card_hash_in_sql_and_params(): + """INSERT SQL содержит колонку card_hash + EXCLUDED в DO UPDATE SET; + params несут вычисленный hash.""" + db = _mock_db(inserted=True) + lot = _base_lot() + + save_listings(db, [lot]) + + sql_text = _get_listings_insert_sql(db) + assert "card_hash" in sql_text + assert "card_hash = EXCLUDED.card_hash" in sql_text + + params = _get_execute_params(db) + assert params["card_hash"] == lot.compute_card_hash() + assert len(params["card_hash"]) == 64 + + +def test_save_listings_card_hash_on_update_path(): + """ON CONFLICT (update) — card_hash тоже передаётся в params.""" + db = _mock_db(inserted=False) + lot = _cian_lot(source_id="ch-upd") + + inserted, updated = save_listings(db, [lot]) + assert inserted == 0 + assert updated == 1 + + params = _get_execute_params(db) + assert params["card_hash"] == lot.compute_card_hash() + + +def test_save_listings_pre_reads_existing_card_hash(): + """Перед upsert'ом делается лёгкий SELECT card_hash FROM listings WHERE dedup_hash.""" + db = _mock_db(inserted=True) + lot = _base_lot() + + save_listings(db, [lot]) + + pre_reads = [ + c + for c in db.execute.call_args_list + if "SELECT card_hash FROM listings WHERE dedup_hash" in str(c.args[0]) + ] + assert len(pre_reads) == 1 + assert pre_reads[0].args[1]["dedup"] == lot.compute_dedup_hash() + + +def test_save_listings_skips_snapshot_when_card_unchanged(): + """Если prior card_hash == new card_hash И snapshot за сегодня уже есть — + upsert_listing_snapshot НЕ вызывается (snapshot-dedup).""" + lot = _base_lot() + new_hash = lot.compute_card_hash() + + # row для INSERT RETURNING (id, inserted) + insert_row = MagicMock() + insert_row.id = 42 + insert_row.inserted = False # update path + + # pre-read row с тем же card_hash → карточка не изменилась + prior_row = MagicMock() + prior_row.card_hash = new_hash + + # today-snapshot EXISTS row (любой truthy) + today_row = MagicMock() + + db = MagicMock() + + def _execute(sql, params=None): + s = str(sql) + res = MagicMock() + if "SELECT card_hash FROM listings WHERE dedup_hash" in s: + res.fetchone.return_value = prior_row + elif "FROM listings_snapshots" in s: + res.fetchone.return_value = today_row + elif "INSERT INTO listings (" in s: + res.fetchone.return_value = insert_row + else: + res.fetchone.return_value = None + return res + + db.execute.side_effect = _execute + + @contextmanager + def _nested(): + yield MagicMock() + + db.begin_nested.side_effect = _nested + + with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap: + save_listings(db, [lot]) + m_snap.assert_not_called() + + +def test_save_listings_writes_snapshot_when_card_changed(): + """Если card_hash изменился — snapshot пишется даже если сегодня уже был + (когда в сомнениях — пишем).""" + lot = _base_lot(price_rub=4_300_000) # новая цена → новый hash + + insert_row = MagicMock() + insert_row.id = 42 + insert_row.inserted = False + + prior_row = MagicMock() + prior_row.card_hash = "оld-different-hash" # отличается + + db = MagicMock() + + def _execute(sql, params=None): + s = str(sql) + res = MagicMock() + if "SELECT card_hash FROM listings WHERE dedup_hash" in s: + res.fetchone.return_value = prior_row + elif "INSERT INTO listings (" in s: + res.fetchone.return_value = insert_row + else: + res.fetchone.return_value = None + return res + + db.execute.side_effect = _execute + + @contextmanager + def _nested(): + yield MagicMock() + + db.begin_nested.side_effect = _nested + + with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap: + save_listings(db, [lot]) + assert m_snap.call_count == 1 + + +def test_save_listings_writes_snapshot_on_fresh_insert(): + """Свежий insert (prior_card_hash is None) → snapshot всегда пишется.""" + lot = _base_lot() + + insert_row = MagicMock() + insert_row.id = 42 + insert_row.inserted = True + + db = MagicMock() + + def _execute(sql, params=None): + s = str(sql) + res = MagicMock() + if "SELECT card_hash FROM listings WHERE dedup_hash" in s: + res.fetchone.return_value = None # row не существует + elif "INSERT INTO listings (" in s: + res.fetchone.return_value = insert_row + else: + res.fetchone.return_value = None + return res + + db.execute.side_effect = _execute + + @contextmanager + def _nested(): + yield MagicMock() + + db.begin_nested.side_effect = _nested + + with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap: + save_listings(db, [lot]) + assert m_snap.call_count == 1 + + +def test_save_listings_writes_snapshot_when_unchanged_but_no_today_snapshot(): + """card_hash не изменился, НО snapshot за сегодня ещё нет (первый скрейп дня) → + snapshot всё равно пишется (не теряем сегодняшнее наблюдение).""" + lot = _base_lot() + new_hash = lot.compute_card_hash() + + insert_row = MagicMock() + insert_row.id = 42 + insert_row.inserted = False + + prior_row = MagicMock() + prior_row.card_hash = new_hash # тот же + + db = MagicMock() + + def _execute(sql, params=None): + s = str(sql) + res = MagicMock() + if "SELECT card_hash FROM listings WHERE dedup_hash" in s: + res.fetchone.return_value = prior_row + elif "FROM listings_snapshots" in s: + res.fetchone.return_value = None # сегодня snapshot'а ещё нет + elif "INSERT INTO listings (" in s: + res.fetchone.return_value = insert_row + else: + res.fetchone.return_value = None + return res + + db.execute.side_effect = _execute + + @contextmanager + def _nested(): + yield MagicMock() + + db.begin_nested.side_effect = _nested + + with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap: + save_listings(db, [lot]) + assert m_snap.call_count == 1