fix(tradein/scrapers): bump scraped_at on live re-confirm + backfill (#2206)

Upsert ре-подтверждал листинг живым (last_seen_at=NOW, обновлял цену/поля),
но scraped_at не трогал никогда. Эстиматор фильтрует свежесть по
scraped_at > NOW()-14d → видел лишь ~45.6% активного инвентаря (CIAN 5.9%).

- scraped_at=NOW() в ON CONFLICT DO UPDATE + dedup-drift reconcile UPDATE,
  в обоих зеркальных base.py (app + scraper-kit, kit — боевой путь)
- 161_backfill_scraped_at_active_recent.sql: ретро scraped_at=last_seen_at
  для active avito/cian/yandex (domklik исключён — испорченный bulk-touch
  last_seen_at ложно освежил бы нескрейпленные строки)
- regression-тесты: оба пути × оба модуля + свойства миграции 161

Refs #2206

tradein-mvp/backend/app/services/scrapers/base.py

@@ -454,6 +454,9 @@ def save_listings(
             -- а NULL не годится как conflict-target → дубли вернулись бы (#1773).
             ON CONFLICT (dedup_hash) DO UPDATE
                 SET last_seen_at = NOW(),
+                    -- #2206: ре-подтверждение живым = свежий скрейп; без bump'а
+                    -- эстиматор (scraped_at > NOW()-14d) терял ~55% живого инвентаря.
+                    scraped_at = NOW(),
                     is_active = true,
                     -- если цена изменилась — обновляем
                     price_rub = EXCLUDED.price_rub,
@@ -564,6 +567,9 @@ def save_listings(
                         UPDATE listings
                            SET dedup_hash = :dedup,
                                last_seen_at = NOW(),
+                               -- #2206: ре-подтверждение живым = свежий скрейп;
+                               -- без bump'а эстиматор терял ~55% живого инвентаря.
+                               scraped_at = NOW(),
                                is_active = true,
                                price_rub = :price_rub,
                                price_per_m2 = :ppm2,

tradein-mvp/backend/data/sql/161_backfill_scraped_at_active_recent.sql

@@ -0,0 +1,54 @@
+-- 161_backfill_scraped_at_active_recent.sql
+-- Issue #2206 — retro-backfill scraped_at for active listings whose scraped_at
+-- lagged behind last_seen_at.
+--
+-- Root cause (fixed in scrapers/base.py + scraper-kit/base.py this same PR):
+--   The listings upsert bumped last_seen_at=NOW() on every live re-confirmation
+--   (ON CONFLICT DO UPDATE + dedup-drift reconcile UPDATE) but NEVER touched
+--   scraped_at. The estimator filters LISTINGS_FRESH_DAYS via `scraped_at >
+--   NOW() - INTERVAL '14 days'`, so re-confirmed-but-not-re-inserted listings
+--   fell out of the estimator's visible inventory even while genuinely live.
+--
+-- Impact (verified on prod 2026-07-02, estimator-visible / total is_active):
+--   avito:  6353 / 8474
+--   yandex: 4561 / 8878
+--   cian:    934 / 15759   (only 5.9% visible)
+--   Overall only ~45.6% of the active inventory reached valuations.
+--
+-- Fix going forward: the upsert now sets scraped_at=NOW() alongside
+--   last_seen_at=NOW(). This migration retro-corrects the already-stale rows.
+--
+-- Why last_seen_at is the honest freshness signal HERE (avito/cian/yandex):
+--   For these three sources last_seen_at only advances when a real scrape run
+--   re-sees the listing (per-lot upsert), so it faithfully marks the last time
+--   the ad was genuinely observed live. Copying it into scraped_at restores the
+--   true "last scraped" instant for rows that were re-confirmed before the code
+--   fix landed.
+--
+-- Why domklik is DELIBERATELY EXCLUDED:
+--   domklik's last_seen_at is advanced every day by an UNtracked bulk-touch
+--   UPDATE (all 6296 rows share one identical timestamp; verified prod
+--   2026-07-02, see migration 160). Backfilling scraped_at from that poisoned
+--   last_seen_at would falsely "refresh" rows that were never actually
+--   re-scraped, re-inflating the estimator with dead inventory. domklik
+--   freshness is tracked via scraped_at directly (deactivate_stale_domklik,
+--   #2204) — nothing to backfill. n1 is a dead source and irrelevant here.
+--
+-- Idempotent by construction: after the first run every matched row has
+--   scraped_at = last_seen_at, so `last_seen_at > scraped_at` matches 0 rows on
+--   any re-apply. Non-destructive: scraped_at only moves forward (last_seen_at
+--   is >= scraped_at for every matched row), is_active rows only.
+--
+-- Expected volume ~6-7k rows (fast single UPDATE, indexable via is_active).
+--
+-- Dependencies: listings table with scraped_at + last_seen_at + is_active + source.
+
+BEGIN;
+
+UPDATE listings
+   SET scraped_at = last_seen_at
+ WHERE is_active
+   AND source IN ('avito', 'cian', 'yandex')
+   AND last_seen_at > scraped_at;
+
+COMMIT;

tradein-mvp/backend/tests/test_scraped_at_bump_2206.py

@@ -0,0 +1,268 @@
+"""#2206: ре-подтверждение живого листинга двигает scraped_at (не только last_seen_at).
+
+Корень бага: upsert листингов при ре-подтверждении живым ставил last_seen_at=NOW(),
+обновлял цену/поля, но scraped_at НЕ трогал никогда. Эстиматор фильтрует свежесть
+через `scraped_at > NOW() - INTERVAL '14 days'`, поэтому ре-подтверждённые (но не
+пере-вставленные) листинги выпадали из видимого инвентаря — на проде до фикса
+эстиматору было видно лишь ~45.6% активного инвентаря (CIAN — 5.9%).
+
+Фикс: и ON CONFLICT DO UPDATE, и dedup-drift reconcile UPDATE теперь выставляют
+scraped_at = NOW() рядом с last_seen_at = NOW(). Проверяем оба пути в ОБОИХ
+зеркальных модулях (app.services.scrapers.base + scraper_kit.base) и свойства
+ретро-бэкфилл-миграции 161.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import psycopg.errors
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from scraper_kit.base import ScrapedLot as KitLot
+from scraper_kit.base import save_listings as kit_save_listings
+
+from app.services.scrapers.base import ScrapedLot as AppLot
+from app.services.scrapers.base import save_listings as app_save_listings
+
+
+@contextmanager
+def _nested_ctx() -> Any:
+    yield MagicMock()
+
+
+def _mock_db_update_path(inserted: bool = False, listing_id: int = 42) -> MagicMock:
+    """Session mock для ON CONFLICT DO UPDATE (xmax != 0 → updated)."""
+    insert_row = MagicMock()
+    insert_row.id = listing_id
+    insert_row.inserted = inserted
+
+    db = MagicMock()
+
+    def _execute(sql: Any, params: dict[str, Any] | None = None) -> MagicMock:
+        s = str(sql)
+        res = MagicMock()
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
+            res.fetchone.return_value = None
+        elif "FROM listings_snapshots" in s:
+            res.fetchone.return_value = None
+        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
+    db.begin_nested.side_effect = _nested_ctx
+    return db
+
+
+def _mock_db_reconcile(reconcile_id: int = 77) -> MagicMock:
+    """Session mock: INSERT → UniqueViolation → reconcile UPDATE by (source, source_id)."""
+    uv_orig = psycopg.errors.UniqueViolation()
+    integrity_err = IntegrityError("INSERT INTO listings ...", {}, uv_orig)
+
+    rec_row = MagicMock()
+    rec_row.id = reconcile_id
+
+    db = MagicMock()
+
+    def _execute(sql: Any, params: dict[str, Any] | None = None) -> MagicMock:
+        s = str(sql)
+        res = MagicMock()
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
+            res.fetchone.return_value = None
+        elif "FROM listings_snapshots" in s:
+            res.fetchone.return_value = None
+        elif "INSERT INTO listings (" in s:
+            raise integrity_err
+        elif "UPDATE listings" in s and "SET dedup_hash" in s:
+            res.fetchone.return_value = rec_row
+        else:
+            res.fetchone.return_value = None
+        return res
+
+    db.execute.side_effect = _execute
+
+    @contextmanager
+    def _nested() -> Any:
+        try:
+            yield MagicMock()
+        except IntegrityError:
+            raise
+
+    db.begin_nested.side_effect = _nested
+    return db
+
+
+def _find_sql(db: MagicMock, needle: str) -> str:
+    for call in db.execute.call_args_list:
+        sql = str(call.args[0])
+        if needle in sql:
+            return sql
+    raise AssertionError(f"SQL containing {needle!r} not found")
+
+
+# ── app.services.scrapers.base ────────────────────────────────────────────────
+
+
+@pytest.fixture
+def _patch_app_matching() -> Any:
+    with (
+        patch("app.services.scrapers.base.match_or_create_house") as m_house,
+        patch("app.services.scrapers.base.upsert_listing_source") as m_link,
+        patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap,
+    ):
+        m_house.return_value = (101, 1.0, "new")
+        m_link.return_value = None
+        m_snap.return_value = None
+        yield
+
+
+def test_app_on_conflict_bumps_scraped_at(_patch_app_matching: Any) -> None:
+    """ON CONFLICT DO UPDATE (ре-подтверждение) двигает scraped_at = NOW()."""
+    db = _mock_db_update_path(inserted=False)
+    lot = AppLot(
+        source="cian",
+        source_url="https://ekb.cian.ru/sale/flat/1/",
+        source_id="1",
+        price_rub=5_000_000,
+    )
+
+    inserted, updated = app_save_listings(db, [lot])
+
+    assert (inserted, updated) == (0, 1)
+    sql = _find_sql(db, "INSERT INTO listings (")
+    # scraped_at должен обновляться в DO UPDATE SET рядом с last_seen_at
+    assert "scraped_at = NOW()" in sql
+    assert "last_seen_at = NOW()" in sql
+
+
+def test_app_reconcile_bumps_scraped_at(_patch_app_matching: Any) -> None:
+    """dedup-drift reconcile UPDATE by (source, source_id) двигает scraped_at = NOW()."""
+    db = _mock_db_reconcile(reconcile_id=88)
+    lot = AppLot(
+        source="avito",
+        source_url="https://www.avito.ru/ekaterinburg/kvartiry/1-k_1",
+        source_id="7960764619",
+        price_rub=5_000_000,
+    )
+
+    app_save_listings(db, [lot])
+
+    sql = _find_sql(db, "SET dedup_hash")
+    assert "scraped_at = NOW()" in sql
+    assert "last_seen_at = NOW()" in sql
+
+
+# ── scraper_kit.base (БОЕВОЙ путь: USE_KIT_SCHEDULER=true) ─────────────────────
+
+
+def _kit_matcher() -> MagicMock:
+    matcher = MagicMock()
+    matcher.match_or_create_house.return_value = (101, 1.0, "new")
+    matcher.upsert_listing_source.return_value = None
+    return matcher
+
+
+def test_kit_on_conflict_bumps_scraped_at() -> None:
+    """kit ON CONFLICT DO UPDATE двигает scraped_at = NOW()."""
+    db = _mock_db_update_path(inserted=False)
+    lot = KitLot(
+        source="cian",
+        source_url="https://ekb.cian.ru/sale/flat/1/",
+        source_id="1",
+        price_rub=5_000_000,
+    )
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        inserted, updated = kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66)
+
+    assert (inserted, updated) == (0, 1)
+    sql = _find_sql(db, "INSERT INTO listings (")
+    assert "scraped_at = NOW()" in sql
+    assert "last_seen_at = NOW()" in sql
+
+
+def test_kit_reconcile_bumps_scraped_at() -> None:
+    """kit dedup-drift reconcile UPDATE двигает scraped_at = NOW()."""
+    db = _mock_db_reconcile(reconcile_id=88)
+    lot = KitLot(
+        source="avito",
+        source_url="https://www.avito.ru/ekaterinburg/kvartiry/1-k_1",
+        source_id="7960764619",
+        price_rub=5_000_000,
+    )
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66)
+
+    sql = _find_sql(db, "SET dedup_hash")
+    assert "scraped_at = NOW()" in sql
+    assert "last_seen_at = NOW()" in sql
+
+
+# ── Migration 161: retro-backfill scraped_at ──────────────────────────────────
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_161 = _SQL_DIR / "161_backfill_scraped_at_active_recent.sql"
+
+
+def test_migration_161_exists() -> None:
+    assert _MIGRATION_161.is_file(), f"missing migration: {_MIGRATION_161}"
+
+
+def test_migration_161_is_transactional() -> None:
+    sql = _MIGRATION_161.read_text("utf-8")
+    assert "BEGIN;" in sql
+    assert "COMMIT;" in sql
+
+
+def test_migration_161_sets_scraped_at_from_last_seen_at() -> None:
+    sql = _MIGRATION_161.read_text("utf-8")
+    assert "SET scraped_at = last_seen_at" in sql
+
+
+def test_migration_161_is_idempotent_by_construction() -> None:
+    """Идемпотентность: WHERE last_seen_at > scraped_at → после первого прогона 0 строк."""
+    sql = _MIGRATION_161.read_text("utf-8")
+    assert "last_seen_at > scraped_at" in sql
+
+
+def test_migration_161_only_active_rows() -> None:
+    sql = _MIGRATION_161.read_text("utf-8")
+    assert "is_active" in sql
+
+
+def test_migration_161_scoped_to_scraped_sources() -> None:
+    """avito/cian/yandex включены; domklik/n1 — НЕТ (поломанный bulk-touch last_seen_at)."""
+    sql = _MIGRATION_161.read_text("utf-8")
+    assert "IN ('avito', 'cian', 'yandex')" in sql
+
+
+def test_migration_161_excludes_domklik() -> None:
+    """domklik намеренно исключён: last_seen_at bulk-touch'ится ежедневно → ложное освежение."""
+    sql = _MIGRATION_161.read_text("utf-8")
+    # domklik не в списке источников UPDATE (упоминание только в комментарии-обосновании)
+    assert "'domklik'" not in sql
+    assert "domklik" in sql  # обоснование исключения зафиксировано в комментарии
+
+
+def test_migration_161_no_psycopg_trap() -> None:
+    """psycopg v3: никаких :param::type."""
+    sql = _MIGRATION_161.read_text("utf-8")
+    assert not re.search(r":\w+::", sql)
+
+
+def test_migration_161_non_destructive_update_only() -> None:
+    sql = _MIGRATION_161.read_text("utf-8")
+    assert "UPDATE listings" in sql
+    assert "DELETE" not in sql.upper()

tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py

@@ -472,6 +472,9 @@ def save_listings(
             -- а NULL не годится как conflict-target → дубли вернулись бы (#1773).
             ON CONFLICT (dedup_hash) DO UPDATE
                 SET last_seen_at = NOW(),
+                    -- #2206: ре-подтверждение живым = свежий скрейп; без bump'а
+                    -- эстиматор (scraped_at > NOW()-14d) терял ~55% живого инвентаря.
+                    scraped_at = NOW(),
                     is_active = true,
                     -- если цена изменилась — обновляем
                     price_rub = EXCLUDED.price_rub,
@@ -582,6 +585,9 @@ def save_listings(
                         UPDATE listings
                            SET dedup_hash = :dedup,
                                last_seen_at = NOW(),
+                               -- #2206: ре-подтверждение живым = свежий скрейп;
+                               -- без bump'а эстиматор терял ~55% живого инвентаря.
+                               scraped_at = NOW(),
                                is_active = true,
                                price_rub = :price_rub,
                                price_per_m2 = :ppm2,