feat(tradein): exhaustive Cian EKB secondary load — room×price partitioning (#926)

Co-authored-by: bot-backend <bot-backend@gendsgn.local>
Co-committed-by: bot-backend <bot-backend@gendsgn.local>

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

@@ -26,6 +26,7 @@ from app.services.scheduler import has_running_run
 from app.services.scrape_pipeline import (
     run_avito_city_sweep,
     run_cian_city_sweep,
+    run_cian_full_load,
     run_n1_city_sweep,
     run_yandex_city_sweep,
 )
@@ -943,6 +944,94 @@ def cancel_cian_city_sweep(
     return {"ok": True, "run_id": run_id, "cancelled": cancelled}
 
 
+# ── Cian exhaustive full load (room × price partitioning) ────────────────────
+
+
+class CianFullLoadRequest(BaseModel):
+    price_cap_per_bucket: int = Field(
+        default=1400,
+        ge=200,
+        le=1500,
+        description=(
+            "Максимум офферов на price-бакет. При totalOffers > cap — бакет делится пополам. "
+            "Ниже Cian SERP-cap ~1500; запас 100 на variance."
+        ),
+    )
+    request_delay_sec: float = Field(
+        default=4.0,
+        ge=3.0,
+        le=15.0,
+        description="Задержка между SERP-запросами (сек). Увеличить при бане.",
+    )
+    enrich_detail: bool = Field(
+        default=False,
+        description="Если True — после сбора обогатить detail-страницы (медленно).",
+    )
+    detail_top_n: int = Field(
+        default=0,
+        ge=0,
+        le=50,
+        description="Сколько свежих листингов обогатить detail (актуально при enrich_detail=True).",
+    )
+
+
+@router.post("/scrape/cian-full-load", response_model=CitySweepStartResponse)
+async def start_cian_full_load(
+    payload: CianFullLoadRequest,
+    background_tasks: BackgroundTasks,
+    db: Annotated[Session, Depends(get_db)],
+) -> CitySweepStartResponse:
+    """Запустить exhaustive Cian ЕКБ full load в background.
+
+    Обходит Cian SERP-cap (~54 стр/запрос ≈ 1500 результатов): адаптивно бьёт
+    цену на бакеты внутри каждой комнатности (totalOffers-driven бинарное деление),
+    пагинирует каждый бакет полностью, дедуп по source_id.
+
+    Один региональный проход БЕЗ anchor'ов (Cian SERP region-wide — anchors избыточны).
+    IP-ротация через changeip при timeout/captcha.
+
+    Coop cancel: POST /scrape/cian-full-load/{run_id}/cancel.
+    Статус прогона: смотреть через GET /scrape/cian-city-sweep/runs
+    (или добавить source='cian_full_load' в /admin SQL).
+    """
+    run_id = runs_mod.create_run(db, source="cian_full_load", params=payload.model_dump())
+
+    async def _full_load_task() -> None:
+        task_db = SessionLocal()
+        try:
+            await run_cian_full_load(
+                task_db,
+                run_id=run_id,
+                price_cap_per_bucket=payload.price_cap_per_bucket,
+                request_delay_sec=payload.request_delay_sec,
+                enrich_detail=payload.enrich_detail,
+                detail_top_n=payload.detail_top_n,
+            )
+        except Exception:
+            logger.exception("cian-full-load background task run_id=%d crashed", run_id)
+        finally:
+            task_db.close()
+
+    background_tasks.add_task(_full_load_task)
+    logger.info("cian-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=payload.detail_top_n,
+    )
+
+
+@router.post("/scrape/cian-full-load/{run_id}/cancel")
+def cancel_cian_full_load(
+    run_id: int,
+    db: Annotated[Session, Depends(get_db)],
+) -> dict[str, object]:
+    """Отменить running Cian 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/services/scrape_pipeline.py

@@ -1526,12 +1526,186 @@ async def run_cian_city_sweep(
         raise
 
 
+# ── Cian exhaustive full load (region-wide, без anchor'ов) ─────────────────
+
+
+@dataclass
+class CianFullLoadCounters:
+    """Aggregate counters для run_cian_full_load.
+
+    Без anchor-полей — единственный региональный проход.
+    detail_* — опциональное обогащение detail-страниц (если enrich_detail=True).
+    """
+
+    unique_fetched: int = 0
+    saved_inserted: int = 0
+    saved_updated: int = 0
+    detail_attempted: int = 0
+    detail_enriched: int = 0
+    detail_failed: 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_cian_full_load(
+    db: Session,
+    *,
+    run_id: int,
+    price_cap_per_bucket: int = 1400,
+    detail_top_n: int = 0,
+    request_delay_sec: float = 4.0,
+    enrich_detail: bool = False,
+) -> CianFullLoadCounters:
+    """Exhaustive региональный сбор Cian ЕКБ вторички (БЕЗ anchor'ов).
+
+    Cian SERP — region-wide, не geo-bbox. Anchor-loop избыточен (5 anchor'ов
+    возвращают одну и ту же выдачу ×5). Один проход с партиционированием по
+    КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление) охватывает весь ЕКБ.
+
+    Параметры:
+      price_cap_per_bucket: максимум офферов на price-бакет (< Cian SERP-cap ~1500).
+      detail_top_n: сколько свежих cian-листингов без detail_enriched_at обогащать
+                    (актуально только при enrich_detail=True).
+      request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default).
+      enrich_detail: включить detail-обогащение (по умолчанию отключено — тяжело).
+
+    Cooperative cancel: scrape_runs.is_cancelled проверяется в on_progress callback
+    (раз в room-bucket). При cancel — возвращаем частичный результат, mark_done.
+    """
+    from app.services.scrapers.cian import CianScraper
+
+    counters = CianFullLoadCounters()
+    _cancelled = False
+
+    def _on_progress(unique_count: int) -> None:
+        nonlocal _cancelled
+        counters.unique_fetched = unique_count
+        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+        if scrape_runs.is_cancelled(db, run_id):
+            _cancelled = True
+            logger.info("cian-full-load run_id=%d: cancel detected in on_progress", run_id)
+
+    try:
+        async with CianScraper() as scraper:
+            scraper.request_delay_sec = request_delay_sec
+
+            lots = await scraper.fetch_all_secondary(
+                price_cap_per_bucket=price_cap_per_bucket,
+                on_progress=_on_progress,
+            )
+
+        counters.unique_fetched = len(lots)
+        logger.info(
+            "cian-full-load run_id=%d: fetch done — unique=%d cancelled=%s",
+            run_id,
+            len(lots),
+            _cancelled,
+        )
+
+        if lots:
+            inserted, updated = save_listings(db, lots, run_id=run_id)
+            counters.saved_inserted = inserted
+            counters.saved_updated = updated
+            logger.info(
+                "cian-full-load run_id=%d: save done — ins=%d upd=%d",
+                run_id,
+                inserted,
+                updated,
+            )
+        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+
+        # ── Опциональный detail-энричмент ─────────────────────────────────────
+        if enrich_detail and detail_top_n > 0 and not _cancelled:
+            from sqlalchemy import text as _text
+
+            from app.services.scrapers.cian_detail import (
+                fetch_detail,
+                save_detail_enrichment,
+            )
+
+            priority_rows = (
+                db.execute(
+                    _text("""
+                        SELECT id, source_url
+                        FROM listings
+                        WHERE source = 'cian'
+                          AND source_url IS NOT NULL
+                          AND detail_enriched_at IS NULL
+                          AND scraped_at > NOW() - INTERVAL '4 hours'
+                        ORDER BY scraped_at DESC NULLS LAST
+                        LIMIT :lim
+                    """),
+                    {"lim": detail_top_n},
+                )
+                .mappings()
+                .all()
+            )
+            for didx, row in enumerate(priority_rows):
+                if scrape_runs.is_cancelled(db, run_id):
+                    logger.info("cian-full-load run_id=%d: cancelled during detail enrich", run_id)
+                    _cancelled = True
+                    break
+                listing_id: int = row["id"]
+                source_url: str = row["source_url"]
+                counters.detail_attempted += 1
+                try:
+                    enrichment = await fetch_detail(source_url)
+                    if enrichment is not None:
+                        save_detail_enrichment(db, listing_id, enrichment)
+                        counters.detail_enriched += 1
+                    else:
+                        counters.detail_failed += 1
+                except Exception as exc:
+                    counters.detail_failed += 1
+                    counters.errors_count += 1
+                    logger.warning(
+                        "cian-full-load run_id=%d: detail failed listing_id=%d: %s",
+                        run_id,
+                        listing_id,
+                        exc,
+                    )
+                if didx < len(priority_rows) - 1:
+                    await asyncio.sleep(request_delay_sec * random.uniform(0.8, 1.2))
+
+            logger.info(
+                "cian-full-load run_id=%d: detail done — enriched=%d/%d failed=%d",
+                run_id,
+                counters.detail_enriched,
+                counters.detail_attempted,
+                counters.detail_failed,
+            )
+            scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+
+        scrape_runs.mark_done(db, run_id, counters.to_dict())
+        logger.info(
+            "cian-full-load run_id=%d done: unique=%d ins=%d upd=%d "
+            "detail=%d/%d errors=%d cancelled=%s",
+            run_id,
+            counters.unique_fetched,
+            counters.saved_inserted,
+            counters.saved_updated,
+            counters.detail_enriched,
+            counters.detail_attempted,
+            counters.errors_count,
+            _cancelled,
+        )
+        return counters
+
+    except Exception as exc:
+        logger.exception("cian-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",
     "CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES",
     "EKB_ANCHORS",
     "CianCitySweepCounters",
+    "CianFullLoadCounters",
     "CitySweepCounters",
     "PipelineCounters",
     "PipelineResult",
@@ -1539,5 +1713,6 @@ __all__ = [
     "run_avito_city_sweep",
     "run_avito_pipeline",
     "run_cian_city_sweep",
+    "run_cian_full_load",
     "run_yandex_city_sweep",
 ]

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

@@ -11,12 +11,21 @@
 
 НЕ используем anchor jitter: offer.geo.coordinates.{lat,lng} — точные координаты
 прямо из SERP state. Jitter запрещён per implementation plan.
+
+Exhaustive load (fetch_all_secondary):
+- Cian SERP — регион-wide (не geo-bbox). Anchor-loop в существующем city-sweep
+  избыточен для Cian (все anchor'ы дают одну выдачу). Полный сбор = ОДИН проход
+  с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление диапазона).
+- IP-ротация через _rotate_ip() (changeip, зеркало avito.py).
 """
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import logging
+import math
+from collections.abc import Callable
 from datetime import UTC, datetime
 from typing import Any
 from urllib.parse import urlencode
@@ -42,6 +51,33 @@ CIAN_EKB_REGION_ID = 4743
 _MFE_SERP = "frontend-serp"
 _STATE_KEY = "initialState"
 
+# ── Константы exhaustive-загрузки ──────────────────────────────────────────────
+
+# Верхняя граница цены при первом рекурсивном делении (нет явного hi).
+# 200 млн ₽ заведомо выше ТОПа ЕКБ — дальние бакеты дадут 0 офферов.
+_MAX_PRICE = 200_000_000
+
+# Минимальный ценовой диапазон для рекурсии; при hi - lo < MIN_BRACKET
+# прекращаем деление и пагинируем как есть (принимаем возможный «хвост»).
+_MIN_BRACKET = 50_000
+
+# Cian SERP показывает ~28 офферов на страницу.
+_CIAN_OFFERS_PER_PAGE = 28
+
+# Комнатности по умолчанию для exhaustive-загрузки.
+# Коды Cian: room1=1к, room2=2к, room3=3к, room4=4к, room5=5к, room6=6+.
+# TODO: студии — Cian использует отдельный тип flat_type=studio, не room-param.
+#   Если добавить room9 — запрос может давать 0 офферов или смешивать типы.
+#   До верификации кода студии — студии НЕ включаем в exhaustive-загрузку.
+_DEFAULT_ROOMS_BUCKETS: list[tuple[int, ...]] = [
+    (1,),
+    (2,),
+    (3,),
+    (4,),
+    (5,),
+    (6,),
+]
+
 
 class CianScraper(BaseScraper):
     """Cian SERP scraper. Использует curl_cffi для обхода TLS fingerprint.
@@ -110,6 +146,29 @@ class CianScraper(BaseScraper):
             await self._cffi.close()
         await super().__aexit__(*args)
 
+    # ── Anti-block: IP rotation (зеркало avito.py _rotate_ip) ─────────────────
+
+    async def _rotate_ip(self) -> bool:
+        """Сменить мобильный IP через changeip-ссылку mobileproxy.
+
+        Дёргается напрямую (без прокси) — это API провайдера, не Cian. Ждём ~9с:
+        мобильному модему нужно время поднять новый IP. Returns True при успехе.
+        Зеркало AvitoScraper._rotate_ip (#623).
+        """
+        rotate_url = settings.cian_proxy_rotate_url or settings.avito_proxy_rotate_url
+        if not rotate_url:
+            return False
+        sep = "&" if "?" in rotate_url else "?"
+        try:
+            async with AsyncSession(timeout=30) as rot:
+                await rot.get(f"{rotate_url}{sep}format=json")
+            await asyncio.sleep(9)
+            logger.info("cian proxy: IP rotated via changeip")
+            return True
+        except Exception:
+            logger.warning("cian proxy: IP rotation failed", exc_info=True)
+            return False
+
     async def fetch_around(
         self,
         lat: float,
@@ -187,11 +246,14 @@ class CianScraper(BaseScraper):
         self,
         rooms: tuple[int, ...] | None = None,
         page: int = 1,
+        min_price: int | None = None,
+        max_price: int | None = None,
     ) -> str:
         """URL для Cian каталога вторички ЕКБ.
 
         Используем ekb.cian.ru (city-specific subdomain).
         Регион задаётся через region= param как fallback для reliability.
+        min_price / max_price — опциональные ценовые границы (руб.) для price-бакетов.
         """
         params: list[tuple[str, Any]] = [
             ("deal_type", "sale"),
@@ -203,10 +265,320 @@ class CianScraper(BaseScraper):
         if rooms:
             for r in rooms:
                 params.append((f"room{r}", "1"))
+        if min_price is not None:
+            params.append(("minprice", min_price))
+        if max_price is not None:
+            params.append(("maxprice", max_price))
         if page > 1:
             params.append(("p", page))
         return f"{self.base_url}/cat.php?{urlencode(params)}"
 
+    def _extract_total_offers(self, html: str) -> int | None:
+        """Извлечь totalOffers из Redux state Cian SERP.
+
+        Использует тот же extract_state, что и _parse_serp_html.
+        Возвращает None при captcha/ошибке парсинга.
+        """
+        state = extract_state(html, mfe=_MFE_SERP, key=_STATE_KEY)
+        if state is None:
+            return None
+        total = state.get("results", {}).get("totalOffers")
+        if total is None:
+            return None
+        try:
+            return int(total)
+        except (TypeError, ValueError):
+            return None
+
+    async def _fetch_page_html(
+        self,
+        rooms: tuple[int, ...] | None,
+        page: int,
+        min_price: int | None,
+        max_price: int | None,
+    ) -> str | None:
+        """GET одной SERP-страницы, возвращает HTML или None при ошибке."""
+        assert self._cffi is not None
+        url = self._build_url(rooms, page, min_price, max_price)
+        try:
+            resp = await self._cffi.get(url)
+        except Exception:
+            logger.exception("cian: GET failed url=%s", url)
+            return None
+        if resp.status_code != 200:
+            logger.warning("cian: HTTP %d for url=%s", resp.status_code, url)
+            return None
+        return resp.text
+
+    async def fetch_all_secondary(
+        self,
+        *,
+        rooms_buckets: list[tuple[int, ...]] | None = None,
+        price_cap_per_bucket: int = 1400,
+        max_pages_per_bucket: int = 54,
+        on_progress: Callable[[int], None] | None = None,
+    ) -> list[ScrapedLot]:
+        """Exhaustive-загрузка Cian ЕКБ вторички через партиционирование КОМНАТНОСТЬ × ЦЕНА.
+
+        Обходит Cian SERP-cap (~54 стр/запрос ≈ 1500 результатов на запрос):
+        внутри каждой комнатности адаптивно бьёт диапазон цены на бакеты так, чтобы
+        в каждом totalOffers < price_cap_per_bucket → пагинирует бакет полностью.
+        Дедуп по source_id (dict seen).
+
+        Параметры:
+          rooms_buckets: список room-кодов Cian для перебора (default: 1-6).
+          price_cap_per_bucket: максимум офферов в бакете перед делением (< 1500).
+          max_pages_per_bucket: Cian hard cap ~54; не превышать.
+          on_progress: опциональный callback(unique_count) для heartbeat.
+
+        Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url).
+        """
+        _buckets = rooms_buckets if rooms_buckets is not None else _DEFAULT_ROOMS_BUCKETS
+        seen: dict[str, ScrapedLot] = {}
+
+        for rooms in _buckets:
+            room_label = f"room{'_'.join(str(r) for r in rooms)}"
+            logger.info(
+                "cian exhaustive: starting %s (price_cap=%d)", room_label, price_cap_per_bucket
+            )
+            before = len(seen)
+            await self._walk_price_range(
+                rooms=rooms,
+                lo=0,
+                hi=None,
+                seen=seen,
+                price_cap_per_bucket=price_cap_per_bucket,
+                max_pages_per_bucket=max_pages_per_bucket,
+            )
+            room_collected = len(seen) - before
+            logger.info(
+                "cian exhaustive: %s done — collected %d (total unique=%d)",
+                room_label,
+                room_collected,
+                len(seen),
+            )
+            if on_progress is not None:
+                on_progress(len(seen))
+
+        logger.info("cian exhaustive: DONE — total unique=%d lots", len(seen))
+        return list(seen.values())
+
+    async def _walk_price_range(
+        self,
+        *,
+        rooms: tuple[int, ...],
+        lo: int,
+        hi: int | None,
+        seen: dict[str, ScrapedLot],
+        price_cap_per_bucket: int,
+        max_pages_per_bucket: int,
+        _depth: int = 0,
+    ) -> None:
+        """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi].
+
+        Алгоритм:
+          1. Запросить page=1 с min_price=lo, max_price=hi → totalOffers.
+          2. Если totalOffers <= cap → пагинировать бакет полностью.
+          3. Если totalOffers > cap → разбить бакет пополам (рекурсия).
+          Guard: hi - lo < _MIN_BRACKET → пагинировать как есть (логируем WARNING).
+        """
+        # Нормализация: None-hi → _MAX_PRICE для первого уровня
+        effective_hi = hi if hi is not None else _MAX_PRICE
+
+        room_label = f"room{'_'.join(str(r) for r in rooms)}"
+
+        # ── Шаг 1: probe page 1 ────────────────────────────────────────────────
+        html = await self._fetch_page_html(rooms, 1, lo if lo > 0 else None, hi)
+        await self.sleep_between_requests()
+
+        total: int | None = None
+        if html is not None:
+            total = self._extract_total_offers(html)
+
+        # Ретрай на captcha/ошибку: rotate IP + 1 retry
+        if total is None:
+            logger.warning(
+                "cian: totalOffers=None for %s [%d, %s] depth=%d — rotating IP + retry",
+                room_label,
+                lo,
+                hi,
+                _depth,
+            )
+            rotated = await self._rotate_ip()
+            if rotated:
+                html = await self._fetch_page_html(rooms, 1, lo if lo > 0 else None, hi)
+                await self.sleep_between_requests()
+                if html is not None:
+                    total = self._extract_total_offers(html)
+
+        if total is None:
+            logger.error(
+                "cian: skipping bucket %s [%d, %s] — totalOffers unavailable after retry",
+                room_label,
+                lo,
+                hi,
+            )
+            return
+
+        logger.info(
+            "cian: %s [%d, %s] totalOffers=%d depth=%d",
+            room_label,
+            lo,
+            hi,
+            total,
+            _depth,
+        )
+
+        if total == 0:
+            return
+
+        # ── Шаг 2: деление или пагинация ──────────────────────────────────────
+        bracket_size = effective_hi - lo
+        need_split = total > price_cap_per_bucket
+        too_narrow = bracket_size < _MIN_BRACKET
+
+        if need_split and too_narrow:
+            logger.warning(
+                "cian: %s [%d, %s] totalOffers=%d > cap=%d but bracket=%d < MIN_BRACKET=%d "
+                "— paginating as-is (tail loss ~%d)",
+                room_label,
+                lo,
+                hi,
+                total,
+                price_cap_per_bucket,
+                bracket_size,
+                _MIN_BRACKET,
+                max(0, total - price_cap_per_bucket),
+            )
+            need_split = False  # принудительно пагинируем
+
+        if need_split:
+            # При hi=None: сначала устанавливаем hi = _MAX_PRICE и рекурсируем
+            if hi is None:
+                await self._walk_price_range(
+                    rooms=rooms,
+                    lo=lo,
+                    hi=_MAX_PRICE,
+                    seen=seen,
+                    price_cap_per_bucket=price_cap_per_bucket,
+                    max_pages_per_bucket=max_pages_per_bucket,
+                    _depth=_depth + 1,
+                )
+                return
+            mid = (lo + effective_hi) // 2
+            # [lo, mid]
+            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,
+                _depth=_depth + 1,
+            )
+            # [mid+1, hi]
+            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,
+                _depth=_depth + 1,
+            )
+            return
+
+        # ── Пагинация бакета ───────────────────────────────────────────────────
+        max_pages = min(
+            math.ceil(total / _CIAN_OFFERS_PER_PAGE),
+            max_pages_per_bucket,
+        )
+
+        # Страница 1 уже есть (html из probe выше)
+        collected_this_bucket = 0
+        if html:
+            page1_lots = self._parse_serp_html(html)
+            for lot in page1_lots:
+                key = lot.source_id or lot.source_url
+                if key:
+                    seen[key] = lot
+            collected_this_bucket += len(page1_lots)
+            if not page1_lots:
+                logger.info("cian: %s [%d, %s] page=1 empty — early stop", room_label, lo, hi)
+                return
+
+        for page in range(2, max_pages + 1):
+            page_html: str | None = None
+            try:
+                page_html = await self._fetch_page_html(rooms, page, lo if lo > 0 else None, hi)
+                await self.sleep_between_requests()
+            except Exception:
+                logger.exception(
+                    "cian: fetch failed %s [%d, %s] page=%d — rotating IP",
+                    room_label,
+                    lo,
+                    hi,
+                    page,
+                )
+                await self._rotate_ip()
+                try:
+                    page_html = await self._fetch_page_html(rooms, page, lo if lo > 0 else None, hi)
+                    await self.sleep_between_requests()
+                except Exception:
+                    logger.error(
+                        "cian: retry failed %s [%d, %s] page=%d — breaking bucket",
+                        room_label,
+                        lo,
+                        hi,
+                        page,
+                    )
+                    break
+
+            if page_html is None:
+                # Повторная попытка после ротации
+                rotated = await self._rotate_ip()
+                if rotated:
+                    try:
+                        page_html = await self._fetch_page_html(
+                            rooms, page, lo if lo > 0 else None, hi
+                        )
+                        await self.sleep_between_requests()
+                    except Exception:
+                        pass
+                if page_html is None:
+                    logger.error(
+                        "cian: page_html=None %s [%d, %s] page=%d — breaking bucket",
+                        room_label,
+                        lo,
+                        hi,
+                        page,
+                    )
+                    break
+
+            page_lots = self._parse_serp_html(page_html)
+            if not page_lots:
+                logger.info(
+                    "cian: %s [%d, %s] page=%d empty — early stop", room_label, lo, hi, page
+                )
+                break
+
+            for lot in page_lots:
+                key = lot.source_id or lot.source_url
+                if key:
+                    seen[key] = lot
+            collected_this_bucket += len(page_lots)
+
+        logger.info(
+            "cian: %s [%d, %s] paginated=%d/%d pages collected=%d unique_total=%d",
+            room_label,
+            lo,
+            hi,
+            min(max_pages, page if "page" in locals() else 1),
+            max_pages,
+            collected_this_bucket,
+            len(seen),
+        )
+
     def _parse_serp_html(self, html: str) -> list[ScrapedLot]:
         """Извлечь offers из Cian Redux state.
 

tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py

@@ -0,0 +1,206 @@
+"""Тесты exhaustive-загрузки Cian: fetch_all_secondary + _walk_price_range.
+
+Мокаем сетевые вызовы (_fetch_page_html, _extract_total_offers, sleep_between_requests)
+для детерминированного контроля поведения без реального HTTP.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from app.services.scrapers.base import ScrapedLot
+from app.services.scrapers.cian import _MAX_PRICE, _MIN_BRACKET, CianScraper
+
+
+def _make_lot(source_id: str, price: int = 3_000_000) -> ScrapedLot:
+    """Минимальный ScrapedLot для теста."""
+    return ScrapedLot(
+        source="cian",
+        source_url=f"https://ekb.cian.ru/sale/flat/{source_id}/",
+        source_id=source_id,
+        address="Екатеринбург",
+        price_rub=price,
+    )
+
+
+def _make_html_stub(total_offers: int, lot_ids: list[str]) -> str:
+    """Заглушка HTML — не парсится напрямую; используется через мокнутые методы."""
+    return f"<html>total={total_offers} ids={lot_ids}</html>"
+
+
+@pytest.fixture
+def scraper() -> CianScraper:
+    """CianScraper без реального __aenter__ (без HTTP warm-up)."""
+    s = CianScraper.__new__(CianScraper)
+    s.name = "cian"
+    s.base_url = "https://ekb.cian.ru"
+    s.request_delay_sec = 0.0
+    s._cffi = MagicMock()  # не None → assert self._cffi is not None пройдёт
+    return s
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_secondary_splits_on_cap(scraper: CianScraper) -> None:
+    """totalOffers > cap → бакет разбивается; в итоге оба суб-бакета пагинируются."""
+    # Настройка: rooms=(1,), один бакет [0, None]
+    # Probe [0, None]: totalOffers=3000 > cap=1400 → split на [0, MAX_PRICE]
+    # Probe [0, MAX_PRICE]: totalOffers=3000 > 1400 → split на [0, mid] и [mid+1, MAX_PRICE]
+    # Probe [0, mid] (mid = MAX_PRICE//2 = 100_000_000): totalOffers=800 <= cap → пагинировать
+    # Probe [mid+1, MAX_PRICE]: totalOffers=700 <= cap → пагинировать
+
+    mid = _MAX_PRICE // 2
+
+    call_log: list[tuple] = []
+
+    async def fake_fetch_page_html(
+        rooms: tuple, page: int, min_price: int | None, max_price: int | None
+    ) -> str:
+        call_log.append((rooms, page, min_price, max_price))
+        return f"<html>page={page} min={min_price} max={max_price}</html>"
+
+    def fake_extract_total_offers(html: str) -> int | None:
+        # Парсим min/max из fake html
+        import re
+
+        min_m = re.search(r"min=(\w+)", html)
+        max_m = re.search(r"max=(\w+)", html)
+        min_p = int(min_m.group(1)) if min_m and min_m.group(1) != "None" else 0
+        max_p = int(max_m.group(1)) if max_m and max_m.group(1) != "None" else _MAX_PRICE
+
+        # Имитируем разделение: весь диапазон > 1400, левая и правая половины <= 1400
+        if max_p is None or max_p >= _MAX_PRICE:
+            if min_p == 0:
+                return 3000  # весь диапазон — нужно split
+        if max_p == mid:
+            return 800  # левая половина — пагинировать
+        if min_p == mid + 1:
+            return 700  # правая половина — пагинировать
+        return 3000  # всё остальное — split
+
+    def fake_parse_serp_html(html: str) -> list[ScrapedLot]:
+        import re
+
+        page_m = re.search(r"page=(\d+)", html)
+        min_m = re.search(r"min=(\w+)", html)
+        max_m = re.search(r"max=(\w+)", html)
+        page = int(page_m.group(1)) if page_m else 1
+        min_p = min_m.group(1) if min_m else "0"
+        max_p = max_m.group(1) if max_m else "MAX"
+        # Возвращаем по 2 лота на первой странице каждого пагинируемого бакета
+        if page == 1:
+            return [
+                _make_lot(f"lot_{min_p}_{max_p}_p{page}_1"),
+                _make_lot(f"lot_{min_p}_{max_p}_p{page}_2"),
+            ]
+        return []  # страница 2+ — пусто → early stop
+
+    async def fake_sleep(*_: object, **__: object) -> None:
+        pass
+
+    with (
+        patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html),
+        patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers),
+        patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html),
+        patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock),
+        patch.object(scraper, "_rotate_ip", return_value=False),
+    ):
+        lots = await scraper.fetch_all_secondary(
+            rooms_buckets=[(1,)],
+            price_cap_per_bucket=1400,
+        )
+
+    # Должны получить лоты из ОБОИХ суб-бакетов после split
+    assert len(lots) > 0, "Должны получить лоты после разделения бакета"
+
+    # Проверяем что был probe с большим диапазоном И с sub-бакетами
+    probed_ranges = [(min_p, max_p) for (_, page, min_p, max_p) in call_log if page == 1]
+    # Должен быть probe [0, None] или [0, MAX] — начальный probe всего диапазона
+    initial_probes = [r for r in probed_ranges if r[0] is None or r[0] == 0]
+    assert (
+        len(initial_probes) >= 1
+    ), f"Ожидался probe всего диапазона, probed_ranges={probed_ranges}"
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_secondary_paginates_when_under_cap(scraper: CianScraper) -> None:
+    """totalOffers <= cap → пагинирует бакет полностью (страница за страницей)."""
+    pages_fetched: list[int] = []
+
+    async def fake_fetch_page_html(
+        rooms: tuple, page: int, min_price: int | None, max_price: int | None
+    ) -> str:
+        pages_fetched.append(page)
+        return f"<html>page={page}</html>"
+
+    def fake_extract_total_offers(html: str) -> int | None:
+        return 56  # 56 офферов → ceil(56/28) = 2 страницы
+
+    def fake_parse_serp_html(html: str) -> list[ScrapedLot]:
+        import re
+
+        page_m = re.search(r"page=(\d+)", html)
+        page = int(page_m.group(1)) if page_m else 1
+        if page <= 2:
+            return [_make_lot(f"lot_p{page}_{i}") for i in range(28)]
+        return []
+
+    with (
+        patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html),
+        patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers),
+        patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html),
+        patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock),
+        patch.object(scraper, "_rotate_ip", return_value=False),
+    ):
+        lots = await scraper.fetch_all_secondary(
+            rooms_buckets=[(2,)],
+            price_cap_per_bucket=1400,
+        )
+
+    # 28 лотов на стр 1 + 28 на стр 2 = 56, дедуп по source_id
+    assert len(lots) == 56, f"Ожидалось 56 уникальных лотов, получено {len(lots)}"
+    # Должно быть 2 страницы пагинации (page=1 probe + page=2)
+    assert 2 in pages_fetched, "Должна быть запрошена страница 2"
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_secondary_min_bracket_guard(scraper: CianScraper) -> None:
+    """Если hi - lo < MIN_BRACKET → пагинируем как есть (не делим бесконечно)."""
+    call_count = 0
+
+    async def fake_fetch_page_html(
+        rooms: tuple, page: int, min_price: int | None, max_price: int | None
+    ) -> str:
+        nonlocal call_count
+        call_count += 1
+        if call_count > 100:
+            raise RuntimeError("Бесконечная рекурсия — MIN_BRACKET guard не сработал")
+        return f"<html>page={page}</html>"
+
+    def fake_extract_total_offers(html: str) -> int | None:
+        # Всегда > cap → без guard делилось бы бесконечно
+        return 5000
+
+    def fake_parse_serp_html(html: str) -> list[ScrapedLot]:
+        return []  # пустой → early stop в пагинации
+
+    with (
+        patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html),
+        patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers),
+        patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html),
+        patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock),
+        patch.object(scraper, "_rotate_ip", return_value=False),
+    ):
+        # Запускаем с очень узким диапазоном — guard должен остановить деление
+        await scraper._walk_price_range(
+            rooms=(1,),
+            lo=1_000_000,
+            hi=1_000_000 + _MIN_BRACKET - 1,  # bracket < MIN_BRACKET
+            seen={},
+            price_cap_per_bucket=1400,
+            max_pages_per_bucket=54,
+        )
+
+    # Тест прошёл если нет RuntimeError (guard сработал)
+    assert call_count <= 10, f"Слишком много вызовов ({call_count}) — guard не сработал"