From bdedac3012f0f0a34d8f0263c35e027c4176092e Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 18 Jun 2026 22:01:46 +0300 Subject: [PATCH] feat(avito): incremental SERP pagination with date early-stop (param-gated, default off) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds since=date mode to fetch_all_secondary: per room x seed-bracket, paginate sequentially newest-first and stop once cards fall below the watermark, skipping deep pagination (and the 403 clusters it triggers). since=None keeps the exact exhaustive bisection walk. No scheduler/schedule change here — wiring follows in a separate PR. --- .../backend/app/services/scrapers/avito.py | 167 +++++++++- .../test_avito_incremental_pagination.py | 315 ++++++++++++++++++ 2 files changed, 468 insertions(+), 14 deletions(-) create mode 100644 tradein-mvp/backend/tests/scrapers/test_avito_incremental_pagination.py diff --git a/tradein-mvp/backend/app/services/scrapers/avito.py b/tradein-mvp/backend/app/services/scrapers/avito.py index 2f83aacd..e1ba79f4 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito.py +++ b/tradein-mvp/backend/app/services/scrapers/avito.py @@ -797,8 +797,9 @@ class AvitoScraper(BaseScraper): on_bucket: Callable[..., Any] | None = None, on_progress: Callable[[int], None] | None = None, skip_buckets: set[str] | None = None, + since: date | None = None, ) -> list[ScrapedLot]: - """Exhaustive-загрузка Avito ЕКБ вторички через КОМНАТНОСТЬ × ЦЕНА. + """Exhaustive- ИЛИ инкрементальная загрузка Avito ЕКБ вторички (КОМНАТНОСТЬ × ЦЕНА). Обходит Avito SERP-cap (~5000 результатов на запрос). Вместо корень-бисекции [0, _AVITO_MAX_PRICE] на каждую комнатность (≈5 холостых probe-спусков через @@ -825,6 +826,11 @@ class AvitoScraper(BaseScraper): on_progress: опциональный callback(unique_count) для heartbeat (per room-bucket). skip_buckets: множество ключей «room_label:lo:hi» уже завершённых бакетов — пагинация и on_bucket для них пропускаются. Probe-запросы выполняются. + since: если None (default) — EXHAUSTIVE bisection-обход (поведение без + изменений). Если задана date — INCREMENTAL: на каждый (комнатность × + seed-брекет) последовательная пагинация newest-first с ранней остановкой, + как только страница уходит ниже watermark `since`. Бисекция по цене НЕ + выполняется (читаем только верхние свежие страницы, до SERP-cap далеко). Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url). """ @@ -847,19 +853,34 @@ class AvitoScraper(BaseScraper): # в общем `seen` страхует на стыках. for br_lo, br_hi in _AVITO_PRICE_SEED_BRACKETS: walk_hi = br_hi - 1 if br_hi is not None else None - await self._walk_price_range( - room_slug=slug, - room_label=room_label, - lo=br_lo, - hi=walk_hi, - seen=seen, - price_cap_per_bucket=price_cap_per_bucket, - max_pages_per_bucket=max_pages_per_bucket, - concurrency=concurrency, - secondary_only=secondary_only, - on_bucket=on_bucket, - skip_buckets=skip_buckets, - ) + if since is not None: + # INCREMENTAL: последовательная newest-first пагинация с ранней + # остановкой по watermark `since`; без бисекции по цене. + await self._paginate_incremental_bracket( + room_slug=slug, + room_label=room_label, + lo=br_lo, + hi=walk_hi, + since=since, + seen=seen, + max_pages_per_bucket=max_pages_per_bucket, + secondary_only=secondary_only, + on_bucket=on_bucket, + ) + else: + await self._walk_price_range( + room_slug=slug, + room_label=room_label, + lo=br_lo, + hi=walk_hi, + seen=seen, + price_cap_per_bucket=price_cap_per_bucket, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + ) room_collected = len(seen) - before logger.info( "avito exhaustive: %s done — collected %d (total unique=%d)", @@ -1264,6 +1285,124 @@ class AvitoScraper(BaseScraper): if inspect.isawaitable(res_cb): await res_cb + async def _paginate_incremental_bracket( + self, + *, + room_slug: str, + room_label: str, + lo: int, + hi: int | None, + since: date, + seen: dict[str, ScrapedLot], + max_pages_per_bucket: int, + secondary_only: bool, + on_bucket: Callable[..., Any] | None, + ) -> None: + """INCREMENTAL пагинация одного (комнатность × seed-брекет) с ранней остановкой. + + SERP отсортирован newest-first (s=104). Идём ПОСЛЕДОВАТЕЛЬНО со страницы 1; + как только страница не содержит ни одной карточки с listing_date >= since — + весь остаток бакета старше watermark → останавливаемся. Бисекция по цене НЕ + выполняется: читаем только верхние свежие страницы, до SERP-cap (~5000) далеко. + + Early-stop semantics: + - На каждой странице берём max(listing_date) среди датированных карточек. + Если ни одна карточка не имеет listing_date >= since → страница «старая». + - Карточки с listing_date is None НЕ триггерят стоп сами по себе (трактуем + как «продолжить», чтобы не потерять свежие-но-недатированные карточки). + - Но страница, целиком состоящая из None-или-старых карточек, считается + «не-свежей»; 2 ПОДРЯД не-свежие страницы → стоп (grace против runaway, + когда весь хвост недатирован). + - page_html is None или parse == [] → конец пагинации, стоп. + + bucket_key, secondary_only-фильтр и дедуп в seen — идентичны _paginate_leaf_bucket. + on_bucket вызывается один раз для собранного бакета (async/sync-aware). + AvitoBlockedError/AvitoRateLimitedError из page-фетчей пробрасываются наверх. + """ + _lo_param = lo if lo > 0 else None + _hi_param = hi # None → _build_rooms_url не ставит pmax + _hi_repr = "open" if hi is None else str(hi) + bucket_key = f"{room_label}:{lo}:{_hi_repr}" + + bucket_lots: list[ScrapedLot] = [] + pages_fetched = 0 + not_fresh_streak = 0 # подряд идущие не-свежие (all None-или-старые) страницы + stop_reason = "max-pages" # перетирается ниже на реальную причину + + for p in range(1, max_pages_per_bucket + 1): + # Последовательный фетч (НЕ asyncio.gather): early-stop требует читать + # страницы по порядку. Тот же helper, что и в _paginate_leaf_bucket. + page_html = await self._fetch_rooms_page_html(room_slug, p, _lo_param, _hi_param) + await asyncio.sleep(self.request_delay_sec) + pages_fetched += 1 + + if page_html is None: + stop_reason = "end-of-pages (page_html=None)" + break + + page_url = self._build_rooms_url(room_slug, p, _lo_param, _hi_param) + page_lots = self._parse_html(page_html, source_url_base=page_url) + if not page_lots: + stop_reason = "end-of-pages (empty parse)" + break + + bucket_lots.extend(page_lots) + + # Свежесть страницы: есть ли хоть одна карточка с датой >= since. + dated = [lot.listing_date for lot in page_lots if lot.listing_date is not None] + has_fresh = any(d >= since for d in dated) + if has_fresh: + not_fresh_streak = 0 + continue + + # Страница не содержит свежих датированных карточек. + if dated: + # Есть даты, но все < since → newest-first гарантирует, что дальше + # только старее → стоп немедленно. + stop_reason = "early-stop (page all older than since)" + break + # Все карточки undated (None) → не стопим сразу (могут быть свежие без + # даты), но копим streak; 2 подряд недатированные страницы → grace-стоп. + not_fresh_streak += 1 + if not_fresh_streak >= 2: + stop_reason = "early-stop (2 consecutive all-undated pages, grace)" + break + + collected_this_bucket = len(bucket_lots) + + # ── Фильтр новостроек (secondary_only) — идентично _paginate_leaf_bucket ── + dropped_nb = 0 + if secondary_only: + filtered = [lot for lot in bucket_lots if lot.listing_segment != "novostroyki"] + dropped_nb = collected_this_bucket - len(filtered) + bucket_lots = filtered + + # Дедуп в общий seen + for lot in bucket_lots: + key = lot.source_id or lot.source_url + if key: + seen[key] = lot + + logger.info( + "avito: %s [%d, %s] paginated=%d pages collected=%d dropped_nb=%d " + "unique_total=%d incremental since=%s stop=%s", + room_label, + lo, + _hi_repr, + pages_fetched, + collected_this_bucket, + dropped_nb, + len(seen), + since.isoformat(), + stop_reason, + ) + + # ── on_bucket callback: инкрементальный save ────────────────────────── + if on_bucket is not None and bucket_lots: + res_cb = on_bucket(bucket_key, bucket_lots) + if inspect.isawaitable(res_cb): + await res_cb + # ── T6: citywide / byrooms fetch methods ────────────────────────────────── async def fetch_city_wide( diff --git a/tradein-mvp/backend/tests/scrapers/test_avito_incremental_pagination.py b/tradein-mvp/backend/tests/scrapers/test_avito_incremental_pagination.py new file mode 100644 index 00000000..b112e7cc --- /dev/null +++ b/tradein-mvp/backend/tests/scrapers/test_avito_incremental_pagination.py @@ -0,0 +1,315 @@ +"""INCREMENTAL SERP-пагинация Avito с date early-stop (param-gated, default off). + +`fetch_all_secondary(since=date)` на каждый (комнатность × seed-брекет) идёт +ПОСЛЕДОВАТЕЛЬНО newest-first и останавливается, как только страница уходит ниже +watermark `since` (SERP s=104 newest-first). `since=None` сохраняет точный +exhaustive bisection-обход (_walk_price_range) — без изменений. + +Тестируем через monkeypatch page-fetch seam (`_fetch_rooms_page_html`) + +`_parse_html`, отдавая контролируемые страницы ScrapedLot с заданными listing_date. +asyncio.sleep мокается в no-op для скорости (паттерн из +test_avito_403_backconnect_reconnect.py). +""" + +from datetime import date +from unittest.mock import AsyncMock, patch + +import pytest + +from app.services.scrapers import avito as avito_mod +from app.services.scrapers.avito import AvitoScraper +from app.services.scrapers.avito_exceptions import AvitoBlockedError +from app.services.scrapers.base import ScrapedLot + +SINCE = date(2026, 6, 10) + + +def _lot(item_id: str, listing_date: date | None, segment: str = "vtorichka") -> ScrapedLot: + """Минимальный валидный ScrapedLot с заданной датой/сегментом.""" + return ScrapedLot( + source="avito", + source_url=f"https://www.avito.ru/ekaterinburg/kvartiry/{item_id}", + source_id=item_id, + rooms=2, + price_rub=4_500_000, + listing_date=listing_date, + listing_segment=segment, + ) + + +def _make_scraper() -> AvitoScraper: + scraper = AvitoScraper() + scraper.request_delay_sec = 0.0 + return scraper + + +def _install_pages( + scraper: AvitoScraper, + monkeypatch: pytest.MonkeyPatch, + pages: list[list[ScrapedLot] | None], +) -> dict[str, int]: + """Подменяет page-fetch seam: страница p (1-based) → pages[p-1]. + + pages[i] is None → _fetch_rooms_page_html вернёт None (end-of-pages). + pages[i] == [] → HTML непустой, но _parse_html вернёт [] (end-of-pages). + Возвращает счётчик {"fetched": N} — сколько раз _fetch_rooms_page_html вызван. + """ + counter = {"fetched": 0} + + async def fake_fetch(room_slug: str, page: int, lo: int | None, hi: int | None) -> str | None: + counter["fetched"] += 1 + if page > len(pages): + return None + canned = pages[page - 1] + if canned is None: + return None + # Непустой HTML-маркер; реальный парс подменён ниже. + return f"" + + def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]: + # Достаём номер страницы из маркера, отдаём соответствующие лоты. + marker = html.split("data-page='")[1].split("'")[0] + page = int(marker) + canned = pages[page - 1] + return list(canned) if canned else [] + + monkeypatch.setattr(scraper, "_fetch_rooms_page_html", fake_fetch) + monkeypatch.setattr(scraper, "_parse_html", fake_parse) + return counter + + +@pytest.mark.asyncio +async def test_early_stop_on_all_older_page(monkeypatch: pytest.MonkeyPatch) -> None: + """page1 fresh, page2 mixed (fresh+old), page3 all older → стоп на page3, page4 НЕ фетчится.""" + scraper = _make_scraper() + pages: list[list[ScrapedLot] | None] = [ + [_lot("p1a", date(2026, 6, 15)), _lot("p1b", date(2026, 6, 14))], # все свежие + [_lot("p2a", date(2026, 6, 11)), _lot("p2b", date(2026, 6, 5))], # mixed (один >= since) + [_lot("p3a", date(2026, 6, 3)), _lot("p3b", date(2026, 6, 1))], # все старше since + [_lot("p4a", date(2026, 5, 20))], # НЕ должна фетчиться + ] + counter = _install_pages(scraper, monkeypatch, pages) + seen: dict[str, ScrapedLot] = {} + captured: list[tuple[str, list[ScrapedLot]]] = [] + + with patch.object(avito_mod.asyncio, "sleep", AsyncMock()): + await scraper._paginate_incremental_bracket( + room_slug="rooms-slug", + room_label="2к", + lo=0, + hi=None, + since=SINCE, + seen=seen, + max_pages_per_bucket=100, + secondary_only=True, + on_bucket=lambda k, lots: captured.append((k, lots)), + ) + + # page3 обработана (она old → break ПОСЛЕ неё), page4 не фетчилась. + assert counter["fetched"] == 3 + # Собраны лоты страниц 1-3 (включая boundary-страницу 3), но не 4. + assert set(seen.keys()) == {"p1a", "p1b", "p2a", "p2b", "p3a", "p3b"} + assert "p4a" not in seen + assert captured and captured[0][0] == "2к:0:open" + + +@pytest.mark.asyncio +async def test_no_early_stop_when_all_fresh(monkeypatch: pytest.MonkeyPatch) -> None: + """Все страницы свежее since до естественного конца (page возвращает None) → walk до конца.""" + scraper = _make_scraper() + pages: list[list[ScrapedLot] | None] = [ + [_lot("p1", date(2026, 6, 18))], + [_lot("p2", date(2026, 6, 16))], + [_lot("p3", date(2026, 6, 12))], + None, # end-of-pages + ] + counter = _install_pages(scraper, monkeypatch, pages) + seen: dict[str, ScrapedLot] = {} + + with patch.object(avito_mod.asyncio, "sleep", AsyncMock()): + await scraper._paginate_incremental_bracket( + room_slug="rooms-slug", + room_label="2к", + lo=0, + hi=None, + since=SINCE, + seen=seen, + max_pages_per_bucket=100, + secondary_only=True, + on_bucket=None, + ) + + # 3 страницы с лотами + 4-я (None) → 4 фетча, затем стоп. + assert counter["fetched"] == 4 + assert set(seen.keys()) == {"p1", "p2", "p3"} + + +@pytest.mark.asyncio +async def test_all_none_dates_grace_stop(monkeypatch: pytest.MonkeyPatch) -> None: + """Страницы all-None дат: page1 не стопит сразу, но 2 подряд undated → grace-стоп на page2.""" + scraper = _make_scraper() + pages: list[list[ScrapedLot] | None] = [ + [_lot("p1", None)], # undated → не стоп, streak=1 + [_lot("p2", None)], # undated → streak=2 → grace-стоп ПОСЛЕ page2 + [_lot("p3", None)], # НЕ должна фетчиться + ] + counter = _install_pages(scraper, monkeypatch, pages) + seen: dict[str, ScrapedLot] = {} + + with patch.object(avito_mod.asyncio, "sleep", AsyncMock()): + await scraper._paginate_incremental_bracket( + room_slug="rooms-slug", + room_label="2к", + lo=0, + hi=None, + since=SINCE, + seen=seen, + max_pages_per_bucket=100, + secondary_only=True, + on_bucket=None, + ) + + assert counter["fetched"] == 2 # page3 не фетчилась (grace bound) + assert set(seen.keys()) == {"p1", "p2"} + + +@pytest.mark.asyncio +async def test_none_date_does_not_stop_then_fresh_continues( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Одна undated-страница между свежими не триггерит стоп (streak сбрасывается).""" + scraper = _make_scraper() + pages: list[list[ScrapedLot] | None] = [ + [_lot("p1", date(2026, 6, 15))], # fresh + [_lot("p2", None)], # undated → streak=1, продолжаем + [_lot("p3", date(2026, 6, 12))], # fresh → streak сброшен + [_lot("p4", date(2026, 6, 2))], # all older → стоп после page4 + [_lot("p5", date(2026, 5, 1))], # НЕ фетчится + ] + counter = _install_pages(scraper, monkeypatch, pages) + seen: dict[str, ScrapedLot] = {} + + with patch.object(avito_mod.asyncio, "sleep", AsyncMock()): + await scraper._paginate_incremental_bracket( + room_slug="rooms-slug", + room_label="2к", + lo=0, + hi=None, + since=SINCE, + seen=seen, + max_pages_per_bucket=100, + secondary_only=True, + on_bucket=None, + ) + + assert counter["fetched"] == 4 + assert set(seen.keys()) == {"p1", "p2", "p3", "p4"} + + +@pytest.mark.asyncio +async def test_secondary_only_filters_novostroyki(monkeypatch: pytest.MonkeyPatch) -> None: + """secondary_only=True отбрасывает novostroyki до дедупа в seen (как leaf-bucket).""" + scraper = _make_scraper() + pages: list[list[ScrapedLot] | None] = [ + [ + _lot("vt1", date(2026, 6, 15), segment="vtorichka"), + _lot("nb1", date(2026, 6, 15), segment="novostroyki"), + ], + None, + ] + _install_pages(scraper, monkeypatch, pages) + seen: dict[str, ScrapedLot] = {} + + with patch.object(avito_mod.asyncio, "sleep", AsyncMock()): + await scraper._paginate_incremental_bracket( + room_slug="rooms-slug", + room_label="2к", + lo=0, + hi=None, + since=SINCE, + seen=seen, + max_pages_per_bucket=100, + secondary_only=True, + on_bucket=None, + ) + + assert set(seen.keys()) == {"vt1"} # novostroyki отфильтрована + + +@pytest.mark.asyncio +async def test_blocked_error_propagates(monkeypatch: pytest.MonkeyPatch) -> None: + """AvitoBlockedError из page-фетча пробрасывается наверх (не глушится).""" + scraper = _make_scraper() + + async def fake_fetch(room_slug: str, page: int, lo: int | None, hi: int | None) -> str | None: + raise AvitoBlockedError("hard block") + + monkeypatch.setattr(scraper, "_fetch_rooms_page_html", fake_fetch) + seen: dict[str, ScrapedLot] = {} + + with patch.object(avito_mod.asyncio, "sleep", AsyncMock()): + with pytest.raises(AvitoBlockedError): + await scraper._paginate_incremental_bracket( + room_slug="rooms-slug", + room_label="2к", + lo=0, + hi=None, + since=SINCE, + seen=seen, + max_pages_per_bucket=100, + secondary_only=True, + on_bucket=None, + ) + + +@pytest.mark.asyncio +async def test_since_none_uses_exhaustive_path(monkeypatch: pytest.MonkeyPatch) -> None: + """since=None → НЕ зовёт _paginate_incremental_bracket; идёт бисекция (_walk_price_range).""" + scraper = _make_scraper() + + walk_calls = {"n": 0} + incr_calls = {"n": 0} + + async def fake_walk(**kwargs: object) -> None: + walk_calls["n"] += 1 + + async def fake_incr(**kwargs: object) -> None: + incr_calls["n"] += 1 + + monkeypatch.setattr(scraper, "_walk_price_range", fake_walk) + monkeypatch.setattr(scraper, "_paginate_incremental_bracket", fake_incr) + + await scraper.fetch_all_secondary( + rooms_buckets=[("2-комнатные", "rooms-slug")], + since=None, + ) + + # Один room × N seed-брекетов → _walk_price_range зовётся, incremental — нет. + assert walk_calls["n"] == len(avito_mod._AVITO_PRICE_SEED_BRACKETS) + assert incr_calls["n"] == 0 + + +@pytest.mark.asyncio +async def test_since_date_uses_incremental_path(monkeypatch: pytest.MonkeyPatch) -> None: + """since=date → зовёт _paginate_incremental_bracket вместо _walk_price_range.""" + scraper = _make_scraper() + + walk_calls = {"n": 0} + incr_calls = {"n": 0} + + async def fake_walk(**kwargs: object) -> None: + walk_calls["n"] += 1 + + async def fake_incr(**kwargs: object) -> None: + incr_calls["n"] += 1 + + monkeypatch.setattr(scraper, "_walk_price_range", fake_walk) + monkeypatch.setattr(scraper, "_paginate_incremental_bracket", fake_incr) + + await scraper.fetch_all_secondary( + rooms_buckets=[("2-комнатные", "rooms-slug")], + since=SINCE, + ) + + assert incr_calls["n"] == len(avito_mod._AVITO_PRICE_SEED_BRACKETS) + assert walk_calls["n"] == 0