Merge pull request 'feat(listings): card_hash column + snapshot-dedup on unchanged cards' (#1767) from feat/listings-card-hash into main

Reviewed-on: #1767

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(

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;

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