diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index b33f6142..434d26ac 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -175,6 +175,45 @@ _AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT: int = 5 # паттерн avito_detail_backfill (max_consecutive_blocks). _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: int = 3 +# #2160: константы для расчёта watchdog-таймаута Cian city sweep. При +# USE_PROXY_POOL_BROWSER=true каждый SERP-фетч идёт через camoufox с relaunch при смене +# прокси (page.goto timeout 60s + overhead) = 13-45s/страница, а якорь cian = 4 room-buckets +# × pages_per_anchor SERP-фетчей + detail_top_n detail-фетчей + houses-enrich. Фиксированный +# ANCHOR_TIMEOUT_SEC=240 гильотинит якорь ещё в SERP-фазе → save_listings не успевает, +# lots_fetched=0, после N подряд → mark_banned. Таймаут масштабируется: +# _cian_anchor_timeout = max( +# ANCHOR_TIMEOUT_SEC, +# _CIAN_ROOM_BUCKETS × pages_per_anchor × (request_delay_sec + _CIAN_PER_SERP_FETCH_S) +# + detail_top_n × _CIAN_PER_DETAIL_S +# + (_CIAN_HOUSES_BUDGET_S if enrich_houses else 0)) +# Пример (дефолты): 4×3×(5+70) + 10×50 + 180 = 900 + 500 + 180 = 1580s/anchor (worst-case, +# это watchdog от зависания, не ожидаемая длительность). +_CIAN_ROOM_BUCKETS: int = 4 # fetch_around_multi_room итерирует ((1,),(2,),(3,),(4,)), +# см. providers/cian/serp.py:215 +_CIAN_PER_SERP_FETCH_S: float = 70.0 # browser fetch через пул: page.goto 60s + camoufox relaunch +_CIAN_PER_DETAIL_S: float = 50.0 # секунд на один detail-fetch (как avito) +_CIAN_HOUSES_BUDGET_S: float = 180.0 # бюджет на house-enrich фазу (как avito) + + +def _cian_anchor_timeout_s( + pages_per_anchor: int, + request_delay_sec: float, + detail_top_n: int, + enrich_houses: bool, +) -> float: + """Масштабируемый watchdog-таймаут одного Cian anchor'а (секунды). + + Чистая функция для тестируемости формулы (#2160). Никогда не опускается ниже + ANCHOR_TIMEOUT_SEC (max-ветка): при малых параметрах фиксированный минимум остаётся. + """ + return max( + float(ANCHOR_TIMEOUT_SEC), + _CIAN_ROOM_BUCKETS * pages_per_anchor * (request_delay_sec + _CIAN_PER_SERP_FETCH_S) + + detail_top_n * _CIAN_PER_DETAIL_S + + (_CIAN_HOUSES_BUDGET_S if enrich_houses else 0.0), + ) + + # Default anchors ЕКБ — 5 точек покрытия города EKB_ANCHORS: list[tuple[float, float, str]] = [ (56.8400, 60.6050, "Центр"), @@ -2108,6 +2147,27 @@ async def run_cian_city_sweep( consecutive_failures = 0 cian_rotations_done = 0 # #1848: бюджет IP-ротаций на весь sweep + # #2160: масштабируемый watchdog-таймаут для cian anchor'а. При proxy-pool browser каждый + # SERP-фетч 13-45s (camoufox relaunch), фиксированный ANCHOR_TIMEOUT_SEC=240 гильотинит + # якорь в SERP-фазе. См. _cian_anchor_timeout_s + константы выше. + _cian_anchor_timeout = _cian_anchor_timeout_s( + pages_per_anchor, request_delay_sec, detail_top_n, enrich_houses + ) + logger.info( + "cian-sweep run_id=%d: anchor_timeout=%.0fs " + "(base=%d, serp=%d×%d×(%.0f+%.0f)s + detail_top_n=%d×%.0fs + houses=%.0fs)", + run_id, + _cian_anchor_timeout, + ANCHOR_TIMEOUT_SEC, + _CIAN_ROOM_BUCKETS, + pages_per_anchor, + request_delay_sec, + _CIAN_PER_SERP_FETCH_S, + detail_top_n, + _CIAN_PER_DETAIL_S, + _CIAN_HOUSES_BUDGET_S if enrich_houses else 0.0, + ) + try: for idx, (lat, lon, name) in enumerate(_anchors, start=1): # Cooperative cancel перед каждым anchor @@ -2381,7 +2441,7 @@ async def run_cian_city_sweep( ) try: - await asyncio.wait_for(_cian_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC) + await asyncio.wait_for(_cian_anchor_phases(), timeout=_cian_anchor_timeout) except TimeoutError: logger.warning( "cian-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " @@ -2392,7 +2452,7 @@ async def run_cian_city_sweep( name, lat, lon, - ANCHOR_TIMEOUT_SEC, + _cian_anchor_timeout, ) counters.errors_count += 1 consecutive_failures += 1 @@ -2413,7 +2473,7 @@ async def run_cian_city_sweep( db, run_id, f"cian sweep aborted: {consecutive_failures} consecutive " - f"anchor failures (last: anchor timeout {ANCHOR_TIMEOUT_SEC}s)", + f"anchor failures (last: anchor timeout {_cian_anchor_timeout:.0f}s)", counters.to_dict(), ) return counters diff --git a/tradein-mvp/backend/tests/test_cian_city_sweep.py b/tradein-mvp/backend/tests/test_cian_city_sweep.py index a82dcacd..aeba4bda 100644 --- a/tradein-mvp/backend/tests/test_cian_city_sweep.py +++ b/tradein-mvp/backend/tests/test_cian_city_sweep.py @@ -785,3 +785,59 @@ def test_cian_list_runs_endpoint(app_with_admin) -> None: assert body[0]["run_id"] == 10 assert body[0]["source"] == "cian_city_sweep" assert "2025-05-01" in body[0]["started_at"] + + +# ── #2160: масштабируемый per-anchor watchdog для proxy-pool browser ────────── + + +def test_cian_anchor_timeout_defaults() -> None: + """Формула watchdog даёт worst-case ~1580s при дефолтных параметрах. + + 4 room-buckets × 3 pages × (5s delay + 70s serp-fetch) + 10 detail × 50s + 180s houses + = 900 + 500 + 180 = 1580s. Фиксированный ANCHOR_TIMEOUT_SEC=240 гильотинил бы якорь в + SERP-фазе (#2160): каждый SERP-фетч через camoufox 13-45s. + """ + t = scrape_pipeline._cian_anchor_timeout_s( + pages_per_anchor=3, + request_delay_sec=5.0, + detail_top_n=10, + enrich_houses=True, + ) + assert t == pytest.approx(1580.0) + assert t > scrape_pipeline.ANCHOR_TIMEOUT_SEC + + +def test_cian_anchor_timeout_no_houses() -> None: + """enrich_houses=False убирает houses-бюджет (180s) из формулы.""" + t = scrape_pipeline._cian_anchor_timeout_s( + pages_per_anchor=3, + request_delay_sec=5.0, + detail_top_n=10, + enrich_houses=False, + ) + assert t == pytest.approx(1400.0) # 900 + 500 + 0 + + +def test_cian_anchor_timeout_floor_at_anchor_timeout_sec() -> None: + """Малые параметры не опускают таймаут ниже ANCHOR_TIMEOUT_SEC (max-ветка).""" + t = scrape_pipeline._cian_anchor_timeout_s( + pages_per_anchor=0, + request_delay_sec=0.0, + detail_top_n=0, + enrich_houses=False, + ) + assert t == float(scrape_pipeline.ANCHOR_TIMEOUT_SEC) + + +def test_cian_anchor_timeout_parity_with_scraper_kit() -> None: + """Strangler-инвариант: backend-хелпер ≡ scraper_kit-хелпер (обе копии в синхроне).""" + from scraper_kit.orchestration.pipeline import ( + _cian_anchor_timeout_s as kit_timeout_s, + ) + + for args in ( + dict(pages_per_anchor=3, request_delay_sec=5.0, detail_top_n=10, enrich_houses=True), + dict(pages_per_anchor=5, request_delay_sec=9.0, detail_top_n=20, enrich_houses=False), + dict(pages_per_anchor=0, request_delay_sec=0.0, detail_top_n=0, enrich_houses=False), + ): + assert scrape_pipeline._cian_anchor_timeout_s(**args) == kit_timeout_s(**args) diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py index 30784467..e766c5aa 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py @@ -235,6 +235,44 @@ _AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT: int = 5 # прерывают enrichment-фазу. Listings из SEARCH+SAVE сохраняются всегда. _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: int = 3 +# #2160: константы для расчёта watchdog-таймаута Cian city sweep. При +# USE_PROXY_POOL_BROWSER=true каждый SERP-фетч идёт через camoufox с relaunch при смене +# прокси (page.goto timeout 60s + overhead) = 13-45s/страница, а якорь cian = 4 room-buckets +# × pages_per_anchor SERP-фетчей + detail_top_n detail-фетчей + houses-enrich. Фиксированный +# ANCHOR_TIMEOUT_SEC=240 гильотинит якорь ещё в SERP-фазе → save_listings не успевает, +# lots_fetched=0, после N подряд → mark_banned. Таймаут масштабируется: +# _cian_anchor_timeout = max( +# ANCHOR_TIMEOUT_SEC, +# _CIAN_ROOM_BUCKETS × pages_per_anchor × (request_delay_sec + _CIAN_PER_SERP_FETCH_S) +# + detail_top_n × _CIAN_PER_DETAIL_S +# + (_CIAN_HOUSES_BUDGET_S if enrich_houses else 0)) +# Пример (дефолты): 4×3×(5+70) + 10×50 + 180 = 900 + 500 + 180 = 1580s/anchor (worst-case, +# это watchdog от зависания, не ожидаемая длительность). +_CIAN_ROOM_BUCKETS: int = 4 # fetch_around_multi_room итерирует ((1,),(2,),(3,),(4,)), +# см. providers/cian/serp.py:215 +_CIAN_PER_SERP_FETCH_S: float = 70.0 # browser fetch через пул: page.goto 60s + camoufox relaunch +_CIAN_PER_DETAIL_S: float = 50.0 # секунд на один detail-fetch (как avito) +_CIAN_HOUSES_BUDGET_S: float = 180.0 # бюджет на house-enrich фазу (как avito) + + +def _cian_anchor_timeout_s( + pages_per_anchor: int, + request_delay_sec: float, + detail_top_n: int, + enrich_houses: bool, +) -> float: + """Масштабируемый watchdog-таймаут одного Cian anchor'а (секунды). + + Чистая функция для тестируемости формулы (#2160). Никогда не опускается ниже + ANCHOR_TIMEOUT_SEC (max-ветка): при малых параметрах фиксированный минимум остаётся. + """ + return max( + float(ANCHOR_TIMEOUT_SEC), + _CIAN_ROOM_BUCKETS * pages_per_anchor * (request_delay_sec + _CIAN_PER_SERP_FETCH_S) + + detail_top_n * _CIAN_PER_DETAIL_S + + (_CIAN_HOUSES_BUDGET_S if enrich_houses else 0.0), + ) + # Default anchors ЕКБ — 5 точек покрытия города EKB_ANCHORS: list[tuple[float, float, str]] = [ (56.8400, 60.6050, "Центр"), @@ -2168,6 +2206,27 @@ async def run_cian_city_sweep( consecutive_failures = 0 cian_rotations_done = 0 # #1848: бюджет IP-ротаций на весь sweep + # #2160: масштабируемый watchdog-таймаут для cian anchor'а. При proxy-pool browser каждый + # SERP-фетч 13-45s (camoufox relaunch), фиксированный ANCHOR_TIMEOUT_SEC=240 гильотинит + # якорь в SERP-фазе. См. _cian_anchor_timeout_s + константы выше. + _cian_anchor_timeout = _cian_anchor_timeout_s( + pages_per_anchor, request_delay_sec, detail_top_n, enrich_houses + ) + logger.info( + "cian-sweep run_id=%d: anchor_timeout=%.0fs " + "(base=%d, serp=%d×%d×(%.0f+%.0f)s + detail_top_n=%d×%.0fs + houses=%.0fs)", + run_id, + _cian_anchor_timeout, + ANCHOR_TIMEOUT_SEC, + _CIAN_ROOM_BUCKETS, + pages_per_anchor, + request_delay_sec, + _CIAN_PER_SERP_FETCH_S, + detail_top_n, + _CIAN_PER_DETAIL_S, + _CIAN_HOUSES_BUDGET_S if enrich_houses else 0.0, + ) + try: for idx, (lat, lon, name) in enumerate(_anchors, start=1): # Cooperative cancel перед каждым anchor @@ -2435,7 +2494,7 @@ async def run_cian_city_sweep( ) try: - await asyncio.wait_for(_cian_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC) + await asyncio.wait_for(_cian_anchor_phases(), timeout=_cian_anchor_timeout) except TimeoutError: logger.warning( "cian-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " @@ -2446,7 +2505,7 @@ async def run_cian_city_sweep( name, lat, lon, - ANCHOR_TIMEOUT_SEC, + _cian_anchor_timeout, ) counters.errors_count += 1 consecutive_failures += 1 @@ -2467,7 +2526,7 @@ async def run_cian_city_sweep( db, run_id, f"cian sweep aborted: {consecutive_failures} consecutive " - f"anchor failures (last: anchor timeout {ANCHOR_TIMEOUT_SEC}s)", + f"anchor failures (last: anchor timeout {_cian_anchor_timeout:.0f}s)", counters.to_dict(), ) return counters