feat(tradein/scrapers): full_load пропускает листинги обновлённые сегодня при повторном прогоне #1828

Merged
lekss361 merged 1 commit from feat/tradein-skip-seen-today into main 2026-06-20 09:16:34 +00:00
4 changed files with 208 additions and 18 deletions

View file

@ -278,6 +278,11 @@ class Settings(BaseSettings):
"""Прокси для Yandex-скраперов. YANDEX_PROXY_URL > scraper_proxy_url (fallback).""" """Прокси для Yandex-скраперов. YANDEX_PROXY_URL > scraper_proxy_url (fallback)."""
return self.yandex_proxy_url_env or self.scraper_proxy_url 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 ─────────────────────────────── # ── #759: deactivate stale avito listings ───────────────────────────────
# Avito-объявления, не виденные scraper'ом более avito_stale_ttl_days дней, # Avito-объявления, не виденные scraper'ом более avito_stale_ttl_days дней,
# помечаются is_active=false ночной задачей deactivate_stale_avito_listings. # помечаются is_active=false ночной задачей deactivate_stale_avito_listings.

View file

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

View file

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

View file

@ -932,7 +932,7 @@ def test_save_listings_card_hash_on_update_path():
def test_save_listings_pre_reads_existing_card_hash(): 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) db = _mock_db(inserted=True)
lot = _base_lot() lot = _base_lot()
@ -941,7 +941,7 @@ def test_save_listings_pre_reads_existing_card_hash():
pre_reads = [ pre_reads = [
c c
for c in db.execute.call_args_list 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 len(pre_reads) == 1
assert pre_reads[0].args[1]["dedup"] == lot.compute_dedup_hash() 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 → карточка не изменилась # pre-read row с тем же card_hash → карточка не изменилась
prior_row = MagicMock() prior_row = MagicMock()
prior_row.card_hash = new_hash prior_row.card_hash = new_hash
prior_row.last_seen_at = None # не сегодня → skip_seen_today не срабатывает
# today-snapshot EXISTS row (любой truthy) # today-snapshot EXISTS row (любой truthy)
today_row = MagicMock() today_row = MagicMock()
@ -970,7 +971,7 @@ def test_save_listings_skips_snapshot_when_card_unchanged():
def _execute(sql, params=None): def _execute(sql, params=None):
s = str(sql) s = str(sql)
res = MagicMock() 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 res.fetchone.return_value = prior_row
elif "FROM listings_snapshots" in s: elif "FROM listings_snapshots" in s:
res.fetchone.return_value = today_row res.fetchone.return_value = today_row
@ -1010,7 +1011,7 @@ def test_save_listings_writes_snapshot_when_card_changed():
def _execute(sql, params=None): def _execute(sql, params=None):
s = str(sql) s = str(sql)
res = MagicMock() 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 res.fetchone.return_value = prior_row
elif "INSERT INTO listings (" in s: elif "INSERT INTO listings (" in s:
res.fetchone.return_value = insert_row res.fetchone.return_value = insert_row
@ -1044,7 +1045,7 @@ def test_save_listings_writes_snapshot_on_fresh_insert():
def _execute(sql, params=None): def _execute(sql, params=None):
s = str(sql) s = str(sql)
res = MagicMock() 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 не существует res.fetchone.return_value = None # row не существует
elif "INSERT INTO listings (" in s: elif "INSERT INTO listings (" in s:
res.fetchone.return_value = insert_row 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): def _execute(sql, params=None):
s = str(sql) s = str(sql)
res = MagicMock() 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 res.fetchone.return_value = prior_row
elif "FROM listings_snapshots" in s: elif "FROM listings_snapshots" in s:
res.fetchone.return_value = None # сегодня snapshot'а ещё нет 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: with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap:
save_listings(db, [lot]) save_listings(db, [lot])
assert m_snap.call_count == 1 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()