diff --git a/tradein-mvp/backend/app/services/scrapers/avito.py b/tradein-mvp/backend/app/services/scrapers/avito.py index e1ba79f4..bd39749c 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito.py +++ b/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: 1 178. # Текст может содержать NBSP (\xa0),  -entity или обычные пробелы как разделители # разрядов — чистятся в _extract_total_count. [^<]* нежадно ловит весь текст до . @@ -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,34 +886,61 @@ 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 - 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, + try: + 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, + ) + 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( diff --git a/tradein-mvp/backend/tests/scrapers/test_avito_429_bucket_skip.py b/tradein-mvp/backend/tests/scrapers/test_avito_429_bucket_skip.py new file mode 100644 index 00000000..6b40270e --- /dev/null +++ b/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 = ( + "" + '
' + '' + "2-к. квартира, 54,3 м², 5/9 эт." + "" + '' + "
" + "" +) + + +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 diff --git a/tradein-mvp/backend/tests/scrapers/test_avito_anti_bot.py b/tradein-mvp/backend/tests/scrapers/test_avito_anti_bot.py index f6ebb3d8..3cc12537 100644 --- a/tradein-mvp/backend/tests/scrapers/test_avito_anti_bot.py +++ b/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 (