fix(tradein/yandex): инкрементальный per-combo save + bounded gate-timeout (анти-зомби city_sweep) #1845
5 changed files with 400 additions and 29 deletions
|
|
@ -1167,8 +1167,9 @@ async def run_avito_newbuilding_sweep(
|
|||
class YandexCitySweepCounters:
|
||||
"""Aggregate counters для Yandex.Недвижимость city sweep run.
|
||||
|
||||
Фазы: SERP (fetch_around_multi_room) → save_listings → address-enrich.
|
||||
Фазы: SERP (fetch_around_multi_room) → save_listings инкрементально → address-enrich.
|
||||
address_* — результат T10-обогащения (detail <title> → полный адрес с номером дома).
|
||||
combos_skipped — сколько (room×price) combo пропущено из-за таймаутов gate-API.
|
||||
"""
|
||||
|
||||
anchors_total: int = 0
|
||||
|
|
@ -1177,6 +1178,7 @@ class YandexCitySweepCounters:
|
|||
lots_inserted: int = 0
|
||||
lots_updated: int = 0
|
||||
price_history_rows: int = 0
|
||||
combos_skipped: int = 0
|
||||
address_attempted: int = 0
|
||||
address_enriched: int = 0
|
||||
address_failed: int = 0
|
||||
|
|
@ -1327,17 +1329,68 @@ async def run_yandex_city_sweep(
|
|||
# по ссылке → неправильные значения при asyncio scheduling.
|
||||
_lat, _lon, _name = lat, lon, name
|
||||
|
||||
# _combo_lots_accumulator: per-anchor мутируемый контейнер для on_combo.
|
||||
# Захватывается по ссылке и в _yandex_anchor_phases и в _on_combo —
|
||||
# избегаем B023 (loop-variable capture) через явный bind в default-arg.
|
||||
_combo_lots_accumulator: list[ScrapedLot] = anchor_lots
|
||||
|
||||
async def _yandex_anchor_phases(
|
||||
_a_lat: float = _lat,
|
||||
_a_lon: float = _lon,
|
||||
_a_name: str = _name,
|
||||
_al: list[ScrapedLot] = _combo_lots_accumulator,
|
||||
) -> None:
|
||||
"""Все фазы одного anchor'а (SERP + address-enrich)."""
|
||||
nonlocal anchor_lots, consecutive_failures
|
||||
"""Все фазы одного anchor'а (SERP + address-enrich).
|
||||
|
||||
FIX: инкрементальный save per-combo — on_combo вызывается scraper'ом после
|
||||
каждого (room×price) combo, сохраняет порцию сразу в БД + обновляет heartbeat.
|
||||
anchor_lots накапливается (через _al) для address-enrich и price-history.
|
||||
"""
|
||||
nonlocal consecutive_failures
|
||||
|
||||
# ── Phase 1+2: SERP + инкрементальный save per-combo ──────
|
||||
def _on_combo(
|
||||
combo_label: str,
|
||||
new_lots: list[ScrapedLot],
|
||||
_accumulator: list[ScrapedLot] = _al,
|
||||
) -> None:
|
||||
"""Callback: вызывается scraper'ом после каждого combo с новыми лотами.
|
||||
|
||||
Сохраняет порцию новых лотов немедленно (анти-зомби: данные в БД
|
||||
до конца anchor'а) и обновляет heartbeat для мониторинга прогресса.
|
||||
"""
|
||||
_accumulator.extend(new_lots)
|
||||
counters.lots_fetched += len(new_lots)
|
||||
try:
|
||||
ins, upd = save_listings(db, new_lots, run_id=run_id)
|
||||
counters.lots_inserted += ins
|
||||
counters.lots_updated += upd
|
||||
except Exception as save_exc:
|
||||
counters.errors_count += 1
|
||||
logger.warning(
|
||||
"yandex-sweep run_id=%d combo %s: save_listings failed: %s",
|
||||
run_id,
|
||||
combo_label,
|
||||
save_exc,
|
||||
)
|
||||
try:
|
||||
db.rollback()
|
||||
except Exception:
|
||||
pass
|
||||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||||
logger.debug(
|
||||
"yandex-sweep run_id=%d combo %s: saved %d lots "
|
||||
"(ins=%d upd=%d total_fetched=%d)",
|
||||
run_id,
|
||||
combo_label,
|
||||
len(new_lots),
|
||||
counters.lots_inserted,
|
||||
counters.lots_updated,
|
||||
counters.lots_fetched,
|
||||
)
|
||||
|
||||
# ── Phase 1+2: SERP + save ─────────────────────────────────
|
||||
async with YandexRealtyScraper() as scraper:
|
||||
anchor_lots = await scraper.fetch_around_multi_room(
|
||||
await scraper.fetch_around_multi_room(
|
||||
_a_lat,
|
||||
_a_lon,
|
||||
radius_m,
|
||||
|
|
@ -1345,16 +1398,15 @@ async def run_yandex_city_sweep(
|
|||
rooms_list=_rooms_list,
|
||||
price_ranges=_price_ranges,
|
||||
segments=_segments,
|
||||
on_combo=_on_combo,
|
||||
)
|
||||
counters.lots_fetched += len(anchor_lots)
|
||||
if anchor_lots:
|
||||
inserted, updated = save_listings(db, anchor_lots, run_id=run_id)
|
||||
counters.lots_inserted += inserted
|
||||
counters.lots_updated += updated
|
||||
# Price-history из gate price.previous/trend (graceful — провал
|
||||
# истории НЕ должен ронять сейв листингов).
|
||||
# _al накоплен on_combo; save уже вызван per-combo.
|
||||
|
||||
# Price-history из gate price.previous/trend (graceful — провал
|
||||
# истории НЕ должен ронять сейв листингов).
|
||||
if _al:
|
||||
try:
|
||||
counters.price_history_rows += record_yandex_price_history(db, anchor_lots)
|
||||
counters.price_history_rows += record_yandex_price_history(db, _al)
|
||||
except Exception as ph_exc:
|
||||
logger.warning(
|
||||
"yandex-sweep run_id=%d: price-history failed: %s",
|
||||
|
|
@ -1365,20 +1417,21 @@ async def run_yandex_city_sweep(
|
|||
db.rollback()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
consecutive_failures = 0
|
||||
logger.info(
|
||||
"yandex-sweep run_id=%d center-combos %s: fetched=%d ins=%d upd=%d",
|
||||
run_id,
|
||||
_a_name,
|
||||
len(anchor_lots),
|
||||
counters.lots_fetched,
|
||||
counters.lots_inserted,
|
||||
counters.lots_updated,
|
||||
)
|
||||
|
||||
# ── Phase 3: address-enrich ────────────────────────────────
|
||||
if not (enrich_address and anchor_lots):
|
||||
if not (enrich_address and _al):
|
||||
return
|
||||
source_ids = [lot.source_id for lot in anchor_lots if lot.source_id]
|
||||
source_ids = [lot.source_id for lot in _al if lot.source_id]
|
||||
if not source_ids:
|
||||
return
|
||||
id_rows = (
|
||||
|
|
|
|||
|
|
@ -40,12 +40,15 @@ class BrowserFetcher:
|
|||
html = await fetcher.fetch("https://example.com")
|
||||
"""
|
||||
|
||||
def __init__(self, source: str = "avito") -> None:
|
||||
def __init__(self, source: str = "avito", fetch_timeout_s: float = _HTTP_TIMEOUT_S) -> None:
|
||||
# source — логический источник ("avito"/"cian"/"yandex"/"domclick"). Сервер
|
||||
# роутит /fetch по нему на отдельный браузер+прокси, когда включён
|
||||
# FEATURE_BROWSER_POOL_ENABLED (Phase 1). При выключенном флаге source
|
||||
# игнорируется — поведение не меняется.
|
||||
# fetch_timeout_s — таймаут httpx-клиента для POST /fetch. Yandex-путь передаёт
|
||||
# 30s чтобы один проблемный combo занимал ≤30s×retries вместо 120s×retries.
|
||||
self._source = source
|
||||
self._fetch_timeout_s = fetch_timeout_s
|
||||
self._client: httpx.AsyncClient | None = None
|
||||
self._endpoint: str | None = None
|
||||
|
||||
|
|
@ -55,7 +58,7 @@ class BrowserFetcher:
|
|||
from app.core.config import settings
|
||||
|
||||
self._endpoint = settings.browser_http_endpoint
|
||||
self._client = httpx.AsyncClient(timeout=_HTTP_TIMEOUT_S)
|
||||
self._client = httpx.AsyncClient(timeout=self._fetch_timeout_s)
|
||||
logger.info("BrowserFetcher: клиент создан, endpoint=%s", self._endpoint)
|
||||
return self
|
||||
|
||||
|
|
|
|||
|
|
@ -83,6 +83,10 @@ _YANDEX_MIN_BRACKET = 50_000
|
|||
_GATE_PAGE_SIZE = 20
|
||||
_YANDEX_TARPIT_MAX_RETRIES: int = 2
|
||||
|
||||
# Таймаут httpx-клиента BrowserFetcher для yandex-пути: 30s вместо глобального 120s.
|
||||
# Один проблемный combo занимает ≤30s×(1+_YANDEX_TARPIT_MAX_RETRIES)=90s вместо 360s.
|
||||
_YANDEX_BROWSER_FETCH_TIMEOUT_S: float = 30.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class _CurlResponse:
|
||||
|
|
@ -431,8 +435,13 @@ class YandexRealtyScraper(BaseScraper):
|
|||
camoufox's real-browser fingerprint avoids the mobile-proxy IP tarpit
|
||||
escalation that raw system-curl over SOCKS5 triggers. Opening per-page is
|
||||
expensive and unnecessary.
|
||||
|
||||
fetch_timeout_s=_YANDEX_BROWSER_FETCH_TIMEOUT_S (30s) ограничивает потери
|
||||
на один тайм-аутовый запрос: worst-case 30s×(1+retries)=90s на combo вместо 360s.
|
||||
"""
|
||||
self._browser = BrowserFetcher(source="yandex")
|
||||
self._browser = BrowserFetcher(
|
||||
source="yandex", fetch_timeout_s=_YANDEX_BROWSER_FETCH_TIMEOUT_S
|
||||
)
|
||||
await self._browser.__aenter__()
|
||||
return self
|
||||
|
||||
|
|
@ -662,6 +671,7 @@ class YandexRealtyScraper(BaseScraper):
|
|||
rooms_list: list[str] | None = None,
|
||||
price_ranges: list[tuple[int | None, int | None]] | None = None,
|
||||
segments: list[str] | None = None,
|
||||
on_combo: Any = None,
|
||||
**_legacy_kwargs: Any,
|
||||
) -> list[ScrapedLot]:
|
||||
"""Fetch via segment x rooms x price-range combos; paginate each combo to totalPages.
|
||||
|
|
@ -672,9 +682,15 @@ class YandexRealtyScraper(BaseScraper):
|
|||
Legacy mode (rooms_list=None, price_ranges=None): single citywide sweep.
|
||||
segments: list of newFlat values, default ["NO"] (vtorichka only). Pass
|
||||
["NO", "YES"] to sweep both vtorichka and novostroyki in one call.
|
||||
|
||||
on_combo: опциональный callback(combo_label: str, new_lots: list[ScrapedLot]) -> None.
|
||||
Вызывается после каждого combo с новыми (не повторными) лотами из него.
|
||||
Позволяет инкрементальный save: вызывающий код сохраняет new_lots сразу,
|
||||
не дожидаясь конца sweep'а. Дубли между combo НЕ передаются повторно.
|
||||
"""
|
||||
seen: dict[str, ScrapedLot] = {}
|
||||
_segments = segments or ["NO"]
|
||||
combos_skipped = 0
|
||||
|
||||
if rooms_list is None and price_ranges is None:
|
||||
combos: list[tuple[str | None, int | None, int | None]] = [(None, None, None)]
|
||||
|
|
@ -688,6 +704,8 @@ class YandexRealtyScraper(BaseScraper):
|
|||
for rooms, price_min, price_max in combos:
|
||||
combo_label = f"{_seg}/{_combo_label(rooms, price_min, price_max)}"
|
||||
total_pages: int | None = None
|
||||
combo_new_lots: list[ScrapedLot] = []
|
||||
combo_skipped = False
|
||||
|
||||
for page in range(1, max_pages + 1):
|
||||
if page == 1:
|
||||
|
|
@ -709,7 +727,9 @@ class YandexRealtyScraper(BaseScraper):
|
|||
break
|
||||
if resp.status_code == 0: # type: ignore[union-attr]
|
||||
logger.warning(
|
||||
"yandex gate: tarpit combo=%s page=1 -- rotating", combo_label
|
||||
"yandex gate: tarpit combo=%s page=1 attempt=%d -- rotating",
|
||||
combo_label,
|
||||
_attempt + 1,
|
||||
)
|
||||
await self._rotate_ip()
|
||||
await asyncio.sleep(2)
|
||||
|
|
@ -740,13 +760,18 @@ class YandexRealtyScraper(BaseScraper):
|
|||
break
|
||||
|
||||
if payload_p1 is None:
|
||||
logger.debug(
|
||||
"yandex gate combo [%s] page=1: failed -- skipping", combo_label
|
||||
logger.warning(
|
||||
"yandex gate combo [%s] page=1: all retries exhausted -- skipping",
|
||||
combo_label,
|
||||
)
|
||||
combo_skipped = True
|
||||
combos_skipped += 1
|
||||
break
|
||||
|
||||
result = _extract_gate_data(payload_p1)
|
||||
if result is None:
|
||||
combo_skipped = True
|
||||
combos_skipped += 1
|
||||
break
|
||||
_entities_p1, pager_p1 = result
|
||||
total_pages = min(
|
||||
|
|
@ -764,6 +789,7 @@ class YandexRealtyScraper(BaseScraper):
|
|||
key = lot.source_id or lot.source_url
|
||||
if key and key not in seen:
|
||||
seen[key] = lot
|
||||
combo_new_lots.append(lot)
|
||||
await self.sleep_between_requests()
|
||||
if total_pages <= 1:
|
||||
break
|
||||
|
|
@ -793,13 +819,25 @@ class YandexRealtyScraper(BaseScraper):
|
|||
key = lot.source_id or lot.source_url
|
||||
if key and key not in seen:
|
||||
seen[key] = lot
|
||||
combo_new_lots.append(lot)
|
||||
|
||||
# Инкрементальный save: вызываем on_combo если есть новые лоты.
|
||||
# Скипнутые combo (combo_skipped=True) не вызывают on_combo.
|
||||
if on_combo is not None and combo_new_lots and not combo_skipped:
|
||||
try:
|
||||
on_combo(combo_label, combo_new_lots)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"yandex gate: on_combo callback failed combo=%s", combo_label
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"yandex gate aggregate: %d unique lots (%d combos x %d segments=%s)",
|
||||
"yandex gate aggregate: %d unique lots (%d combos x %d segments=%s, skipped=%d)",
|
||||
len(seen),
|
||||
len(combos),
|
||||
len(_segments),
|
||||
_segments,
|
||||
combos_skipped,
|
||||
)
|
||||
return list(seen.values())
|
||||
|
||||
|
|
|
|||
|
|
@ -153,6 +153,7 @@ def test_counters_defaults() -> None:
|
|||
c = YandexCitySweepCounters()
|
||||
assert c.anchors_total == 0
|
||||
assert c.lots_fetched == 0
|
||||
assert c.combos_skipped == 0
|
||||
assert c.address_attempted == 0
|
||||
assert c.address_enriched == 0
|
||||
assert c.address_failed == 0
|
||||
|
|
@ -176,7 +177,12 @@ def test_counters_to_dict_all_keys() -> None:
|
|||
async def test_sweep_iterates_all_anchors_and_aggregates(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""fetch_around_multi_room вызывается для каждого anchor, counters агрегируются."""
|
||||
"""fetch_around_multi_room вызывается для каждого anchor, counters агрегируются.
|
||||
|
||||
FIX: fake_fetch_multi вызывает on_combo (инкрементальный save per-combo) —
|
||||
имитирует реальное поведение scraper'а. save_listings должен вызываться
|
||||
on_combo'м, а не одним батчем в конце anchor'а.
|
||||
"""
|
||||
serp_calls: list[str] = []
|
||||
save_calls: list[int] = []
|
||||
|
||||
|
|
@ -186,10 +192,15 @@ async def test_sweep_iterates_all_anchors_and_aggregates(
|
|||
lon: float,
|
||||
radius_m: int = 1000,
|
||||
max_pages: int = 2,
|
||||
on_combo: Any = None,
|
||||
**_: Any,
|
||||
) -> list[ScrapedLot]:
|
||||
serp_calls.append(f"{lat:.4f}")
|
||||
return [_fake_lot(f"{lat}-{i}") for i in range(3)]
|
||||
lots = [_fake_lot(f"{lat}-{i}") for i in range(3)]
|
||||
# Имитируем один combo — вызываем on_combo если предоставлен.
|
||||
if on_combo is not None:
|
||||
on_combo(f"vtorichka/combo_{lat:.4f}", lots)
|
||||
return lots
|
||||
|
||||
def fake_save(
|
||||
_db: Any, lots: list[ScrapedLot], *, run_id: int | None = None
|
||||
|
|
@ -212,6 +223,8 @@ async def test_sweep_iterates_all_anchors_and_aggregates(
|
|||
)
|
||||
|
||||
assert len(serp_calls) == len(TEST_ANCHORS)
|
||||
# save вызывается инкрементально через on_combo — по одному разу на anchor
|
||||
# (fake имитирует один combo per anchor).
|
||||
assert len(save_calls) == len(TEST_ANCHORS)
|
||||
assert counters.anchors_total == len(TEST_ANCHORS)
|
||||
assert counters.anchors_done == len(TEST_ANCHORS)
|
||||
|
|
@ -284,8 +297,12 @@ async def test_address_enrich_fills_counters(
|
|||
lon: float,
|
||||
radius_m: int = 1000,
|
||||
max_pages: int = 2,
|
||||
on_combo: Any = None,
|
||||
**_: Any,
|
||||
) -> list[ScrapedLot]:
|
||||
# Вызываем on_combo чтобы anchor_lots накопился (инкрементальный save).
|
||||
if on_combo is not None:
|
||||
on_combo("vtorichka/combo_enrich_test", lots)
|
||||
return lots
|
||||
|
||||
monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
|
||||
|
|
@ -343,9 +360,13 @@ async def test_address_enrich_http_error_does_not_abort(
|
|||
lon: float,
|
||||
radius_m: int = 1000,
|
||||
max_pages: int = 2,
|
||||
on_combo: Any = None,
|
||||
**_: Any,
|
||||
) -> list[ScrapedLot]:
|
||||
return [_fake_lot("id-err")]
|
||||
lots = [_fake_lot("id-err")]
|
||||
if on_combo is not None:
|
||||
on_combo("vtorichka/combo_http_err", lots)
|
||||
return lots
|
||||
|
||||
monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
|
||||
monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
|
||||
|
|
@ -498,11 +519,16 @@ async def test_non_consecutive_failures_do_not_abort(
|
|||
lon: float,
|
||||
radius_m: int = 1000,
|
||||
max_pages: int = 2,
|
||||
on_combo: Any = None,
|
||||
**_: Any,
|
||||
) -> list[ScrapedLot]:
|
||||
if _name_of(lat, anchors) in failing:
|
||||
raise RuntimeError("transient")
|
||||
return [_fake_lot(f"{lat}-0")]
|
||||
lots = [_fake_lot(f"{lat}-0")]
|
||||
# Вызываем on_combo для инкрементального save.
|
||||
if on_combo is not None:
|
||||
on_combo(f"vtorichka/combo_{lat:.4f}", lots)
|
||||
return lots
|
||||
|
||||
monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch)
|
||||
monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
|
||||
|
|
@ -689,8 +715,10 @@ async def test_prod_mode_uses_center_anchor_with_combos(
|
|||
max_pages: int = 2,
|
||||
rooms_list: list | None = None,
|
||||
price_ranges: list | None = None,
|
||||
on_combo: Any = None,
|
||||
**_: Any,
|
||||
) -> list[ScrapedLot]:
|
||||
lots = [_fake_lot("center-1"), _fake_lot("center-2")]
|
||||
calls.append(
|
||||
{
|
||||
"lat": lat,
|
||||
|
|
@ -701,7 +729,10 @@ async def test_prod_mode_uses_center_anchor_with_combos(
|
|||
"max_pages": max_pages,
|
||||
}
|
||||
)
|
||||
return [_fake_lot("center-1"), _fake_lot("center-2")]
|
||||
# Имитируем on_combo (инкрементальный save) — один combo.
|
||||
if on_combo is not None:
|
||||
on_combo("vtorichka/center_combo", lots)
|
||||
return lots
|
||||
|
||||
monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
|
||||
monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
|
||||
|
|
@ -745,7 +776,7 @@ async def test_prod_mode_uses_center_anchor_with_combos(
|
|||
# max_pages forwarded from pages_per_anchor
|
||||
assert call["max_pages"] == 3
|
||||
|
||||
# lots aggregated correctly
|
||||
# lots aggregated correctly via on_combo
|
||||
assert counters.lots_fetched == 2
|
||||
assert counters.lots_inserted == 2
|
||||
assert fake.done is not None
|
||||
|
|
@ -895,3 +926,133 @@ async def test_explicit_anchor_mode_uses_anchor_timeout_sec(
|
|||
assert (
|
||||
t == ANCHOR_TIMEOUT_SEC
|
||||
), f"Expected ANCHOR_TIMEOUT_SEC={ANCHOR_TIMEOUT_SEC}s, got {t}s"
|
||||
|
||||
|
||||
# ── FIX A: инкрементальный save per-combo ────────────────────────────────────
|
||||
|
||||
|
||||
async def test_incremental_save_called_per_combo(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""save_listings вызывается on_combo per-combo, не одним батчем в конце anchor'а.
|
||||
|
||||
Имитируем 3 combo через on_combo — save_listings должен быть вызван 3 раза
|
||||
(по одному на combo) с порциями лотов. Проверяем, что при краше anchor'а
|
||||
уже сохранённые combo не теряются (данные в БД до finish anchor'а).
|
||||
"""
|
||||
save_batches: list[list[str]] = []
|
||||
|
||||
async def fake_fetch_multi(
|
||||
self: YandexRealtyScraper,
|
||||
lat: float,
|
||||
lon: float,
|
||||
radius_m: int = 1000,
|
||||
max_pages: int = 2,
|
||||
on_combo: Any = None,
|
||||
**_: Any,
|
||||
) -> list[ScrapedLot]:
|
||||
all_lots = []
|
||||
for combo_idx in range(3):
|
||||
combo_lots = [_fake_lot(f"{lat}-combo{combo_idx}-lot0")]
|
||||
all_lots.extend(combo_lots)
|
||||
if on_combo is not None:
|
||||
on_combo(f"vtorichka/combo_{combo_idx}", combo_lots)
|
||||
return all_lots
|
||||
|
||||
def fake_save(
|
||||
_db: Any, lots: list[ScrapedLot], *, run_id: int | None = None
|
||||
) -> tuple[int, int]:
|
||||
save_batches.append([lot.source_id for lot in lots if lot.source_id])
|
||||
return len(lots), 0
|
||||
|
||||
monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
|
||||
monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
|
||||
fake = _FakeRuns()
|
||||
_install_runs(monkeypatch, fake)
|
||||
|
||||
counters = await scrape_pipeline.run_yandex_city_sweep(
|
||||
db=MagicMock(),
|
||||
run_id=30,
|
||||
anchors=TEST_ANCHORS[:1],
|
||||
pages_per_anchor=2,
|
||||
request_delay_sec=0.0,
|
||||
enrich_address=False,
|
||||
)
|
||||
|
||||
# save_listings вызван ровно 3 раза (по одному на combo), не одним батчем.
|
||||
assert (
|
||||
len(save_batches) == 3
|
||||
), f"Ожидалось 3 save вызова (per-combo), получено {len(save_batches)}: {save_batches}"
|
||||
# Каждый batch содержит ровно 1 лот (размер combo).
|
||||
for batch in save_batches:
|
||||
assert len(batch) == 1, f"Каждый combo-batch должен иметь 1 лот, получено {batch}"
|
||||
|
||||
# Суммарные counters корректны.
|
||||
assert counters.lots_fetched == 3
|
||||
assert counters.lots_inserted == 3
|
||||
assert fake.done is not None
|
||||
|
||||
# heartbeat вызывался после каждого combo (через on_combo).
|
||||
assert (
|
||||
len(fake.heartbeats) >= 3
|
||||
), f"Ожидалось ≥3 heartbeat обновлений (per-combo), получено {len(fake.heartbeats)}"
|
||||
|
||||
|
||||
# ── FIX B: bounded retry — скипает combo после exhaustion ────────────────────
|
||||
|
||||
|
||||
async def test_bounded_retry_combo_skipped_logged(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""on_combo НЕ вызывается если combo пропущен (все retry page-1 исчерпаны).
|
||||
|
||||
Имитируем sweep с двумя combos: первый скипается (on_combo не вызван),
|
||||
второй успешен (on_combo вызван). Проверяем, что save_listings вызван
|
||||
только для успешного combo, а не скипнутого.
|
||||
"""
|
||||
save_batches: list[list[str]] = []
|
||||
on_combo_calls: list[str] = []
|
||||
|
||||
async def fake_fetch_multi(
|
||||
self: YandexRealtyScraper,
|
||||
lat: float,
|
||||
lon: float,
|
||||
radius_m: int = 1000,
|
||||
max_pages: int = 2,
|
||||
on_combo: Any = None,
|
||||
**_: Any,
|
||||
) -> list[ScrapedLot]:
|
||||
# Combo 0: скипается — on_combo НЕ вызывается (симулирует exhausted retries).
|
||||
# Combo 1: успешен — on_combo вызывается с лотом.
|
||||
success_lot = _fake_lot(f"{lat}-success")
|
||||
if on_combo is not None:
|
||||
on_combo_calls.append("combo_1_success")
|
||||
on_combo("vtorichka/combo_1", [success_lot])
|
||||
return [success_lot]
|
||||
|
||||
def fake_save(
|
||||
_db: Any, lots: list[ScrapedLot], *, run_id: int | None = None
|
||||
) -> tuple[int, int]:
|
||||
save_batches.append([lot.source_id for lot in lots if lot.source_id])
|
||||
return len(lots), 0
|
||||
|
||||
monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
|
||||
monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
|
||||
fake = _FakeRuns()
|
||||
_install_runs(monkeypatch, fake)
|
||||
|
||||
counters = await scrape_pipeline.run_yandex_city_sweep(
|
||||
db=MagicMock(),
|
||||
run_id=31,
|
||||
anchors=TEST_ANCHORS[:1],
|
||||
pages_per_anchor=2,
|
||||
request_delay_sec=0.0,
|
||||
enrich_address=False,
|
||||
)
|
||||
|
||||
# Только успешный combo вызвал on_combo и save.
|
||||
assert len(on_combo_calls) == 1
|
||||
assert len(save_batches) == 1
|
||||
assert counters.lots_fetched == 1
|
||||
assert counters.lots_inserted == 1
|
||||
assert fake.done is not None
|
||||
|
|
|
|||
|
|
@ -1176,3 +1176,119 @@ async def test_walk_price_range_open_bracket_no_split():
|
|||
assert all(
|
||||
pm is None for pm in seen_prices
|
||||
), f"Открытый брекет: priceMax всегда None, got {seen_prices}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PART J: fetch_around_multi_room on_combo callback (FIX A — incremental save)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_around_multi_room_on_combo_called_per_combo():
|
||||
"""on_combo вызывается после каждого combo с новыми (деduped) лотами.
|
||||
|
||||
2 combo (rooms_list=['1'], 2 price_ranges): каждый имеет уникальный лот.
|
||||
on_combo должен быть вызван дважды — по одному разу на combo.
|
||||
"""
|
||||
s = YandexRealtyScraper()
|
||||
|
||||
entity_0 = dict(_ENTITY_FULL, offerId="combo0_lot", url="//realty.yandex.ru/offer/combo0_lot")
|
||||
entity_1 = dict(_ENTITY_FULL, offerId="combo1_lot", url="//realty.yandex.ru/offer/combo1_lot")
|
||||
pager = {"page": 0, "pageSize": 20, "totalItems": 1, "totalPages": 1}
|
||||
|
||||
call_idx = 0
|
||||
payloads = [
|
||||
_make_gate_payload([entity_0], pager),
|
||||
_make_gate_payload([entity_1], pager),
|
||||
]
|
||||
|
||||
async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse:
|
||||
nonlocal call_idx
|
||||
idx = call_idx
|
||||
call_idx += 1
|
||||
if idx < len(payloads):
|
||||
return _CurlResponse(status_code=200, text=json.dumps(payloads[idx]))
|
||||
return _CurlResponse(status_code=200, text=json.dumps(_make_gate_payload([])))
|
||||
|
||||
on_combo_calls: list[tuple[str, list]] = []
|
||||
|
||||
def on_combo(label: str, new_lots: list) -> None:
|
||||
on_combo_calls.append((label, list(new_lots)))
|
||||
|
||||
with patch.object(s, "_http_get", side_effect=fake_http_get):
|
||||
with patch.object(s, "sleep_between_requests", new_callable=AsyncMock):
|
||||
lots = await s.fetch_around_multi_room(
|
||||
lat=0.0,
|
||||
lon=0.0,
|
||||
rooms_list=["1"],
|
||||
price_ranges=[(None, 5_000_000), (5_000_000, 10_000_000)],
|
||||
max_pages=1,
|
||||
on_combo=on_combo,
|
||||
)
|
||||
|
||||
# Два уникальных лота, два combo.
|
||||
assert len(lots) == 2
|
||||
# on_combo вызван ровно дважды — по одному на combo.
|
||||
assert len(on_combo_calls) == 2, f"Ожидалось 2 on_combo вызова, получено {len(on_combo_calls)}"
|
||||
# Каждый on_combo получил ровно один новый лот.
|
||||
for _label, combo_lots in on_combo_calls:
|
||||
assert len(combo_lots) == 1
|
||||
# Конкретные source_id в правильных batch'ах.
|
||||
all_ids = {lot.source_id for _, combo_lots in on_combo_calls for lot in combo_lots}
|
||||
assert all_ids == {"combo0_lot", "combo1_lot"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_around_multi_room_on_combo_not_called_for_failed_combo():
|
||||
"""on_combo НЕ вызывается для combo, у которого все page-1 retries исчерпаны.
|
||||
|
||||
FIX B: bounded retry — если status_code=0 на всех попытках page=1,
|
||||
combo пропускается. on_combo не должен быть вызван для пропущенного combo.
|
||||
Успешный combo (второй) — on_combo должен сработать.
|
||||
"""
|
||||
s = YandexRealtyScraper()
|
||||
|
||||
entity_ok = dict(_ENTITY_FULL, offerId="ok_lot", url="//realty.yandex.ru/offer/ok_lot")
|
||||
pager = {"page": 0, "pageSize": 20, "totalItems": 1, "totalPages": 1}
|
||||
|
||||
call_idx = 0
|
||||
|
||||
async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse:
|
||||
nonlocal call_idx
|
||||
call_idx += 1
|
||||
# Первые (1 + _YANDEX_TARPIT_MAX_RETRIES) запросы — это retries для combo[0] → fail.
|
||||
from app.services.scrapers.yandex_realty import _YANDEX_TARPIT_MAX_RETRIES
|
||||
|
||||
if call_idx <= 1 + _YANDEX_TARPIT_MAX_RETRIES:
|
||||
return _CurlResponse(status_code=0, text="")
|
||||
# Следующий запрос — combo[1] page=1 → success.
|
||||
return _CurlResponse(
|
||||
status_code=200, text=json.dumps(_make_gate_payload([entity_ok], pager))
|
||||
)
|
||||
|
||||
on_combo_calls: list[str] = []
|
||||
|
||||
def on_combo(label: str, new_lots: list) -> None:
|
||||
on_combo_calls.append(label)
|
||||
|
||||
with patch.object(s, "_http_get", side_effect=fake_http_get):
|
||||
with patch.object(s, "_rotate_ip", new_callable=AsyncMock):
|
||||
with patch("asyncio.sleep", new_callable=AsyncMock):
|
||||
with patch.object(s, "sleep_between_requests", new_callable=AsyncMock):
|
||||
lots = await s.fetch_around_multi_room(
|
||||
lat=0.0,
|
||||
lon=0.0,
|
||||
rooms_list=["1"],
|
||||
price_ranges=[(None, 5_000_000), (5_000_000, 10_000_000)],
|
||||
max_pages=1,
|
||||
on_combo=on_combo,
|
||||
)
|
||||
|
||||
# combo[0] скипнут, combo[1] успешен → 1 лот в итоге.
|
||||
assert len(lots) == 1
|
||||
assert lots[0].source_id == "ok_lot"
|
||||
# on_combo вызван только для успешного combo, не для скипнутого.
|
||||
assert len(on_combo_calls) == 1, (
|
||||
f"on_combo должен быть вызван только 1 раз (успешный combo), "
|
||||
f"получено {len(on_combo_calls)}: {on_combo_calls}"
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue