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

Merged
lekss361 merged 1 commit from fix/1949-cian-full-load-zombie-watchdog into main 2026-07-03 22:45:48 +00:00
3 changed files with 122 additions and 11 deletions

View file

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

View file

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

View file

@ -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