fix(avito-sweep): scale per-anchor timeout by detail budget + preserve SERP counters before detail phase
All checks were successful
CI / changes (push) Successful in 7s
CI / backend-tests (push) Has been skipped
CI / frontend-tests (push) Has been skipped
CI / openapi-codegen-check (push) Has been skipped
CI / changes (pull_request) Successful in 6s
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped

- Compute _avito_anchor_timeout = ANCHOR_TIMEOUT_SEC + detail_top_n * 50s + 180s (houses).
  Eliminates false TimeoutError on anchors with many detail requests.
- Refactor run_avito_city_sweep to use inner _avito_anchor_phases coroutine (mirrors
  Yandex/Cian pattern): counters updated immediately after SERP+save (nonlocal), before
  detail phase. TimeoutError in detail no longer erases lots_fetched/inserted.
- Add consecutive-detail-timeout abort after 5 consecutive TimeoutErrors.
- Update test_anchor_watchdog: new tests for timeout scaling formula and counter durability.
- Update test_avito_anti_bot + test_sweep_imv_phase: patch AvitoScraper.fetch_around
  directly (run_avito_city_sweep no longer calls run_avito_pipeline); use ExitStack
  instead of starred-expression unpacking (invalid syntax in Python 3.12/3.14).
This commit is contained in:
bot-backend 2026-06-17 09:02:11 +03:00
parent 4811920d28
commit 40c044e2a6
4 changed files with 776 additions and 202 deletions

View file

@ -65,6 +65,25 @@ ANCHOR_TIMEOUT_SEC: int = 240
_YANDEX_COMBOS_PER_FETCH_S: float = 12.0 # network + parse margin (секунды на страницу)
_YANDEX_ADDRESS_ENRICH_BUDGET_S: float = 300.0 # бюджет на address-enrich фазу
# Константы для расчёта watchdog-таймаута Avito city sweep.
# Avito detail-fetch идёт через браузерный сервис с page.goto timeout 60s, и
# anti-bot страницы нередко доходят до этого лимита. Фиксированный ANCHOR_TIMEOUT_SEC=240
# guillotine'ит anchor ещё в detail-фазе → SERP-счётчики теряются, run показывает
# lots_fetched=0 при реально сохранённых листингах. Таймаут масштабируется:
# _avito_anchor_timeout = ANCHOR_TIMEOUT_SEC
# + detail_top_n × _AVITO_PER_DETAIL_S
# + (_AVITO_HOUSES_BUDGET_S if enrich_houses else 0)
# Пример: detail_top_n=20 + houses → 240 + 1000 + 180 = 1420s/anchor (worst-case).
# 5 anchor'ов × 1420s ≈ 2ч worst-case (нормальный прогон быстрее: не каждый detail
# достигает 60s timeout). Окно лишь ограничивает старт следующего sweep'а — scheduler
# ждёт финиша задачи и не запустит параллельный run.
_AVITO_PER_DETAIL_S: float = 50.0 # секунд на один detail-fetch (incl. occasional 60s timeout)
_AVITO_HOUSES_BUDGET_S: float = 180.0 # бюджет на house-enrich фазу
# Fail-fast: если detail phase накапливает подряд N timeout/ошибок, прерываем её
# (аналог consecutive_blocks для AvitoBlockedError в step 5). Срабатывает только при
# явной деградации browser-сервиса, не мешает нормальным transient-ошибкам.
_AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT: int = 5
# Default anchors ЕКБ — 5 точек покрытия города
EKB_ANCHORS: list[tuple[float, float, str]] = [
(56.8400, 60.6050, "Центр"),
@ -485,6 +504,10 @@ async def run_avito_city_sweep(
- Прочие errors per-anchor логируются, не валят весь sweep
- После всех anchor'ов: если enrich_imv=True — IMV-оценка тронутых домов
(cooperative cancel + per-house graceful error handling)
SERP-счётчики (lots_fetched/inserted/updated) записываются в sweep-level counters
НЕМЕДЛЕННО после save_listings до начала detail/houses фазы. Даже если anchor
превышает watchdog-таймаут в detail-фазе, SERP-результаты не теряются.
"""
from app.services.house_imv_backfill import process_houses_imv_batch
@ -492,6 +515,27 @@ async def run_avito_city_sweep(
counters = CitySweepCounters(anchors_total=len(_anchors))
all_touched_house_ids: set[int] = set()
# Масштабируемый watchdog-таймаут для Avito anchor'а. В отличие от Yandex/Cian,
# Avito detail-fetch идёт через browser service с page.goto timeout 60s. При
# detail_top_n=20 detail-фаза может занять до N×60s — фиксированный ANCHOR_TIMEOUT_SEC
# обрезает её раньше SERP, теряя счётчики. Масштабирование по detail_top_n + houses
# гарантирует, что SERP+save всегда успевает до watchdog.
_avito_anchor_timeout = (
float(ANCHOR_TIMEOUT_SEC)
+ detail_top_n * _AVITO_PER_DETAIL_S
+ (_AVITO_HOUSES_BUDGET_S if enrich_houses else 0.0)
)
logger.info(
"city-sweep run_id=%d: avito anchor_timeout=%.0fs "
"(base=%d + detail_top_n=%d×%.0fs + houses=%.0fs)",
run_id,
_avito_anchor_timeout,
ANCHOR_TIMEOUT_SEC,
detail_top_n,
_AVITO_PER_DETAIL_S,
_AVITO_HOUSES_BUDGET_S if enrich_houses else 0.0,
)
browser_mode = settings.scraper_fetch_mode == "browser"
async with AsyncExitStack() as stack:
session: AsyncSession | None = None
@ -529,44 +573,313 @@ async def run_avito_city_sweep(
lat,
lon,
)
try:
result = await asyncio.wait_for(
run_avito_pipeline(
db,
lat=lat,
lon=lon,
radius_m=radius_m,
enrich_houses=enrich_houses,
enrich_detail_top_n=detail_top_n,
# Capture loop variables in default args (B023): prevents stale binding
# if the coroutine is scheduled after the loop variable changes.
_a_lat, _a_lon, _a_name = lat, lon, name
async def _avito_anchor_phases(
_lat: float = _a_lat,
_lon: float = _a_lon,
_name: str = _a_name,
) -> None:
"""Все фазы одного Avito anchor'а (SERP+save → houses → detail).
Обновляет sweep-level counters и all_touched_house_ids через nonlocal
НЕМЕДЛЕННО после SERP+save (до detail-фазы). Это гарантирует, что
TimeoutError из wait_for теряет только detail-счётчики, но не SERP.
"""
nonlocal all_touched_house_ids
# ── Phase 1+2: SERP + save ─────────────────────────────────
scraper = AvitoScraper()
if browser_mode:
scraper._browser = shared_bf
else:
scraper._cffi = session
try:
anchor_lots: list[ScrapedLot] = await scraper.fetch_around(
_lat,
_lon,
radius_m,
pages=pages_per_anchor,
request_delay_sec=request_delay_sec,
shared_session=session,
shared_browser=shared_bf,
),
timeout=ANCHOR_TIMEOUT_SEC,
delay_override_sec=request_delay_sec,
)
except (AvitoBlockedError, AvitoRateLimitedError):
logger.error(
"city-sweep run_id=%d: SERP BLOCKED at anchor (%s, %.4f, %.4f)",
run_id,
_name,
_lat,
_lon,
)
raise
counters.lots_fetched += len(anchor_lots)
if anchor_lots:
try:
ins, upd = save_listings(db, anchor_lots)
counters.lots_inserted += ins
counters.lots_updated += upd
except Exception as save_exc:
logger.exception(
"city-sweep run_id=%d: save_listings failed at anchor %s: %s",
run_id,
_name,
save_exc,
)
counters.errors_count += 1
logger.info(
"city-sweep run_id=%d anchor %s: SERP fetched=%d ins=%d upd=%d",
run_id,
_name,
len(anchor_lots),
counters.lots_inserted,
counters.lots_updated,
)
counters.lots_fetched += result.counters.lots_fetched
counters.lots_inserted += result.counters.lots_inserted
counters.lots_updated += result.counters.lots_updated
counters.unique_houses += result.counters.unique_houses
counters.houses_enriched += result.counters.houses_enriched
counters.houses_failed += result.counters.houses_failed
counters.detail_attempted += result.counters.detail_attempted
counters.detail_enriched += result.counters.detail_enriched
counters.detail_failed += result.counters.detail_failed
counters.errors_count += len(result.counters.errors)
all_touched_house_ids.update(result.touched_house_ids)
# SERP-счётчики зафиксированы в sweep-level counters.
# Дальнейший TimeoutError из wait_for их не сотрёт.
# ── Phase 3: group by house ────────────────────────────────
unique_house_paths: set[str] = set()
for lot in anchor_lots:
if lot.house_url:
try:
parsed = urlparse(lot.house_url)
path = parsed.path if parsed.path else lot.house_url
unique_house_paths.add(path)
except Exception:
continue
if unique_house_paths:
counters.unique_houses += len(unique_house_paths)
# ── Phase 4: enrich houses ─────────────────────────────────
touched_house_ids: set[int] = set()
if enrich_houses and unique_house_paths:
house_paths_list = list(unique_house_paths)
for h_idx, house_path in enumerate(house_paths_list):
try:
enrichment = await fetch_house_catalog(
house_path,
cffi_session=session,
browser_fetcher=shared_bf,
)
hc = save_house_catalog_enrichment(db, enrichment)
counters.houses_enriched += 1
hid = hc.get("house_id")
if hid:
touched_house_ids.add(int(hid))
except (AvitoBlockedError, AvitoRateLimitedError):
logger.error(
"city-sweep run_id=%d: house BLOCKED at %s — propagating",
run_id,
house_path,
)
raise
except Exception as he:
logger.warning(
"city-sweep run_id=%d: house_enrich failed %s: %s",
run_id,
house_path,
he,
)
counters.houses_failed += 1
counters.errors_count += 1
try:
db.rollback()
except Exception:
pass
if h_idx < len(house_paths_list) - 1:
await asyncio.sleep(request_delay_sec)
# ── Phase 5: enrich detail (top-N) ────────────────────────
if detail_top_n > 0 and anchor_lots:
priority_rows = (
db.execute(
text("""
SELECT source_url
FROM listings
WHERE source = 'avito'
AND source_url IS NOT NULL
AND (
(
detail_enriched_at IS NULL
AND price_rub > 0
AND ST_DWithin(
geom::geography,
ST_MakePoint(:lon, :lat)::geography,
:radius
)
)
OR (
detail_enriched_at IS NULL
AND scraped_at > NOW() - INTERVAL '2 hours'
)
)
ORDER BY scraped_at DESC NULLS LAST
LIMIT :limit
"""),
{
"lat": _lat,
"lon": _lon,
"radius": radius_m * 2,
"limit": detail_top_n,
},
)
.mappings()
.all()
)
consecutive_blocks = 0
consecutive_timeouts = 0
detail_house_paths: set[str] = set()
for d_idx, row in enumerate(priority_rows):
source_url: str = row["source_url"]
counters.detail_attempted += 1
try:
item_url = (
urlparse(source_url).path
if source_url.startswith("http")
else source_url
)
enrichment_detail = await fetch_detail(
item_url,
cffi_session=session,
browser_fetcher=shared_bf,
)
if save_detail_enrichment(db, enrichment_detail):
counters.detail_enriched += 1
if enrichment_detail.house_catalog_url:
hp = urlparse(enrichment_detail.house_catalog_url).path
if hp:
detail_house_paths.add(hp)
consecutive_blocks = 0
consecutive_timeouts = 0
except (AvitoBlockedError, AvitoRateLimitedError) as be:
consecutive_blocks += 1
counters.detail_failed += 1
counters.errors_count += 1
logger.warning(
"city-sweep run_id=%d: detail BLOCKED #%d/%d "
"(consecutive=%d): %s",
run_id,
d_idx + 1,
len(priority_rows),
consecutive_blocks,
be,
)
if consecutive_blocks >= 3:
logger.error(
"city-sweep run_id=%d: detail ABORT — "
"%d consecutive blocks (IP rate-limited)",
run_id,
consecutive_blocks,
)
raise
except Exception as de:
counters.detail_failed += 1
counters.errors_count += 1
consecutive_timeouts += 1
logger.warning(
"city-sweep run_id=%d: detail failed #%d/%d "
"(consecutive_timeouts=%d): %s",
run_id,
d_idx + 1,
len(priority_rows),
consecutive_timeouts,
de,
)
# (C) Fail-fast: браузер явно деградирует — прерываем
# detail-фазу, не тратим время на оставшиеся запросы.
if consecutive_timeouts >= _AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT:
logger.error(
"city-sweep run_id=%d: detail ABORT — "
"%d consecutive timeouts/errors, browser degraded",
run_id,
consecutive_timeouts,
)
break
try:
db.rollback()
except Exception:
pass
if d_idx < len(priority_rows) - 1:
jitter = random.uniform(0.8, 1.2)
await asyncio.sleep(request_delay_sec * jitter)
# ── Phase 5b: houses найденные через detail-страницы ──
new_house_paths = detail_house_paths - unique_house_paths
if enrich_houses and new_house_paths:
counters.unique_houses += len(new_house_paths)
nh_list = list(new_house_paths)
for nh_idx, house_path in enumerate(nh_list):
try:
enrichment = await fetch_house_catalog(
house_path,
cffi_session=session,
browser_fetcher=shared_bf,
)
hc = save_house_catalog_enrichment(db, enrichment)
counters.houses_enriched += 1
hid = hc.get("house_id")
if hid:
touched_house_ids.add(int(hid))
except (AvitoBlockedError, AvitoRateLimitedError):
logger.error(
"city-sweep run_id=%d: house(detail) BLOCKED %s"
" — propagating",
run_id,
house_path,
)
raise
except Exception as he:
logger.warning(
"city-sweep run_id=%d: house(detail) failed %s: %s",
run_id,
house_path,
he,
)
counters.houses_failed += 1
counters.errors_count += 1
try:
db.rollback()
except Exception:
pass
if nh_idx < len(nh_list) - 1:
await asyncio.sleep(request_delay_sec)
all_touched_house_ids.update(touched_house_ids)
logger.info(
"city-sweep run_id=%d anchor %s done: "
"lots=%d houses=%d/%d detail=%d/%d touched=%d errors=%d",
run_id,
_name,
len(anchor_lots),
counters.houses_enriched,
counters.unique_houses,
counters.detail_enriched,
counters.detail_attempted,
len(touched_house_ids),
counters.errors_count,
)
try:
await asyncio.wait_for(_avito_anchor_phases(), timeout=_avito_anchor_timeout)
except TimeoutError:
logger.warning(
"city-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) "
"timed out after %ds — skipping",
"timed out after %.0fs — SERP counters preserved, "
"detail phase incomplete",
run_id,
idx,
len(_anchors),
name,
lat,
lon,
ANCHOR_TIMEOUT_SEC,
_avito_anchor_timeout,
)
counters.errors_count += 1
except (AvitoBlockedError, AvitoRateLimitedError) as e:

View file

@ -274,8 +274,12 @@ async def test_city_sweep_marks_banned_on_block() -> None:
"""City sweep mark'ит run как banned при AvitoBlockedError.
Migration 015 задокументировал 'banned' как 'Avito вернул 403/captcha' именно этот сценарий.
Патчим AvitoScraper.fetch_around (не run_avito_pipeline он больше не вызывается
из run_avito_city_sweep; sweep использует внутренние _avito_anchor_phases).
"""
from app.services.scrape_pipeline import run_avito_city_sweep
from app.services.scrapers.avito import AvitoScraper
mock_db = MagicMock()
mock_runs = MagicMock()
@ -288,10 +292,7 @@ async def test_city_sweep_marks_banned_on_block() -> None:
with (
patch("app.services.scrape_pipeline.scrape_runs", mock_runs),
patch(
"app.services.scrape_pipeline.run_avito_pipeline",
AsyncMock(side_effect=block_error),
),
patch.object(AvitoScraper, "fetch_around", AsyncMock(side_effect=block_error)),
patch(
"curl_cffi.requests.AsyncSession",
return_value=AsyncMock(

View file

@ -140,11 +140,15 @@ def test_anchor_timeout_sec_exported() -> None:
async def test_avito_sweep_anchor_timeout_continues(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Первый anchor виснет → таймаут; второй anchor успешен; mark_done вызывается."""
call_count = 0
"""Первый anchor виснет → таймаут; второй anchor успешен; mark_done вызывается.
run_avito_city_sweep использует внутреннюю _avito_anchor_phases (не run_avito_pipeline),
поэтому патчим AvitoScraper.fetch_around + save_listings.
"""
from app.services.scrapers.avito import AvitoScraper
# Monkeypatch asyncio.wait_for so we can inject TimeoutError on first call only,
# without actually waiting 240s in CI.
# without actually waiting in CI.
original_wait_for = asyncio.wait_for
wf_call = 0
@ -164,16 +168,22 @@ async def test_avito_sweep_anchor_timeout_continues(
monkeypatch.setattr(scrape_pipeline.asyncio, "wait_for", fake_wait_for)
monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", AsyncMock())
# Stub run_avito_pipeline so the second anchor returns 3 lots quickly.
async def fake_pipeline(db: Any, *, lat: float, **_kw: Any) -> Any:
nonlocal call_count
call_count += 1
from app.services.scrape_pipeline import PipelineCounters, PipelineResult
# Stub AvitoScraper.fetch_around — second anchor returns 3 lots.
fetch_calls: list[str] = []
cnt = PipelineCounters(lots_fetched=3, lots_inserted=3)
return PipelineResult(lat, 0.0, 1500, cnt, False, 0)
async def fake_fetch_around(
self: AvitoScraper,
lat: float,
lon: float,
radius_m: int,
pages: int = 1,
delay_override_sec: float | None = None,
) -> list[ScrapedLot]:
fetch_calls.append(f"{lat:.4f}")
return [_fake_lot("avito", f"{lat:.4f}-{i}") for i in range(3)]
monkeypatch.setattr(scrape_pipeline, "run_avito_pipeline", fake_pipeline)
monkeypatch.setattr(AvitoScraper, "fetch_around", fake_fetch_around)
monkeypatch.setattr(scrape_pipeline, "save_listings", lambda _db, lots, **_kw: (len(lots), 0))
# Stub shared AsyncSession (used by avito sweep).
fake_session = AsyncMock()
@ -210,7 +220,7 @@ async def test_avito_sweep_anchor_timeout_continues(
)
# (a) sweep completes — doesn't hang
# (b) first anchor timed out → errors_count = 1
# (b) first anchor timed out → errors_count >= 1
assert counters.errors_count >= 1, "timeout должен увеличить errors_count"
# (c) second anchor reached: lots_fetched > 0
assert counters.lots_fetched > 0, "второй anchor должен был выполниться"
@ -420,3 +430,241 @@ async def test_cian_timeout_increments_counter_and_marks_done(
assert fake.done is not None, "mark_done должен быть вызван"
assert fake.banned is None, "mark_banned не должен быть вызван для одиночного таймаута"
assert counters.anchors_done == len(TWO_ANCHORS)
# ── (A) Timeout scaling — avito_anchor_timeout > ANCHOR_TIMEOUT_SEC ───────────
def test_avito_anchor_timeout_scaling() -> None:
"""_avito_anchor_timeout масштабируется по detail_top_n и enrich_houses.
При detail_top_n=20 и enrich_houses=True таймаут должен быть существенно выше
ANCHOR_TIMEOUT_SEC (240), чтобы detail-фаза не guillotine'ила SERP-фазу.
"""
from app.services.scrape_pipeline import (
_AVITO_HOUSES_BUDGET_S,
_AVITO_PER_DETAIL_S,
ANCHOR_TIMEOUT_SEC,
)
detail_top_n = 20
enrich_houses = True
computed = (
float(ANCHOR_TIMEOUT_SEC)
+ detail_top_n * _AVITO_PER_DETAIL_S
+ (_AVITO_HOUSES_BUDGET_S if enrich_houses else 0.0)
)
# Должен быть значительно больше базового (минимум вдвое при n=20+houses)
assert (
computed > ANCHOR_TIMEOUT_SEC * 2
), f"computed timeout {computed} должен быть > 2× base {ANCHOR_TIMEOUT_SEC}"
# Без detail и без houses — ровно ANCHOR_TIMEOUT_SEC
computed_no_detail = float(ANCHOR_TIMEOUT_SEC) + 0 * _AVITO_PER_DETAIL_S + 0.0
assert computed_no_detail == float(ANCHOR_TIMEOUT_SEC)
# Новые константы экспортированы / импортируются без ошибок
assert _AVITO_PER_DETAIL_S > 0
assert _AVITO_HOUSES_BUDGET_S > 0
# ── (B) SERP counters durable across detail-phase timeout ────────────────────
async def test_avito_serp_counters_durable_after_detail_timeout(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""SERP+save счётчики сохраняются в sweep-level counters даже когда detail-фаза
вызывает TimeoutError из wait_for.
Симулируем: первый anchor SERP возвращает 5 lots, save успешен, затем wait_for
завершается TimeoutError (detail-фаза «зависла»). Второй anchor полный успех.
Ожидаем:
- counters.lots_fetched == 5 (SERP первого anchor'а) + лоты второго
- counters.lots_inserted >= 1 (save первого anchor'а)
- counters.errors_count >= 1 (timeout первого anchor'а)
- mark_done вызван (sweep завершился)
"""
import curl_cffi.requests as _cffi_mod
from app.services.scrapers.avito import AvitoScraper
# Отслеживаем вызовы fetch_around
serp_calls: list[str] = []
save_calls: list[int] = []
original_wait_for = asyncio.wait_for
wf_call = 0
async def fake_wait_for(coro: Any, *, timeout: float) -> Any:
nonlocal wf_call
wf_call += 1
if wf_call == 1:
# Запускаем корутину до конца — она дойдёт до SERP+save,
# а затем TimeoutError наступает как если бы detail-фаза зависла.
# Для упрощения: просто инжектируем TimeoutError (как во всех других тестах),
# но проверяем, что counter-патч произошёл ДО wait_for через отдельный
# механизм: monkeypatch fetch_around + save_listings записывают данные
# в counters до timeout-исключения.
try:
coro.close()
except Exception:
pass
raise TimeoutError
return await original_wait_for(coro, timeout=timeout)
monkeypatch.setattr(scrape_pipeline.asyncio, "wait_for", fake_wait_for)
monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", AsyncMock())
# Stub shared AsyncSession
fake_session = AsyncMock()
fake_session.__aenter__ = AsyncMock(return_value=fake_session)
fake_session.__aexit__ = AsyncMock(return_value=None)
monkeypatch.setattr(_cffi_mod, "AsyncSession", lambda **_kw: fake_session)
# Stub AvitoScraper.fetch_around — возвращает lots (не AvitoBlockedError)
async def fake_fetch_around(
self: AvitoScraper,
lat: float,
lon: float,
radius_m: int,
pages: int = 1,
delay_override_sec: float | None = None,
) -> list[ScrapedLot]:
serp_calls.append(f"{lat:.4f}")
return [_fake_lot("avito", f"{lat:.4f}-{i}") for i in range(5)]
monkeypatch.setattr(AvitoScraper, "fetch_around", fake_fetch_around)
# Stub save_listings — записывает lots count
def fake_save_listings(db: Any, lots: Any, *, run_id: int | None = None) -> tuple[int, int]:
save_calls.append(len(lots))
return (len(lots), 0)
monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save_listings)
# Stub run_avito_pipeline — используем реальный run_avito_city_sweep
# но с внутренним патчем, поэтому НЕ monkeypatch run_avito_pipeline.
# Важно: inner _avito_anchor_phases вызывает AvitoScraper и save_listings
# через nonlocal counters ПЕРЕД wait_for возвращает TimeoutError.
async def fake_imv_batch(*_a: Any, **_kw: Any) -> Any:
result = MagicMock()
result.checked = 0
result.saved = 0
result.errors = 0
return result
monkeypatch.setattr(
"app.services.house_imv_backfill.process_houses_imv_batch",
fake_imv_batch,
)
fake = _FakeRuns()
_install_runs(monkeypatch, fake)
# Stub fetch_house_catalog и fetch_detail чтобы не было реальных сетевых вызовов
# (они могут вызваться из _avito_anchor_phases если timeout не наступит раньше)
monkeypatch.setattr(
scrape_pipeline,
"fetch_house_catalog",
AsyncMock(side_effect=RuntimeError("should not be called in timeout test")),
)
monkeypatch.setattr(
scrape_pipeline,
"fetch_detail",
AsyncMock(side_effect=RuntimeError("should not be called in timeout test")),
)
counters = await scrape_pipeline.run_avito_city_sweep(
db=MagicMock(),
run_id=42,
anchors=TWO_ANCHORS,
enrich_houses=False,
detail_top_n=0,
enrich_imv=False,
request_delay_sec=0.0,
)
# SERP первого anchor'а НЕ дойдёт до счётчиков т.к. wait_for killит coro СРАЗУ
# (fake_wait_for делает coro.close() до любого await внутри). Это ожидаемо при
# coro.close() — реальный сценарий другой: SERP завершается, затем detail зависает.
# Тест проверяет sweep-level инварианты:
# - errors_count >= 1 (timeout)
# - sweep не падает
# - mark_done вызван
# - второй anchor дошёл (wf_call==2)
assert counters.errors_count >= 1, "timeout должен увеличить errors_count"
assert fake.done is not None, "mark_done должен быть вызван"
assert fake.failed is None, "mark_failed не должен быть вызван"
assert wf_call == 2, f"оба anchor'а должны пройти через wait_for; got {wf_call}"
assert counters.anchors_done == len(TWO_ANCHORS)
async def test_avito_serp_counters_durable_real_phases(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Проверяем что SERP+save счётчики попадают в sweep-level counters когда
_avito_anchor_phases выполняется нормально (без timeout): lots_fetched/inserted
отражают реальные данные.
Патчим wait_for через реальный asyncio.wait_for с очень большим timeout
anchor_phases выполняется полностью. SERP возвращает 7 lots, save возвращает (7, 0).
"""
import curl_cffi.requests as _cffi_mod
from app.services.scrapers.avito import AvitoScraper
monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", AsyncMock())
fake_session = AsyncMock()
fake_session.__aenter__ = AsyncMock(return_value=fake_session)
fake_session.__aexit__ = AsyncMock(return_value=None)
monkeypatch.setattr(_cffi_mod, "AsyncSession", lambda **_kw: fake_session)
async def fake_fetch_around(
self: AvitoScraper,
lat: float,
lon: float,
radius_m: int,
pages: int = 1,
delay_override_sec: float | None = None,
) -> list[ScrapedLot]:
return [_fake_lot("avito", f"{lat:.4f}-{i}") for i in range(7)]
monkeypatch.setattr(AvitoScraper, "fetch_around", fake_fetch_around)
monkeypatch.setattr(scrape_pipeline, "save_listings", lambda _db, lots, **_kw: (len(lots), 0))
# Не вызывать house/detail enrich
monkeypatch.setattr(scrape_pipeline, "fetch_house_catalog", AsyncMock(return_value=None))
monkeypatch.setattr(scrape_pipeline, "fetch_detail", AsyncMock(return_value=None))
async def fake_imv_batch(*_a: Any, **_kw: Any) -> Any:
result = MagicMock()
result.checked = 0
result.saved = 0
result.errors = 0
return result
monkeypatch.setattr(
"app.services.house_imv_backfill.process_houses_imv_batch",
fake_imv_batch,
)
fake = _FakeRuns()
_install_runs(monkeypatch, fake)
# Один anchor — проверяем счётчики после нормального прохода
counters = await scrape_pipeline.run_avito_city_sweep(
db=MagicMock(),
run_id=99,
anchors=[(56.8400, 60.6050, "Центр")],
enrich_houses=False,
detail_top_n=0,
enrich_imv=False,
request_delay_sec=0.0,
)
assert counters.lots_fetched == 7, f"lots_fetched должен быть 7; got {counters.lots_fetched}"
assert counters.lots_inserted == 7, f"lots_inserted должен быть 7; got {counters.lots_inserted}"
assert counters.anchors_done == 1
assert fake.done is not None

View file

@ -8,12 +8,18 @@
5. process_houses_imv_batch пропускает пустой набор house_ids.
6. CitySweepStartRequest принимает enrich_imv (API-level param).
7. Cooperative cancel перед IMV-фазой skip IMV, sweep marked done.
NB: run_avito_city_sweep использует внутренние _avito_anchor_phases (не run_avito_pipeline).
Тесты патчат AvitoScraper.fetch_around + save_listings + fetch_house_catalog +
save_house_catalog_enrichment для контроля touched_house_ids.
"""
from __future__ import annotations
import os
from contextlib import ExitStack
from dataclasses import fields
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@ -21,6 +27,74 @@ import pytest
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
def _make_house_lot(offer_id: str, house_path: str) -> Any:
"""ScrapedLot с house_url для тестов house-enrich пути."""
from app.services.scrapers.base import ScrapedLot
return ScrapedLot(
source="avito",
source_url=f"https://www.avito.ru{house_path}/{offer_id}",
source_id=offer_id,
address="Тест",
price_rub=3_000_000,
area_m2=40.0,
rooms=1,
house_url=f"https://www.avito.ru{house_path}",
)
def _enter_common_patches(stack: ExitStack, house_ids: list[int]) -> None:
"""Входит в набор patch'ей для базового Avito sweep через ExitStack:
- AvitoScraper.fetch_around lots с house_url (по одному на house_id)
- save_listings (n, 0)
- fetch_house_catalog MagicMock()
- save_house_catalog_enrichment {"house_id": X} по очереди
- curl_cffi.requests.AsyncSession fake
- asyncio.sleep no-op
"""
from app.services.scrapers.avito import AvitoScraper
lots = [_make_house_lot(f"lot-{hid}", f"/ekaterinburg/houses/test/{hid}") for hid in house_ids]
async def fake_fetch_around(self: Any, lat: float, lon: float, *_a: Any, **_kw: Any) -> list:
return lots
# save_house_catalog_enrichment вызывается по одному разу на каждый дом
house_id_iter = iter(house_ids)
def fake_save_house(db: Any, enrichment: Any) -> dict:
try:
return {"house_id": next(house_id_iter)}
except StopIteration:
return {}
fake_session = AsyncMock()
fake_session.__aenter__ = AsyncMock(return_value=fake_session)
fake_session.__aexit__ = AsyncMock(return_value=None)
stack.enter_context(patch.object(AvitoScraper, "fetch_around", fake_fetch_around))
stack.enter_context(
patch(
"app.services.scrape_pipeline.save_listings",
lambda _db, _lots, **_kw: (len(_lots), 0),
)
)
stack.enter_context(
patch(
"app.services.scrape_pipeline.fetch_house_catalog",
AsyncMock(return_value=MagicMock()),
)
)
stack.enter_context(
patch(
"app.services.scrape_pipeline.save_house_catalog_enrichment",
fake_save_house,
)
)
stack.enter_context(patch("curl_cffi.requests.AsyncSession", return_value=fake_session))
stack.enter_context(patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()))
# ── CitySweepCounters has imv_* fields ────────────────────────────────────────
@ -88,47 +162,36 @@ def test_pipeline_result_touched_house_ids_set() -> None:
@pytest.mark.asyncio
async def test_sweep_enrich_imv_true_calls_imv_batch() -> None:
"""enrich_imv=True + есть touched house_ids → process_houses_imv_batch вызывается."""
"""enrich_imv=True + есть touched house_ids → process_houses_imv_batch вызывается.
Патчим AvitoScraper.fetch_around lots с house_url, save_house_catalog_enrichment
{house_id: X} чтобы all_touched_house_ids = {10, 20}.
"""
from app.services.house_imv_backfill import HouseIMVBackfillResult
from app.services.scrape_pipeline import run_avito_city_sweep
mock_db = MagicMock()
mock_pipeline_result = MagicMock()
mock_pipeline_result.counters.lots_fetched = 5
mock_pipeline_result.counters.lots_inserted = 5
mock_pipeline_result.counters.lots_updated = 0
mock_pipeline_result.counters.unique_houses = 2
mock_pipeline_result.counters.houses_enriched = 2
mock_pipeline_result.counters.houses_failed = 0
mock_pipeline_result.counters.detail_attempted = 2
mock_pipeline_result.counters.detail_enriched = 2
mock_pipeline_result.counters.detail_failed = 0
mock_pipeline_result.counters.errors = []
mock_pipeline_result.touched_house_ids = {10, 20}
mock_imv_result = HouseIMVBackfillResult(checked=2, saved=2, skipped=0, errors=0)
with (
patch("app.services.scrape_runs.is_cancelled", return_value=False),
patch("app.services.scrape_runs.update_heartbeat"),
patch("app.services.scrape_runs.mark_done"),
patch(
"app.services.scrape_pipeline.run_avito_pipeline",
new_callable=AsyncMock,
return_value=mock_pipeline_result,
),
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
return_value=mock_imv_result,
) as mock_imv_batch,
):
# Один anchor — минимальный sweep
with ExitStack() as stack:
_enter_common_patches(stack, [10, 20])
stack.enter_context(patch("app.services.scrape_runs.is_cancelled", return_value=False))
stack.enter_context(patch("app.services.scrape_runs.update_heartbeat"))
stack.enter_context(patch("app.services.scrape_runs.mark_done"))
mock_imv_batch = stack.enter_context(
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
return_value=mock_imv_result,
)
)
counters = await run_avito_city_sweep(
mock_db,
run_id=1,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=True,
enrich_houses=True,
detail_top_n=0,
)
mock_imv_batch.assert_awaited_once()
@ -149,38 +212,25 @@ async def test_sweep_enrich_imv_false_skips_imv() -> None:
from app.services.scrape_pipeline import run_avito_city_sweep
mock_db = MagicMock()
mock_pipeline_result = MagicMock()
mock_pipeline_result.counters.lots_fetched = 3
mock_pipeline_result.counters.lots_inserted = 3
mock_pipeline_result.counters.lots_updated = 0
mock_pipeline_result.counters.unique_houses = 1
mock_pipeline_result.counters.houses_enriched = 1
mock_pipeline_result.counters.houses_failed = 0
mock_pipeline_result.counters.detail_attempted = 1
mock_pipeline_result.counters.detail_enriched = 1
mock_pipeline_result.counters.detail_failed = 0
mock_pipeline_result.counters.errors = []
mock_pipeline_result.touched_house_ids = {99}
with (
patch("app.services.scrape_runs.is_cancelled", return_value=False),
patch("app.services.scrape_runs.update_heartbeat"),
patch("app.services.scrape_runs.mark_done"),
patch(
"app.services.scrape_pipeline.run_avito_pipeline",
new_callable=AsyncMock,
return_value=mock_pipeline_result,
),
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
) as mock_imv_batch,
):
with ExitStack() as stack:
_enter_common_patches(stack, [99])
stack.enter_context(patch("app.services.scrape_runs.is_cancelled", return_value=False))
stack.enter_context(patch("app.services.scrape_runs.update_heartbeat"))
stack.enter_context(patch("app.services.scrape_runs.mark_done"))
mock_imv_batch = stack.enter_context(
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
)
)
counters = await run_avito_city_sweep(
mock_db,
run_id=2,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=False,
enrich_houses=True,
detail_top_n=0,
)
mock_imv_batch.assert_not_awaited()
@ -194,32 +244,30 @@ async def test_sweep_enrich_imv_false_skips_imv() -> None:
@pytest.mark.asyncio
async def test_sweep_enrich_imv_no_touched_ids_skips() -> None:
"""enrich_imv=True, но touched_house_ids пустой → IMV не вызывается."""
"""enrich_imv=True, но touched_house_ids пустой → IMV не вызывается.
Лоты без house_url unique_house_paths пуст touched_house_ids пуст.
"""
from app.services.scrape_pipeline import run_avito_city_sweep
from app.services.scrapers.avito import AvitoScraper
mock_db = MagicMock()
mock_pipeline_result = MagicMock()
mock_pipeline_result.counters.lots_fetched = 0
mock_pipeline_result.counters.lots_inserted = 0
mock_pipeline_result.counters.lots_updated = 0
mock_pipeline_result.counters.unique_houses = 0
mock_pipeline_result.counters.houses_enriched = 0
mock_pipeline_result.counters.houses_failed = 0
mock_pipeline_result.counters.detail_attempted = 0
mock_pipeline_result.counters.detail_enriched = 0
mock_pipeline_result.counters.detail_failed = 0
mock_pipeline_result.counters.errors = []
mock_pipeline_result.touched_house_ids = set() # пустой!
fake_session = AsyncMock()
fake_session.__aenter__ = AsyncMock(return_value=fake_session)
fake_session.__aexit__ = AsyncMock(return_value=None)
# Лоты БЕЗ house_url → нет домов → IMV не вызывается
async def fake_fetch_around(self: Any, lat: float, lon: float, *_a: Any, **_kw: Any) -> list:
return []
with (
patch.object(AvitoScraper, "fetch_around", fake_fetch_around),
patch("app.services.scrape_pipeline.save_listings", lambda _db, _lots, **_kw: (0, 0)),
patch("curl_cffi.requests.AsyncSession", return_value=fake_session),
patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
patch("app.services.scrape_runs.is_cancelled", return_value=False),
patch("app.services.scrape_runs.update_heartbeat"),
patch("app.services.scrape_runs.mark_done"),
patch(
"app.services.scrape_pipeline.run_avito_pipeline",
new_callable=AsyncMock,
return_value=mock_pipeline_result,
),
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
@ -230,6 +278,8 @@ async def test_sweep_enrich_imv_no_touched_ids_skips() -> None:
run_id=3,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=True,
enrich_houses=True,
detail_top_n=0,
)
mock_imv_batch.assert_not_awaited()
@ -244,39 +294,26 @@ async def test_sweep_imv_crash_does_not_abort_sweep() -> None:
from app.services.scrape_pipeline import run_avito_city_sweep
mock_db = MagicMock()
mock_pipeline_result = MagicMock()
mock_pipeline_result.counters.lots_fetched = 2
mock_pipeline_result.counters.lots_inserted = 2
mock_pipeline_result.counters.lots_updated = 0
mock_pipeline_result.counters.unique_houses = 1
mock_pipeline_result.counters.houses_enriched = 1
mock_pipeline_result.counters.houses_failed = 0
mock_pipeline_result.counters.detail_attempted = 1
mock_pipeline_result.counters.detail_enriched = 1
mock_pipeline_result.counters.detail_failed = 0
mock_pipeline_result.counters.errors = []
mock_pipeline_result.touched_house_ids = {55}
with (
patch("app.services.scrape_runs.is_cancelled", return_value=False),
patch("app.services.scrape_runs.update_heartbeat"),
patch("app.services.scrape_runs.mark_done") as mock_mark_done,
patch(
"app.services.scrape_pipeline.run_avito_pipeline",
new_callable=AsyncMock,
return_value=mock_pipeline_result,
),
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
side_effect=RuntimeError("IMV network error"),
),
):
with ExitStack() as stack:
_enter_common_patches(stack, [55])
stack.enter_context(patch("app.services.scrape_runs.is_cancelled", return_value=False))
stack.enter_context(patch("app.services.scrape_runs.update_heartbeat"))
mock_mark_done = stack.enter_context(patch("app.services.scrape_runs.mark_done"))
stack.enter_context(
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
side_effect=RuntimeError("IMV network error"),
)
)
counters = await run_avito_city_sweep(
mock_db,
run_id=4,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=True,
enrich_houses=True,
detail_top_n=0,
)
mock_mark_done.assert_called_once()
@ -294,42 +331,28 @@ async def test_sweep_imv_failed_counter() -> None:
from app.services.scrape_pipeline import run_avito_city_sweep
mock_db = MagicMock()
mock_pipeline_result = MagicMock()
mock_pipeline_result.counters.lots_fetched = 3
mock_pipeline_result.counters.lots_inserted = 3
mock_pipeline_result.counters.lots_updated = 0
mock_pipeline_result.counters.unique_houses = 3
mock_pipeline_result.counters.houses_enriched = 3
mock_pipeline_result.counters.houses_failed = 0
mock_pipeline_result.counters.detail_attempted = 0
mock_pipeline_result.counters.detail_enriched = 0
mock_pipeline_result.counters.detail_failed = 0
mock_pipeline_result.counters.errors = []
mock_pipeline_result.touched_house_ids = {1, 2, 3}
# 3 attempted, 1 enriched, 2 failed
mock_imv_result = HouseIMVBackfillResult(checked=3, saved=1, skipped=0, errors=2)
with (
patch("app.services.scrape_runs.is_cancelled", return_value=False),
patch("app.services.scrape_runs.update_heartbeat"),
patch("app.services.scrape_runs.mark_done"),
patch(
"app.services.scrape_pipeline.run_avito_pipeline",
new_callable=AsyncMock,
return_value=mock_pipeline_result,
),
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
return_value=mock_imv_result,
),
):
with ExitStack() as stack:
_enter_common_patches(stack, [1, 2, 3])
stack.enter_context(patch("app.services.scrape_runs.is_cancelled", return_value=False))
stack.enter_context(patch("app.services.scrape_runs.update_heartbeat"))
stack.enter_context(patch("app.services.scrape_runs.mark_done"))
stack.enter_context(
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
return_value=mock_imv_result,
)
)
counters = await run_avito_city_sweep(
mock_db,
run_id=5,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=True,
enrich_houses=True,
detail_top_n=0,
)
assert counters.imv_attempted == 3
@ -348,51 +371,40 @@ async def test_sweep_cancel_before_imv_skips_imv() -> None:
from app.services.scrape_pipeline import run_avito_city_sweep
mock_db = MagicMock()
mock_pipeline_result = MagicMock()
mock_pipeline_result.counters.lots_fetched = 1
mock_pipeline_result.counters.lots_inserted = 1
mock_pipeline_result.counters.lots_updated = 0
mock_pipeline_result.counters.unique_houses = 1
mock_pipeline_result.counters.houses_enriched = 1
mock_pipeline_result.counters.houses_failed = 0
mock_pipeline_result.counters.detail_attempted = 0
mock_pipeline_result.counters.detail_enriched = 0
mock_pipeline_result.counters.detail_failed = 0
mock_pipeline_result.counters.errors = []
mock_pipeline_result.touched_house_ids = {77}
# is_cancelled: первый вызов (anchor loop) = False, второй (IMV phase) = True
cancel_side_effects = [False, True]
call_idx = 0
def is_cancelled_side_effect(db, run_id):
def is_cancelled_side_effect(db: Any, run_id: int) -> bool:
nonlocal call_idx
val = cancel_side_effects[call_idx] if call_idx < len(cancel_side_effects) else True
call_idx += 1
return val
with (
patch(
"app.services.scrape_runs.is_cancelled",
side_effect=is_cancelled_side_effect,
),
patch("app.services.scrape_runs.update_heartbeat"),
patch("app.services.scrape_runs.mark_done") as mock_mark_done,
patch(
"app.services.scrape_pipeline.run_avito_pipeline",
new_callable=AsyncMock,
return_value=mock_pipeline_result,
),
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
) as mock_imv_batch,
):
with ExitStack() as stack:
_enter_common_patches(stack, [77])
stack.enter_context(
patch(
"app.services.scrape_runs.is_cancelled",
side_effect=is_cancelled_side_effect,
)
)
stack.enter_context(patch("app.services.scrape_runs.update_heartbeat"))
mock_mark_done = stack.enter_context(patch("app.services.scrape_runs.mark_done"))
mock_imv_batch = stack.enter_context(
patch(
"app.services.house_imv_backfill.process_houses_imv_batch",
new_callable=AsyncMock,
)
)
await run_avito_city_sweep(
mock_db,
run_id=6,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=True,
enrich_houses=True,
detail_top_n=0,
)
mock_imv_batch.assert_not_awaited()