fix(scrapers): ban/stuck-ротация для cian+yandex (provider-aware _rotate_proxy_ip)

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

@@ -359,6 +359,9 @@ class Settings(BaseSettings):
     # changeip-ссылка для Cian-прокси (ротация IP при бане/таймауте). Если не задан —
     # fallback на avito_proxy_rotate_url. ENV: CIAN_PROXY_ROTATE_URL.
     cian_proxy_rotate_url: str | None = None
+    # Максимум IP-ротаций для Cian на один sweep-прогон. Аналог avito_proxy_max_rotations.
+    # ENV: CIAN_PROXY_MAX_ROTATIONS.
+    cian_proxy_max_rotations: int = 4
 
     @property
     def cian_proxy_url(self) -> str | None:
@@ -373,6 +376,9 @@ class Settings(BaseSettings):
     # changeip-ссылка для Yandex-прокси (ротация IP при капче/таймауте). Если не задан —
     # fallback на avito_proxy_rotate_url. ENV: YANDEX_PROXY_ROTATE_URL.
     yandex_proxy_rotate_url: str | None = None
+    # Максимум IP-ротаций для Yandex на один sweep-прогон. Аналог avito_proxy_max_rotations.
+    # ENV: YANDEX_PROXY_MAX_ROTATIONS.
+    yandex_proxy_max_rotations: int = 4
 
     @property
     def yandex_proxy_url(self) -> str | None:

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

@@ -53,15 +53,29 @@ 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).
+async def _rotate_proxy_ip(*, reason: str, rotations_done: int, source: str = "avito") -> bool:
+    """Вызвать changeip-ссылку mobileproxy и подождать settle (#1790/#1848).
+
+    Provider-aware: выбирает rotate_url и max_rotations по параметру `source`
+    (avito / cian / yandex). Fallback-цепочка для rotate_url:
+      avito: avito_proxy_rotate_url
+      cian:  cian_proxy_rotate_url → avito_proxy_rotate_url
+      yandex: yandex_proxy_rotate_url → avito_proxy_rotate_url
 
     Аналог AvitoScraper._rotate_ip, но на уровне pipeline — используется в
-    enrichment-фазах (houses + detail) где нет доступа к экземпляру scraper'а.
+    enrichment-фазах где нет доступа к экземпляру scraper'а.
 
     Returns True при успешной смене IP, False если rotate_url не задан или ошибка.
-    Логирует каждую ротацию (причина, порядковый номер, остаток лимита).
+    Логирует каждую ротацию (причина, порядковый номер, source, остаток лимита).
     """
+    if source == "cian":
+        rotate_url = settings.cian_proxy_rotate_url or settings.avito_proxy_rotate_url
+        max_rot = settings.cian_proxy_max_rotations
+    elif source == "yandex":
+        rotate_url = settings.yandex_proxy_rotate_url or settings.avito_proxy_rotate_url
+        max_rot = settings.yandex_proxy_max_rotations
+    else:
+        # avito (default) — backward compat
         rotate_url = settings.avito_proxy_rotate_url
         max_rot = settings.avito_proxy_max_rotations
     if not rotate_url:
@@ -78,7 +92,9 @@ async def _rotate_proxy_ip(*, reason: str, rotations_done: int) -> bool:
             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",
+            "pipeline: IP rotated via changeip — source=%s reason=%s "
+            "rotation=#%d remaining=%d new_ip=%s",
+            source,
             reason,
             rotations_done + 1,
             max_rot - rotations_done - 1,
@@ -87,7 +103,8 @@ async def _rotate_proxy_ip(*, reason: str, rotations_done: int) -> bool:
         return True
     except Exception:
         logger.warning(
-            "pipeline: IP rotation failed (reason=%s, rotation=#%d)",
+            "pipeline: IP rotation failed (source=%s reason=%s rotation=#%d)",
+            source,
             reason,
             rotations_done + 1,
             exc_info=True,
@@ -759,9 +776,7 @@ async def run_avito_city_sweep(
                         scraper._browser = shared_bf
                         # Shared-browser режим: _cffi=None → curl_cffi-fallback на
                         # firewall пропускается (by design). Логируем для observability.
-                        logger.debug(
-                            "avito pipeline-mode: shared browser only, no cffi fallback"
-                        )
+                        logger.debug("avito pipeline-mode: shared browser only, no cffi fallback")
                     else:
                         scraper._cffi = session
 
@@ -1482,6 +1497,7 @@ async def run_yandex_city_sweep(
     enrich_delay = request_delay_sec if request_delay_sec is not None else 3.0
     _resolved_delay = request_delay_sec if request_delay_sec is not None else 9.0
     consecutive_failures = 0
+    yandex_rotations_done = 0  # #1848: бюджет IP-ротаций на весь sweep
 
     # Вычисляем watchdog-таймаут для combos-режима (центр, anchors=None).
     # В combos-режиме один "anchor" выполняет num_segments × num_combos × max_pages
@@ -1565,7 +1581,7 @@ async def run_yandex_city_sweep(
                 каждого (room×price) combo, сохраняет порцию сразу в БД + обновляет heartbeat.
                 anchor_lots накапливается (через _al) для address-enrich и price-history.
                 """
-                nonlocal consecutive_failures
+                nonlocal consecutive_failures, yandex_rotations_done
 
                 # ── Phase 1+2: SERP + инкрементальный save per-combo ──────
                 def _on_combo(
@@ -1675,19 +1691,25 @@ async def run_yandex_city_sweep(
                 if items:
                     from curl_cffi.requests import AsyncSession as _AsyncSession
 
+                    # #1848: rotation tracking для Yandex address-enrich (stuck-proxy)
+                    _yandex_enrich_consec_fail = 0
+                    _yandex_enrich_abort = 3  # abort address-enrich после N подряд
+
                     async with _AsyncSession(
                         impersonate="chrome120",
                         timeout=30.0,
                         proxies=_proxies,
                         headers={"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8"},
                     ) as enrich_session:
-                        for eidx, item in enumerate(items):
+                        eidx = 0
+                        while eidx < len(items):
+                            item = items[eidx]
                             lid: int = item["id"]
                             old_addr: str = item["address"]
                             url: str = item["source_url"]
                             try:
                                 resp = await enrich_session.get(url, allow_redirects=True)
-                                if resp.status_code != 200:
+                                if resp.status_code not in {200}:
                                     logger.warning(
                                         "yandex-sweep run_id=%d address-enrich:"
                                         " HTTP %d listing_id=%d",
@@ -1696,6 +1718,11 @@ async def run_yandex_city_sweep(
                                         lid,
                                     )
                                     counters.address_failed += 1
+                                    # 503/429 = прокси заблокирован → считаем как failure
+                                    if resp.status_code in {429, 503}:
+                                        _yandex_enrich_consec_fail += 1
+                                    else:
+                                        _yandex_enrich_consec_fail = 0
                                 else:
                                     new_addr = _extract_address_from_title(resp.text)
                                     if (
@@ -1741,17 +1768,53 @@ async def run_yandex_city_sweep(
                                                 db.rollback()
                                             except Exception:
                                                 pass
+                                    _yandex_enrich_consec_fail = 0
                             except Exception as fetch_exc:
                                 counters.address_failed += 1
+                                _yandex_enrich_consec_fail += 1
                                 logger.warning(
                                     "yandex-sweep run_id=%d: address-enrich "
-                                    "fetch failed listing_id=%d: %s",
+                                    "fetch failed listing_id=%d "
+                                    "(consecutive=%d): %s",
                                     run_id,
                                     lid,
+                                    _yandex_enrich_consec_fail,
                                     fetch_exc,
                                 )
+                            # Ротация Yandex-прокси при stuck/ban (#1848)
+                            if _yandex_enrich_consec_fail >= _yandex_enrich_abort:
+                                if yandex_rotations_done < settings.yandex_proxy_max_rotations:
+                                    rotated = await _rotate_proxy_ip(
+                                        reason=(
+                                            f"yandex enrich consecutive="
+                                            f"{_yandex_enrich_consec_fail}"
+                                        ),
+                                        rotations_done=yandex_rotations_done,
+                                        source="yandex",
+                                    )
+                                    yandex_rotations_done += 1
+                                    if rotated:
+                                        _yandex_enrich_consec_fail = 0
+                                        logger.info(
+                                            "yandex-sweep run_id=%d: enrich ROTATE — "
+                                            "IP changed, reset consecutive "
+                                            "(rotation #%d/%d)",
+                                            run_id,
+                                            yandex_rotations_done,
+                                            settings.yandex_proxy_max_rotations,
+                                        )
+                                        # retry текущего item после ротации
+                                        continue
+                                logger.error(
+                                    "yandex-sweep run_id=%d: address-enrich ABORT — "
+                                    "%d consecutive failures (proxy likely stuck/banned)",
+                                    run_id,
+                                    _yandex_enrich_consec_fail,
+                                )
+                                break
                             if eidx < len(items) - 1:
                                 await asyncio.sleep(enrich_delay)
+                            eidx += 1
 
                     logger.info(
                         "yandex-sweep run_id=%d anchor %s: address enrich=%d/%d failed=%d",
@@ -1954,6 +2017,7 @@ async def run_cian_city_sweep(
     _anchors = anchors if anchors is not None else EKB_ANCHORS
     counters = CianCitySweepCounters(anchors_total=len(_anchors))
     consecutive_failures = 0
+    cian_rotations_done = 0  # #1848: бюджет IP-ротаций на весь sweep
 
     try:
         for idx, (lat, lon, name) in enumerate(_anchors, start=1):
@@ -1994,7 +2058,7 @@ async def run_cian_city_sweep(
                 _a_name: str = _c_name,
             ) -> None:
                 """Все фазы одного cian anchor'а (SERP + detail + houses)."""
-                nonlocal anchor_lots, consecutive_failures
+                nonlocal anchor_lots, consecutive_failures, cian_rotations_done
                 from sqlalchemy import text as _text
 
                 # ── Phase 1+2: SERP + save ─────────────────────────────────
@@ -2051,7 +2115,12 @@ async def run_cian_city_sweep(
                         .mappings()
                         .all()
                     )
-                    for didx, row in enumerate(priority_rows):
+                    # #1848: Cian bans/stuck-proxy tracking + rotation (зеркало avito)
+                    _cian_detail_consec_failures = 0
+                    _cian_detail_abort = 3  # abort detail-фазы после N подряд
+                    didx = 0
+                    while didx < len(priority_rows):
+                        row = priority_rows[didx]
                         listing_id: int = row["id"]
                         source_url: str = row["source_url"]
                         counters.detail_attempted += 1
@@ -2069,18 +2138,54 @@ async def run_cian_city_sweep(
                                     listing_id,
                                     source_url,
                                 )
+                            _cian_detail_consec_failures = 0
                         except Exception as exc:
                             counters.detail_failed += 1
                             counters.errors_count += 1
+                            _cian_detail_consec_failures += 1
                             logger.warning(
-                                "cian-sweep run_id=%d: detail failed listing_id=%d: %s",
+                                "cian-sweep run_id=%d: detail failed listing_id=%d "
+                                "(consecutive=%d): %s",
                                 run_id,
                                 listing_id,
+                                _cian_detail_consec_failures,
                                 exc,
                             )
+                            if _cian_detail_consec_failures >= _cian_detail_abort:
+                                # Попытка ротации Cian-прокси перед abort'ом (#1848)
+                                if cian_rotations_done < settings.cian_proxy_max_rotations:
+                                    rotated = await _rotate_proxy_ip(
+                                        reason=(
+                                            f"cian detail consecutive="
+                                            f"{_cian_detail_consec_failures}"
+                                        ),
+                                        rotations_done=cian_rotations_done,
+                                        source="cian",
+                                    )
+                                    cian_rotations_done += 1
+                                    if rotated:
+                                        _cian_detail_consec_failures = 0
+                                        logger.info(
+                                            "cian-sweep run_id=%d: detail ROTATE — "
+                                            "IP changed, reset consecutive "
+                                            "(rotation #%d/%d)",
+                                            run_id,
+                                            cian_rotations_done,
+                                            settings.cian_proxy_max_rotations,
+                                        )
+                                        # retry текущего detail (не инкрементируем didx)
+                                        continue
+                                logger.error(
+                                    "cian-sweep run_id=%d: detail ABORT — "
+                                    "%d consecutive failures (proxy likely banned/stuck)",
+                                    run_id,
+                                    _cian_detail_consec_failures,
+                                )
+                                break
                         if didx < len(priority_rows) - 1:
                             jitter = random.uniform(0.8, 1.2)
                             await asyncio.sleep(request_delay_sec * jitter)
+                        didx += 1
                     logger.info(
                         "cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d",
                         run_id,

tradein-mvp/backend/tests/test_provider_rotation_1848.py

@@ -0,0 +1,242 @@
+"""Тесты Fix 2: provider-aware _rotate_proxy_ip (#1848).
+
+Тест 1: source='cian' → cian_proxy_rotate_url используется.
+Тест 2: source='yandex' → yandex_proxy_rotate_url используется.
+Тест 3: source='avito' (default) → avito_proxy_rotate_url (backward compat).
+Тест 4: cian без cian url → fallback на avito_proxy_rotate_url.
+Тест 5: rotate_url=None → returns False (no HTTP call).
+Тест 6: network error → returns False (no crash).
+Тест 7: yandex fallback на avito.
+"""
+
+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 _rotate_proxy_ip
+
+# ── Helpers ──────────────────────────────────────────────────────────────────
+
+
+def _async_session_ctx(resp_json: dict | None = None):
+    """Возвращает (async_cm_class, session_mock) для curl_cffi AsyncSession."""
+    session = AsyncMock()
+    if resp_json is not None:
+        mock_resp = MagicMock()
+        mock_resp.json.return_value = resp_json
+        session.get = AsyncMock(return_value=mock_resp)
+    else:
+        session.get = AsyncMock(side_effect=RuntimeError("network error"))
+
+    # curl_cffi AsyncSession используется как `async with AsyncSession(...) as rot:`
+    # Нам нужен callable класс-замена, вызов которого возвращает async context manager.
+    class _MockAsyncSession:
+        def __init__(self, timeout=30):
+            pass
+
+        async def __aenter__(self):
+            return session
+
+        async def __aexit__(self, *_):
+            pass
+
+    return _MockAsyncSession, session
+
+
+# ── Тест 1: cian_proxy_rotate_url ────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_cian_uses_cian_url() -> None:
+    """source='cian' → cian_proxy_rotate_url (не avito_proxy_rotate_url)."""
+    mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "1.2.3.4"})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+    ):
+        s.cian_proxy_rotate_url = "http://cian-rotate.test/"
+        s.avito_proxy_rotate_url = "http://avito-rotate.test/"
+        s.yandex_proxy_rotate_url = None
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian")
+
+    assert result is True
+    called_url: str = session.get.call_args[0][0]
+    assert "cian-rotate.test" in called_url
+    assert "avito-rotate.test" not in called_url
+
+
+# ── Тест 2: yandex_proxy_rotate_url ─────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_yandex_uses_yandex_url() -> None:
+    """source='yandex' → yandex_proxy_rotate_url."""
+    mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "5.6.7.8"})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+    ):
+        s.yandex_proxy_rotate_url = "http://yandex-rotate.test/"
+        s.avito_proxy_rotate_url = "http://avito-rotate.test/"
+        s.cian_proxy_rotate_url = None
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        result = await _rotate_proxy_ip(reason="timeout", rotations_done=0, source="yandex")
+
+    assert result is True
+    called_url: str = session.get.call_args[0][0]
+    assert "yandex-rotate.test" in called_url
+    assert "avito-rotate.test" not in called_url
+
+
+# ── Тест 3: avito backward compat ────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_avito_backward_compat() -> None:
+    """source не передан (дефолт 'avito') → avito_proxy_rotate_url."""
+    mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "9.9.9.9"})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+    ):
+        s.avito_proxy_rotate_url = "http://avito-only.test/"
+        s.cian_proxy_rotate_url = None
+        s.yandex_proxy_rotate_url = None
+        s.avito_proxy_max_rotations = 4
+        s.cian_proxy_max_rotations = 4
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        # Вызов без source= — должен работать как раньше
+        result = await _rotate_proxy_ip(reason="block", rotations_done=0)
+
+    assert result is True
+    called_url: str = session.get.call_args[0][0]
+    assert "avito-only.test" in called_url
+
+
+# ── Тест 4: cian fallback на avito когда нет cian url ─────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_cian_fallback_to_avito() -> None:
+    """cian_proxy_rotate_url=None → fallback на avito_proxy_rotate_url."""
+    mock_session_cls, session = _async_session_ctx(resp_json={})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+    ):
+        s.cian_proxy_rotate_url = None
+        s.avito_proxy_rotate_url = "http://avito-fallback.test/"
+        s.yandex_proxy_rotate_url = None
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian")
+
+    assert result is True
+    called_url: str = session.get.call_args[0][0]
+    assert "avito-fallback.test" in called_url
+
+
+# ── Тест 5: rotate_url=None → returns False ──────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_returns_false_when_no_url() -> None:
+    """Никакой URL не задан → False, HTTP-запрос не делается."""
+    mock_session_cls, session = _async_session_ctx(resp_json={})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+    ):
+        s.cian_proxy_rotate_url = None
+        s.avito_proxy_rotate_url = None
+        s.yandex_proxy_rotate_url = None
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.yandex_proxy_max_rotations = 4
+
+        result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian")
+
+    assert result is False
+    # HTTP-запрос не делался
+    session.get.assert_not_called()
+
+
+# ── Тест 6: network error → returns False ────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_returns_false_on_network_error() -> None:
+    """Ошибка сети → returns False (не крашит caller)."""
+    mock_session_cls, _session = _async_session_ctx(resp_json=None)  # raises RuntimeError
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+    ):
+        s.cian_proxy_rotate_url = "http://cian-rotate.test/"
+        s.avito_proxy_rotate_url = None
+        s.yandex_proxy_rotate_url = None
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian")
+
+    assert result is False
+
+
+# ── Тест 7: yandex fallback на avito ─────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_yandex_fallback_to_avito() -> None:
+    """yandex_proxy_rotate_url=None → fallback на avito_proxy_rotate_url."""
+    mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "11.22.33.44"})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+    ):
+        s.yandex_proxy_rotate_url = None
+        s.avito_proxy_rotate_url = "http://shared-proxy.test/"
+        s.cian_proxy_rotate_url = None
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        result = await _rotate_proxy_ip(reason="503", rotations_done=0, source="yandex")
+
+    assert result is True
+    called_url: str = session.get.call_args[0][0]
+    assert "shared-proxy.test" in called_url