lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(tradein): avito city-sweep — IMV enrichment phase (один прогон = SERP+detail+houses+IMV)

Browse source 
Встраивает IMV-оценку домов как финальную фазу run_avito_city_sweep.
Один запуск sweep'а теперь наполняет: listings (SERP) + detail-обогащение
+ houses (catalog/reviews/sellers) + IMV-оценку тронутых домов.

- scrape_pipeline: добавлен enrich_imv param, CitySweepCounters.imv_attempted/
  imv_enriched/imv_failed, PipelineResult.touched_house_ids (собирается при
  houses-enrichment из save_house_catalog_enrichment['house_id']).
  После всех anchor'ов — IMV-фаза: process_houses_imv_batch для touched_house_ids,
  cooperative cancel check, graceful per-house error handling (не валит sweep).
- house_imv_backfill: добавлена process_houses_imv_batch — тонкая обёртка для
  обработки конкретного set[house_id] (пропускает imv_status=ok, дома без coords).
- admin: CitySweepStartRequest.enrich_imv: bool = True + проброс в sweep.
- tests: 14 новых тестов (enrich_imv=True/False, counters, cancel-before-IMV,
  IMV crash graceful, process_houses_imv_batch empty, API param).
This commit is contained in:
bot-backend 2026-05-31 10:06:37 +03:00
parent b16a45ef87
commit 7ce758419a
4 changed files with 668 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
tradein-mvp/backend/app/api/v1/admin.py
View file

@ -587,6 +587,13 @@ class CitySweepStartRequest(BaseModel):
detail_top_n: int = Field(default=10, ge=0, le=30)
request_delay_sec: float = Field(default=7.0, ge=3.0, le=15.0)
enrich_houses: bool = True
enrich_imv: bool = Field(
default=True,
description=(
"Если True — после всех anchor'ов запускает IMV-оценку тронутых домов "
"(финальная фаза sweep'а). Требует enrich_houses=True для сбора house_id."
),
)
radius_m: int = Field(default=1500, ge=500, le=5000)
@ -636,6 +643,7 @@ async def start_avito_city_sweep(
enrich_houses=payload.enrich_houses,
detail_top_n=payload.detail_top_n,
request_delay_sec=payload.request_delay_sec,
enrich_imv=payload.enrich_imv,
)
except Exception:
logger.exception("city-sweep background task run_id=%d crashed", run_id)

105
tradein-mvp/backend/app/services/house_imv_backfill.py
View file

@ -455,6 +455,111 @@ async def backfill_house_imv(
return result
async def process_houses_imv_batch(
db: Session,
house_ids: set[int],
*,
request_delay_sec: float = 5.0,
) -> HouseIMVBackfillResult:
"""Обработать конкретный набор house_id (для sweep-интеграции).
Зеркало backfill_house_imv, но работает по явному списку house_id
(не по imv_status-фильтру). Предназначен для финальной IMV-фазы
run_avito_city_sweep — обрабатываем только дома, тронутые за этот sweep.
Дома без lat/lon или address пропускаются (статус no_params/no_address).
Дома уже с imv_status='ok' тоже пропускаются — UPSERT всё равно обновит,
поэтому лучше не гонять лишних API-запросов, если данные свежие.
Для форс-обновления — оставь это решение caller'у (sweep по умолчанию
обходит дома с ok, обновляя только pending/not_found/error/transient_error).
"""
result = HouseIMVBackfillResult()
t0 = time.time()
if not house_ids:
result.duration_sec = time.time() - t0
return result
ids_list = list(house_ids)
# Загружаем только дома, у которых есть координаты + адрес + imv_status != 'ok'
# (уже оцененные дома пропускаем — не тратим API-квоту повторно)
placeholders = ", ".join(f":hid_{i}" for i in range(len(ids_list)))
params: dict[str, object] = {f"hid_{i}": hid for i, hid in enumerate(ids_list)}
rows = (
db.execute(
text(f"""
SELECT id, address, full_address, lat, lon
FROM houses
WHERE id IN ({placeholders})
AND lat IS NOT NULL
AND lon IS NOT NULL
AND address IS NOT NULL
AND COALESCE(imv_status, 'pending') != 'ok'
ORDER BY id
"""),
params,
)
.mappings()
.all()
)
result.checked = len(rows)
if not rows:
logger.info(
"process_houses_imv_batch: nothing to process (%d house_ids, all ok/no-coords)",
len(house_ids),
)
result.duration_sec = time.time() - t0
return result
logger.info(
"process_houses_imv_batch: %d/%d houses eligible (delay=%.1fs)",
result.checked,
len(house_ids),
request_delay_sec,
)
for i, house in enumerate(rows):
hid: int = house["id"]
status_str = await _process_one_house(db, dict(house))
result.status_counts[status_str] = result.status_counts.get(status_str, 0) + 1
if status_str == "ok":
result.saved += 1
elif status_str in ("no_params", "no_address"):
result.skipped += 1
else:
result.errors += 1
logger.info(
"[%d/%d] imv_batch house_id=%d status=%s",
i + 1,
result.checked,
hid,
status_str,
)
if i < len(rows) - 1:
if status_str == "auth_error":
logger.warning("IMV auth error — extra delay 60s before next house")
await asyncio.sleep(60)
else:
await asyncio.sleep(request_delay_sec)
result.duration_sec = time.time() - t0
logger.info(
"process_houses_imv_batch done: checked=%d saved=%d skipped=%d errors=%d %.1fs %s",
result.checked,
result.saved,
result.skipped,
result.errors,
result.duration_sec,
result.status_counts,
)
return result
async def _process_one_house(db: Session, house: dict) -> str:
"""Process a single house. Returns final status string."""
hid: int = house["id"]

85
tradein-mvp/backend/app/services/scrape_pipeline.py
View file

@ -19,7 +19,10 @@ Anti-bot hardening (2023-05-23):
Graceful degradation на каждом step: exception в одной house/listing
не валит весь pipeline (кроме blocked exceptions — они abort весь sweep).
IMV ОТСУТСТВУЕТ — он on-demand (Stage 3 estimator integration), не cron.
IMV-фаза (финальная, после всех anchor'ов): если enrich_imv=True —
для домов, тронутых в этом sweep'е (touched house_id из houses-enrichment),
вызывается house_imv_backfill.process_houses_imv_batch. Ошибки per-house
логируются и считаются, но не валят sweep.
"""
from __future__ import annotations
@ -111,6 +114,7 @@ class PipelineResult:
counters: PipelineCounters
enrich_houses: bool
enrich_detail_top_n: int
touched_house_ids: set[int] = field(default_factory=set)
# ── Main orchestrator ───────────────────────────────────────────
@ -203,13 +207,17 @@ async def run_avito_pipeline(
logger.info("pipeline:group_by_house unique=%d", counters.unique_houses)
# ── Step 4: enrich houses с shared session + sleep ──────
touched_house_ids: set[int] = set()
if enrich_houses and unique_house_paths:
house_paths_list = list(unique_house_paths)
for idx, house_path in enumerate(house_paths_list):
try:
enrichment = await fetch_house_catalog(house_path, cffi_session=session)
save_house_catalog_enrichment(db, enrichment)
house_counters = save_house_catalog_enrichment(db, enrichment)
counters.houses_enriched += 1
hid = house_counters.get("house_id")
if hid:
touched_house_ids.add(int(hid))
except (AvitoBlockedError, AvitoRateLimitedError):
logger.error("pipeline:house BLOCKED at %s — propagating", house_path)
raise
@ -304,7 +312,7 @@ async def run_avito_pipeline(
logger.info(
"pipeline:done anchor=(%.4f,%.4f) lots=%d (ins=%d/upd=%d) "
"houses=%d/%d detail=%d/%d errors=%d",
"houses=%d/%d detail=%d/%d touched_house_ids=%d errors=%d",
lat,
lon,
counters.lots_fetched,
@ -314,9 +322,18 @@ async def run_avito_pipeline(
counters.unique_houses,
counters.detail_enriched,
counters.detail_attempted,
len(touched_house_ids),
len(counters.errors),
)
return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n)
return PipelineResult(
lat,
lon,
radius_m,
counters,
enrich_houses,
enrich_detail_top_n,
touched_house_ids=touched_house_ids,
)
finally:
if own_session:
@ -338,6 +355,9 @@ class CitySweepCounters:
detail_attempted: int = 0
detail_enriched: int = 0
detail_failed: int = 0
imv_attempted: int = 0
imv_enriched: int = 0
imv_failed: int = 0
errors_count: int = 0
def to_dict(self) -> dict[str, int]:
@ -354,15 +374,21 @@ async def run_avito_city_sweep(
enrich_houses: bool = True,
detail_top_n: int = 10,
request_delay_sec: float = 7.0,
enrich_imv: bool = True,
) -> CitySweepCounters:
"""Full city sweep: iterate anchors × pages → save → enrich houses + detail.
"""Full city sweep: iterate anchors × pages → save → enrich houses + detail → IMV.
- Single shared AsyncSession на весь sweep (один TLS fingerprint)
- AvitoBlockedError/AvitoRateLimitedError → ABORT sweep + mark_banned (status='banned')
- Прочие errors per-anchor логируются, не валят весь sweep
- После всех anchor'ов: если enrich_imv=True — IMV-оценка тронутых домов
(cooperative cancel + per-house graceful error handling)
"""
from app.services.house_imv_backfill import process_houses_imv_batch
_anchors = anchors if anchors is not None else EKB_ANCHORS
counters = CitySweepCounters(anchors_total=len(_anchors))
all_touched_house_ids: set[int] = set()
async with AsyncSession(
impersonate="chrome120",
@ -414,6 +440,7 @@ async def run_avito_city_sweep(
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)
except (AvitoBlockedError, AvitoRateLimitedError) as e:
logger.error(
"city-sweep run_id=%d ABORT at anchor #%d/%d (%s) — blocked: %s",
@ -435,10 +462,54 @@ async def run_avito_city_sweep(
counters.anchors_done = idx
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
# ── IMV-фаза: финальный обход тронутых домов ──────────
if enrich_imv and all_touched_house_ids:
if scrape_runs.is_cancelled(db, run_id):
logger.info(
"city-sweep run_id=%d: cancelled before IMV phase (%d houses skipped)",
run_id,
len(all_touched_house_ids),
)
scrape_runs.mark_done(db, run_id, counters.to_dict())
return counters
logger.info(
"city-sweep run_id=%d: IMV phase — %d touched houses",
run_id,
len(all_touched_house_ids),
)
try:
imv_result = await process_houses_imv_batch(
db,
all_touched_house_ids,
request_delay_sec=request_delay_sec,
)
counters.imv_attempted += imv_result.checked
counters.imv_enriched += imv_result.saved
counters.imv_failed += imv_result.errors
counters.errors_count += imv_result.errors
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
logger.info(
"city-sweep run_id=%d: IMV phase done — "
"attempted=%d enriched=%d failed=%d",
run_id,
imv_result.checked,
imv_result.saved,
imv_result.errors,
)
except Exception as exc:
logger.error(
"city-sweep run_id=%d: IMV phase crashed — %r (sweep still done)",
run_id,
exc,
)
counters.errors_count += 1
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
scrape_runs.mark_done(db, run_id, counters.to_dict())
logger.info(
"city-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) "
"houses=%d/%d detail=%d/%d errors=%d",
"houses=%d/%d detail=%d/%d imv=%d/%d errors=%d",
run_id,
counters.anchors_done,
counters.anchors_total,
@ -449,6 +520,8 @@ async def run_avito_city_sweep(
counters.unique_houses,
counters.detail_enriched,
counters.detail_attempted,
counters.imv_enriched,
counters.imv_attempted,
counters.errors_count,
)
return counters

476
tradein-mvp/backend/tests/test_sweep_imv_phase.py Normal file
View file

@ -0,0 +1,476 @@
"""Tests for IMV enrichment phase integrated into Avito city sweep.
Проверяет:
1. enrich_imv=True — sweep вызывает IMV-фазу для touched house_id.
2. enrich_imv=False — IMV не вызывается.
3. IMV counters (imv_attempted, imv_enriched, imv_failed) заполняются.
4. CitySweepCounters содержит новые imv_* поля.
5. process_houses_imv_batch пропускает пустой набор house_ids.
6. CitySweepStartRequest принимает enrich_imv (API-level param).
7. Cooperative cancel перед IMV-фазой — skip IMV, sweep marked done.
"""
from __future__ import annotations
import os
from dataclasses import fields
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
# ── CitySweepCounters has imv_* fields ────────────────────────────────────────
def test_city_sweep_counters_has_imv_fields() -> None:
from app.services.scrape_pipeline import CitySweepCounters
c = CitySweepCounters()
assert hasattr(c, "imv_attempted")
assert hasattr(c, "imv_enriched")
assert hasattr(c, "imv_failed")
assert c.imv_attempted == 0
assert c.imv_enriched == 0
assert c.imv_failed == 0
def test_city_sweep_counters_to_dict_has_imv_keys() -> None:
from app.services.scrape_pipeline import CitySweepCounters
c = CitySweepCounters(imv_attempted=5, imv_enriched=3, imv_failed=2)
d = c.to_dict()
assert d["imv_attempted"] == 5
assert d["imv_enriched"] == 3
assert d["imv_failed"] == 2
# Все dataclass-поля должны быть в to_dict
expected_keys = {f.name for f in fields(c)}
assert set(d.keys()) == expected_keys
# ── PipelineResult имеет touched_house_ids ────────────────────────────────────
def test_pipeline_result_has_touched_house_ids() -> None:
from app.services.scrape_pipeline import PipelineCounters, PipelineResult
r = PipelineResult(
anchor_lat=56.84,
anchor_lon=60.6,
radius_m=1500,
counters=PipelineCounters(),
enrich_houses=True,
enrich_detail_top_n=10,
)
assert hasattr(r, "touched_house_ids")
assert isinstance(r.touched_house_ids, set)
assert len(r.touched_house_ids) == 0
def test_pipeline_result_touched_house_ids_set() -> None:
from app.services.scrape_pipeline import PipelineCounters, PipelineResult
r = PipelineResult(
anchor_lat=56.84,
anchor_lon=60.6,
radius_m=1500,
counters=PipelineCounters(),
enrich_houses=True,
enrich_detail_top_n=10,
touched_house_ids={1, 2, 3},
)
assert r.touched_house_ids == {1, 2, 3}
# ── enrich_imv=True → IMV phase вызывается ───────────────────────────────────
@pytest.mark.asyncio
async def test_sweep_enrich_imv_true_calls_imv_batch() -> None:
"""enrich_imv=True + есть touched house_ids → process_houses_imv_batch вызывается."""
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
counters = await run_avito_city_sweep(
mock_db,
run_id=1,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=True,
)
mock_imv_batch.assert_awaited_once()
# Первый позиционный аргумент — db, второй — house_ids
call_args = mock_imv_batch.call_args
assert call_args.args[1] == {10, 20}
assert counters.imv_attempted == 2
assert counters.imv_enriched == 2
assert counters.imv_failed == 0
# ── enrich_imv=False → IMV не вызывается ────────────────────────────────────
@pytest.mark.asyncio
async def test_sweep_enrich_imv_false_skips_imv() -> None:
"""enrich_imv=False → process_houses_imv_batch НЕ вызывается."""
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,
):
counters = await run_avito_city_sweep(
mock_db,
run_id=2,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=False,
)
mock_imv_batch.assert_not_awaited()
assert counters.imv_attempted == 0
assert counters.imv_enriched == 0
assert counters.imv_failed == 0
# ── enrich_imv=True но нет touched house_ids → IMV не вызывается ────────────
@pytest.mark.asyncio
async def test_sweep_enrich_imv_no_touched_ids_skips() -> None:
"""enrich_imv=True, но touched_house_ids пустой → IMV не вызывается."""
from app.services.scrape_pipeline import run_avito_city_sweep
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() # пустой!
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,
):
await run_avito_city_sweep(
mock_db,
run_id=3,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=True,
)
mock_imv_batch.assert_not_awaited()
# ── IMV-ошибки не валят sweep ────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_sweep_imv_crash_does_not_abort_sweep() -> None:
"""Если process_houses_imv_batch raises — sweep всё равно завершается (mark_done)."""
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"),
),
):
counters = await run_avito_city_sweep(
mock_db,
run_id=4,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=True,
)
mock_mark_done.assert_called_once()
# IMV crash → ошибка считается, но sweep завершается как done
assert counters.errors_count >= 1
# ── IMV counters накапливают failed ────────────────────────────────────────────
@pytest.mark.asyncio
async def test_sweep_imv_failed_counter() -> None:
"""imv_failed отражает errors из HouseIMVBackfillResult."""
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 = 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,
),
):
counters = await run_avito_city_sweep(
mock_db,
run_id=5,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=True,
)
assert counters.imv_attempted == 3
assert counters.imv_enriched == 1
assert counters.imv_failed == 2
# errors_count должен включить imv_failed
assert counters.errors_count >= 2
# ── Cancel перед IMV-фазой ────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_sweep_cancel_before_imv_skips_imv() -> None:
"""Если is_cancelled=True в IMV-фазе → IMV пропускается, sweep всё равно done."""
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):
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,
):
await run_avito_city_sweep(
mock_db,
run_id=6,
anchors=[(56.84, 60.6, "TestAnchor")],
enrich_imv=True,
)
mock_imv_batch.assert_not_awaited()
mock_mark_done.assert_called_once()
# ── process_houses_imv_batch с пустым set ────────────────────────────────────
@pytest.mark.asyncio
async def test_process_houses_imv_batch_empty_ids() -> None:
"""process_houses_imv_batch с пустым house_ids → result.checked=0, быстрый return."""
from app.services.house_imv_backfill import process_houses_imv_batch
mock_db = MagicMock()
result = await process_houses_imv_batch(mock_db, set(), request_delay_sec=1.0)
assert result.checked == 0
assert result.saved == 0
assert result.errors == 0
# DB не должна вызываться при пустом наборе
mock_db.execute.assert_not_called()
# ── CitySweepStartRequest принимает enrich_imv ───────────────────────────────
def test_city_sweep_start_request_enrich_imv_default_true() -> None:
"""enrich_imv по умолчанию True."""
from app.api.v1.admin import CitySweepStartRequest
req = CitySweepStartRequest()
assert req.enrich_imv is True
def test_city_sweep_start_request_enrich_imv_false() -> None:
"""enrich_imv=False принимается."""
from app.api.v1.admin import CitySweepStartRequest
req = CitySweepStartRequest(enrich_imv=False)
assert req.enrich_imv is False
# ── API endpoint передаёт enrich_imv в sweep ────────────────────────────────
def test_sweep_endpoint_passes_enrich_imv(app_with_admin) -> None:
"""POST /scrape/avito-city-sweep с enrich_imv=False → sweep вызывается без IMV."""
with (
patch("app.services.scrape_runs.create_run", return_value=42),
patch(
"app.services.scrape_pipeline.run_avito_city_sweep",
new_callable=AsyncMock,
),
):
r = app_with_admin.post(
"/api/v1/admin/scrape/avito-city-sweep",
json={"pages_per_anchor": 2, "enrich_imv": False},
)
assert r.status_code == 200
body = r.json()
assert body["run_id"] == 42
@pytest.fixture
def app_with_admin():
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.api.v1 import admin as admin_module
from app.core.db import get_db
app = FastAPI()
app.include_router(admin_module.router, prefix="/api/v1/admin")
def fake_db():
yield MagicMock()
app.dependency_overrides[get_db] = fake_db
return TestClient(app)