diff --git a/.forgejo/workflows/deploy-tradein.yml b/.forgejo/workflows/deploy-tradein.yml index 71ee7f53..1c10cf23 100644 --- a/.forgejo/workflows/deploy-tradein.yml +++ b/.forgejo/workflows/deploy-tradein.yml @@ -47,6 +47,7 @@ jobs: - '.forgejo/workflows/deploy-tradein.yml' scraper: - 'tradein-mvp/backend/app/services/scrapers/**' + - 'tradein-mvp/backend/app/services/scrape_pipeline.py' - 'tradein-mvp/backend/app/services/scheduler.py' - 'tradein-mvp/backend/app/scheduler_main.py' - 'tradein-mvp/backend/app/tasks/**' diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 48a2a015..90e2d26c 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -56,6 +56,15 @@ logger = logging.getLogger(__name__) # Диапазон 180-300 с; 240 — разумный середина (4 мин > любой нормальный anchor). ANCHOR_TIMEOUT_SEC: int = 240 +# Константы для расчёта watchdog-таймаута combos-режима Yandex sweep. +# В combos-режиме единственный "anchor" делает num_combos × max_pages fetch'ей; +# каждый fetch занимает request_delay_sec + сетевой overhead. ADDRESS_ENRICH_BUDGET_S +# добавляет буфер на address-enrich фазу (per-listing HTTP + sleep × N листингов). +# Пример: 30 combos × 3 pages × (9+12) + 300 ≈ 2190s (~37 мин) — надёжно укладывается +# в окно 02:00-05:00 без риска перекрытия следующего sweep'а. +_YANDEX_COMBOS_PER_FETCH_S: float = 12.0 # network + parse margin (секунды на страницу) +_YANDEX_ADDRESS_ENRICH_BUDGET_S: float = 300.0 # бюджет на address-enrich фазу + # Default anchors ЕКБ — 5 точек покрытия города EKB_ANCHORS: list[tuple[float, float, str]] = [ (56.8400, 60.6050, "Центр"), @@ -760,8 +769,26 @@ async def run_yandex_city_sweep( counters = YandexCitySweepCounters(anchors_total=len(_anchors)) inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0 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 + # Вычисляем watchdog-таймаут для combos-режима (центр, anchors=None). + # В combos-режиме один "anchor" выполняет num_combos × max_pages fetch'ей — + # каждый занимает примерно _resolved_delay + _YANDEX_COMBOS_PER_FETCH_S секунд. + # Для 30 combos × 3 pages × (9+12) + 300 ≈ 2190s (~37 мин). + # В explicit-anchor (тест/override) режиме оставляем ANCHOR_TIMEOUT_SEC (240s) — + # там каждый anchor небольшой и watchdog работает как старый защитный барьер. + _num_combos = len(_rooms_list) * len(_price_ranges) + if anchors is None and _num_combos > 0: + _sweep_timeout = max( + ANCHOR_TIMEOUT_SEC, + _num_combos * pages_per_anchor * (_resolved_delay + _YANDEX_COMBOS_PER_FETCH_S) + + _YANDEX_ADDRESS_ENRICH_BUDGET_S, + ) + else: + # Explicit anchors (backward-compat тесты или ручной override) — legacy timeout. + _sweep_timeout = ANCHOR_TIMEOUT_SEC + # Прокси для address-enrich curl_cffi сессии (зеркало yandex_address_backfill). _proxy_url = settings.scraper_proxy_url _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None @@ -954,18 +981,18 @@ async def run_yandex_city_sweep( ) try: - await asyncio.wait_for(_yandex_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC) + await asyncio.wait_for(_yandex_anchor_phases(), timeout=_sweep_timeout) except TimeoutError: logger.warning( "yandex-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " - "timed out after %ds — skipping", + "timed out after %.0fs — skipping", run_id, idx, len(_anchors), name, lat, lon, - ANCHOR_TIMEOUT_SEC, + _sweep_timeout, ) counters.errors_count += 1 consecutive_failures += 1 @@ -987,7 +1014,7 @@ async def run_yandex_city_sweep( db, run_id, f"yandex sweep aborted: {consecutive_failures} consecutive " - f"anchor failures (last: anchor timeout {ANCHOR_TIMEOUT_SEC}s)", + f"anchor failures (last: anchor timeout {_sweep_timeout:.0f}s)", counters.to_dict(), ) return counters diff --git a/tradein-mvp/backend/tests/test_yandex_city_sweep.py b/tradein-mvp/backend/tests/test_yandex_city_sweep.py index fb6aa586..ca8ec585 100644 --- a/tradein-mvp/backend/tests/test_yandex_city_sweep.py +++ b/tradein-mvp/backend/tests/test_yandex_city_sweep.py @@ -797,3 +797,101 @@ async def test_explicit_anchors_still_works( assert c["price_ranges"] is not None assert fake.done is not None + + +# ── Watchdog timeout: combos vs explicit-anchor mode ─────────────────────── + + +def test_combos_sweep_timeout_substantially_larger_than_anchor_timeout() -> None: + """В combos/center mode (anchors=None) _sweep_timeout намного > ANCHOR_TIMEOUT_SEC. + + Проверяет что вычисленный таймаут для дефолтных 30 combos × 3 pages > 1500s, + т.е. не равен ANCHOR_TIMEOUT_SEC=240 и обеспечивает полный прогон sweep'а. + Не нужны мокапы — только публичные константы и арифметика из модуля. + """ + from app.services.scrape_pipeline import ( + _YANDEX_ADDRESS_ENRICH_BUDGET_S, + _YANDEX_COMBOS_PER_FETCH_S, + ANCHOR_TIMEOUT_SEC, + ) + from app.services.scrapers.yandex_realty import DEFAULT_PRICE_RANGES, ROOM_PATH + + # Параметры prod-режима + rooms_list = list(ROOM_PATH.keys()) # 5 комнатностей + price_ranges = DEFAULT_PRICE_RANGES # 6 ценовых диапазонов + max_pages = 3 + request_delay_sec = 9.0 # продакшн default + + num_combos = len(rooms_list) * len(price_ranges) + assert num_combos == 30, f"Expected 30 combos, got {num_combos}" + + sweep_timeout = max( + ANCHOR_TIMEOUT_SEC, + num_combos * max_pages * (request_delay_sec + _YANDEX_COMBOS_PER_FETCH_S) + + _YANDEX_ADDRESS_ENRICH_BUDGET_S, + ) + + # Должен быть существенно > 1500s (≈ 25 мин), что покрывает полный run. + assert sweep_timeout > 1500, ( + f"sweep_timeout={sweep_timeout:.0f}s unexpectedly small — " + f"combos-mode will still timeout mid-sweep" + ) + # И намного > ANCHOR_TIMEOUT_SEC (240s) + assert ( + sweep_timeout > ANCHOR_TIMEOUT_SEC * 4 + ), f"sweep_timeout={sweep_timeout:.0f}s should be >> ANCHOR_TIMEOUT_SEC={ANCHOR_TIMEOUT_SEC}s" + + +async def test_explicit_anchor_mode_uses_anchor_timeout_sec( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """anchors=[...] (explicit override) → wait_for используется с ANCHOR_TIMEOUT_SEC. + + Проверяет что в legacy/test режиме (explicit anchors) sweep_timeout не раздут + до combos-значения — он равен ANCHOR_TIMEOUT_SEC, т.к. каждый anchor маленький. + """ + import asyncio as _asyncio + + from app.services.scrape_pipeline import ANCHOR_TIMEOUT_SEC + + captured_timeouts: list[float] = [] + + _orig_wait_for = _asyncio.wait_for + + async def _capturing_wait_for(coro, *, timeout): # type: ignore[no-untyped-def] + captured_timeouts.append(timeout) + # На самом деле запустить — мокнутый scraper вернёт [] мгновенно. + return await _orig_wait_for(coro, timeout=timeout) + + monkeypatch.setattr(_asyncio, "wait_for", _capturing_wait_for) + + async def fake_fetch_multi( + self: YandexRealtyScraper, + lat: float, + lon: float, + radius_m: int = 1000, + max_pages: int = 2, + **_: object, + ) -> list[ScrapedLot]: + return [] + + monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi) + monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save) + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + await scrape_pipeline.run_yandex_city_sweep( + db=MagicMock(), + run_id=20, + anchors=TEST_ANCHORS[:2], # explicit anchors — legacy mode + pages_per_anchor=2, + enrich_address=False, + request_delay_sec=0.0, + ) + + # В explicit-anchor режиме каждый anchor должен использовать ANCHOR_TIMEOUT_SEC + assert len(captured_timeouts) == 2 + for t in captured_timeouts: + assert ( + t == ANCHOR_TIMEOUT_SEC + ), f"Expected ANCHOR_TIMEOUT_SEC={ANCHOR_TIMEOUT_SEC}s, got {t}s"