Merge pull request 'fix(scrapers): avito ban-triggered IP rotation (#1790) разблокирует houses-enrichment (#1789) + rating_score' (#1848) from fix/avito-ban-rotation-houses into main

Reviewed-on: #1848

tradein-mvp/backend/app/core/config.py

@@ -277,6 +277,9 @@ class Settings(BaseSettings):
     # #1731: 2→4 — больше шансов восстановиться mid-sweep после проактивной
     # ротации на старте (Datadome ban recovery).
     avito_proxy_max_rotations: int = 4
+    # Settle-sleep после changeip-вызова: мобильный модем поднимает новый IP.
+    # ~9с по умолчанию (эмпирика mobileproxy.space). ENV: AVITO_PROXY_ROTATE_SETTLE_S.
+    avito_proxy_rotate_settle_s: float = 9.0
 
     # ── Cian dedicated mobile proxy (separate egress from Avito) ──────────────
     # Cian и Avito делят один мобильный IP при общем scraper_proxy_url → конкуренция

tradein-mvp/backend/app/services/scrape_pipeline.py

@@ -52,6 +52,49 @@ from app.services.yandex_price_history import record_yandex_price_history
 
 logger = logging.getLogger(__name__)
 
+
+async def _rotate_proxy_ip(*, reason: str, rotations_done: int) -> bool:
+    """Вызвать changeip-ссылку mobileproxy и подождать settle (#1790).
+
+    Аналог AvitoScraper._rotate_ip, но на уровне pipeline — используется в
+    enrichment-фазах (houses + detail) где нет доступа к экземпляру scraper'а.
+
+    Returns True при успешной смене IP, False если rotate_url не задан или ошибка.
+    Логирует каждую ротацию (причина, порядковый номер, остаток лимита).
+    """
+    rotate_url = settings.avito_proxy_rotate_url
+    max_rot = settings.avito_proxy_max_rotations
+    if not rotate_url:
+        return False
+    sep = "&" if "?" in rotate_url else "?"
+    try:
+        async with AsyncSession(timeout=30) as rot:
+            resp = await rot.get(f"{rotate_url}{sep}format=json")
+        new_ip: str = ""
+        try:
+            data = resp.json()
+            new_ip = str(data.get("new_ip", ""))
+        except Exception:
+            pass
+        await asyncio.sleep(settings.avito_proxy_rotate_settle_s)
+        logger.info(
+            "pipeline: IP rotated via changeip — reason=%s rotation=#%d remaining=%d new_ip=%s",
+            reason,
+            rotations_done + 1,
+            max_rot - rotations_done - 1,
+            new_ip or "unknown",
+        )
+        return True
+    except Exception:
+        logger.warning(
+            "pipeline: IP rotation failed (reason=%s, rotation=#%d)",
+            reason,
+            rotations_done + 1,
+            exc_info=True,
+        )
+        return False
+
+
 # Per-anchor watchdog timeout (секунды). Если обработка одного anchor'а занимает
 # дольше этого значения (зависший HTTP-запрос, прокси hang), asyncio.wait_for
 # прерывает её с TimeoutError, sweep продолжается со следующим anchor'ом.
@@ -200,8 +243,10 @@ async def run_avito_pipeline(
     lots: list[ScrapedLot] = []
     detail_delay = request_delay_sec if request_delay_sec is not None else 7.0
     # #1820: общий счётчик ПОСЛЕДОВАТЕЛЬНЫХ блоков на все enrichment-фазы (4/5/5b).
-    # Сбрасывается на любом успешном fetch; при достижении порога — enrichment_aborted.
+    # Сбрасывается на любом успешном fetch; при достижении порога — ротация IP (#1790)
+    # или abort если бюджет ротаций исчерпан.
     enrich_consecutive_blocks = 0
+    enrich_rotations_done = 0  # #1790: сколько раз уже сменили IP в этом run
     enrichment_aborted = False
 
     browser_mode = settings.scraper_fetch_mode == "browser"
@@ -302,6 +347,24 @@ async def run_avito_pipeline(
                         enrich_consecutive_blocks,
                     )
                     if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
+                        # #1790: Перед abort'ом пробуем сменить IP (changeip).
+                        # После ротации сбрасываем счётчик и продолжаем — НЕ abort.
+                        # Если ротации исчерпаны — abort enrichment (listings уже сохранены).
+                        if enrich_rotations_done < settings.avito_proxy_max_rotations:
+                            rotated = await _rotate_proxy_ip(
+                                reason=f"house consecutive={enrich_consecutive_blocks}",
+                                rotations_done=enrich_rotations_done,
+                            )
+                            enrich_rotations_done += 1
+                            if rotated:
+                                enrich_consecutive_blocks = 0
+                                logger.info(
+                                    "pipeline:enrich ROTATE — IP changed, "
+                                    "reset consecutive blocks (rotation #%d/%d)",
+                                    enrich_rotations_done,
+                                    settings.avito_proxy_max_rotations,
+                                )
+                                continue
                         logger.error(
                             "pipeline:enrich ABORT — %d consecutive blocks (IP rate-limited), "
                             "listings preserved",
@@ -400,6 +463,24 @@ async def run_avito_pipeline(
                         e,
                     )
                     if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
+                        # #1790: Перед abort'ом пробуем сменить IP (changeip).
+                        # После ротации сбрасываем счётчик — продолжаем detail-обход.
+                        # Если ротации исчерпаны — abort detail-фазы (listings уже сохранены).
+                        if enrich_rotations_done < settings.avito_proxy_max_rotations:
+                            rotated = await _rotate_proxy_ip(
+                                reason=f"detail consecutive={enrich_consecutive_blocks}",
+                                rotations_done=enrich_rotations_done,
+                            )
+                            enrich_rotations_done += 1
+                            if rotated:
+                                enrich_consecutive_blocks = 0
+                                logger.info(
+                                    "pipeline:enrich ROTATE — IP changed, "
+                                    "reset consecutive blocks (rotation #%d/%d)",
+                                    enrich_rotations_done,
+                                    settings.avito_proxy_max_rotations,
+                                )
+                                continue
                         logger.error(
                             "pipeline:detail ABORT — %d consecutive blocks (IP rate-limited), "
                             "listings preserved",
@@ -464,6 +545,22 @@ async def run_avito_pipeline(
                             enrich_consecutive_blocks,
                         )
                         if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
+                            # #1790: попытка ротации перед abort'ом (общий бюджет run).
+                            if enrich_rotations_done < settings.avito_proxy_max_rotations:
+                                rotated = await _rotate_proxy_ip(
+                                    reason=f"house-detail consecutive={enrich_consecutive_blocks}",
+                                    rotations_done=enrich_rotations_done,
+                                )
+                                enrich_rotations_done += 1
+                                if rotated:
+                                    enrich_consecutive_blocks = 0
+                                    logger.info(
+                                        "pipeline:enrich ROTATE — IP changed, "
+                                        "reset consecutive blocks (rotation #%d/%d)",
+                                        enrich_rotations_done,
+                                        settings.avito_proxy_max_rotations,
+                                    )
+                                    continue
                             logger.error(
                                 "pipeline:enrich ABORT — %d consecutive blocks "
                                 "(IP rate-limited), listings preserved",
@@ -599,6 +696,9 @@ async def run_avito_city_sweep(
     )
 
     browser_mode = settings.scraper_fetch_mode == "browser"
+    # #1790: общий бюджет IP-ротаций на весь sweep (разделяется по всем anchor'ам).
+    sweep_rotations_done = 0
+
     async with AsyncExitStack() as stack:
         session: AsyncSession | None = None
         shared_bf: BrowserFetcher | None = None
@@ -651,7 +751,7 @@ async def run_avito_city_sweep(
                     НЕМЕДЛЕННО после SERP+save (до detail-фазы). Это гарантирует, что
                     TimeoutError из wait_for теряет только detail-счётчики, но не SERP.
                     """
-                    nonlocal all_touched_house_ids
+                    nonlocal all_touched_house_ids, sweep_rotations_done
 
                     # ── Phase 1+2: SERP + save ─────────────────────────────────
                     scraper = AvitoScraper()
@@ -719,9 +819,12 @@ async def run_avito_city_sweep(
 
                     # ── Phase 4: enrich houses ─────────────────────────────────
                     touched_house_ids: set[int] = set()
+                    sweep_consecutive_house_blocks = 0  # #1790: per-anchor счётчик
                     if enrich_houses and unique_house_paths:
                         house_paths_list = list(unique_house_paths)
-                        for h_idx, house_path in enumerate(house_paths_list):
+                        h_idx = 0
+                        while h_idx < len(house_paths_list):
+                            house_path = house_paths_list[h_idx]
                             try:
                                 enrichment = await fetch_house_catalog(
                                     house_path,
@@ -730,16 +833,60 @@ async def run_avito_city_sweep(
                                 )
                                 hc = save_house_catalog_enrichment(db, enrichment)
                                 counters.houses_enriched += 1
+                                sweep_consecutive_house_blocks = 0
                                 hid = hc.get("house_id")
                                 if hid:
                                     touched_house_ids.add(int(hid))
-                            except (AvitoBlockedError, AvitoRateLimitedError):
-                                logger.error(
-                                    "city-sweep run_id=%d: house BLOCKED at %s — propagating",
+                                if h_idx < len(house_paths_list) - 1:
+                                    await asyncio.sleep(request_delay_sec)
+                                h_idx += 1
+                            except (AvitoBlockedError, AvitoRateLimitedError) as be:
+                                # #1790: пробуем сменить IP перед re-raise.
+                                sweep_consecutive_house_blocks += 1
+                                counters.houses_failed += 1
+                                counters.errors_count += 1
+                                logger.warning(
+                                    "city-sweep run_id=%d: house BLOCKED at %s "
+                                    "(consecutive=%d) — %s",
                                     run_id,
                                     house_path,
+                                    sweep_consecutive_house_blocks,
+                                    be,
                                 )
-                                raise
+                                if (
+                                    sweep_consecutive_house_blocks
+                                    >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT
+                                    and sweep_rotations_done < settings.avito_proxy_max_rotations
+                                ):
+                                    rotated = await _rotate_proxy_ip(
+                                        reason=(
+                                            f"city-sweep house "
+                                            f"consecutive={sweep_consecutive_house_blocks}"
+                                        ),
+                                        rotations_done=sweep_rotations_done,
+                                    )
+                                    sweep_rotations_done += 1
+                                    if rotated:
+                                        sweep_consecutive_house_blocks = 0
+                                        logger.info(
+                                            "city-sweep run_id=%d: house ROTATE — "
+                                            "IP changed, reset consecutive "
+                                            "(rotation #%d/%d)",
+                                            run_id,
+                                            sweep_rotations_done,
+                                            settings.avito_proxy_max_rotations,
+                                        )
+                                        # Не инкрементируем h_idx — retry текущего дома
+                                        continue
+                                elif sweep_consecutive_house_blocks >= 3:
+                                    logger.error(
+                                        "city-sweep run_id=%d: house ABORT — "
+                                        "%d consecutive blocks, propagating",
+                                        run_id,
+                                        sweep_consecutive_house_blocks,
+                                    )
+                                    raise
+                                h_idx += 1
                             except Exception as he:
                                 logger.warning(
                                     "city-sweep run_id=%d: house_enrich failed %s: %s",
@@ -753,8 +900,7 @@ async def run_avito_city_sweep(
                                     db.rollback()
                                 except Exception:
                                     pass
-                            if h_idx < len(house_paths_list) - 1:
-                                await asyncio.sleep(request_delay_sec)
+                                h_idx += 1
 
                     # ── Phase 5: enrich detail (top-N) ────────────────────────
                     if detail_top_n > 0 and anchor_lots:
@@ -833,6 +979,27 @@ async def run_avito_city_sweep(
                                     be,
                                 )
                                 if consecutive_blocks >= 3:
+                                    # #1790: попытка ротации перед abort'ом.
+                                    if sweep_rotations_done < settings.avito_proxy_max_rotations:
+                                        rotated = await _rotate_proxy_ip(
+                                            reason=(
+                                                f"city-sweep detail "
+                                                f"consecutive={consecutive_blocks}"
+                                            ),
+                                            rotations_done=sweep_rotations_done,
+                                        )
+                                        sweep_rotations_done += 1
+                                        if rotated:
+                                            consecutive_blocks = 0
+                                            logger.info(
+                                                "city-sweep run_id=%d: detail ROTATE — "
+                                                "IP changed, reset consecutive "
+                                                "(rotation #%d/%d)",
+                                                run_id,
+                                                sweep_rotations_done,
+                                                settings.avito_proxy_max_rotations,
+                                            )
+                                            continue
                                     logger.error(
                                         "city-sweep run_id=%d: detail ABORT — "
                                         "%d consecutive blocks (IP rate-limited)",
@@ -877,7 +1044,10 @@ async def run_avito_city_sweep(
                         if enrich_houses and new_house_paths:
                             counters.unique_houses += len(new_house_paths)
                             nh_list = list(new_house_paths)
-                            for nh_idx, house_path in enumerate(nh_list):
+                            nh_idx = 0
+                            sweep_consecutive_detail_house_blocks = 0  # #1790
+                            while nh_idx < len(nh_list):
+                                house_path = nh_list[nh_idx]
                                 try:
                                     enrichment = await fetch_house_catalog(
                                         house_path,
@@ -886,17 +1056,59 @@ async def run_avito_city_sweep(
                                     )
                                     hc = save_house_catalog_enrichment(db, enrichment)
                                     counters.houses_enriched += 1
+                                    sweep_consecutive_detail_house_blocks = 0
                                     hid = hc.get("house_id")
                                     if hid:
                                         touched_house_ids.add(int(hid))
-                                except (AvitoBlockedError, AvitoRateLimitedError):
-                                    logger.error(
+                                    if nh_idx < len(nh_list) - 1:
+                                        await asyncio.sleep(request_delay_sec)
+                                    nh_idx += 1
+                                except (AvitoBlockedError, AvitoRateLimitedError) as beh:
+                                    # #1790: ротация перед re-raise.
+                                    sweep_consecutive_detail_house_blocks += 1
+                                    counters.houses_failed += 1
+                                    counters.errors_count += 1
+                                    logger.warning(
                                         "city-sweep run_id=%d: house(detail) BLOCKED %s"
-                                        " — propagating",
+                                        " (consecutive=%d) — %s",
                                         run_id,
                                         house_path,
+                                        sweep_consecutive_detail_house_blocks,
+                                        beh,
                                     )
-                                    raise
+                                    if (
+                                        sweep_consecutive_detail_house_blocks >= 3
+                                        and sweep_rotations_done
+                                        < settings.avito_proxy_max_rotations
+                                    ):
+                                        rotated = await _rotate_proxy_ip(
+                                            reason=(
+                                                "city-sweep house(detail) "
+                                                f"consecutive="
+                                                f"{sweep_consecutive_detail_house_blocks}"
+                                            ),
+                                            rotations_done=sweep_rotations_done,
+                                        )
+                                        sweep_rotations_done += 1
+                                        if rotated:
+                                            sweep_consecutive_detail_house_blocks = 0
+                                            logger.info(
+                                                "city-sweep run_id=%d: house(detail) "
+                                                "ROTATE — IP changed (rotation #%d/%d)",
+                                                run_id,
+                                                sweep_rotations_done,
+                                                settings.avito_proxy_max_rotations,
+                                            )
+                                            continue
+                                    elif sweep_consecutive_detail_house_blocks >= 3:
+                                        logger.error(
+                                            "city-sweep run_id=%d: house(detail) ABORT"
+                                            " — %d consecutive blocks, propagating",
+                                            run_id,
+                                            sweep_consecutive_detail_house_blocks,
+                                        )
+                                        raise
+                                    nh_idx += 1
                                 except Exception as he:
                                     logger.warning(
                                         "city-sweep run_id=%d: house(detail) failed %s: %s",
@@ -910,8 +1122,7 @@ async def run_avito_city_sweep(
                                         db.rollback()
                                     except Exception:
                                         pass
-                                if nh_idx < len(nh_list) - 1:
-                                    await asyncio.sleep(request_delay_sec)
+                                    nh_idx += 1
 
                     all_touched_house_ids.update(touched_house_ids)
                     logger.info(

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

@@ -793,6 +793,35 @@ def parse_houses_state(state: dict[str, Any], house_url: str) -> HouseCatalogEnr
         reviews, next_page_url = _parse_reviews(reviews_widget)
         if not reviews:
             logger.warning("Виджет 'reviews' найден, но 0 отзывов распарсено — проверь шейп props")
+        # #1789: рейтинг переехал из housePage.developmentData.ratingPreview в
+        # reviews.filterEntries[type=="score"]. Применяем как fallback — только
+        # если housePage не дал рейтинг (чтобы не затирать уже правильный источник).
+        if house_info.rating_score is None:
+            reviews_payload = _widget_payload(reviews_widget, "reviews")
+            filter_entries: list[dict[str, Any]] = reviews_payload.get("filterEntries", [])
+            for fe in filter_entries:
+                if fe.get("type") == "score":
+                    score_value = fe.get("value", {})
+                    if isinstance(score_value, dict):
+                        score_float = score_value.get("scoreFloat")
+                        if score_float is not None:
+                            house_info.rating_score = float(score_float)
+                            house_info.rating_string = score_value.get("scoreString")
+                            rc = score_value.get("reviewCount")
+                            if rc is not None:
+                                house_info.reviews_count = int(rc)
+                            rating_stat = score_value.get("ratingStat")
+                            if isinstance(rating_stat, list):
+                                house_info.rating_distribution = rating_stat
+                            logger.info(
+                                "parse_houses_state: rating из reviews.filterEntries "
+                                "url=%s score=%.3f string=%s reviews=%s",
+                                house_url,
+                                house_info.rating_score,
+                                house_info.rating_string,
+                                house_info.reviews_count,
+                            )
+                    break
     else:
         logger.debug("Виджет 'reviews' не найден — пропускаем")
 

tradein-mvp/backend/tests/test_ban_rotation_1790.py

@@ -0,0 +1,216 @@
+"""#1790 — ban-triggered IP rotation в enrichment-фазах scrape_pipeline.
+
+Тест 1: AvitoBlockedError при доступных ротациях → вызывается _rotate_proxy_ip
+        (changeip) + consecutive счётчик сбрасывается → enrichment продолжается.
+
+Тест 2: Ротаций больше нет (бюджет исчерпан) → abort enrichment (bounded N ротаций).
+
+Тест 3: Graceful per-item (#1820) сохранён — один блок без порога ротации НЕ вызывает.
+
+Без сети, без БД.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from app.services.scrape_pipeline import (
+    _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT,
+    PipelineResult,
+    run_avito_pipeline,
+)
+from app.services.scrapers.avito_exceptions import AvitoBlockedError
+
+
+def _make_lot(
+    item_id: str,
+    house_url: str | None = None,
+) -> object:
+    from app.services.scrapers.base import ScrapedLot
+
+    return ScrapedLot(
+        source="avito",
+        source_url=f"https://www.avito.ru/ekaterinburg/kvartiry/test_{item_id}",
+        source_id=item_id,
+        price_rub=5_000_000,
+        address="Test address",
+        house_source="avito" if house_url else None,
+        house_ext_id=house_url.rstrip("/").split("/")[-1] if house_url else None,
+        house_url=house_url,
+        listing_segment="vtorichka",
+    )
+
+
+def _block_scraper(lots: list) -> MagicMock:  # type: ignore[type-arg]
+    scraper = MagicMock()
+    scraper.__aenter__ = AsyncMock(return_value=scraper)
+    scraper.__aexit__ = AsyncMock(return_value=None)
+    scraper.fetch_around = AsyncMock(return_value=lots)
+    scraper._cffi = None
+    return scraper
+
+
+# ---------------------------------------------------------------------------
+# Тест 1: ротация срабатывает при consecutive blocks == порог
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_rotation_called_on_consecutive_blocks() -> None:
+    """При N подряд-блоках (порог) вызывается _rotate_proxy_ip — IP меняется.
+
+    После успешной ротации consecutive счётчик сбрасывается и enrichment продолжается:
+    следующий (N+1)-й дом фетчится успешно.
+    """
+    n = _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT
+    lots = [
+        _make_lot(str(i), f"https://www.avito.ru/catalog/houses/ekb/h{i}/{100 + i}")
+        for i in range(n + 2)  # n блокируются, потом ротация, потом успех
+    ]
+    scraper = _block_scraper(lots)
+
+    # Первые n обращений — блок, (n+1)-й — успех
+    fetch_house = AsyncMock(side_effect=[AvitoBlockedError("banned")] * n + [MagicMock()])
+    rotate_mock = AsyncMock(return_value=True)  # ротация успешна
+
+    with (
+        patch("app.services.scrape_pipeline.AvitoScraper", return_value=scraper),
+        patch("app.services.scrape_pipeline.save_listings", return_value=(len(lots), 0)),
+        patch("app.services.scrape_pipeline.fetch_house_catalog", fetch_house),
+        patch(
+            "app.services.scrape_pipeline.save_house_catalog_enrichment",
+            return_value={"house_id": 99},
+        ),
+        patch("app.services.scrape_pipeline._rotate_proxy_ip", rotate_mock),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+        patch(
+            "curl_cffi.requests.AsyncSession",
+            return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()),
+        ),
+    ):
+        result = await run_avito_pipeline(
+            MagicMock(), 56.8, 60.6, 1000, enrich_houses=True, enrich_detail_top_n=0
+        )
+
+    assert isinstance(result, PipelineResult)
+    # _rotate_proxy_ip вызвана хотя бы один раз
+    assert rotate_mock.await_count >= 1
+    # Enrichment продолжился после ротации — (n+1)-й дом обогащён
+    assert result.counters.houses_enriched >= 1
+    # Listings сохранены
+    assert result.counters.lots_inserted == len(lots)
+
+
+# ---------------------------------------------------------------------------
+# Тест 2: бюджет ротаций исчерпан → abort
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_rotation_bounded_abort_when_budget_exhausted() -> None:
+    """После max_rotations попыток ротации enrichment abort'ится (bounded).
+
+    Используем avito_proxy_max_rotations=1 → только одна попытка ротации.
+    После неё блоки продолжаются → abort.
+    """
+    n = _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT
+    # Достаточно домов чтобы вызвать и ротацию и повторные блоки
+    lots = [
+        _make_lot(str(i), f"https://www.avito.ru/catalog/houses/ekb/h{i}/{200 + i}")
+        for i in range(n * 3)
+    ]
+    scraper = _block_scraper(lots)
+
+    # Все дома блокируются (не будет успешного фетча)
+    fetch_house = AsyncMock(side_effect=AvitoBlockedError("banned"))
+    # Ротация успешна, но дом после неё тоже блокируется
+    rotate_mock = AsyncMock(return_value=True)
+
+    with (
+        patch("app.services.scrape_pipeline.AvitoScraper", return_value=scraper),
+        patch("app.services.scrape_pipeline.save_listings", return_value=(len(lots), 0)),
+        patch("app.services.scrape_pipeline.fetch_house_catalog", fetch_house),
+        patch("app.services.scrape_pipeline.save_house_catalog_enrichment", return_value={}),
+        patch("app.services.scrape_pipeline._rotate_proxy_ip", rotate_mock),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+        patch("app.services.scrape_pipeline.settings") as mock_settings,
+        patch(
+            "curl_cffi.requests.AsyncSession",
+            return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()),
+        ),
+    ):
+        # Лимит 1 ротация: после неё бюджет исчерпан → abort на следующей серии блоков
+        mock_settings.avito_proxy_max_rotations = 1
+        mock_settings.avito_proxy_rotate_url = "http://rotate.test/"
+        mock_settings.avito_proxy_rotate_settle_s = 0.0
+        mock_settings.scraper_fetch_mode = "curl_cffi"
+        mock_settings.scraper_proxy_url = None
+        mock_settings.scraper_skip_seen_today = False
+
+        result = await run_avito_pipeline(
+            MagicMock(), 56.8, 60.6, 1000, enrich_houses=True, enrich_detail_top_n=0
+        )
+
+    assert isinstance(result, PipelineResult)
+    # Ровно 1 ротация (бюджет = 1)
+    assert rotate_mock.await_count == 1
+    # Enrichment abort'нут (есть AVITO_BLOCKED в ошибках)
+    assert any("AVITO_BLOCKED" in e for e in result.counters.errors)
+    # Listings всё равно сохранены
+    assert result.counters.lots_inserted == len(lots)
+
+
+# ---------------------------------------------------------------------------
+# Тест 3: один блок (ниже порога) — ротация НЕ вызывается (#1820 graceful)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_single_block_below_threshold_no_rotation() -> None:
+    """Один блок (< порога) — _rotate_proxy_ip НЕ вызывается.
+
+    Graceful per-item #1820: единственный блок пишется в errors, enrichment
+    продолжается, ротация не тратится.
+    """
+    n = _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT
+    # Только 2 дома: первый блокируется, второй успешен
+    lots = [
+        _make_lot("1", "https://www.avito.ru/catalog/houses/ekb/a/100"),
+        _make_lot("2", "https://www.avito.ru/catalog/houses/ekb/b/200"),
+    ]
+    assert n >= 2, "Порог должен быть >= 2 чтобы тест был корректен"
+    scraper = _block_scraper(lots)
+
+    fetch_house = AsyncMock(side_effect=[AvitoBlockedError("HTTP 403"), MagicMock()])
+    rotate_mock = AsyncMock(return_value=True)
+
+    with (
+        patch("app.services.scrape_pipeline.AvitoScraper", return_value=scraper),
+        patch("app.services.scrape_pipeline.save_listings", return_value=(2, 0)),
+        patch("app.services.scrape_pipeline.fetch_house_catalog", fetch_house),
+        patch(
+            "app.services.scrape_pipeline.save_house_catalog_enrichment",
+            return_value={"house_id": 55},
+        ),
+        patch("app.services.scrape_pipeline._rotate_proxy_ip", rotate_mock),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+        patch(
+            "curl_cffi.requests.AsyncSession",
+            return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock()),
+        ),
+    ):
+        result = await run_avito_pipeline(
+            MagicMock(), 56.8, 60.6, 1000, enrich_houses=True, enrich_detail_top_n=0
+        )
+
+    # Ротация НЕ вызвана — один блок ниже порога
+    assert rotate_mock.await_count == 0
+    # Первый дом упал, второй обогащён
+    assert result.counters.houses_failed == 1
+    assert result.counters.houses_enriched == 1
+    assert any("BLOCKED_HOUSE" in e for e in result.counters.errors)

tradein-mvp/backend/tests/test_houses_rating_filter_entries_1789.py

@@ -0,0 +1,189 @@
+"""#1789 — rating_score из reviews.filterEntries[type==score] (fallback).
+
+Avito перенёс сводный рейтинг из housePage.developmentData.ratingPreview в виджет
+reviews.filterEntries[{type: 'score', value: {scoreFloat, scoreString, reviewCount, ratingStat}}].
+
+Тест 1: когда housePage НЕ даёт рейтинг (ratingBadge пуст), rating берётся из
+        reviews.filterEntries — проверяем rating_score/rating_string/reviews_count/
+        rating_distribution.
+
+Тест 2: housePage даёт рейтинг (старый путь ratingBadge) → filterEntries НЕ
+        перезаписывает (приоритет housePage).
+
+Тест 3: новостройка (developmentPage) без reviews-виджета → graceful, нет краша.
+
+Без сети, без БД.
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from app.services.scrapers.avito_houses import parse_houses_state
+
+
+def _make_state(
+    *,
+    house_page_rating: dict | None = None,
+    reviews_filter_entries: list | None = None,
+) -> dict:
+    """Минимальный __preloadedState__ с housePage и опциональными reviews.
+
+    house_page_rating — если задан, проставляем в developmentData (старый путь).
+    reviews_filter_entries — если задан, проставляем в reviews.props.reviews.filterEntries
+    """
+    dev_data: dict = {
+        "avitoId": 555666,
+        "id": "aGFzaA==",
+        "title": "Test House",
+        "address": "ул. Тестовая, 1",
+        "coords": {"lat": 56.8, "lng": 60.6},
+        "aboutDevelopment": {"expandParams": {"items": []}},
+        "developer": {},
+        "mapPreview": {},
+    }
+    if house_page_rating is not None:
+        dev_data.update(house_page_rating)
+
+    placeholders: list = [
+        {
+            "type": "housePage",
+            "props": {"developmentData": dev_data},
+        },
+    ]
+
+    if reviews_filter_entries is not None:
+        placeholders.append(
+            {
+                "type": "reviews",
+                "props": {
+                    # MFE-форма: данные под ключом "reviews" внутри props
+                    "reviews": {
+                        "entries": [],
+                        "filterEntries": reviews_filter_entries,
+                    }
+                },
+            }
+        )
+
+    return {"data": {"data": {"page": {"placeholders": placeholders}}}}
+
+
+# ---------------------------------------------------------------------------
+# Тест 1: рейтинг из reviews.filterEntries (housePage пуст)
+# ---------------------------------------------------------------------------
+
+
+def test_rating_from_filter_entries_when_housepage_empty() -> None:
+    """Когда housePage не даёт рейтинг — берём из reviews.filterEntries."""
+    state = _make_state(
+        reviews_filter_entries=[
+            {
+                "type": "score",
+                "value": {
+                    "scoreFloat": 4.666,
+                    "scoreString": "4,7",
+                    "reviewCount": 3,
+                    "ratingStat": [
+                        {"score": 5, "count": 2},
+                        {"score": 4, "count": 1},
+                    ],
+                },
+            }
+        ]
+    )
+
+    e = parse_houses_state(state, "/catalog/houses/ekaterinburg/test/555666")
+    h = e.house
+
+    assert h.rating_score == pytest.approx(4.666, rel=1e-3)
+    assert h.rating_string == "4,7"
+    assert h.reviews_count == 3
+    assert len(h.rating_distribution) == 2
+    assert h.rating_distribution[0]["score"] == 5
+
+
+def test_rating_filter_entries_float_type() -> None:
+    """scoreFloat из filterEntries конвертируется в float (не остаётся int/str)."""
+    state = _make_state(
+        reviews_filter_entries=[
+            {
+                "type": "score",
+                "value": {"scoreFloat": 5, "scoreString": "5,0", "reviewCount": 1},
+            }
+        ]
+    )
+    e = parse_houses_state(state, "/catalog/houses/ekaterinburg/test/555666")
+    assert isinstance(e.house.rating_score, float)
+    assert e.house.rating_score == 5.0
+
+
+# ---------------------------------------------------------------------------
+# Тест 2: housePage даёт рейтинг → filterEntries НЕ перезаписывает
+# ---------------------------------------------------------------------------
+
+
+def test_housepage_rating_takes_priority_over_filter_entries() -> None:
+    """Если housePage.ratingBadge уже заполнен — filterEntries не перезаписывает."""
+    state = _make_state(
+        house_page_rating={
+            "ratingBadge": {"info": {"score": 3.9, "scoreString": "3,9"}},
+            "ratingSummaryStat": {"reviewCount": 10, "ratingStat": []},
+        },
+        reviews_filter_entries=[
+            {
+                "type": "score",
+                "value": {
+                    "scoreFloat": 4.9,
+                    "scoreString": "4,9",
+                    "reviewCount": 99,
+                    "ratingStat": [],
+                },
+            }
+        ],
+    )
+
+    e = parse_houses_state(state, "/catalog/houses/ekaterinburg/test/555666")
+    h = e.house
+
+    # Значение из housePage должно сохраниться
+    assert h.rating_score == pytest.approx(3.9, rel=1e-3)
+    assert h.rating_string == "3,9"
+    assert h.reviews_count == 10
+
+
+# ---------------------------------------------------------------------------
+# Тест 3: нет reviews-виджета вообще → None, не падаем
+# ---------------------------------------------------------------------------
+
+
+def test_no_reviews_widget_rating_is_none() -> None:
+    """Без виджета reviews и без ratingBadge → rating_score=None, graceful."""
+    state = _make_state()  # нет reviews_filter_entries
+
+    e = parse_houses_state(state, "/catalog/houses/ekaterinburg/test/555666")
+    h = e.house
+
+    assert h.rating_score is None
+    assert h.rating_string is None
+    assert h.reviews_count is None
+
+
+# ---------------------------------------------------------------------------
+# Тест 4: filterEntries без entry type==score → rating остаётся None
+# ---------------------------------------------------------------------------
+
+
+def test_filter_entries_without_score_type() -> None:
+    """filterEntries без type==score → рейтинг не заполняется."""
+    state = _make_state(
+        reviews_filter_entries=[
+            {"type": "something_else", "value": {"scoreFloat": 9.9}},
+        ]
+    )
+    e = parse_houses_state(state, "/catalog/houses/ekaterinburg/test/555666")
+    assert e.house.rating_score is None