diff --git a/tradein-mvp/backend/app/services/scrapers/avito_detail.py b/tradein-mvp/backend/app/services/scrapers/avito_detail.py index 65055665..f9a8d451 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito_detail.py +++ b/tradein-mvp/backend/app/services/scrapers/avito_detail.py @@ -303,8 +303,8 @@ async def warm_up_session(session: AsyncSession) -> bool: """ try: await session.get(_AVITO_WARM_YANDEX_URL) # органический referer-источник, best-effort - except Exception: - pass + except Exception as exc: + logger.debug("avito warm-up: yandex GET failed (non-fatal): %s", exc) await asyncio.sleep(random.uniform(2.0, 3.5)) try: r = await session.get( diff --git a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py index 1c7359a5..f736856b 100644 --- a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py +++ b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py @@ -99,6 +99,7 @@ async def run_avito_detail_backfill( warm_batch = int(params.get("warm_batch", 500)) research_every = int(params.get("research_every", 50)) block_cooldown_sec = float(params.get("block_cooldown_sec", 30.0)) + warm_timeout_s = float(params.get("warm_timeout_s", 90.0)) # #1950: hard-timeout'ы на блокирующие await'ы внутри loop'а. Без них один # зависший fetch_detail/rotate ронял весь run в zombie (run 423 завис 7.7ч). @@ -264,20 +265,43 @@ async def run_avito_detail_backfill( # (re-GET search той же сессией, освежить куки). On-block — cooldown ниже. if use_curl and session is not None: if warm_batch and items_since_warm >= warm_batch: - # периодический полный re-warm на том же sticky exit-IP (свежий TLS+куки) + # периодический полный re-warm на том же sticky exit-IP (свежий TLS+куки). + # #1950: bound сетевой re-warm; строим НОВУЮ сессию до close старой — на + # timeout/fail сохраняем текущую рабочую сессию (graceful degrade). try: - await session.close() + new_session = await asyncio.wait_for( + build_warmed_session(), timeout=warm_timeout_s + ) + try: + await session.close() + except Exception: + pass + session = new_session except Exception: - pass - session = await build_warmed_session() + logger.warning( + "avito_detail_backfill: run_id=%d re-warm timed out/failed (>%.0fs) " + "-- keeping current session", + run_id, + warm_timeout_s, + exc_info=True, + ) items_since_warm = 0 elif ( research_every and items_since_warm > 0 and items_since_warm % research_every == 0 ): - # лёгкий in-session перепоиск: тот же exit-IP, освежить куки - await research_in_session(session) + # лёгкий in-session перепоиск: тот же exit-IP, освежить куки. + # #1950: bound сетевой re-search. + try: + await asyncio.wait_for(research_in_session(session), timeout=warm_timeout_s) + except Exception: + logger.warning( + "avito_detail_backfill: run_id=%d in-session re-search " + "timed out/failed", + run_id, + exc_info=True, + ) try: # #1950: hard-timeout — зависший fetch (browser hang / curl-stall) не @@ -344,6 +368,18 @@ async def run_avito_detail_backfill( consecutive_blocks, e, ) + # abort FIRST — на последнем блоке не тратим cooldown+research/rotate + # (consecutive_blocks уже инкрементнут выше). + if consecutive_blocks >= max_consecutive_blocks: + logger.error( + "avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, " + "IP rate-limited. enriched=%d attempted=%d", + run_id, + consecutive_blocks, + counters.enriched, + counters.attempted, + ) + break # МГТС sticky-IP: один фикс. exit-IP, per-connection ротации нет (проверено: # 6/6 свежих сессий = тот же IP 109.252.125.80; ротация только вручную # кнопкой). Уйти на свежий IP софтом нельзя → блок = rate-limit текущего IP: @@ -353,7 +389,18 @@ async def run_avito_detail_backfill( if use_curl: await asyncio.sleep(block_cooldown_sec) if session is not None: - await research_in_session(session) + # #1950: bound сетевой re-search. + try: + await asyncio.wait_for( + research_in_session(session), timeout=warm_timeout_s + ) + except Exception: + logger.warning( + "avito_detail_backfill: run_id=%d on-block re-search " + "timed out/failed", + run_id, + exc_info=True, + ) else: # #1950: bound rotate — _rotate_ip ждёт settle + до 3 changeip-попыток; # на блокирующем changeip (зависшее соединение) без timeout loop виснет. @@ -367,16 +414,6 @@ async def run_avito_detail_backfill( rotate_timeout_s, exc_info=True, ) - if consecutive_blocks >= max_consecutive_blocks: - logger.error( - "avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, " - "IP rate-limited. enriched=%d attempted=%d", - run_id, - consecutive_blocks, - counters.enriched, - counters.attempted, - ) - break except TimeoutError: # asyncio.wait_for → TimeoutError (py3.12: asyncio.TimeoutError — alias). diff --git a/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py b/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py index 332265fd..394e570d 100644 --- a/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py +++ b/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py @@ -140,7 +140,11 @@ async def test_backfill_processes_snapshot_to_completion() -> None: @pytest.mark.asyncio async def test_backfill_blocked_abort_after_max_consecutive() -> None: - """5 consecutive AvitoBlockedError -> abort, mark_done (NOT mark_failed), rotate_ip x5.""" + """5 consecutive AvitoBlockedError -> abort, mark_done (NOT mark_failed). + + #1950 abort-reorder: abort-check ПЕРЕД recovery → на 5-м (аборт-)блоке rotate_ip + НЕ дёргается (не тратим recovery на финальном блоке). rotate_ip x4 (блоки 1-4). + """ from app.services.scrapers.avito_exceptions import AvitoBlockedError snapshot = _make_snapshot(10) @@ -170,7 +174,8 @@ async def test_backfill_blocked_abort_after_max_consecutive() -> None: assert result.enriched == 0 runs.mark_done.assert_called_once() runs.mark_failed.assert_not_called() - assert mock_scraper.return_value._rotate_ip.call_count == 5 + # abort-check до recovery → 5-й блок абортит без rotate; rotate только на блоках 1-4. + assert mock_scraper.return_value._rotate_ip.call_count == 4 @pytest.mark.asyncio