feat(tradein/scrapers): full_load пропускает листинги обновлённые сегодня при повторном прогоне #1828
4 changed files with 208 additions and 18 deletions
|
|
@ -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.
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
|
||||||
|
|
@ -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),
|
||||||
|
|
|
||||||
|
|
@ -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()
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue