diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 29297cf2..0997e4c6 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -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 → полный адрес с номером дома). + 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 = ( diff --git a/tradein-mvp/backend/app/services/scrapers/browser_fetcher.py b/tradein-mvp/backend/app/services/scrapers/browser_fetcher.py index b175041c..730415e2 100644 --- a/tradein-mvp/backend/app/services/scrapers/browser_fetcher.py +++ b/tradein-mvp/backend/app/services/scrapers/browser_fetcher.py @@ -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 diff --git a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py index 1103f875..c8e5463e 100644 --- a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py +++ b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py @@ -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()) diff --git a/tradein-mvp/backend/tests/test_yandex_city_sweep.py b/tradein-mvp/backend/tests/test_yandex_city_sweep.py index ca8ec585..3eeaa61a 100644 --- a/tradein-mvp/backend/tests/test_yandex_city_sweep.py +++ b/tradein-mvp/backend/tests/test_yandex_city_sweep.py @@ -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 diff --git a/tradein-mvp/backend/tests/test_yandex_realty_serp.py b/tradein-mvp/backend/tests/test_yandex_realty_serp.py index 6757846a..a74d6084 100644 --- a/tradein-mvp/backend/tests/test_yandex_realty_serp.py +++ b/tradein-mvp/backend/tests/test_yandex_realty_serp.py @@ -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}" + )