diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index e45faa26..543276e8 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/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. diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 56aa2299..29297cf2 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/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 diff --git a/tradein-mvp/backend/app/services/scrapers/base.py b/tradein-mvp/backend/app/services/scrapers/base.py index 30980a7e..a05cf5ef 100644 --- a/tradein-mvp/backend/app/services/scrapers/base.py +++ b/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), diff --git a/tradein-mvp/backend/tests/test_base_save_listings.py b/tradein-mvp/backend/tests/test_base_save_listings.py index 65ee349d..5ef70dde 100644 --- a/tradein-mvp/backend/tests/test_base_save_listings.py +++ b/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()