diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index 88a2fbf2..c08cf6b0 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -513,6 +513,13 @@ class Settings(BaseSettings): # ENV: CIAN_FULL_LOAD_PER_FETCH_TIMEOUT_S. cian_full_load_per_fetch_timeout_s: float = 90.0 + # #1949: если run_cian_city_sweep уже сохранил лоты (ins+upd > 0) и abort случился + # из-за anchor-timeout/consecutive SERP failures ПОСЛЕ этого, ставим 'done' а не + # 'banned' — основной сбор состоялся, 'banned' лишний шум в мониторинге. + # Зеркалит avito_serp_ok_not_banned (#1950), но отдельный флаг — не смешиваем + # cian/avito семантику. False = старое поведение. ENV: CIAN_SWEEP_OK_NOT_BANNED. + cian_sweep_ok_not_banned: bool = True + @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 e3312b69..a46ed941 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -2315,6 +2315,38 @@ class CianCitySweepCounters: CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES = 3 +def _mark_cian_sweep_abort( + db: Session, + run_id: int, + counters: CianCitySweepCounters, + reason: str, +) -> None: + """#1949: честный финальный статус при defensive abort run_cian_city_sweep. + + Зеркалит avito_serp_ok_not_banned (#1950): если SERP-фаза уже сохранила лоты + (lots_inserted + lots_updated > 0) до момента abort'а, основной сбор состоялся — + 'banned' лишний шум в мониторинге. Ставим 'done' с заметкой о причине abort'а. + Если лотов нет (SERP сам заблокирован/завис на первом же anchor'е) — 'banned' + как раньше. За флагом settings.cian_sweep_ok_not_banned (default True). + """ + _serp_ok = (counters.lots_inserted + counters.lots_updated) > 0 + if settings.cian_sweep_ok_not_banned and _serp_ok: + logger.warning( + "cian-sweep run_id=%d: SERP OK (ins=%d upd=%d) before abort — " "DONE not banned. %s", + run_id, + counters.lots_inserted, + counters.lots_updated, + reason, + ) + scrape_runs.mark_done( + db, + run_id, + {**counters.to_dict(), "abort_note": reason}, # type: ignore[arg-type] + ) + else: + scrape_runs.mark_banned(db, run_id, reason, counters.to_dict()) + + async def run_cian_city_sweep( db: Session, *, @@ -2351,7 +2383,8 @@ async def run_cian_city_sweep( Anti-bot: - CianScraper управляет своей curl_cffi-сессией (прокси уже внутри). - asyncio.sleep(request_delay_sec) между detail-запросами (jitter ±20%). - - Defensive abort при CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES SERP-ошибок подряд. + - Defensive abort при CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES SERP-ошибок подряд → + _mark_cian_sweep_abort (#1949): 'done' если SERP уже сохранил лоты, иначе 'banned'. Cooperative cancel: scrape_runs.is_cancelled(db, run_id) перед каждым anchor. mark_done вызывается ВСЕГДА (finally outer). @@ -2690,13 +2723,11 @@ async def run_cian_city_sweep( ) counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) - scrape_runs.mark_banned( - db, - run_id, + _abort_reason = ( f"cian sweep aborted: {consecutive_failures} consecutive " - f"anchor failures (last: anchor timeout {_cian_anchor_timeout:.0f}s)", - counters.to_dict(), + f"anchor failures (last: anchor timeout {_cian_anchor_timeout:.0f}s)" ) + _mark_cian_sweep_abort(db, run_id, counters, _abort_reason) return counters counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) @@ -2718,13 +2749,11 @@ async def run_cian_city_sweep( ) counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) - scrape_runs.mark_banned( - db, - run_id, + _abort_reason = ( f"cian sweep aborted: {consecutive_failures} consecutive " - f"anchor SERP failures (last: {e})", - counters.to_dict(), + f"anchor SERP failures (last: {e})" ) + _mark_cian_sweep_abort(db, run_id, counters, _abort_reason) return counters counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) 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 e659e375..ff96d687 100644 --- a/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py +++ b/tradein-mvp/backend/tests/test_scraper_resilience_1949_1950.py @@ -12,6 +12,11 @@ Fix C (#1949): run_cian_full_load применяет monkey-patch _browser.fetch Fix D (#1949): run_cian_full_load создаёт фоновый heartbeat-task и отменяет его в finally при любом завершении. + +Fix E (#1949): run_cian_city_sweep при defensive abort (consecutive anchor + timeout/failure) вызывает mark_done (не mark_banned) когда + SERP уже сохранил лоты (lots_inserted+lots_updated>0) и + cian_sweep_ok_not_banned=True. """ from __future__ import annotations @@ -556,3 +561,73 @@ async def test_cian_full_load_heartbeat_cancelled_on_exception() -> None: task.cancelled() or task.done() ), f"Heartbeat task должен быть отменён в finally даже при RuntimeError: {task}" assert mock_runs.mark_failed.called, "mark_failed должен быть вызван при RuntimeError" + + +# ────────────────────────────────────────────────────────────────────────────── +# Fix E: run_cian_city_sweep честный статус при defensive abort (#1949) +# ────────────────────────────────────────────────────────────────────────────── + + +def test_mark_cian_sweep_abort_lots_present_marks_done() -> None: + """#1949-E: SERP уже сохранил лоты (ins+upd>0) до abort → mark_done, не mark_banned.""" + from app.services.scrape_pipeline import CianCitySweepCounters, _mark_cian_sweep_abort + + mock_db = MagicMock() + mock_runs = MagicMock() + counters = CianCitySweepCounters( + anchors_total=5, anchors_done=2, lots_inserted=12, lots_updated=3 + ) + + with ( + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_sweep_ok_not_banned = True + _mark_cian_sweep_abort(mock_db, 42, counters, "cian sweep aborted: 3 consecutive timeouts") + + assert mock_runs.mark_done.called, "mark_done должен быть вызван: SERP OK (ins=12 upd=3)" + assert not mock_runs.mark_banned.called, "mark_banned НЕ должен вызываться когда лоты сохранены" + passed_counters: dict = mock_runs.mark_done.call_args[0][2] + assert "abort_note" in passed_counters, "abort_note должен присутствовать в counters" + + +def test_mark_cian_sweep_abort_no_lots_marks_banned() -> None: + """#1949-E: лотов ещё не было (ins=upd=0) на момент abort → mark_banned как раньше.""" + from app.services.scrape_pipeline import CianCitySweepCounters, _mark_cian_sweep_abort + + mock_db = MagicMock() + mock_runs = MagicMock() + counters = CianCitySweepCounters( + anchors_total=5, anchors_done=1, lots_inserted=0, lots_updated=0 + ) + + with ( + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_sweep_ok_not_banned = True + _mark_cian_sweep_abort(mock_db, 43, counters, "cian sweep aborted: SERP blocked at anchor") + + assert mock_runs.mark_banned.called, "mark_banned должен быть вызван: SERP не собрал лотов" + assert not mock_runs.mark_done.called, "mark_done НЕ должен вызываться при lots ins=upd=0" + + +def test_mark_cian_sweep_abort_flag_off_still_marks_banned() -> None: + """#1949-E: cian_sweep_ok_not_banned=False → mark_banned даже при lots>0 (backward-compat).""" + from app.services.scrape_pipeline import CianCitySweepCounters, _mark_cian_sweep_abort + + mock_db = MagicMock() + mock_runs = MagicMock() + counters = CianCitySweepCounters( + anchors_total=5, anchors_done=3, lots_inserted=20, lots_updated=5 + ) + + with ( + patch("app.services.scrape_pipeline.scrape_runs", mock_runs), + patch("app.services.scrape_pipeline.settings") as s, + ): + s.cian_sweep_ok_not_banned = False + _mark_cian_sweep_abort(mock_db, 44, counters, "cian sweep aborted: 3 consecutive timeouts") + + assert mock_runs.mark_banned.called, "flag=False → mark_banned (backward-compat)" + assert not mock_runs.mark_done.called