From 757f6486f561d2e761a6bcd46b1d39e47f7db0e7 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 27 Jun 2026 11:24:54 +0300 Subject: [PATCH 1/2] =?UTF-8?q?fix(scrapers):=20avito=20changeip-resilienc?= =?UTF-8?q?e=20(retry=20=D0=BA=D0=BE=D1=80=D0=BE=D1=82=D0=BA=D0=B8=D0=BC?= =?UTF-8?q?=D0=B8=20=D0=BF=D0=BE=D0=BF=D1=8B=D1=82=D0=BA=D0=B0=D0=BC=D0=B8?= =?UTF-8?q?)=20+=20=D1=87=D0=B5=D1=81=D1=82=D0=BD=D1=8B=D0=B9=20=D1=81?= =?UTF-8?q?=D1=82=D0=B0=D1=82=D1=83=D1=81=20=D0=BA=D0=BE=D0=B3=D0=B4=D0=B0?= =?UTF-8?q?=20SERP=20=D0=BE=D1=82=D1=80=D0=B0=D0=B1=D0=BE=D1=82=D0=B0?= =?UTF-8?q?=D0=BB=20(#1950)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Два связанных фикса из диагностики 2026-06-27: ФИКС A (_rotate_proxy_ip): вместо одношотного GET с timeout=30s — retry-loop (proxy_rotate_attempts=3 попыток по proxy_rotate_attempt_timeout_s=8s каждая). Зависший changeip-сервис больше не блокирует весь sweep на 30s. Первый успешный attempt → True; все провалились → False (логируется как error). ФИКС B (run_avito_city_sweep): когда SERP-фаза уже сохранила лоты (lots_inserted + lots_updated > 0) и упала только detail/houses-фаза (AvitoBlockedError / RateLimitedError), помечаем run как done (не banned) при avito_serp_ok_not_banned=True (default). Метрика banned зарезервирована для случаев где SERP сам заблокирован (0 лотов). Новые настройки (ENV): PROXY_ROTATE_ATTEMPT_TIMEOUT_S=8.0 — timeout одной changeip-попытки PROXY_ROTATE_ATTEMPTS=3 — число попыток перед отказом AVITO_SERP_OK_NOT_BANNED=true — флаг нового поведения (default on) --- tradein-mvp/backend/app/core/config.py | 11 + .../backend/app/services/scrape_pipeline.py | 99 ++++-- .../tests/test_provider_rotation_1848.py | 3 + .../test_scraper_resilience_1949_1950.py | 336 ++++++++++++++++++ 4 files changed, 421 insertions(+), 28 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index d5500e2a..4be0645b 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -359,6 +359,17 @@ class Settings(BaseSettings): # Settle-sleep после changeip-вызова: мобильный модем поднимает новый IP. # ~9с по умолчанию (эмпирика mobileproxy.space). ENV: AVITO_PROXY_ROTATE_SETTLE_S. avito_proxy_rotate_settle_s: float = 9.0 + # #1950: retry-параметры changeip-GET (_rotate_proxy_ip). Вместо одношотного 30s-timeout + # делаем proxy_rotate_attempts попыток по proxy_rotate_attempt_timeout_s каждая. + # Короткий timeout (8s) означает, что зависший changeip не блокирует весь run на 30s. + # ENV: PROXY_ROTATE_ATTEMPT_TIMEOUT_S / PROXY_ROTATE_ATTEMPTS. + proxy_rotate_attempt_timeout_s: float = 8.0 + proxy_rotate_attempts: int = 3 + + # #1950: если SERP уже сохранил лоты (ins+upd > 0) и упали только detail/houses, + # ставим 'done' а не 'banned' — partial intake сохранён, 'banned' лишний. + # False = старое поведение. ENV: AVITO_SERP_OK_NOT_BANNED. + avito_serp_ok_not_banned: bool = True # ── Cian dedicated mobile proxy (separate egress from Avito) ────────────── # Cian и Avito делят один мобильный IP при общем scraper_proxy_url → конкуренция diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index d5632d4f..6141710e 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -81,35 +81,55 @@ async def _rotate_proxy_ip(*, reason: str, rotations_done: int, source: str = "a 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 = "" + # #1950: retry changeip-GET короткими попытками вместо одношотного 30s-timeout. + # Если changeip-сервер завис на 30s → весь прогон ждёт зря + ротация считается + # успешной (False возвращался). Теперь: proxy_rotate_attempts попыток по + # proxy_rotate_attempt_timeout_s каждая; возвращаем True при первом успехе. + _attempts = settings.proxy_rotate_attempts + _attempt_timeout = settings.proxy_rotate_attempt_timeout_s + for attempt in range(_attempts): try: - data = resp.json() - new_ip = str(data.get("new_ip", "")) + async with AsyncSession(timeout=_attempt_timeout) 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 — source=%s reason=%s " + "rotation=#%d attempt=%d/%d remaining=%d new_ip=%s", + source, + reason, + rotations_done + 1, + attempt + 1, + _attempts, + max_rot - rotations_done - 1, + new_ip or "unknown", + ) + return True except Exception: - pass - await asyncio.sleep(settings.avito_proxy_rotate_settle_s) - logger.info( - "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, - new_ip or "unknown", - ) - return True - except Exception: - logger.warning( - "pipeline: IP rotation failed (source=%s reason=%s rotation=#%d)", - source, - reason, - rotations_done + 1, - exc_info=True, - ) - return False + logger.warning( + "pipeline: IP rotation attempt %d/%d failed " "(source=%s reason=%s rotation=#%d)", + attempt + 1, + _attempts, + source, + reason, + rotations_done + 1, + exc_info=True, + ) + if attempt < _attempts - 1: + await asyncio.sleep(1.0) + logger.error( + "pipeline: all %d changeip attempts failed " "(source=%s reason=%s rotation=#%d)", + _attempts, + source, + reason, + rotations_done + 1, + ) + return False # Per-anchor watchdog timeout (секунды). Если обработка одного anchor'а занимает @@ -1187,7 +1207,30 @@ async def run_avito_city_sweep( counters.errors_count += 1 counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) - scrape_runs.mark_banned(db, run_id, str(e), counters.to_dict()) + # #1950: если SERP уже собрал лоты и заблокировало только detail/houses, + # ставим 'done' (не 'banned') — partial intake сохранён. + # За флагом avito_serp_ok_not_banned (default True). + _serp_ok = (counters.lots_inserted + counters.lots_updated) > 0 + if settings.avito_serp_ok_not_banned and _serp_ok: + _note = ( + f"detail enrichment aborted ({e}) но SERP intake завершён: " + f"{counters.lots_inserted} ins {counters.lots_updated} upd" + ) + logger.warning( + "city-sweep run_id=%d: SERP OK (ins=%d upd=%d), " + "detail blocked — DONE not banned. %s", + run_id, + counters.lots_inserted, + counters.lots_updated, + _note, + ) + scrape_runs.mark_done( + db, + run_id, + {**counters.to_dict(), "enrichment_abort_note": _note}, # type: ignore[arg-type] + ) + else: + scrape_runs.mark_banned(db, run_id, str(e), counters.to_dict()) return counters except Exception: logger.exception("city-sweep run_id=%d: anchor %s failed", run_id, name) diff --git a/tradein-mvp/backend/tests/test_provider_rotation_1848.py b/tradein-mvp/backend/tests/test_provider_rotation_1848.py index c9021119..2ef3cdce 100644 --- a/tradein-mvp/backend/tests/test_provider_rotation_1848.py +++ b/tradein-mvp/backend/tests/test_provider_rotation_1848.py @@ -200,6 +200,7 @@ async def test_rotate_proxy_ip_returns_false_on_network_error() -> None: 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 = None @@ -208,6 +209,8 @@ async def test_rotate_proxy_ip_returns_false_on_network_error() -> None: s.avito_proxy_max_rotations = 4 s.yandex_proxy_max_rotations = 4 s.avito_proxy_rotate_settle_s = 0.0 + s.proxy_rotate_attempts = 1 # одна попытка, без retry-sleep (#1950) + s.proxy_rotate_attempt_timeout_s = 8.0 result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian") diff --git a/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py b/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py new file mode 100644 index 00000000..07313c56 --- /dev/null +++ b/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py @@ -0,0 +1,336 @@ +"""Тесты устойчивости скрейперов: avito changeip-retry (#1950) + cian anti-zombie (#1949). + +Fix A (#1950): _rotate_proxy_ip теперь делает до proxy_rotate_attempts попыток + по proxy_rotate_attempt_timeout_s каждая. + +Fix B (#1950): run_avito_city_sweep при AvitoBlockedError/RateLimited с + lots_inserted+lots_updated>0 вызывает mark_done (не mark_banned) + когда avito_serp_ok_not_banned=True. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from app.services.scrapers.avito_exceptions import AvitoBlockedError +from app.services.scrapers.base import ScrapedLot + +# ────────────────────────────────────────────────────────────────────────────── +# Fix A: _rotate_proxy_ip retry loop (#1950) +# ────────────────────────────────────────────────────────────────────────────── + + +def _make_session_mock(get_side_effect: Any) -> AsyncMock: + """Вспомогательный: AsyncMock-сессия с заданным поведением .get().""" + mock_session = AsyncMock() + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + mock_session.get = AsyncMock(side_effect=get_side_effect) + return mock_session + + +@pytest.mark.asyncio +async def test_changeip_retry_success_on_second_attempt() -> None: + """#1950-A: первая попытка таймаут, вторая успех → возвращает True; ровно 2 GET-вызова.""" + from app.services.scrape_pipeline import _rotate_proxy_ip + + get_call_count = 0 + + async def fake_get(url: str) -> MagicMock: + nonlocal get_call_count + get_call_count += 1 + if get_call_count == 1: + raise TimeoutError("connect timeout") + resp = MagicMock() + resp.json.return_value = {"new_ip": "1.2.3.4"} + return resp + + mock_session = _make_session_mock(fake_get) + + with ( + patch("app.services.scrape_pipeline.AsyncSession", return_value=mock_session), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.proxy_rotate_attempts = 3 + s.proxy_rotate_attempt_timeout_s = 8.0 + s.avito_proxy_rotate_settle_s = 0.0 + s.avito_proxy_rotate_url = "http://changeip.example.com/rotate?token=X" + s.avito_proxy_max_rotations = 3 + + result = await _rotate_proxy_ip( + source="avito", + reason="test-detail-block", + rotations_done=0, + ) + + assert result is True + assert get_call_count == 2, f"expected 2 GET calls, got {get_call_count}" + + +@pytest.mark.asyncio +async def test_changeip_retry_all_fail_returns_false() -> None: + """#1950-A: все 3 попытки падают с TimeoutError → False; ровно 3 GET-вызова.""" + from app.services.scrape_pipeline import _rotate_proxy_ip + + get_call_count = 0 + + async def failing_get(url: str) -> None: + nonlocal get_call_count + get_call_count += 1 + raise TimeoutError("connect timeout") + + mock_session = _make_session_mock(failing_get) + + with ( + patch("app.services.scrape_pipeline.AsyncSession", return_value=mock_session), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.proxy_rotate_attempts = 3 + s.proxy_rotate_attempt_timeout_s = 8.0 + s.avito_proxy_rotate_settle_s = 0.0 + s.avito_proxy_rotate_url = "http://changeip.example.com/rotate?token=X" + s.avito_proxy_max_rotations = 3 + + result = await _rotate_proxy_ip( + source="avito", + reason="test-detail-block", + rotations_done=0, + ) + + assert result is False + assert ( + get_call_count == 3 + ), f"expected exactly 3 attempts (proxy_rotate_attempts=3), got {get_call_count}" + + +@pytest.mark.asyncio +async def test_changeip_retry_bounded_by_setting() -> None: + """#1950-A: proxy_rotate_attempts=2 → ровно 2 попытки, не больше.""" + from app.services.scrape_pipeline import _rotate_proxy_ip + + call_count = 0 + + async def failing_get(url: str) -> None: + nonlocal call_count + call_count += 1 + raise TimeoutError("timeout") + + mock_session = _make_session_mock(failing_get) + + with ( + patch("app.services.scrape_pipeline.AsyncSession", return_value=mock_session), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.proxy_rotate_attempts = 2 + s.proxy_rotate_attempt_timeout_s = 8.0 + s.avito_proxy_rotate_settle_s = 0.0 + s.cian_proxy_rotate_url = "http://changeip.example.com/rotate" + s.avito_proxy_rotate_url = None + s.cian_proxy_max_rotations = 3 + + result = await _rotate_proxy_ip( + source="cian", + reason="serp-block", + rotations_done=1, + ) + + assert result is False + assert call_count == 2, f"proxy_rotate_attempts=2 should make exactly 2 calls, got {call_count}" + + +# ────────────────────────────────────────────────────────────────────────────── +# Fix B: честный статус run_avito_city_sweep (#1950) +# ────────────────────────────────────────────────────────────────────────────── + + +def _make_city_sweep_mocks() -> tuple[MagicMock, MagicMock, MagicMock]: + """Возвращает (mock_db, mock_runs, mock_scraper) для тестов run_avito_city_sweep.""" + mock_db = MagicMock() + + # priority_rows для detail-фазы: 3 строки с source_url + class _DetailRow: + def __getitem__(self, key: str) -> str: + return "https://www.avito.ru/ekaterinburg/kvartiry/test-1" + + mock_db.execute.return_value.mappings.return_value.all.return_value = [ + _DetailRow(), + _DetailRow(), + _DetailRow(), + ] + + mock_runs = MagicMock() + mock_runs.is_cancelled.return_value = False + mock_runs.mark_done = MagicMock() + mock_runs.mark_banned = MagicMock() + mock_runs.mark_failed = MagicMock() + mock_runs.update_heartbeat = MagicMock() + + mock_scraper = MagicMock() + # AvitoScraper НЕ используется как async CM в _avito_anchor_phases (просто instantiate) + mock_scraper.fetch_around = AsyncMock( + return_value=[ + ScrapedLot( + source="avito", + source_url=f"https://www.avito.ru/e/k/{i}", + source_id=str(i), + price_rub=5_000_000, + address="ЕКБ test", + listing_segment="vtorichka", + ) + for i in range(3) + ] + ) + + return mock_db, mock_runs, mock_scraper + + +@pytest.mark.asyncio +async def test_avito_city_sweep_serp_ok_marks_done_not_banned() -> None: + """#1950-B: SERP собрал лоты (ins=3), detail заблокировало → mark_done не mark_banned. + + Условие: avito_serp_ok_not_banned=True (default), lots_inserted+lots_updated > 0. + """ + from app.services.scrape_pipeline import run_avito_city_sweep + + mock_db, mock_runs, mock_scraper = _make_city_sweep_mocks() + + # Shared AsyncSession mock (browser_mode=False) + mock_session = AsyncMock() + + with ( + patch("app.services.scrape_pipeline.AvitoScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.AsyncSession", return_value=mock_session), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch("app.services.scrape_pipeline.save_listings", return_value=(3, 0)), + patch( + "app.services.scrape_pipeline.fetch_detail", + AsyncMock(side_effect=AvitoBlockedError("HTTP 403 banned")), + ), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.scraper_fetch_mode = "curl_cffi" + s.scraper_proxy_url = None + s.avito_proxy_max_rotations = 0 # no rotation budget + s.avito_serp_ok_not_banned = True # ← наш флаг + s.avito_proxy_rotate_settle_s = 0.0 + s.proxy_rotate_attempts = 3 + s.proxy_rotate_attempt_timeout_s = 8.0 + + await run_avito_city_sweep( + mock_db, + run_id=1, + anchors=[(56.84, 60.61, "test-anchor")], + detail_top_n=3, + enrich_houses=False, + enrich_imv=False, + ) + + assert ( + mock_runs.mark_done.called + ), "mark_done должен быть вызван: SERP OK (ins=3), detail blocked" + assert ( + not mock_runs.mark_banned.called + ), "mark_banned НЕ должен быть вызван когда SERP уже сохранил лоты" + # enrichment_abort_note передан в counters + call_args = mock_runs.mark_done.call_args + # mark_done(db, run_id, counters_dict) + passed_counters: dict = call_args[0][2] if call_args[0] else {} + assert ( + "enrichment_abort_note" in passed_counters + ), "enrichment_abort_note должен присутствовать в counters переданных mark_done" + + +@pytest.mark.asyncio +async def test_avito_city_sweep_serp_zero_lots_marks_banned() -> None: + """#1950-B: SERP вернул 0 лотов, SERP-блок → mark_banned (не mark_done). + + Условие: lots_inserted+lots_updated == 0. + """ + from app.services.scrape_pipeline import run_avito_city_sweep + + mock_db, mock_runs, _ = _make_city_sweep_mocks() + + # SERP сам поднимает AvitoBlockedError → anchor_lots не сохраняется + mock_scraper_serp_block = MagicMock() + mock_scraper_serp_block.fetch_around = AsyncMock(side_effect=AvitoBlockedError("SERP HTTP 403")) + + mock_session = AsyncMock() + + with ( + patch("app.services.scrape_pipeline.AvitoScraper", return_value=mock_scraper_serp_block), + patch("app.services.scrape_pipeline.AsyncSession", return_value=mock_session), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.scraper_fetch_mode = "curl_cffi" + s.scraper_proxy_url = None + s.avito_proxy_max_rotations = 0 + s.avito_serp_ok_not_banned = True # включён флаг, но лотов нет + s.avito_proxy_rotate_settle_s = 0.0 + s.proxy_rotate_attempts = 3 + s.proxy_rotate_attempt_timeout_s = 8.0 + + await run_avito_city_sweep( + mock_db, + run_id=2, + anchors=[(56.84, 60.61, "test-anchor")], + detail_top_n=3, + enrich_houses=False, + enrich_imv=False, + ) + + assert ( + mock_runs.mark_banned.called + ), "mark_banned должен быть вызван: SERP сам заблокирован, lots=0" + assert not mock_runs.mark_done.called, "mark_done НЕ должен быть вызван при lots_inserted == 0" + + +@pytest.mark.asyncio +async def test_avito_city_sweep_flag_off_still_marks_banned() -> None: + """#1950-B: avito_serp_ok_not_banned=False → mark_banned даже при lots>0 (backward-compat).""" + from app.services.scrape_pipeline import run_avito_city_sweep + + mock_db, mock_runs, mock_scraper = _make_city_sweep_mocks() + mock_session = AsyncMock() + + with ( + patch("app.services.scrape_pipeline.AvitoScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.AsyncSession", return_value=mock_session), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch("app.services.scrape_pipeline.save_listings", return_value=(3, 0)), + patch( + "app.services.scrape_pipeline.fetch_detail", + AsyncMock(side_effect=AvitoBlockedError("HTTP 403")), + ), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.scraper_fetch_mode = "curl_cffi" + s.scraper_proxy_url = None + s.avito_proxy_max_rotations = 0 + s.avito_serp_ok_not_banned = False # ← флаг выключен + s.avito_proxy_rotate_settle_s = 0.0 + s.proxy_rotate_attempts = 3 + s.proxy_rotate_attempt_timeout_s = 8.0 + + await run_avito_city_sweep( + mock_db, + run_id=3, + anchors=[(56.84, 60.61, "test-anchor")], + detail_top_n=3, + enrich_houses=False, + enrich_imv=False, + ) + + assert ( + mock_runs.mark_banned.called + ), "avito_serp_ok_not_banned=False → mark_banned (backward-compat)" + assert not mock_runs.mark_done.called -- 2.45.3 From e24d3cf4bee9011ea94e30cd6fe9c386bc603107 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 27 Jun 2026 11:32:42 +0300 Subject: [PATCH 2/2] =?UTF-8?q?fix(scrapers):=20cian=20full=5Fload=20anti-?= =?UTF-8?q?zombie=20=E2=80=94=20per-fetch=20wait=5Ffor=20+=20heartbeat=20?= =?UTF-8?q?=D0=B2=D0=BD=D1=83=D1=82=D1=80=D0=B8=20=D0=B1=D0=B0=D0=BA=D0=B5?= =?UTF-8?q?=D1=82=D0=B0=20(#1949)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Два связанных фикса из диагностики 2026-06-27: ФИКС C (run_cian_full_load, per-fetch timeout): monkey-patch _browser.fetch → asyncio.wait_for(_f(url), timeout=cian_full_load_per_fetch_timeout_s=90s). BrowserFetcher.fetch имеет httpx-timeout 120s, но browser-сервис иногда виснет так что httpx не получает ответ (нет EOF) → fetch висит → asyncio.gather блокирует весь bucket → heartbeat_at не обновляется → reap_zombies убивает живой run. wait_for(90s) отменяет зависший fetch → TimeoutError → _fetch_page_html ловит как Exception → None → _one_page → [] → gather завершается нормально. За флагом (cian_full_load_per_fetch_timeout_s=0 = отключить, >0 = включить). ФИКС D (background heartbeat): asyncio.create_task(_background_heartbeat()) обновляет heartbeat_at каждые 60s независимо от on_bucket/on_progress прогресса. Без него пустые bucket или зависший gather замораживает heartbeat на несколько часов. Task отменяется в finally → не течёт при return/raise/cancel. Новая настройка (ENV): CIAN_FULL_LOAD_PER_FETCH_TIMEOUT_S=90.0 — timeout одного browser-fetch --- tradein-mvp/backend/app/core/config.py | 7 + .../backend/app/services/scrape_pipeline.py | 61 +++++ .../test_scraper_resilience_1949_1950.py | 222 ++++++++++++++++++ 3 files changed, 290 insertions(+) diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index 4be0645b..766959fb 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -384,6 +384,13 @@ class Settings(BaseSettings): # ENV: CIAN_PROXY_MAX_ROTATIONS. cian_proxy_max_rotations: int = 4 + # #1949: per-fetch hard-cancel для browser-fetch'ей внутри bucket-gather Cian full_load. + # asyncio.wait_for(fetch, timeout=X) отменяет зависший fetch → TimeoutError → + # _fetch_page_html ловит как Exception → None → _one_page → [] → gather завершается. + # 0 = отключить (старое поведение). Дефолт 90.0s щедрее nормального fetch ~12s. + # ENV: CIAN_FULL_LOAD_PER_FETCH_TIMEOUT_S. + cian_full_load_per_fetch_timeout_s: float = 90.0 + @property def cian_proxy_url(self) -> str | None: """Прокси для Cian-скраперов. CIAN_PROXY_URL > scraper_proxy_url (fallback).""" diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 6141710e..aa38b5ff 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -33,6 +33,7 @@ import random from contextlib import AsyncExitStack from dataclasses import dataclass, field, fields from datetime import date, timedelta +from typing import Any from urllib.parse import urlparse from curl_cffi.requests import AsyncSession @@ -2535,10 +2536,61 @@ async def run_cian_full_load( """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов).""" scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + # #1949: фоновый heartbeat — обновляет heartbeat_at каждые 60s независимо от on_bucket. + # Без него зависший asyncio.gather внутри bucket-loop замораживает heartbeat на часы + # → reap_zombies помечает живой run zombie. Отменяется в finally при любом завершении. + async def _background_heartbeat() -> None: + """Периодический heartbeat каждые 60s, не завязанный на прогресс бакетов.""" + while True: + await asyncio.sleep(60.0) + try: + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + except Exception: + pass + + _hb_task: asyncio.Task[None] | None = None try: + _hb_task = asyncio.create_task(_background_heartbeat()) + async with CianScraper() as scraper: scraper.request_delay_sec = request_delay_sec + # #1949: per-fetch hard-cancel для browser-fetch'ей внутри bucket-gather. + # BrowserFetcher.fetch имеет httpx-timeout 120s, но browser-сервис иногда + # виснет так что httpx не получает ответ → gather морозит → heartbeat стоп. + # Оборачиваем _browser.fetch в asyncio.wait_for(timeout=X) через monkey-patch: + # TimeoutError → _fetch_page_html ловит как Exception → None → _one_page → [] + # → gather завершается нормально → on_bucket вызывается → heartbeat обновляется. + # За флагом cian_full_load_per_fetch_timeout_s (0 = отключить). + _fetch_timeout = settings.cian_full_load_per_fetch_timeout_s + if _fetch_timeout > 0 and scraper._browser is not None: + _orig_fetch = scraper._browser.fetch + + async def _timed_fetch( + url: str, + _f: Any = _orig_fetch, + _t: float = _fetch_timeout, + _rid: int = run_id, + ) -> str: + try: + return await asyncio.wait_for(_f(url), timeout=_t) + except TimeoutError: + logger.warning( + "cian-full-load run_id=%d: browser-fetch timed out " + "after %.0fs — url=%s", + _rid, + _t, + url, + ) + raise + + scraper._browser.fetch = _timed_fetch # type: ignore[method-assign] + logger.info( + "cian-full-load run_id=%d: per-fetch timeout %.0fs enabled", + run_id, + _fetch_timeout, + ) + await scraper.fetch_all_secondary( price_cap_per_bucket=price_cap_per_bucket, concurrency=concurrency, @@ -2652,6 +2704,15 @@ async def run_cian_full_load( scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise + finally: + # #1949: отменяем фоновый heartbeat при любом завершении (return / raise / cancel). + if _hb_task is not None: + _hb_task.cancel() + try: + await _hb_task + except asyncio.CancelledError: + pass + # ── Yandex exhaustive full load (region-wide, без anchor'ов) ────────────────── diff --git a/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py b/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py index 07313c56..e659e375 100644 --- a/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py +++ b/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py @@ -6,15 +6,26 @@ Fix A (#1950): _rotate_proxy_ip теперь делает до proxy_rotate_atte Fix B (#1950): run_avito_city_sweep при AvitoBlockedError/RateLimited с lots_inserted+lots_updated>0 вызывает mark_done (не mark_banned) когда avito_serp_ok_not_banned=True. + +Fix C (#1949): run_cian_full_load применяет monkey-patch _browser.fetch с + asyncio.wait_for(timeout=cian_full_load_per_fetch_timeout_s). + +Fix D (#1949): run_cian_full_load создаёт фоновый heartbeat-task и отменяет + его в finally при любом завершении. """ from __future__ import annotations +import asyncio +import os +import time from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import pytest +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + from app.services.scrapers.avito_exceptions import AvitoBlockedError from app.services.scrapers.base import ScrapedLot @@ -334,3 +345,214 @@ async def test_avito_city_sweep_flag_off_still_marks_banned() -> None: mock_runs.mark_banned.called ), "avito_serp_ok_not_banned=False → mark_banned (backward-compat)" assert not mock_runs.mark_done.called + + +# ────────────────────────────────────────────────────────────────────────────── +# Fix C: cian per-fetch wait_for (#1949) +# ────────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_cian_full_load_hanging_fetch_cancelled_by_timeout() -> None: + """#1949-C: browser-fetch зависший навсегда отменяется per-fetch timeout. + + Monkey-patch заменяет _browser.fetch на _timed_fetch(wait_for timeout=0.05s). + Зависший fetch отменяется через ~50ms → TimeoutError → run завершается. + """ + from app.services.scrape_pipeline import run_cian_full_load + + fetch_was_called = False + + async def hanging_fetch(url: str) -> str: + nonlocal fetch_was_called + fetch_was_called = True + await asyncio.Event().wait() # висит вечно + return "" + + mock_browser = MagicMock() + mock_browser.fetch = hanging_fetch + + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper._browser = mock_browser + mock_scraper.request_delay_sec = 0.0 + + # fetch_all_secondary симулирует _fetch_page_html: вызывает fetch и ловит Exception + async def fake_fetch_all_secondary(**kwargs: object) -> None: + try: + # После monkey-patch _browser.fetch = _timed_fetch (с timeout=0.05s) + await mock_scraper._browser.fetch("https://ekb.cian.ru/sale/flat/1/") + except Exception: + pass # TimeoutError → аналог _fetch_page_html → returns None + + mock_scraper.fetch_all_secondary = fake_fetch_all_secondary + + mock_db = MagicMock() + mock_runs = MagicMock() + mock_runs.is_cancelled.return_value = False + mock_runs.mark_done = MagicMock() + mock_runs.mark_failed = MagicMock() + mock_runs.update_heartbeat = MagicMock() + + t0 = time.monotonic() + with ( + patch("app.services.scrapers.cian.CianScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_full_load_per_fetch_timeout_s = 0.05 # 50ms — быстрый timeout для теста + s.cian_full_load_detail_top_n = 0 + + await run_cian_full_load(mock_db, run_id=1) + + elapsed = time.monotonic() - t0 + + assert fetch_was_called, "hanging_fetch должен быть вызван через monkey-patched _browser.fetch" + assert ( + elapsed < 3.0 + ), f"run должен завершиться быстро (fetch timeout 50ms), занял {elapsed:.2f}s" + assert mock_runs.mark_done.called, "run_cian_full_load должен завершиться mark_done" + + +@pytest.mark.asyncio +async def test_cian_full_load_per_fetch_timeout_disabled_when_zero() -> None: + """#1949-C: cian_full_load_per_fetch_timeout_s=0 → monkey-patch НЕ применяется.""" + from app.services.scrape_pipeline import run_cian_full_load + + original_fetch = AsyncMock(return_value="") + + mock_browser = MagicMock() + mock_browser.fetch = original_fetch + + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper._browser = mock_browser + mock_scraper.request_delay_sec = 0.0 + mock_scraper.fetch_all_secondary = AsyncMock(return_value=None) + + mock_db = MagicMock() + mock_runs = MagicMock() + mock_runs.is_cancelled.return_value = False + mock_runs.mark_done = MagicMock() + mock_runs.mark_failed = MagicMock() + mock_runs.update_heartbeat = MagicMock() + + with ( + patch("app.services.scrapers.cian.CianScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_full_load_per_fetch_timeout_s = 0.0 # отключено + s.cian_full_load_detail_top_n = 0 + + await run_cian_full_load(mock_db, run_id=2) + + # Когда timeout=0 (отключён), monkey-patch не применяется → fetch остаётся original + assert ( + mock_browser.fetch is original_fetch + ), "cian_full_load_per_fetch_timeout_s=0 → fetch НЕ должен быть заменён monkey-patch" + + +# ────────────────────────────────────────────────────────────────────────────── +# Fix D: cian background heartbeat created and cancelled (#1949) +# ────────────────────────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_cian_full_load_background_heartbeat_created_and_cancelled() -> None: + """#1949-D: background heartbeat task создаётся и отменяется в finally.""" + from app.services.scrape_pipeline import run_cian_full_load + + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper._browser = None # per-fetch timeout не применяется + mock_scraper.fetch_all_secondary = AsyncMock(return_value=None) + + mock_db = MagicMock() + mock_runs = MagicMock() + mock_runs.is_cancelled.return_value = False + mock_runs.mark_done = MagicMock() + mock_runs.mark_failed = MagicMock() + mock_runs.update_heartbeat = MagicMock() + + created_tasks: list[asyncio.Task[None]] = [] + _real_create_task = asyncio.create_task + + def _tracking_create_task(coro: Any, *, name: str | None = None) -> asyncio.Task[None]: + task = _real_create_task(coro, name=name) + created_tasks.append(task) + return task # type: ignore[return-value] + + with ( + patch("app.services.scrapers.cian.CianScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch( + "app.services.scrape_pipeline.asyncio.create_task", + side_effect=_tracking_create_task, + ), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_full_load_per_fetch_timeout_s = 0.0 + s.cian_full_load_detail_top_n = 0 + + await run_cian_full_load(mock_db, run_id=3) + + assert ( + len(created_tasks) >= 1 + ), "asyncio.create_task должен быть вызван хотя бы раз (background heartbeat)" + for task in created_tasks: + assert ( + task.cancelled() or task.done() + ), f"Heartbeat task должен быть отменён в finally: {task}" + + +@pytest.mark.asyncio +async def test_cian_full_load_heartbeat_cancelled_on_exception() -> None: + """#1949-D: heartbeat task отменяется в finally даже при исключении в run.""" + from app.services.scrape_pipeline import run_cian_full_load + + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper._browser = None + mock_scraper.fetch_all_secondary = AsyncMock(side_effect=RuntimeError("fatal-test-error")) + + mock_db = MagicMock() + mock_runs = MagicMock() + mock_runs.is_cancelled.return_value = False + mock_runs.mark_done = MagicMock() + mock_runs.mark_failed = MagicMock() + mock_runs.update_heartbeat = MagicMock() + + created_tasks: list[asyncio.Task[None]] = [] + _real_create_task = asyncio.create_task + + def _tracking_create_task(coro: Any, *, name: str | None = None) -> asyncio.Task[None]: + task = _real_create_task(coro, name=name) + created_tasks.append(task) + return task # type: ignore[return-value] + + with ( + patch("app.services.scrapers.cian.CianScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch( + "app.services.scrape_pipeline.asyncio.create_task", + side_effect=_tracking_create_task, + ), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_full_load_per_fetch_timeout_s = 0.0 + s.cian_full_load_detail_top_n = 0 + + with pytest.raises(RuntimeError, match="fatal-test-error"): + await run_cian_full_load(mock_db, run_id=4) + + assert len(created_tasks) >= 1, "heartbeat task должен быть создан" + for task in created_tasks: + assert ( + task.cancelled() or task.done() + ), f"Heartbeat task должен быть отменён в finally даже при RuntimeError: {task}" + assert mock_runs.mark_failed.called, "mark_failed должен быть вызван при RuntimeError" -- 2.45.3