.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/**'

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

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"