diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index 9e2c7b94..50c9a881 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/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: diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 5f75bf97..d5632d4f 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -53,17 +53,31 @@ 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, остаток лимита). """ - rotate_url = settings.avito_proxy_rotate_url - max_rot = settings.avito_proxy_max_rotations + 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: return False sep = "&" if "?" in rotate_url else "?" @@ -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, diff --git a/tradein-mvp/backend/tests/test_provider_rotation_1848.py b/tradein-mvp/backend/tests/test_provider_rotation_1848.py new file mode 100644 index 00000000..c9021119 --- /dev/null +++ b/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