fix(tradein/scrapers): cian_city_sweep честный статус при defensive abort (#1949)

Два из трёх root cause в #1949 (cian_full_load per-fetch zombie + city_sweep
anchor-timeout scaling) уже были закрыты ранее (#1952, #2272). Оставшийся
gap: mark_banned красил весь run даже когда SERP-фаза уже сохранила лоты —
статус должен быть done, не banned, когда основной сбор состоялся. Зеркалит
уже существующий avito_serp_ok_not_banned (#1950) под отдельным флагом
cian_sweep_ok_not_banned (не смешиваем cian/avito семантику).

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)."""

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())

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