tradein-mvp/backend/app/core/config.py

@@ -278,6 +278,11 @@ class Settings(BaseSettings):
         """Прокси для Yandex-скраперов. YANDEX_PROXY_URL > scraper_proxy_url (fallback)."""
         return self.yandex_proxy_url_env or self.scraper_proxy_url
 
+    # full_load повторный прогон в день пропускает листинги уже обновлённые сегодня
+    # (last_seen_at MSK) — экономит upsert + price-trigger churn; False = всегда
+    # обновлять (старое поведение). ENV: SCRAPER_SKIP_SEEN_TODAY.
+    scraper_skip_seen_today: bool = True
+
     # ── #759: deactivate stale avito listings ───────────────────────────────
     # Avito-объявления, не виденные scraper'ом более avito_stale_ttl_days дней,
     # помечаются is_active=false ночной задачей deactivate_stale_avito_listings.

tradein-mvp/backend/app/services/scrape_pipeline.py

@@ -2091,7 +2091,9 @@ async def run_cian_full_load(
                 db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
             )
             return
-        inserted, updated = save_listings(db, lots, run_id=run_id)
+        inserted, updated = save_listings(
+            db, lots, run_id=run_id, skip_seen_today=settings.scraper_skip_seen_today
+        )
         # save_listings вызывает db.commit() внутри — данные в БД сразу
         counters.saved_inserted += inserted
         counters.saved_updated += updated
@@ -2313,7 +2315,9 @@ async def run_yandex_full_load(
                 db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
             )
             return
-        inserted, updated = save_listings(db, lots, run_id=run_id)
+        inserted, updated = save_listings(
+            db, lots, run_id=run_id, skip_seen_today=settings.scraper_skip_seen_today
+        )
         # save_listings вызывает db.commit() внутри — данные в БД сразу
         counters.saved_inserted += inserted
         counters.saved_updated += updated
@@ -2504,7 +2508,9 @@ async def run_avito_full_load(
                 db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
             )
             return
-        inserted, updated = save_listings(db, lots, run_id=run_id)
+        inserted, updated = save_listings(
+            db, lots, run_id=run_id, skip_seen_today=settings.scraper_skip_seen_today
+        )
         # save_listings вызывает db.commit() внутри — данные в БД сразу
         counters.saved_inserted += inserted
         counters.saved_updated += updated

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

@@ -20,8 +20,9 @@ import hashlib
 import logging
 import random
 from abc import ABC, abstractmethod
-from datetime import date
+from datetime import date, datetime
 from typing import Any
+from zoneinfo import ZoneInfo
 
 import httpx
 from pydantic import BaseModel, Field
@@ -34,6 +35,8 @@ from app.services.scrapers.snapshot_writer import upsert_listing_snapshot
 
 logger = logging.getLogger(__name__)
 
+_MSK = ZoneInfo("Europe/Moscow")
+
 
 # ── Realistic browser User-Agents (Apr 2026 versions) ──────────────────────
 
@@ -253,6 +256,7 @@ def save_listings(
     lots: list[ScrapedLot],
     *,
     run_id: int | None = None,
+    skip_seen_today: bool = False,
 ) -> tuple[int, int]:
     """Пишем list[ScrapedLot] в `listings` с upsert по dedup_hash.
 
@@ -263,6 +267,11 @@ def save_listings(
         db: SQLAlchemy session.
         lots: список распарсенных объявлений.
         run_id: FK scrape_runs.id текущего run'а (опционально; None при ad-hoc вызовах).
+        skip_seen_today: если True и листинг уже обновлялся сегодня по МСК
+            (last_seen_at >= начало сегодняшнего дня), то пропустить upsert и snapshot.
+            Используется full_load для экономии redundant upsert + price-trigger churn
+            при повторном прогоне в тот же день. Новые листинги (prior_row=None) всегда
+            вставляются. False = старое поведение (всегда upsert).
 
     Returns:
         (inserted, updated) — counters для логов.
@@ -272,25 +281,37 @@ def save_listings(
 
     inserted = 0
     updated = 0
+    skipped = 0
     matched = 0
     match_failures = 0
 
+    today_msk = datetime.now(_MSK).date()
+
     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.
+        # Pre-read the existing row's card_hash and last_seen_at (keyed by
+        # dedup_hash) BEFORE the upsert — needed to know the *prior* card
+        # content and to implement skip_seen_today logic.
+        # The upsert's RETURNING can only expose the NEW values (EXCLUDED.*),
+        # never the old ones, so a lightweight SELECT is the clean way.
+        # None → row did not exist yet (fresh insert) → always process.
         prior_row = db.execute(
-            text("SELECT card_hash FROM listings WHERE dedup_hash = :dedup"),
+            text("SELECT card_hash, last_seen_at FROM listings WHERE dedup_hash = :dedup"),
             {"dedup": dedup},
         ).fetchone()
         prior_card_hash = prior_row.card_hash if prior_row is not None else None
 
+        # skip_seen_today: пропускаем листинги уже обновлённые сегодня по МСК.
+        # Только для existing строк (prior_row is not None); новые всегда вставляются.
+        if skip_seen_today and prior_row is not None and prior_row.last_seen_at is not None:
+            ls_msk = prior_row.last_seen_at.astimezone(_MSK).date()
+            if ls_msk == today_msk:
+                skipped += 1
+                continue
+
         result = db.execute(
             text(
                 """
@@ -526,11 +547,12 @@ def save_listings(
 
     db.commit()
     logger.info(
-        "save_listings: source=%s inserted=%d updated=%d matched=%d "
-        "match_failures=%d (total %d)",
+        "save_listings: source=%s inserted=%d updated=%d skipped_seen_today=%d "
+        "matched=%d match_failures=%d (total %d)",
         lots[0].source if lots else "?",
         inserted,
         updated,
+        skipped,
         matched,
         match_failures,
         len(lots),

tradein-mvp/backend/tests/test_base_save_listings.py

@@ -932,7 +932,7 @@ def test_save_listings_card_hash_on_update_path():
 
 
 def test_save_listings_pre_reads_existing_card_hash():
-    """Перед upsert'ом делается лёгкий SELECT card_hash FROM listings WHERE dedup_hash."""
+    """Перед upsert'ом делается лёгкий SELECT card_hash, last_seen_at FROM listings."""
     db = _mock_db(inserted=True)
     lot = _base_lot()
 
@@ -941,7 +941,7 @@ def test_save_listings_pre_reads_existing_card_hash():
     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])
+        if "SELECT card_hash" in str(c.args[0]) and "WHERE dedup_hash" in str(c.args[0])
     ]
     assert len(pre_reads) == 1
     assert pre_reads[0].args[1]["dedup"] == lot.compute_dedup_hash()
@@ -961,6 +961,7 @@ def test_save_listings_skips_snapshot_when_card_unchanged():
     # pre-read row с тем же card_hash → карточка не изменилась
     prior_row = MagicMock()
     prior_row.card_hash = new_hash
+    prior_row.last_seen_at = None  # не сегодня → skip_seen_today не срабатывает
 
     # today-snapshot EXISTS row (любой truthy)
     today_row = MagicMock()
@@ -970,7 +971,7 @@ def test_save_listings_skips_snapshot_when_card_unchanged():
     def _execute(sql, params=None):
         s = str(sql)
         res = MagicMock()
-        if "SELECT card_hash FROM listings WHERE dedup_hash" in s:
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
             res.fetchone.return_value = prior_row
         elif "FROM listings_snapshots" in s:
             res.fetchone.return_value = today_row
@@ -1010,7 +1011,7 @@ def test_save_listings_writes_snapshot_when_card_changed():
     def _execute(sql, params=None):
         s = str(sql)
         res = MagicMock()
-        if "SELECT card_hash FROM listings WHERE dedup_hash" in s:
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
             res.fetchone.return_value = prior_row
         elif "INSERT INTO listings (" in s:
             res.fetchone.return_value = insert_row
@@ -1044,7 +1045,7 @@ def test_save_listings_writes_snapshot_on_fresh_insert():
     def _execute(sql, params=None):
         s = str(sql)
         res = MagicMock()
-        if "SELECT card_hash FROM listings WHERE dedup_hash" in s:
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
             res.fetchone.return_value = None  # row не существует
         elif "INSERT INTO listings (" in s:
             res.fetchone.return_value = insert_row
@@ -1083,7 +1084,7 @@ def test_save_listings_writes_snapshot_when_unchanged_but_no_today_snapshot():
     def _execute(sql, params=None):
         s = str(sql)
         res = MagicMock()
-        if "SELECT card_hash FROM listings WHERE dedup_hash" in s:
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
             res.fetchone.return_value = prior_row
         elif "FROM listings_snapshots" in s:
             res.fetchone.return_value = None  # сегодня snapshot'а ещё нет
@@ -1104,3 +1105,159 @@ def test_save_listings_writes_snapshot_when_unchanged_but_no_today_snapshot():
     with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap:
         save_listings(db, [lot])
         assert m_snap.call_count == 1
+
+
+# ── Tests: skip_seen_today ────────────────────────────────────────────────────
+
+
+def _mock_db_with_last_seen(
+    last_seen_at,
+    inserted: bool = False,
+    listing_id: int = 42,
+) -> MagicMock:
+    """Mock Session где prior SELECT возвращает строку с card_hash и last_seen_at.
+
+    last_seen_at: datetime с tzinfo (timezone-aware) или None.
+    """
+    insert_row = MagicMock()
+    insert_row.id = listing_id
+    insert_row.inserted = inserted
+
+    prior_row = MagicMock()
+    prior_row.card_hash = "somehash"
+    prior_row.last_seen_at = last_seen_at
+
+    db = MagicMock()
+
+    def _execute(sql, params=None):
+        s = str(sql)
+        res = MagicMock()
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
+            res.fetchone.return_value = prior_row
+        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
+
+    @contextmanager
+    def _nested():
+        yield MagicMock()
+
+    db.begin_nested.side_effect = _nested
+    return db
+
+
+def test_skip_seen_today_skips_when_last_seen_is_today(_patch_matching):
+    """skip_seen_today=True + last_seen_at=сегодня по МСК → лот пропускается (ins=upd=0)."""
+    from datetime import datetime
+    from zoneinfo import ZoneInfo
+
+    msk_tz = ZoneInfo("Europe/Moscow")
+    now_msk = datetime.now(msk_tz)
+
+    db = _mock_db_with_last_seen(last_seen_at=now_msk, inserted=False)
+    lot = _base_lot(source_id="skip-today")
+
+    inserted, updated = save_listings(db, [lot], skip_seen_today=True)
+
+    assert inserted == 0
+    assert updated == 0
+    # INSERT INTO listings не должен вызываться
+    insert_calls = [
+        c for c in db.execute.call_args_list if "INSERT INTO listings (" in str(c.args[0])
+    ]
+    assert len(insert_calls) == 0
+    # matching hook не должен срабатывать
+    assert _patch_matching["link"].call_count == 0
+
+
+def test_skip_seen_today_processes_when_last_seen_is_yesterday(_patch_matching):
+    """skip_seen_today=True + last_seen_at=вчера по МСК → лот обрабатывается."""
+    from datetime import datetime, timedelta
+    from zoneinfo import ZoneInfo
+
+    msk_tz = ZoneInfo("Europe/Moscow")
+    yesterday_msk = datetime.now(msk_tz) - timedelta(days=1)
+
+    db = _mock_db_with_last_seen(last_seen_at=yesterday_msk, inserted=False)
+    lot = _base_lot(source_id="process-yesterday")
+
+    inserted, updated = save_listings(db, [lot], skip_seen_today=True)
+
+    # updated=1 (upsert path)
+    assert inserted == 0
+    assert updated == 1
+
+
+def test_skip_seen_today_new_lot_always_inserted(_patch_matching):
+    """skip_seen_today=True + prior_row=None (новый листинг) → вставляется."""
+    insert_row = MagicMock()
+    insert_row.id = 10
+    insert_row.inserted = True
+
+    db = MagicMock()
+
+    def _execute(sql, params=None):
+        s = str(sql)
+        res = MagicMock()
+        if "SELECT card_hash" in s and "WHERE dedup_hash" 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
+
+    @contextmanager
+    def _nested():
+        yield MagicMock()
+
+    db.begin_nested.side_effect = _nested
+
+    lot = _base_lot(source_id="brand-new")
+
+    inserted, updated = save_listings(db, [lot], skip_seen_today=True)
+
+    assert inserted == 1
+    assert updated == 0
+
+
+def test_skip_seen_today_false_always_updates(_patch_matching):
+    """skip_seen_today=False (дефолт) + last_seen_at=сегодня → upsert как обычно."""
+    from datetime import datetime
+    from zoneinfo import ZoneInfo
+
+    msk_tz = ZoneInfo("Europe/Moscow")
+    now_msk = datetime.now(msk_tz)
+
+    db = _mock_db_with_last_seen(last_seen_at=now_msk, inserted=False)
+    lot = _base_lot(source_id="no-skip-flag")
+
+    # skip_seen_today не передаём (дефолт False)
+    inserted, updated = save_listings(db, [lot])
+
+    assert inserted == 0
+    assert updated == 1
+
+
+def test_skip_seen_today_skips_snapshot_for_skipped_lot(_patch_matching):
+    """Пропущенный лот (seen_today) не вызывает upsert_listing_snapshot."""
+    from datetime import datetime
+    from zoneinfo import ZoneInfo
+
+    msk_tz = ZoneInfo("Europe/Moscow")
+    now_msk = datetime.now(msk_tz)
+
+    db = _mock_db_with_last_seen(last_seen_at=now_msk, inserted=False)
+    lot = _base_lot(source_id="no-snap-skip")
+
+    with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap:
+        save_listings(db, [lot], skip_seen_today=True)
+        m_snap.assert_not_called()