tradein-mvp/backend/app/api/v1/admin.py

@@ -29,6 +29,7 @@ from app.services.scrape_pipeline import (
     run_cian_full_load,
     run_n1_city_sweep,
     run_yandex_city_sweep,
+    run_yandex_full_load,
 )
 from app.services.scraper_settings import invalidate_cache
 from app.services.scrapers.avito import AvitoScraper
@@ -1042,6 +1043,93 @@ def cancel_cian_full_load(
     return {"ok": True, "run_id": run_id, "cancelled": cancelled}
 
 
+# ── Yandex exhaustive full load (room × price partitioning) ──────────────────
+
+
+class YandexFullLoadRequest(BaseModel):
+    price_cap_per_bucket: int = Field(
+        default=500,
+        ge=100,
+        le=575,
+        description=(
+            "Максимум офферов на price-бакет. При totalItems > cap — бакет делится пополам. "
+            "Ниже Yandex SERP-cap ~575; запас на variance. "
+            "При недоступности totalItems — fallback на paginate-until-empty."
+        ),
+    )
+    request_delay_sec: float = Field(
+        default=2.0,
+        ge=1.0,
+        le=15.0,
+        description=(
+            "Задержка между SERP-запросами (сек). "
+            "Рекомендуется ≥ 2.0 — Yandex агрессивнее Cian по captcha-детекции."
+        ),
+    )
+    concurrency: int = Field(
+        default=4,
+        ge=1,
+        le=8,
+        description="Параллельных page-фетчей внутри одного leaf-бакета.",
+    )
+
+
+@router.post("/scrape/yandex-full-load", response_model=CitySweepStartResponse)
+async def start_yandex_full_load(
+    payload: YandexFullLoadRequest,
+    background_tasks: BackgroundTasks,
+    db: Annotated[Session, Depends(get_db)],
+) -> CitySweepStartResponse:
+    """Запустить exhaustive Yandex ЕКБ full load в background.
+
+    Обходит Yandex SERP-cap (~575 карточек/запрос): адаптивно бьёт цену на бакеты
+    внутри каждой комнатности, пагинирует каждый бакет полностью, дедуп по source_id.
+    При недоступности totalItems из state — автоматический fallback на paginate-until-empty.
+
+    Один региональный проход БЕЗ anchor'ов (Yandex SERP path-based по городу — anchors
+    избыточны). IP-ротация через changeip при captcha.
+
+    Coop cancel: POST /scrape/yandex-full-load/{run_id}/cancel.
+    Статус прогона: смотреть через GET /scrape/cian-city-sweep/runs
+    (source='yandex_full_load').
+    """
+    run_id = runs_mod.create_run(db, source="yandex_full_load", params=payload.model_dump())
+
+    async def _full_load_task() -> None:
+        task_db = SessionLocal()
+        try:
+            await run_yandex_full_load(
+                task_db,
+                run_id=run_id,
+                price_cap_per_bucket=payload.price_cap_per_bucket,
+                request_delay_sec=payload.request_delay_sec,
+                concurrency=payload.concurrency,
+            )
+        except Exception:
+            logger.exception("yandex-full-load background task run_id=%d crashed", run_id)
+        finally:
+            task_db.close()
+
+    background_tasks.add_task(_full_load_task)
+    logger.info("yandex-full-load queued run_id=%d params=%s", run_id, payload.model_dump())
+    return CitySweepStartResponse(
+        run_id=run_id,
+        status="running",
+        pages_per_anchor=0,  # нет anchor'ов в full load
+        detail_top_n=0,
+    )
+
+
+@router.post("/scrape/yandex-full-load/{run_id}/cancel")
+def cancel_yandex_full_load(
+    run_id: int,
+    db: Annotated[Session, Depends(get_db)],
+) -> dict[str, object]:
+    """Отменить running Yandex full load. Cooperative: проверяется каждый room-bucket."""
+    cancelled = runs_mod.mark_cancelled(db, run_id)
+    return {"ok": True, "run_id": run_id, "cancelled": cancelled}
+
+
 # ── Yandex city sweep (#861) — on-demand trigger + runs list ─────────────────
 
 

tradein-mvp/backend/app/core/config.py

@@ -213,6 +213,20 @@ class Settings(BaseSettings):
         """Прокси для Cian-скраперов. CIAN_PROXY_URL > scraper_proxy_url (fallback)."""
         return self.cian_proxy_url_env or self.scraper_proxy_url
 
+    # ── Yandex dedicated mobile proxy (separate egress from Avito/Cian) ────────
+    # Отдельный прокси для Yandex устраняет contention при параллельных прогонах.
+    # Если не задан — fallback на общий scraper_proxy_url (backward-compat).
+    # ENV: YANDEX_PROXY_URL.
+    yandex_proxy_url_env: str | None = Field(default=None, validation_alias="YANDEX_PROXY_URL")
+    # changeip-ссылка для Yandex-прокси (ротация IP при капче/таймауте). Если не задан —
+    # fallback на avito_proxy_rotate_url. ENV: YANDEX_PROXY_ROTATE_URL.
+    yandex_proxy_rotate_url: str | None = None
+
+    @property
+    def yandex_proxy_url(self) -> str | None:
+        """Прокси для Yandex-скраперов. YANDEX_PROXY_URL > scraper_proxy_url (fallback)."""
+        return self.yandex_proxy_url_env or self.scraper_proxy_url
+
     # ── #759: deactivate stale avito listings ───────────────────────────────
     # Avito-объявления, не виденные scraper'ом более avito_stale_ttl_days дней,
     # помечаются is_active=false ночной задачей deactivate_stale_avito_listings.

tradein-mvp/backend/app/services/scrape_pipeline.py

@@ -879,7 +879,7 @@ async def run_yandex_city_sweep(
                                 await asyncio.sleep(enrich_delay)
 
                     logger.info(
-                        "yandex-sweep run_id=%d anchor %s: " "address enrich=%d/%d failed=%d",
+                        "yandex-sweep run_id=%d anchor %s: address enrich=%d/%d failed=%d",
                         run_id,
                         _a_name,
                         counters.address_enriched,
@@ -1722,6 +1722,129 @@ async def run_cian_full_load(
         raise
 
 
+# ── Yandex exhaustive full load (region-wide, без anchor'ов) ──────────────────
+
+
+@dataclass
+class YandexFullLoadCounters:
+    """Aggregate counters для run_yandex_full_load.
+
+    Без detail_* полей — Yandex full load не делает detail-обогащение.
+    """
+
+    unique_fetched: int = 0
+    saved_inserted: int = 0
+    saved_updated: int = 0
+    errors_count: int = 0
+
+    def to_dict(self) -> dict[str, int]:
+        return {f.name: getattr(self, f.name) for f in fields(self)}
+
+
+async def run_yandex_full_load(
+    db: Session,
+    *,
+    run_id: int,
+    price_cap_per_bucket: int = 500,
+    request_delay_sec: float = 2.0,
+    concurrency: int = 4,
+) -> YandexFullLoadCounters:
+    """Exhaustive региональный сбор Yandex ЕКБ вторички (БЕЗ anchor'ов).
+
+    Yandex SERP — path-based по городу, не geo-bbox. Anchor-loop избыточен
+    (все anchor'ы дают одну и ту же городскую выдачу). Один проход с
+    партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление)
+    охватывает весь ЕКБ.
+
+    Параметры:
+      price_cap_per_bucket: максимум офферов на price-бакет (< Yandex SERP-cap ~575).
+      request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default).
+      concurrency: параллельных page-фетчей в leaf-бакете.
+
+    Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора.
+    Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket.
+    """
+    from app.services.scrapers.yandex_realty import YandexRealtyScraper
+
+    counters = YandexFullLoadCounters()
+
+    def _on_bucket(lots: list) -> None:  # type: ignore[type-arg]
+        """Инкрементальный save после каждого leaf-бакета."""
+        if scrape_runs.is_cancelled(db, run_id):
+            logger.info("yandex-full-load run_id=%d: cancel detected in on_bucket", run_id)
+            raise RuntimeError("cancelled")
+        if not lots:
+            return
+        inserted, updated = save_listings(db, lots, run_id=run_id)
+        # save_listings вызывает db.commit() внутри — данные в БД сразу
+        counters.saved_inserted += inserted
+        counters.saved_updated += updated
+        counters.unique_fetched += len(lots)
+        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+        logger.info(
+            "yandex-full-load run_id=%d: bucket saved ins=%d upd=%d total_unique=%d",
+            run_id,
+            inserted,
+            updated,
+            counters.unique_fetched,
+        )
+
+    def _on_progress(unique_count: int) -> None:
+        """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов)."""
+        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+
+    try:
+        async with YandexRealtyScraper() as scraper:
+            scraper.request_delay_sec = request_delay_sec
+
+            await scraper.fetch_all_secondary(
+                price_cap_per_bucket=price_cap_per_bucket,
+                concurrency=concurrency,
+                on_bucket=_on_bucket,
+                on_progress=_on_progress,
+            )
+
+        logger.info(
+            "yandex-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d",
+            run_id,
+            counters.unique_fetched,
+            counters.saved_inserted,
+            counters.saved_updated,
+        )
+        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+        scrape_runs.mark_done(db, run_id, counters.to_dict())
+        logger.info(
+            "yandex-full-load run_id=%d done: unique=%d ins=%d upd=%d errors=%d",
+            run_id,
+            counters.unique_fetched,
+            counters.saved_inserted,
+            counters.saved_updated,
+            counters.errors_count,
+        )
+        return counters
+
+    except RuntimeError as exc:
+        # on_bucket кидает RuntimeError("cancelled") при cooperative cancel
+        if str(exc) == "cancelled":
+            logger.info(
+                "yandex-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d",
+                run_id,
+                counters.unique_fetched,
+                counters.saved_inserted,
+                counters.saved_updated,
+            )
+            scrape_runs.mark_done(db, run_id, counters.to_dict())
+            return counters
+        logger.exception("yandex-full-load run_id=%d: fatal error", run_id)
+        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
+        raise
+
+    except Exception as exc:
+        logger.exception("yandex-full-load run_id=%d: fatal error", run_id)
+        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
+        raise
+
+
 # ── Public re-exports ───────────────────────────────────────────
 __all__ = [
     "ANCHOR_TIMEOUT_SEC",
@@ -1733,9 +1856,11 @@ __all__ = [
     "PipelineCounters",
     "PipelineResult",
     "YandexCitySweepCounters",
+    "YandexFullLoadCounters",
     "run_avito_city_sweep",
     "run_avito_pipeline",
     "run_cian_city_sweep",
     "run_cian_full_load",
     "run_yandex_city_sweep",
+    "run_yandex_full_load",
 ]

tradein-mvp/backend/app/services/scrapers/yandex_realty.py

@@ -21,8 +21,11 @@ URL pattern (path-based, no geo radius):
 
 from __future__ import annotations
 
+import asyncio
+import inspect
 import json
 import logging
+import math
 import re
 from datetime import date  # noqa: F401  (used in type hints / future helpers)
 from pathlib import Path
@@ -94,6 +97,34 @@ DEFAULT_PRICE_RANGES: list[tuple[int | None, int | None]] = [
     (25_000_000, None),
 ]
 
+# ── Константы exhaustive-загрузки ──────────────────────────────────────────────
+
+# Верхняя граница цены при первом рекурсивном делении (нет явного hi).
+# 200 млн ₽ заведомо выше ТОПа ЕКБ — дальние бакеты дадут 0 офферов.
+_YANDEX_MAX_PRICE = 200_000_000
+
+# Минимальный ценовой диапазон для рекурсии; при hi - lo < MIN_BRACKET
+# прекращаем деление и пагинируем как есть.
+_YANDEX_MIN_BRACKET = 50_000
+
+# Оценка офферов на страницу Yandex SERP.
+# Yandex показывает ~20-30 карточек на страницу (зависит от типа выдачи).
+# Консервативно 30 — гарантирует, что мы не срежем страницы в leaf-bucket.
+_YANDEX_OFFERS_PER_PAGE = 30
+
+# Пути в JSON-state для поиска totalItems (порядок = приоритет).
+_TOTAL_COUNT_PATHS: list[list[str]] = [
+    ["pageData", "pager", "totalItems"],
+    ["pageData", "pager", "total"],
+    ["searchResults", "pager", "totalItems"],
+    ["pager", "totalItems"],
+    ["pager", "total"],
+    ["pageData", "searchQuery", "total"],
+]
+
+# Regex для fallback-поиска количества офферов в тексте DOM.
+_RE_TOTAL_COUNT = re.compile(r"(\d[\d ]{0,8})\s+(?:объявлен|квартир|предложен)", re.IGNORECASE)
+
 
 def _load_cookies_from_file(path: str | None) -> dict[str, str]:
     """Load Yandex cookies from JSON file path (Netscape/browser export format).
@@ -244,8 +275,9 @@ class YandexRealtyScraper(BaseScraper):
         self._cookies = _load_cookies_from_file(settings.yandex_cookies_file)
         # Mobile proxy wiring (#806 follow-up): Yandex blocks datacenter IPs with
         # captcha/shell HTML even with Chrome TLS fingerprint. Route via mobile proxy.
+        # Dedicated YANDEX_PROXY_URL > общий scraper_proxy_url (backward-compat).
         # proxy=None → curl_cffi ходит напрямую (dev без прокси — не падаем).
-        _proxy_url = settings.scraper_proxy_url
+        _proxy_url = settings.yandex_proxy_url
         _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
         self._cffi_session = _CurlCffiSession(
             impersonate="chrome120",
@@ -278,6 +310,404 @@ class YandexRealtyScraper(BaseScraper):
         kwargs.setdefault("timeout", 30)
         return await self._cffi_session.get(url, **kwargs)
 
+    # ── Anti-block: IP rotation (зеркало cian.py _rotate_ip) ──────────────────
+
+    async def _rotate_ip(self) -> bool:
+        """Сменить мобильный IP через changeip-ссылку mobileproxy.
+
+        Дёргается напрямую (без прокси) — это API провайдера, не Yandex. Ждём ~9с:
+        мобильному модему нужно время поднять новый IP. Returns True при успехе.
+        Зеркало CianScraper._rotate_ip.
+        """
+        from app.core.config import settings as _settings
+
+        rotate_url = _settings.yandex_proxy_rotate_url or _settings.avito_proxy_rotate_url
+        if not rotate_url:
+            return False
+        sep = "&" if "?" in rotate_url else "?"
+        try:
+            async with _CurlCffiSession(timeout=30) as rot:
+                await rot.get(f"{rotate_url}{sep}format=json")
+            await asyncio.sleep(9)
+            logger.info("yandex proxy: IP rotated via changeip")
+            return True
+        except Exception:
+            logger.warning("yandex proxy: IP rotation failed", exc_info=True)
+            return False
+
+    # ── Total-count extraction (UNVERIFIED — see fallback note) ────────────────
+
+    def _extract_total_count(self, html: str) -> int | None:
+        """Извлечь totalItems из embedded JSON state Yandex SERP.
+
+        CAVEAT: Yandex state is chunked + evaporates after hydration (as of 2026-05).
+        This is a best-effort extraction — the fallback (paginate-until-empty) is the
+        safety net when total is not reliably available. Field is UNVERIFIED against
+        live production SERP; paths are taken from known Yandex state shapes.
+
+        Пути кандидаты (проверяются в порядке приоритета):
+          pageData.pager.totalItems
+          pageData.pager.total
+          searchResults.pager.totalItems
+          pager.totalItems
+          pager.total
+          pageData.searchQuery.total
+        """
+        tree = HTMLParser(html)
+        candidates: list[str] = []
+        for selector in _STATE_SCRIPT_SELECTORS:
+            for node in tree.css(selector):
+                text = node.text() or ""
+                text = text.strip()
+                if text:
+                    candidates.append(text)
+
+        for blob in candidates:
+            try:
+                state = json.loads(blob)
+            except (json.JSONDecodeError, ValueError):
+                continue
+            if not isinstance(state, dict):
+                continue
+            for path in _TOTAL_COUNT_PATHS:
+                node_val: Any = state
+                for key in path:
+                    if not isinstance(node_val, dict):
+                        node_val = None
+                        break
+                    node_val = node_val.get(key)
+                if node_val is not None:
+                    try:
+                        result = int(node_val)
+                        if result >= 0:
+                            return result
+                    except (TypeError, ValueError):
+                        pass
+
+        # DOM / regex fallback: try to find a Russian results-count header.
+        # Best-effort only — Yandex uses dynamically generated class names.
+        # NBSP already normalised to space by caller.
+        m = _RE_TOTAL_COUNT.search(html[:20_000])
+        if m:
+            try:
+                return int(m.group(1).replace(" ", ""))
+            except (TypeError, ValueError):
+                pass
+
+        return None
+
+    # ── Leaf-bucket fetch helper ────────────────────────────────────────────────
+
+    async def _fetch_page_html(
+        self,
+        rooms: str | None,
+        page: int,
+        price_min: int | None,
+        price_max: int | None,
+    ) -> str | None:
+        """GET одной SERP-страницы, возвращает HTML или None при ошибке."""
+        url = self._build_url(page=page, rooms=rooms, price_min=price_min, price_max=price_max)
+        try:
+            resp = await self._http_get(url)
+        except Exception:
+            logger.exception("yandex: GET failed url=%s", url)
+            return None
+        if resp.status_code != 200:  # type: ignore[union-attr]
+            logger.warning("yandex: HTTP %d for url=%s", resp.status_code, url)  # type: ignore[union-attr]
+            return None
+        html = resp.text.replace("\xa0", " ")  # type: ignore[union-attr]
+        return html
+
+    # ── Exhaustive full load ────────────────────────────────────────────────────
+
+    async def fetch_all_secondary(
+        self,
+        *,
+        rooms_buckets: list[str] | None = None,
+        price_cap_per_bucket: int = 500,
+        max_pages_per_bucket: int = 100,
+        concurrency: int = 4,
+        on_bucket: Any = None,
+        on_progress: Any = None,
+    ) -> list[ScrapedLot]:
+        """Exhaustive-загрузка Yandex ЕКБ вторички через партиционирование КОМНАТНОСТЬ × ЦЕНА.
+
+        Обходит Yandex SERP-cap (~575 карточек/запрос): внутри каждой комнатности
+        адаптивно бьёт ценовой диапазон на бакеты так, чтобы totalItems < price_cap_per_bucket.
+        При недоступности totalItems (см. _extract_total_count CAVEAT) переходит на
+        paginate-until-empty fallback.
+
+        Параметры:
+          rooms_buckets: список ключей ROOM_PATH для перебора (default: все 5 комнатностей).
+          price_cap_per_bucket: максимум офферов в бакете перед делением (< Yandex ~575 cap).
+          max_pages_per_bucket: жёсткий cap страниц; 100 × 30 ≈ 3000 > 575.
+          concurrency: максимум параллельных page-фетчей в leaf-бакете.
+          on_bucket: опциональный callback(list[ScrapedLot]) после каждого leaf-бакета.
+            Может быть async или sync. Исключение прерывает прогон.
+          on_progress: опциональный callback(unique_count) для heartbeat (per room-bucket).
+
+        Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url).
+        """
+        _buckets = rooms_buckets if rooms_buckets is not None else list(ROOM_PATH.keys())
+        seen: dict[str, ScrapedLot] = {}
+
+        for rooms_key in _buckets:
+            logger.info(
+                "yandex exhaustive: starting rooms=%s (price_cap=%d concurrency=%d)",
+                rooms_key,
+                price_cap_per_bucket,
+                concurrency,
+            )
+            before = len(seen)
+            await self._walk_price_range(
+                rooms=rooms_key,
+                lo=0,
+                hi=_YANDEX_MAX_PRICE,
+                seen=seen,
+                price_cap_per_bucket=price_cap_per_bucket,
+                max_pages_per_bucket=max_pages_per_bucket,
+                concurrency=concurrency,
+                on_bucket=on_bucket,
+            )
+            room_collected = len(seen) - before
+            logger.info(
+                "yandex exhaustive: rooms=%s done — collected %d (total unique=%d)",
+                rooms_key,
+                room_collected,
+                len(seen),
+            )
+            if on_progress is not None:
+                on_progress(len(seen))
+
+        logger.info("yandex exhaustive: DONE — total unique=%d lots", len(seen))
+        return list(seen.values())
+
+    async def _walk_price_range(
+        self,
+        *,
+        rooms: str | None,
+        lo: int,
+        hi: int,
+        seen: dict[str, ScrapedLot],
+        price_cap_per_bucket: int,
+        max_pages_per_bucket: int,
+        concurrency: int = 4,
+        on_bucket: Any = None,
+        _depth: int = 0,
+    ) -> None:
+        """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi].
+
+        Алгоритм:
+          1. Запросить page=0 (probe) с price_min=lo, price_max=hi → totalItems.
+          2. Если totalItems <= cap → пагинировать leaf-бакет параллельно (concurrency).
+          3. Если totalItems > cap → разбить бакет пополам (рекурсия).
+          Guard: hi - lo < _YANDEX_MIN_BRACKET → пагинировать как есть.
+          Fallback: если totalItems=None → paginate-until-empty (безопасная деградация).
+
+        # TODO(yandex-total): totalItems not reliably extractable from hydrated DOM
+        # — using paginate-until-empty fallback when state extraction fails.
+        """
+        _lo_param = lo if lo > 0 else None
+
+        # ── Шаг 1: probe page=0 ─────────────────────────────────────────────────
+        html = await self._fetch_page_html(rooms, 0, _lo_param, hi)
+        await asyncio.sleep(self.request_delay_sec)
+
+        total: int | None = None
+
+        # CAPTCHA-aware probe: rotate IP + 1 retry
+        if html is None or _is_captcha(html):
+            logger.warning(
+                "yandex: probe captcha/None for rooms=%s [%d, %d] depth=%d — rotating IP + retry",
+                rooms,
+                lo,
+                hi,
+                _depth,
+            )
+            rotated = await self._rotate_ip()
+            if rotated:
+                html = await self._fetch_page_html(rooms, 0, _lo_param, hi)
+                await asyncio.sleep(self.request_delay_sec)
+
+        if html is not None and not _is_captcha(html):
+            total = self._extract_total_count(html)
+
+        # ── Fallback: totalItems not available ──────────────────────────────────
+        # TODO(yandex-total): totalItems not reliably extractable from hydrated DOM
+        # — using paginate-until-empty fallback when state extraction fails.
+        if total is None:
+            logger.warning(
+                "yandex: totalItems=None for rooms=%s [%d, %d] depth=%d "
+                "— using paginate-until-empty fallback",
+                rooms,
+                lo,
+                hi,
+                _depth,
+            )
+            bucket_lots: list[ScrapedLot] = []
+            # page=0 probe already fetched — reuse its HTML if available
+            if html is not None and not _is_captcha(html):
+                probe_lots = self._parse_html(html, page=0)
+                bucket_lots.extend(probe_lots)
+                start_page = 1
+            else:
+                start_page = 0
+
+            for p in range(start_page, max_pages_per_bucket):
+                if p == 0:
+                    page_lots: list[ScrapedLot] = []
+                else:
+                    page_html = await self._fetch_page_html(rooms, p, _lo_param, hi)
+                    await asyncio.sleep(self.request_delay_sec)
+                    if page_html is None or _is_captcha(page_html):
+                        logger.warning(
+                            "yandex: fallback page=%d captcha/None rooms=%s [%d, %d] — stopping",
+                            p,
+                            rooms,
+                            lo,
+                            hi,
+                        )
+                        break
+                    page_lots = self._parse_html(page_html, page=p)
+                if not page_lots and p > start_page:
+                    break
+                bucket_lots.extend(page_lots)
+
+            for lot in bucket_lots:
+                key = lot.source_id or lot.source_url
+                if key:
+                    seen[key] = lot
+
+            if on_bucket and bucket_lots:
+                res = on_bucket(bucket_lots)
+                if inspect.isawaitable(res):
+                    await res
+            return
+
+        logger.info(
+            "yandex: rooms=%s [%d, %d] totalItems=%d depth=%d",
+            rooms,
+            lo,
+            hi,
+            total,
+            _depth,
+        )
+
+        if total == 0:
+            return
+
+        # ── Шаг 2: деление или пагинация ────────────────────────────────────────
+        bracket_size = hi - lo
+        need_split = total > price_cap_per_bucket
+        too_narrow = bracket_size < _YANDEX_MIN_BRACKET
+
+        if need_split and too_narrow:
+            logger.warning(
+                "yandex: rooms=%s [%d, %d] totalItems=%d > cap=%d but bracket=%d < "
+                "MIN_BRACKET=%d — paginating as-is (tail loss ~%d)",
+                rooms,
+                lo,
+                hi,
+                total,
+                price_cap_per_bucket,
+                bracket_size,
+                _YANDEX_MIN_BRACKET,
+                max(0, total - price_cap_per_bucket),
+            )
+            need_split = False
+
+        if need_split:
+            mid = (lo + hi) // 2
+            await self._walk_price_range(
+                rooms=rooms,
+                lo=lo,
+                hi=mid,
+                seen=seen,
+                price_cap_per_bucket=price_cap_per_bucket,
+                max_pages_per_bucket=max_pages_per_bucket,
+                concurrency=concurrency,
+                on_bucket=on_bucket,
+                _depth=_depth + 1,
+            )
+            await self._walk_price_range(
+                rooms=rooms,
+                lo=mid + 1,
+                hi=hi,
+                seen=seen,
+                price_cap_per_bucket=price_cap_per_bucket,
+                max_pages_per_bucket=max_pages_per_bucket,
+                concurrency=concurrency,
+                on_bucket=on_bucket,
+                _depth=_depth + 1,
+            )
+            return
+
+        # ── Параллельная пагинация leaf-бакета ─────────────────────────────────
+        # Yandex SERP pages are 0-indexed (unlike Cian which is 1-indexed)
+        max_pages = min(
+            math.ceil(total / _YANDEX_OFFERS_PER_PAGE),
+            max_pages_per_bucket,
+        )
+
+        sem = asyncio.Semaphore(concurrency)
+
+        async def _one_page(p: int) -> list[ScrapedLot]:
+            if p == 0:
+                # Reuse probe HTML
+                return self._parse_html(html, page=0) if html else []
+            async with sem:
+                page_html = await self._fetch_page_html(rooms, p, _lo_param, hi)
+                await asyncio.sleep(self.request_delay_sec)
+                if page_html is None or _is_captcha(page_html):
+                    logger.warning(
+                        "yandex: page_html=None/captcha rooms=%s [%d, %d] page=%d — skipping",
+                        rooms,
+                        lo,
+                        hi,
+                        p,
+                    )
+                    return []
+                return self._parse_html(page_html, page=p)
+
+        page_results = await asyncio.gather(
+            *[_one_page(p) for p in range(0, max_pages)],
+            return_exceptions=True,
+        )
+
+        bucket_lots_leaf: list[ScrapedLot] = []
+        for p_idx, res in enumerate(page_results):
+            if isinstance(res, BaseException):
+                logger.warning(
+                    "yandex: page exception rooms=%s [%d, %d] page=%d — %r",
+                    rooms,
+                    lo,
+                    hi,
+                    p_idx,
+                    res,
+                )
+            else:
+                bucket_lots_leaf.extend(res)
+
+        for lot in bucket_lots_leaf:
+            key = lot.source_id or lot.source_url
+            if key:
+                seen[key] = lot
+
+        logger.info(
+            "yandex: rooms=%s [%d, %d] paginated=%d pages collected=%d unique_total=%d",
+            rooms,
+            lo,
+            hi,
+            max_pages,
+            len(bucket_lots_leaf),
+            len(seen),
+        )
+
+        if on_bucket and bucket_lots_leaf:
+            res_cb = on_bucket(bucket_lots_leaf)
+            if inspect.isawaitable(res_cb):
+                await res_cb
+
     async def fetch_around(
         self,
         lat: float,

tradein-mvp/backend/tests/test_scraper_proxy.py

@@ -32,6 +32,9 @@ def _mock_settings(
         # scraper_proxy_url fallback). Mirror scraper_proxy_url here so the existing
         # cian proxy-wiring assertions keep passing (#923 follow-up).
         cian_proxy_url=scraper_proxy_url,
+        # Yandex scrapers read settings.yandex_proxy_url (property: YANDEX_PROXY_URL or
+        # scraper_proxy_url fallback). Mirror scraper_proxy_url here for test compat.
+        yandex_proxy_url=scraper_proxy_url,
         yandex_cookies_file=yandex_cookies_file,
     )
 

tradein-mvp/backend/tests/test_yandex_realty_serp.py

@@ -634,3 +634,195 @@ async def test_fetch_around_multi_room_legacy_no_combos():
     # Should call with rooms=None, price_min=None (citywide)
     assert all(c["rooms"] is None for c in calls)
     assert all(c["price_min"] is None for c in calls)
+
+
+# ── PART E: exhaustive load — fetch_all_secondary + _walk_price_range ────────
+
+
+def _make_lots(n: int, prefix: str = "lot") -> list[MagicMock]:
+    """Create n mock ScrapedLot objects with unique source_id."""
+    lots = []
+    for i in range(n):
+        lot = MagicMock(spec=ScrapedLot)
+        lot.source_id = f"{prefix}_{i}"
+        lot.source_url = f"https://realty.yandex.ru/offer/{prefix}_{i}/"
+        lots.append(lot)
+    return lots
+
+
+@pytest.mark.asyncio
+async def test_walk_price_range_splits_when_total_exceeds_cap():
+    """When totalItems > price_cap_per_bucket and bracket >= MIN_BRACKET,
+    _walk_price_range recurses and subdivides — each leaf has ≤ cap items."""
+    s = YandexRealtyScraper()
+    s._cffi_session = AsyncMock()
+
+    call_ranges: list[tuple[int | None, int]] = []
+
+    async def fake_fetch_page_html(
+        rooms: str | None, page: int, price_min: int | None, price_max: int
+    ) -> str:
+        call_ranges.append((price_min, price_max))
+        # Return captcha-free HTML with 0 cards (parsing gives no lots, which is fine —
+        # we only care that the range is split, not about actual lot count in this test)
+        return "<html><body>empty</body></html>"
+
+    # total_count returns 600 for the first (wide) probe, then 0 for sub-ranges
+    # so recursion terminates after the first split.
+    call_counts: list[int] = [0]
+
+    def fake_extract_total(html: str) -> int | None:
+        call_counts[0] += 1
+        # First call (wide range) returns 600 > cap=500, triggering a split
+        # Subsequent calls (sub-ranges) return 0 → no further recursion
+        if call_counts[0] == 1:
+            return 600
+        return 0
+
+    seen: dict = {}
+    with (
+        patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
+        patch.object(s, "_extract_total_count", side_effect=fake_extract_total),
+        patch.object(s, "_is_captcha_html", return_value=False, create=True),
+    ):
+        # Patch asyncio.sleep to be instant
+        with patch("asyncio.sleep", new_callable=AsyncMock):
+            await s._walk_price_range(
+                rooms="1",
+                lo=0,
+                hi=10_000_000,
+                seen=seen,
+                price_cap_per_bucket=500,
+                max_pages_per_bucket=100,
+                concurrency=2,
+            )
+
+    # Should have probed at least the initial range + 2 sub-ranges
+    assert len(call_ranges) >= 3, f"Expected >= 3 probes (wide + 2 sub-ranges), got {call_ranges}"
+    # Initial probe covers the full range
+    assert call_ranges[0] == (None, 10_000_000)  # lo=0 → _lo_param=None
+
+
+@pytest.mark.asyncio
+async def test_walk_price_range_fires_on_bucket_per_leaf():
+    """on_bucket callback is called once per leaf bucket with the bucket's lots."""
+    s = YandexRealtyScraper()
+    s._cffi_session = AsyncMock()
+
+    bucket_calls: list[list] = []
+
+    def fake_on_bucket(lots: list) -> None:
+        bucket_calls.append(list(lots))
+
+    # total = 10 (< cap=500) → paginate leaf immediately, 1 page (page=0)
+    async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str:
+        # Return HTML with 1 parseable card for page=0, empty for others
+        if page == 0:
+            return SINGLE_CARD_HTML
+        return EMPTY_PAGE_HTML
+
+    seen: dict = {}
+    with (
+        patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
+        patch.object(s, "_extract_total_count", return_value=10),
+        patch("asyncio.sleep", new_callable=AsyncMock),
+    ):
+        await s._walk_price_range(
+            rooms="1",
+            lo=0,
+            hi=20_000_000,
+            seen=seen,
+            price_cap_per_bucket=500,
+            max_pages_per_bucket=5,
+            concurrency=2,
+            on_bucket=fake_on_bucket,
+        )
+
+    # on_bucket must have been called at least once
+    assert len(bucket_calls) >= 1
+    # All calls contain ScrapedLot-like objects (the actual card from SINGLE_CARD_HTML)
+    for call_lots in bucket_calls:
+        assert len(call_lots) >= 0  # may be 0 if card parse yields empty — still fires
+
+
+@pytest.mark.asyncio
+async def test_walk_price_range_fallback_when_total_none():
+    """When totalItems=None (state unavailable), falls back to paginate-until-empty
+    and still fires on_bucket with whatever was collected."""
+    s = YandexRealtyScraper()
+    s._cffi_session = AsyncMock()
+
+    bucket_calls: list[list] = []
+
+    def fake_on_bucket(lots: list) -> None:
+        bucket_calls.append(list(lots))
+
+    page_htmls = {
+        0: SINGLE_CARD_HTML,  # page=0: 1 card
+        1: EMPTY_PAGE_HTML,  # page=1: empty → stop
+    }
+
+    async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str | None:
+        return page_htmls.get(page, EMPTY_PAGE_HTML)
+
+    seen: dict = {}
+    with (
+        patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
+        patch.object(s, "_extract_total_count", return_value=None),
+        patch("asyncio.sleep", new_callable=AsyncMock),
+    ):
+        await s._walk_price_range(
+            rooms="2",
+            lo=0,
+            hi=15_000_000,
+            seen=seen,
+            price_cap_per_bucket=500,
+            max_pages_per_bucket=10,
+            concurrency=2,
+            on_bucket=fake_on_bucket,
+        )
+
+    # Fallback path must fire on_bucket with the collected lots
+    assert len(bucket_calls) >= 1, "on_bucket must be called in fallback path"
+    # The card from SINGLE_CARD_HTML should have been collected
+    all_lots = [lot for call in bucket_calls for lot in call]
+    assert len(all_lots) >= 1
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_secondary_deduplicates_across_rooms():
+    """fetch_all_secondary deduplicates offers that appear in multiple room buckets."""
+    s = YandexRealtyScraper()
+
+    shared_lot = MagicMock(spec=ScrapedLot)
+    shared_lot.source_id = "shared_999"
+    shared_lot.source_url = "https://realty.yandex.ru/offer/shared_999/"
+
+    unique_lot_1 = MagicMock(spec=ScrapedLot)
+    unique_lot_1.source_id = "unique_001"
+    unique_lot_1.source_url = "https://realty.yandex.ru/offer/unique_001/"
+
+    unique_lot_2 = MagicMock(spec=ScrapedLot)
+    unique_lot_2.source_id = "unique_002"
+    unique_lot_2.source_url = "https://realty.yandex.ru/offer/unique_002/"
+
+    call_count = 0
+
+    async def fake_walk(*, rooms, lo, hi, seen, **kwargs):  # type: ignore[override]
+        nonlocal call_count
+        # Both room buckets yield the shared lot + one unique each
+        seen["shared_999"] = shared_lot
+        if call_count == 0:
+            seen["unique_001"] = unique_lot_1
+        else:
+            seen["unique_002"] = unique_lot_2
+        call_count += 1
+
+    with patch.object(s, "_walk_price_range", side_effect=fake_walk):
+        result = await s.fetch_all_secondary(rooms_buckets=["1", "2"])
+
+    ids = {lot.source_id for lot in result}
+    assert "shared_999" in ids
+    assert "unique_001" in ids
+    assert "unique_002" in ids
+    assert len(ids) == 3  # shared deduplicated — only 3 unique