tradein-mvp/backend/app/services/scrapers/avito.py

@@ -250,10 +250,16 @@ _AVITO_OFFERS_PER_PAGE = 50
 
 # HTTP 429 в curl_cffi-режиме через backconnect-прокси (mproxy.site) — НЕ IP-ban, а
 # transient «слишком много одновременных соединений» (лимит 5). Проходит на коротком
-# retry без ротации IP. Делаем до _AVITO_429_MAX_RETRIES коротких пауз перед тем как
-# свалиться в старое поведение (rotation / AvitoRateLimitedError).
-_AVITO_429_MAX_RETRIES = 4
+# retry без ротации IP. Делаем до _AVITO_429_MAX_RETRIES коротких пауз; если они
+# исчерпаны под backconnect — ещё _AVITO_429_RECONNECT_RETRIES попыток через
+# пересоздание сессии (_reset_cffi): свежее соединение освобождает удерживаемый
+# conn-slot и может попасть на менее загруженный путь. Только потом —
+# AvitoRateLimitedError (run 216: deep-pagination 429 на page 36 ронял весь sweep).
+_AVITO_429_MAX_RETRIES = 8
 _AVITO_429_BACKOFF_SEC = 1.5
+# Reconnect-fallback на исчерпанном 429 short-retry (только backconnect): дольше спим
+# (_AVITO_403_BACKOFF_SEC), чтобы дать conn-slot'ам разгрузиться.
+_AVITO_429_RECONNECT_RETRIES = 3
 
 # HTTP 403 / firewall-интерстициал в curl_cffi-режиме через backconnect-прокси
 # (mproxy.site, нет changeip-URL): провайдер выдаёт СВЕЖИЙ exit-IP на каждое НОВОЕ
@@ -266,6 +272,12 @@ _AVITO_429_BACKOFF_SEC = 1.5
 _AVITO_403_MAX_RETRIES = 5
 _AVITO_403_BACKOFF_SEC = 2.0
 
+# Per-bucket block tolerance в fetch_all_secondary: одиночный 429/блок на ОДНОМ
+# (комнатность × ценовой брекет) бакете (напр. deep-pagination dense-бакета) НЕ должен
+# ронять весь sweep — бакет скипается, обход продолжается. Только N ПОДРЯД заблокированных
+# бакетов = реальный hard IP-ban → re-raise (run_avito_full_load → mark_banned).
+_AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED = 4
+
 # Счётчик результатов в шапке SERP: <span data-marker="page-title/count">1&nbsp;178</span>.
 # Текст может содержать NBSP (\xa0), &nbsp;-entity или обычные пробелы как разделители
 # разрядов — чистятся в _extract_total_count. [^<]* нежадно ловит весь текст до </span>.
@@ -445,6 +457,7 @@ class AvitoScraper(BaseScraper):
         rot_done = 0
         transient_left = 2
         r429 = 0
+        r429_reconnect = 0
         r403 = 0
         while True:
             try:
@@ -507,6 +520,22 @@ class AvitoScraper(BaseScraper):
                     await asyncio.sleep(_AVITO_403_BACKOFF_SEC)
                     continue
                 if sc == 429:
+                    # short-retry (тот же session) исчерпан. Под backconnect одна
+                    # фаза reconnect-fallback: пересоздание сессии освобождает
+                    # удерживаемый conn-slot и может лечь на менее загруженный путь
+                    # (deep-pagination 429 = conn-limit, не IP-ban). Только потом raise.
+                    if backconnect and r429_reconnect < _AVITO_429_RECONNECT_RETRIES:
+                        r429_reconnect += 1
+                        logger.info(
+                            "avito page=%d HTTP 429 short-retry exhausted — "
+                            "backconnect reconnect retry %d/%d",
+                            page,
+                            r429_reconnect,
+                            _AVITO_429_RECONNECT_RETRIES,
+                        )
+                        await self._reset_cffi()
+                        await asyncio.sleep(_AVITO_403_BACKOFF_SEC)
+                        continue
                     logger.error("avito SERP HTTP 429 (rate limited) page=%d url=%s", page, url)
                     raise AvitoRateLimitedError(
                         f"Avito SERP returned 429 — rate limited at page={page}"
@@ -836,6 +865,10 @@ class AvitoScraper(BaseScraper):
         """
         _buckets = rooms_buckets if rooms_buckets is not None else _AVITO_DEFAULT_ROOMS
         seen: dict[str, ScrapedLot] = {}
+        # Tolerance к одиночным per-bucket блокам (deep-pagination 429 dense-бакета):
+        # скипаем заблокированный бакет и продолжаем; N ПОДРЯД блоков = hard ban → re-raise.
+        consecutive_blocked = 0
+        skipped_blocked = 0
 
         for name, slug in _buckets:
             room_label = _avito_room_label(name)
@@ -853,6 +886,7 @@ 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
+                try:
                     if since is not None:
                         # INCREMENTAL: последовательная newest-first пагинация с ранней
                         # остановкой по watermark `since`; без бисекции по цене.
@@ -881,6 +915,32 @@ class AvitoScraper(BaseScraper):
                             on_bucket=on_bucket,
                             skip_buckets=skip_buckets,
                         )
+                except (AvitoRateLimitedError, AvitoBlockedError) as exc:
+                    # Одиночный блок на бакете (deep-pagination 429 dense-бакета) НЕ
+                    # роняет sweep: скипаем бакет, продолжаем. N ПОДРЯД блоков = реальный
+                    # hard IP-ban → re-raise (run_avito_full_load → mark_banned).
+                    consecutive_blocked += 1
+                    skipped_blocked += 1
+                    logger.warning(
+                        "avito exhaustive: %s bracket [%s..%s] skipped — blocked (%s); "
+                        "consecutive_blocked=%d/%d",
+                        room_label,
+                        br_lo,
+                        walk_hi,
+                        type(exc).__name__,
+                        consecutive_blocked,
+                        _AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED,
+                    )
+                    if consecutive_blocked >= _AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED:
+                        logger.error(
+                            "avito exhaustive: %d consecutive blocked buckets — "
+                            "treating as hard IP-ban, aborting sweep",
+                            consecutive_blocked,
+                        )
+                        raise
+                    continue
+                # Успешный бакет — сбрасываем счётчик подряд-блоков.
+                consecutive_blocked = 0
             room_collected = len(seen) - before
             logger.info(
                 "avito exhaustive: %s done — collected %d (total unique=%d)",
@@ -891,7 +951,11 @@ class AvitoScraper(BaseScraper):
             if on_progress is not None:
                 on_progress(len(seen))
 
-        logger.info("avito exhaustive: DONE — total unique=%d lots", len(seen))
+        logger.info(
+            "avito exhaustive: DONE — total unique=%d lots skipped_blocked=%d",
+            len(seen),
+            skipped_blocked,
+        )
         return list(seen.values())
 
     async def _walk_price_range(

tradein-mvp/backend/tests/scrapers/test_avito_429_bucket_skip.py

@@ -0,0 +1,242 @@
+"""429-hardening: widened short-retry + reconnect-fallback в _fetch_serp_html и
+per-bucket skip в fetch_all_secondary.
+
+Run 216 (прод): merged 403-reconnect-retry восстановил два 403, но прогон всё равно
+забанился на HTTP 429 на page 36 (deep-pagination dense-бакета). 429 под backconnect =
+transient conn-limit (лимит 5 соединений), не IP-ban.
+
+Две защиты:
+  A. _fetch_serp_html: short-retry того же session расширен до _AVITO_429_MAX_RETRIES=8;
+     на исчерпании под backconnect — ещё _AVITO_429_RECONNECT_RETRIES попыток через
+     _reset_cffi (освобождает conn-slot) перед AvitoRateLimitedError.
+  B. fetch_all_secondary: одиночный заблокированный бакет скипается (continue), sweep
+     продолжается; только _AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED ПОДРЯД блоков = hard ban
+     → re-raise (сохраняет real-ban detection).
+"""
+
+from unittest.mock import AsyncMock, MagicMock, 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, AvitoRateLimitedError
+
+# Валидная карточка SERP (одна) — successful page.
+_CARD_HTML = (
+    "<html><body>"
+    '<div data-marker="item" data-item-id="123456789">'
+    '<a data-marker="item-title" href="/ekaterinburg/kvartiry/test-123456789">'
+    "2-к. квартира, 54,3 м², 5/9 эт."
+    "</a>"
+    '<meta itemprop="price" content="4500000"/>'
+    "</div>"
+    "</body></html>"
+)
+
+
+def _enable_backconnect(monkeypatch: pytest.MonkeyPatch) -> None:
+    """scraper_proxy_url задан (через env-поле), changeip-URL — нет → backconnect."""
+    monkeypatch.setattr(avito_mod.settings, "scraper_proxy_url_env", "http://u:p@mproxy.site:14619")
+    monkeypatch.setattr(avito_mod.settings, "avito_proxy_url", None)
+    monkeypatch.setattr(avito_mod.settings, "avito_proxy_rotate_url", None)
+
+
+def _resp(status: int, text: str = "") -> MagicMock:
+    r = MagicMock()
+    r.status_code = status
+    r.text = text
+    return r
+
+
+# --- A. _fetch_serp_html 429 short-retry + reconnect-fallback -------------------
+
+
+@pytest.mark.asyncio
+async def test_429_short_retries_then_200_within_widened_budget(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """429×6 (< _AVITO_429_MAX_RETRIES=8) → 200: short-retry того же session снимает,
+    _reset_cffi НЕ дёргается (reconnect-fallback не нужен)."""
+    _enable_backconnect(monkeypatch)
+    n_429 = 6
+    assert n_429 < avito_mod._AVITO_429_MAX_RETRIES
+
+    scraper = AvitoScraper()
+    seq = [_resp(429) for _ in range(n_429)] + [_resp(200, _CARD_HTML)]
+    mock_session = AsyncMock()
+    mock_session.get = AsyncMock(side_effect=seq)
+    scraper._cffi = mock_session
+
+    reset = AsyncMock()
+    with (
+        patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()),
+        patch.object(AvitoScraper, "_reset_cffi", reset),
+    ):
+        html = await scraper._fetch_serp_html("https://www.avito.ru/x", page=36)
+
+    assert html == _CARD_HTML
+    reset.assert_not_awaited()  # short-retry хватило, reconnect не понадобился
+    assert mock_session.get.await_count == n_429 + 1
+
+
+@pytest.mark.asyncio
+async def test_429_exhausted_then_reconnect_then_200(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """short-retry исчерпан (429×_AVITO_429_MAX_RETRIES) → reconnect-fallback (_reset_cffi)
+    → 200 успех. Подтверждает: одна reconnect-фаза освобождает conn-slot."""
+    _enable_backconnect(monkeypatch)
+    n_short = avito_mod._AVITO_429_MAX_RETRIES
+
+    scraper = AvitoScraper()
+    # _AVITO_429_MAX_RETRIES коротких retry (все 429), ещё один 429 проваливается в
+    # reconnect-block, после _reset_cffi следующий ответ 200.
+    seq = [_resp(429) for _ in range(n_short + 1)] + [_resp(200, _CARD_HTML)]
+    mock_session = AsyncMock()
+    mock_session.get = AsyncMock(side_effect=seq)
+    scraper._cffi = mock_session
+
+    reset = AsyncMock()
+    with (
+        patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()),
+        patch.object(AvitoScraper, "_reset_cffi", reset),
+    ):
+        html = await scraper._fetch_serp_html("https://www.avito.ru/x", page=36)
+
+    assert html == _CARD_HTML
+    reset.assert_awaited()  # reconnect-fallback сработал
+    assert reset.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_429_fully_exhausted_raises_rate_limited(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """429 на каждый запрос: short-retry + все reconnect-fallback исчерпаны →
+    AvitoRateLimitedError. _reset_cffi дёрнут ровно _AVITO_429_RECONNECT_RETRIES раз."""
+    _enable_backconnect(monkeypatch)
+
+    scraper = AvitoScraper()
+    mock_session = AsyncMock()
+    mock_session.get = AsyncMock(return_value=_resp(429))
+    scraper._cffi = mock_session
+
+    reset = AsyncMock()
+    with (
+        patch("app.services.scrapers.avito.asyncio.sleep", AsyncMock()),
+        patch.object(AvitoScraper, "_reset_cffi", reset),
+    ):
+        with pytest.raises(AvitoRateLimitedError, match="rate limited"):
+            await scraper._fetch_serp_html("https://www.avito.ru/x", page=36)
+
+    assert reset.await_count == avito_mod._AVITO_429_RECONNECT_RETRIES
+    # get: 1 первичный + _AVITO_429_MAX_RETRIES short-retry + _AVITO_429_RECONNECT_RETRIES
+    assert (
+        mock_session.get.await_count
+        == 1 + avito_mod._AVITO_429_MAX_RETRIES + avito_mod._AVITO_429_RECONNECT_RETRIES
+    )
+
+
+# --- B. fetch_all_secondary per-bucket skip -----------------------------------
+
+
+def _single_room() -> list[tuple[str, str]]:
+    """Один room-bucket → число брекетов = len(_AVITO_PRICE_SEED_BRACKETS)."""
+    return [("2-комнатные", "2-komnatnye")]
+
+
+@pytest.mark.asyncio
+async def test_one_blocked_bucket_skipped_sweep_completes(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """_walk_price_range кидает AvitoRateLimitedError на 1-м брекете, успех на остальных →
+    sweep завершается без raise, заблокированный бакет посчитан."""
+    n_brackets = len(avito_mod._AVITO_PRICE_SEED_BRACKETS)
+    assert n_brackets >= 2
+
+    scraper = AvitoScraper()
+    calls = {"n": 0}
+
+    async def fake_walk(self: object, **kwargs: object) -> None:
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise AvitoRateLimitedError("429 deep pagination")
+        # успех: ничего не добавляем в seen (не нужно для теста контракта)
+
+    with patch.object(AvitoScraper, "_walk_price_range", fake_walk):
+        result = await scraper.fetch_all_secondary(rooms_buckets=_single_room())
+
+    assert result == []  # seen пуст, но без raise
+    assert calls["n"] == n_brackets  # все брекеты пройдены (1 скипнут, остальные ок)
+
+
+@pytest.mark.asyncio
+async def test_consecutive_blocked_threshold_reraises(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """_walk_price_range кидает _AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED раз ПОДРЯД →
+    последнее исключение пробрасывается (real hard-ban detection сохранена)."""
+    thr = avito_mod._AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED
+    assert len(avito_mod._AVITO_PRICE_SEED_BRACKETS) >= thr
+
+    scraper = AvitoScraper()
+
+    async def always_block(self: object, **kwargs: object) -> None:
+        raise AvitoBlockedError("IP banned")
+
+    with patch.object(AvitoScraper, "_walk_price_range", always_block):
+        with pytest.raises(AvitoBlockedError, match="IP banned"):
+            await scraper.fetch_all_secondary(rooms_buckets=_single_room())
+
+
+@pytest.mark.asyncio
+async def test_success_between_blocks_resets_counter(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Чередование block/success НЕ срабатывает порог: успех между блоками сбрасывает
+    consecutive_blocked. Паттерн block,success,block,success,... → sweep завершается."""
+    thr = avito_mod._AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED
+    n_brackets = len(avito_mod._AVITO_PRICE_SEED_BRACKETS)
+    # Нужно достаточно брекетов чтобы суммарно было > thr блоков, но не подряд.
+    assert n_brackets > thr
+
+    scraper = AvitoScraper()
+    calls = {"n": 0}
+
+    async def alternate(self: object, **kwargs: object) -> None:
+        calls["n"] += 1
+        # нечётные вызовы блокируются, чётные — успех → никогда не thr подряд
+        if calls["n"] % 2 == 1:
+            raise AvitoRateLimitedError("429 transient")
+
+    with patch.object(AvitoScraper, "_walk_price_range", alternate):
+        result = await scraper.fetch_all_secondary(rooms_buckets=_single_room())
+
+    assert result == []  # без raise — порог не пробит
+    assert calls["n"] == n_brackets
+
+
+@pytest.mark.asyncio
+async def test_incremental_path_also_skips_blocked_bucket(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Per-bucket skip применяется и к incremental-пути (_paginate_incremental_bracket)."""
+    from datetime import date
+
+    n_brackets = len(avito_mod._AVITO_PRICE_SEED_BRACKETS)
+    scraper = AvitoScraper()
+    calls = {"n": 0}
+
+    async def fake_incr(self: object, **kwargs: object) -> None:
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise AvitoRateLimitedError("429")
+
+    with patch.object(AvitoScraper, "_paginate_incremental_bracket", fake_incr):
+        result = await scraper.fetch_all_secondary(
+            rooms_buckets=_single_room(), since=date(2026, 1, 1)
+        )
+
+    assert result == []
+    assert calls["n"] == n_brackets

tradein-mvp/backend/tests/scrapers/test_avito_anti_bot.py

@@ -521,8 +521,11 @@ async def test_fetch_around_429_exhausted_with_rotate_url_rotates(
     ok.status_code = 200
     ok.text = _CARD_HTML
     mock_session = AsyncMock()
-    # 429 на коротких retry, после ротации — 200
-    mock_session.get = AsyncMock(side_effect=[r429, r429, r429, r429, r429, ok])
+    # 429 на всех коротких retry (1 первичный + _AVITO_429_MAX_RETRIES), после
+    # исчерпания short-retry — ротация → 200. backconnect-reconnect здесь НЕ
+    # срабатывает (scraper_proxy_url не задан), поэтому сразу rotate-путь.
+    n_429 = 1 + avito_mod._AVITO_429_MAX_RETRIES
+    mock_session.get = AsyncMock(side_effect=[r429] * n_429 + [ok])
     scraper._cffi = mock_session
 
     with (