feat(avito): incremental SERP pagination with date early-stop (param-gated, default off)

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.

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(

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"<html data-page='{page}'></html>"
+
+    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