perf(tradein): avito seed price-brackets with open top bucket (≤2k blocks, no price ceiling)

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

@@ -229,10 +229,32 @@ NOVOSTROYKA_SLUG = "novostroyka-ASgBAgICAkSSA8YQ5geOUg"
 # ── Exhaustive full-load (room×price bisection) ───────────────────────────────
 # Верхняя граница цены при первом рекурсивном делении (нет явного hi).
 # 200 млн ₽ заведомо выше ТОПа ЕКБ — дальние бакеты дадут 0 результатов.
+# Больше НЕ корень бисекции (см. _AVITO_PRICE_SEED_BRACKETS) — оставлен для
+# совместимости (используется в тестах probe-устойчивости как sentinel).
 _AVITO_MAX_PRICE = 200_000_000
 # Минимальный ценовой диапазон для рекурсии; при hi - lo < MIN_BRACKET прекращаем
 # деление и пагинируем как есть (принимаем возможный «хвост»).
 _AVITO_MIN_BRACKET = 50_000
+
+# Seed price-brackets (руб) — data-driven по прод-распределению вторички ЕКБ
+# (пик 4-8М, bulk 3-16М; самая плотная комнатность — студии 5783, единственный
+# >2к блок был студии 4-6М → пик дроблю по 1М). Каждый (комнатность × брекет)
+# дальше дробится _walk_price_range если total > cap (=2000 → блоки ≤2к).
+# Последний брекет ОТКРЫТ (hi=None → pmax не ставится) — ловит весь хвост люкса
+# без потолка; он крошечный (avito >50М ≈ 73), пагинируется напрямую без бисекции.
+_AVITO_PRICE_SEED_BRACKETS: list[tuple[int, int | None]] = [
+    (0, 4_000_000),
+    (4_000_000, 5_000_000),
+    (5_000_000, 6_000_000),
+    (6_000_000, 7_000_000),
+    (7_000_000, 8_000_000),
+    (8_000_000, 10_000_000),
+    (10_000_000, 13_000_000),
+    (13_000_000, 17_000_000),
+    (17_000_000, 25_000_000),
+    (25_000_000, 250_000_000),
+    (250_000_000, None),  # 250М+ открытый catch-all (без потолка)
+]
 # Avito SERP показывает ~50 карточек на страницу.
 _AVITO_OFFERS_PER_PAGE = 50
 
@@ -639,7 +661,7 @@ class AvitoScraper(BaseScraper):
         return await self._fetch_serp_html(url, page)
 
     async def _probe_total(
-        self, room_slug: str, min_price: int | None, max_price: int
+        self, room_slug: str, min_price: int | None, max_price: int | None
     ) -> tuple[str | None, int | None]:
         """Probe page=1 бакета и извлекает total, устойчиво к browser-ошибкам.
 
@@ -661,7 +683,7 @@ class AvitoScraper(BaseScraper):
             raise
         except (httpx.HTTPStatusError, httpx.TimeoutException, httpx.HTTPError) as exc:
             logger.warning(
-                "avito: probe browser error %s for %s [%s, %d] — treating as total=None",
+                "avito: probe browser error %s for %s [%s, %s] — treating as total=None",
                 type(exc).__name__,
                 room_slug,
                 min_price,
@@ -670,7 +692,7 @@ class AvitoScraper(BaseScraper):
             return None, None
         except Exception:
             logger.warning(
-                "avito: probe unexpected error for %s [%s, %d] — treating as total=None",
+                "avito: probe unexpected error for %s [%s, %s] — treating as total=None",
                 room_slug,
                 min_price,
                 max_price,
@@ -695,11 +717,16 @@ class AvitoScraper(BaseScraper):
     ) -> list[ScrapedLot]:
         """Exhaustive-загрузка Avito ЕКБ вторички через КОМНАТНОСТЬ × ЦЕНА.
 
-        Зеркало CianScraper.fetch_all_secondary. Обходит Avito SERP-cap (~5000
+        Обходит Avito SERP-cap (~5000 результатов на запрос). Вместо корень-бисекции
-        результатов на запрос): внутри каждой комнатности адаптивно бьёт ценовой
+        [0, _AVITO_MAX_PRICE] на каждую комнатность (≈5 холостых probe-спусков через
-        диапазон на бакеты так, чтобы в каждом total < price_cap_per_bucket →
+        пустой верх, т.к. вторичка ЕКБ почти вся <16М) — стартуем с data-driven
-        пагинирует бакет полностью. Страницы leaf-бакета — параллельно (Semaphore).
+        seed-сетки ценовых брекетов (_AVITO_PRICE_SEED_BRACKETS) по прод-распределению.
-        Дедуп по source_id (dict seen).
+        Каждый (комнатность × seed-брекет) дальше дробится _walk_price_range если
+        total > price_cap_per_bucket. Последний seed-брекет ОТКРЫТ (hi=None → pmax
+        не ставится) — ловит весь хвост люкса без потолка.
+
+        Страницы leaf-бакета — параллельно (Semaphore). Дедуп по source_id (dict seen),
+        ОБЩИЙ по всем брекетам и комнатностям.
 
         Параметры:
           rooms_buckets: список (name, slug) комнатностей (default: все ROOM_SLUGS,
@@ -731,19 +758,25 @@ class AvitoScraper(BaseScraper):
                 secondary_only,
             )
             before = len(seen)
-            await self._walk_price_range(
+            # Перебираем seed-брекеты: для ЗАКРЫТЫХ передаём hi = br_hi - 1
-                room_slug=slug,
+            # (pmax эксклюзивный сверху → соседние брекеты не пересекаются), для
-                room_label=room_label,
+            # ОТКРЫТОГО (br_hi is None) — hi=None (без потолка). Дедуп по source_id
-                lo=0,
+            # в общем `seen` страхует на стыках.
-                hi=_AVITO_MAX_PRICE,
+            for br_lo, br_hi in _AVITO_PRICE_SEED_BRACKETS:
-                seen=seen,
+                walk_hi = br_hi - 1 if br_hi is not None else None
-                price_cap_per_bucket=price_cap_per_bucket,
+                await self._walk_price_range(
-                max_pages_per_bucket=max_pages_per_bucket,
+                    room_slug=slug,
-                concurrency=concurrency,
+                    room_label=room_label,
-                secondary_only=secondary_only,
+                    lo=br_lo,
-                on_bucket=on_bucket,
+                    hi=walk_hi,
-                skip_buckets=skip_buckets,
+                    seen=seen,
-            )
+                    price_cap_per_bucket=price_cap_per_bucket,
+                    max_pages_per_bucket=max_pages_per_bucket,
+                    concurrency=concurrency,
+                    secondary_only=secondary_only,
+                    on_bucket=on_bucket,
+                    skip_buckets=skip_buckets,
+                )
             room_collected = len(seen) - before
             logger.info(
                 "avito exhaustive: %s done — collected %d (total unique=%d)",
@@ -763,7 +796,7 @@ class AvitoScraper(BaseScraper):
         room_slug: str,
         room_label: str,
         lo: int,
-        hi: int,
+        hi: int | None,
         seen: dict[str, ScrapedLot],
         price_cap_per_bucket: int,
         max_pages_per_bucket: int,
@@ -775,14 +808,19 @@ class AvitoScraper(BaseScraper):
     ) -> None:
         """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi].
 
-        Зеркало CianScraper._walk_price_range. Алгоритм:
+        Алгоритм для ЗАКРЫТОГО брекета (hi is not None):
           1. Запросить page=1 с pmin=lo, pmax=hi → total (page-title/count).
           2. Если total <= cap → пагинировать leaf-бакет параллельно (concurrency).
           3. Если total > cap → разбить бакет пополам (рекурсия).
           Guard: hi - lo < _AVITO_MIN_BRACKET → пагинировать как есть (WARNING).
 
+        ОТКРЫТЫЙ брекет (hi is None) — верхний seed-брекет без потолка (pmax не
+        ставится, avito отдаёт всё ≥lo). Делить нельзя (mid=(lo+hi)//2 невозможен) —
+        probe + пагинируем leaf напрямую. Хвост 250М+ крошечный (≈73), за cap
+        обычно не выходит; если вышел — пагинируем до max_pages + WARNING (acceptable).
+
         После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots).
-        bucket_key = "room_label:lo:hi" (checkpoint-ключ для resume).
+        bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый).
         on_bucket может быть async или sync. Исключение в on_bucket прерывает прогон.
         skip_buckets: если bucket_key в skip_buckets — пагинация и on_bucket пропускаются.
         secondary_only: новостройки (listing_segment=="novostroyki") отбрасываются после
@@ -790,6 +828,7 @@ class AvitoScraper(BaseScraper):
         AvitoBlockedError/AvitoRateLimitedError из _fetch_serp_html пробрасываются наверх.
         """
         _lo_param = lo if lo > 0 else None
+        _hi_repr = "open" if hi is None else str(hi)
 
         # ── Шаг 1: probe page 1 ────────────────────────────────────────────────
         # _probe_total ловит browser-ошибки (HTTP 500 / timeout от tradein-browser
@@ -800,10 +839,10 @@ class AvitoScraper(BaseScraper):
         # Ретрай на captcha/missing-count/browser-ошибке: rotate IP + 1 retry.
         if total is None:
             logger.warning(
-                "avito: total=None for %s [%d, %d] depth=%d — rotating IP + retry",
+                "avito: total=None for %s [%d, %s] depth=%d — rotating IP + retry",
                 room_label,
                 lo,
-                hi,
+                _hi_repr,
                 _depth,
             )
             rotated = await self._rotate_ip()
@@ -811,6 +850,33 @@ class AvitoScraper(BaseScraper):
                 html, total = await self._probe_total(room_slug, _lo_param, hi)
 
         if total is None:
+            # ОТКРЫТЫЙ брекет: делить нельзя (нет hi). Probe failed → нет total для
+            # ceil(); пагинируем до max_pages_per_bucket (best-effort, хвост 250М+
+            # крошечный) и WARNING. НЕ роняем прогон.
+            if hi is None:
+                logger.warning(
+                    "avito: probe failed for OPEN bucket %s [%d, open] depth=%d — "
+                    "paginating up to max_pages (tail-loss accepted)",
+                    room_label,
+                    lo,
+                    _depth,
+                )
+                await self._paginate_leaf_bucket(
+                    room_slug=room_slug,
+                    room_label=room_label,
+                    lo=lo,
+                    hi=None,
+                    html=None,
+                    max_pages=max_pages_per_bucket,
+                    seen=seen,
+                    price_cap_per_bucket=price_cap_per_bucket,
+                    max_pages_per_bucket=max_pages_per_bucket,
+                    concurrency=concurrency,
+                    secondary_only=secondary_only,
+                    on_bucket=on_bucket,
+                    skip_buckets=skip_buckets,
+                )
+                return
             bracket_size = hi - lo
             # Probe не удался даже после ретрая. Если бакет ещё делим — НЕ теряем
             # его целиком: тяжёлая страница (огромная выдача) скорее всего станет
@@ -867,10 +933,10 @@ class AvitoScraper(BaseScraper):
             return
 
         logger.info(
-            "avito: %s [%d, %d] total=%d depth=%d",
+            "avito: %s [%d, %s] total=%d depth=%d",
             room_label,
             lo,
-            hi,
+            _hi_repr,
             total,
             _depth,
         )
@@ -878,6 +944,39 @@ class AvitoScraper(BaseScraper):
         if total == 0:
             return
 
+        # ── ОТКРЫТЫЙ брекет (hi is None): делить нельзя — пагинируем leaf напрямую ─
+        # total известен из probe → max_pages = min(ceil(total/per_page), cap).
+        # Если total > cap (не должно для 250М+, но guard) — пагинируем как есть до
+        # max_pages_per_bucket с WARNING (хвост люкса крошечный, tail-loss accepted).
+        if hi is None:
+            if total > price_cap_per_bucket:
+                logger.warning(
+                    "avito: OPEN bucket %s [%d, open] total=%d > cap=%d — paginating "
+                    "without split (tail-loss accepted, lux tail tiny)",
+                    room_label,
+                    lo,
+                    total,
+                    price_cap_per_bucket,
+                )
+            pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE)
+            max_pages = min(pages_needed, max_pages_per_bucket)
+            await self._paginate_leaf_bucket(
+                room_slug=room_slug,
+                room_label=room_label,
+                lo=lo,
+                hi=None,
+                html=html,
+                max_pages=max_pages,
+                seen=seen,
+                price_cap_per_bucket=price_cap_per_bucket,
+                max_pages_per_bucket=max_pages_per_bucket,
+                concurrency=concurrency,
+                secondary_only=secondary_only,
+                on_bucket=on_bucket,
+                skip_buckets=skip_buckets,
+            )
+            return
+
         # ── Шаг 2: деление или пагинация ──────────────────────────────────────
         bracket_size = hi - lo
         need_split = total > price_cap_per_bucket
@@ -932,15 +1031,7 @@ class AvitoScraper(BaseScraper):
             )
             return
 
-        # ── Checkpoint-ключ leaf-бакета ──────────────────────────────────────
+        # ── Закрытый leaf-бакет: считаем max_pages + warning, делегируем helper'у ─
-        bucket_key = f"{room_label}:{lo}:{hi}"
-
-        # Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket.
-        if skip_buckets and bucket_key in skip_buckets:
-            logger.info("avito: skip bucket %s — already done (resume)", bucket_key)
-            return
-
-        # ── Параллельная пагинация leaf-бакета ────────────────────────────────
         pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE)
         max_pages = min(pages_needed, max_pages_per_bucket)
 
@@ -959,26 +1050,81 @@ class AvitoScraper(BaseScraper):
                 tail_loss,
             )
 
-        # Страница 1 уже есть (html из probe выше); остальные — параллельно.
+        await self._paginate_leaf_bucket(
+            room_slug=room_slug,
+            room_label=room_label,
+            lo=lo,
+            hi=hi,
+            html=html,
+            max_pages=max_pages,
+            seen=seen,
+            price_cap_per_bucket=price_cap_per_bucket,
+            max_pages_per_bucket=max_pages_per_bucket,
+            concurrency=concurrency,
+            secondary_only=secondary_only,
+            on_bucket=on_bucket,
+            skip_buckets=skip_buckets,
+        )
+
+    async def _paginate_leaf_bucket(
+        self,
+        *,
+        room_slug: str,
+        room_label: str,
+        lo: int,
+        hi: int | None,
+        html: str | None,
+        max_pages: int,
+        seen: dict[str, ScrapedLot],
+        price_cap_per_bucket: int,
+        max_pages_per_bucket: int,
+        concurrency: int,
+        secondary_only: bool,
+        on_bucket: Callable[..., Any] | None,
+        skip_buckets: set[str] | None,
+    ) -> None:
+        """Параллельная пагинация одного leaf-бакета + фильтр + дедуп + on_bucket.
+
+        Общая для ЗАКРЫТЫХ ([lo, hi]) и ОТКРЫТОГО (hi=None → pmax не ставится)
+        брекетов. ``html`` — HTML страницы 1 из probe (переиспользуем, не
+        перезапрашиваем); для открытого-без-probe передаётся None → стр.1 фетчится
+        наравне с остальными. ``max_pages`` уже посчитан вызывающим.
+
+        bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый).
+        skip_buckets: если bucket_key в skip_buckets — пагинация и on_bucket пропускаются.
+        AvitoBlockedError/AvitoRateLimitedError из page-фетчей пробрасываются наверх.
+        """
+        _lo_param = lo if lo > 0 else None
+        _hi_param = hi  # None → _build_rooms_url не ставит pmax
+        _hi_repr = "open" if hi is None else str(hi)
+        bucket_key = f"{room_label}:{lo}:{_hi_repr}"
+
+        # Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket.
+        if skip_buckets and bucket_key in skip_buckets:
+            logger.info("avito: skip bucket %s — already done (resume)", bucket_key)
+            return
+
+        # Страница 1 — из probe-html если он есть (закрытый/открытый с probe);
+        # иначе фетчим её как обычную страницу (открытый брекет без probe-html).
         sem = asyncio.Semaphore(concurrency)
-        first_url = self._build_rooms_url(room_slug, 1, _lo_param, hi)
+        first_url = self._build_rooms_url(room_slug, 1, _lo_param, _hi_param)
 
         async def _one_page(p: int) -> list[ScrapedLot]:
-            if p == 1:
+            if p == 1 and html is not None:
-                return self._parse_html(html, source_url_base=first_url) if html else []
+                return self._parse_html(html, source_url_base=first_url)
             async with sem:
-                page_html = await self._fetch_rooms_page_html(room_slug, p, _lo_param, hi)
+                page_html = await self._fetch_rooms_page_html(room_slug, p, _lo_param, _hi_param)
                 await asyncio.sleep(self.request_delay_sec)
                 if page_html is None:
                     logger.warning(
-                        "avito: page_html=None %s [%d, %d] page=%d — skipping page",
+                        "avito: page_html=None %s [%d, %s] page=%d — skipping page",
                         room_label,
                         lo,
-                        hi,
+                        _hi_repr,
                         p,
                     )
                     return []
-                page_url = self._build_rooms_url(room_slug, p, _lo_param, hi)
+                page_url = self._build_rooms_url(room_slug, p, _lo_param, _hi_param)
                 return self._parse_html(page_html, source_url_base=page_url)
 
         page_results = await asyncio.gather(
@@ -993,10 +1139,10 @@ class AvitoScraper(BaseScraper):
                 if isinstance(res, AvitoBlockedError | AvitoRateLimitedError):
                     raise res
                 logger.warning(
-                    "avito: page exception %s [%d, %d] page=%d — %r",
+                    "avito: page exception %s [%d, %s] page=%d — %r",
                     room_label,
                     lo,
-                    hi,
+                    _hi_repr,
                     p_idx,
                     res,
                 )
@@ -1019,10 +1165,10 @@ class AvitoScraper(BaseScraper):
                 seen[key] = lot
 
         logger.info(
-            "avito: %s [%d, %d] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d",
+            "avito: %s [%d, %s] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d",
             room_label,
             lo,
-            hi,
+            _hi_repr,
             max_pages,
             collected_this_bucket,
             dropped_nb,

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

@@ -23,6 +23,7 @@ from app.services.scrapers.avito import (
     _AVITO_MAX_PRICE,
     _AVITO_MIN_BRACKET,
     _AVITO_OFFERS_PER_PAGE,
+    _AVITO_PRICE_SEED_BRACKETS,
     AvitoScraper,
 )
 from app.services.scrapers.avito_exceptions import AvitoBlockedError
@@ -119,8 +120,12 @@ def test_extract_total_count_missing_returns_none(scraper: AvitoScraper) -> None
 
 @pytest.mark.asyncio
 async def test_fetch_all_secondary_splits_on_cap(scraper: AvitoScraper) -> None:
-    """total > cap → бакет делится; оба суб-бакета пагинируются."""
+    """Внутри seed-брекета total > cap → бакет делится пополам; суб-бакеты пагинируются."""
-    mid = _AVITO_MAX_PRICE // 2
+    # Первый закрытый seed-брекет (0, 4_000_000) → walk_hi=3_999_999.
+    br_lo, br_hi_raw = _AVITO_PRICE_SEED_BRACKETS[0]
+    assert br_hi_raw is not None
+    br_hi = br_hi_raw - 1  # эксклюзивный верх
+    mid = (br_lo + br_hi) // 2
     call_log: list[tuple] = []
 
     async def fake_fetch(
@@ -135,12 +140,16 @@ async def test_fetch_all_secondary_splits_on_cap(scraper: AvitoScraper) -> None:
         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 _AVITO_MAX_PRICE
+        max_p = int(max_m.group(1)) if max_m and max_m.group(1) != "None" else None
-        if max_p == mid:
+        # Внутри первого брекета [0, 3_999_999]: весь брекет > cap → split;
+        # половинки < cap → пагинируются. Прочие seed-брекеты — пусто (0).
+        if min_p == 0 and max_p == br_hi:
+            return 3000  # весь первый брекет — split
+        if min_p == 0 and max_p == mid:
             return 800  # левая половина — пагинировать
-        if min_p == mid + 1:
+        if min_p == mid + 1 and max_p == br_hi:
             return 700  # правая половина — пагинировать
-        return 3000  # весь диапазон — split
+        return 0  # остальные seed-брекеты пусты
 
     def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
         import re
@@ -170,7 +179,7 @@ async def test_fetch_all_secondary_splits_on_cap(scraper: AvitoScraper) -> None:
         )
 
     assert len(lots) > 0
-    # Должны быть пробы суб-бакетов (после split весь диапазон не пагинируется)
+    # Должны быть пробы суб-бакетов (после split весь брекет не пагинируется).
     probed_max = [mx for (p, _mn, mx) in call_log if p == 1]
     assert any(mx == mid for mx in probed_max), f"Ожидался probe левой половины, got {probed_max}"
 
@@ -292,16 +301,23 @@ async def test_fetch_all_secondary_concurrent_pages_deduped(scraper: AvitoScrape
 
 @pytest.mark.asyncio
 async def test_on_bucket_receives_key_and_lots(scraper: AvitoScraper) -> None:
-    """on_bucket получает (bucket_key, lots) — bucket_key = room_label:lo:hi."""
+    """on_bucket получает (bucket_key, lots) на каждый seed-брекет.
+
+    bucket_key закрытого брекета = "room_label:lo:hi" (hi = br_hi-1, эксклюзивный);
+    открытого (последнего) = "room_label:lo:open". Одна on_bucket-нотификация
+    на каждый непустой seed-брекет (здесь все непустые → len(brackets) вызовов).
+    """
     received: list[tuple[str, int]] = []
 
     def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None:
         received.append((bucket_key, len(lots)))
 
+    # Уникальные source_id на каждый брекет, чтобы on_bucket получал непустые
+    # списки во ВСЕХ брекетах (дедуп в seen не должен опустошать поздние брекеты).
     async def fake_fetch(
         room_slug: str, page: int, min_price: int | None, max_price: int | None
     ) -> str:
-        return f"<html>page={page}</html>"
+        return f"<html>page={page} min={min_price} max={max_price}</html>"
 
     def fake_total(html: str) -> int | None:
         return 5  # 1 страница
@@ -309,9 +325,13 @@ async def test_on_bucket_receives_key_and_lots(scraper: AvitoScraper) -> None:
     def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
         import re
 
+        min_m = re.search(r"min=(\w+)", html)
+        max_m = re.search(r"max=(\w+)", html)
+        min_p = min_m.group(1) if min_m else "0"
+        max_p = max_m.group(1) if max_m else "open"
         page_m = re.search(r"page=(\d+)", html)
         page = int(page_m.group(1)) if page_m else 1
-        return [_make_lot(f"lot_p{page}_{i}") for i in range(5)]
+        return [_make_lot(f"lot_{min_p}_{max_p}_p{page}_{i}") for i in range(5)]
 
     with (
         patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
@@ -325,34 +345,55 @@ async def test_on_bucket_receives_key_and_lots(scraper: AvitoScraper) -> None:
             on_bucket=fake_on_bucket,
         )
 
-    assert len(received) == 1, f"Ожидался 1 вызов on_bucket, получено {len(received)}"
+    # Один on_bucket на каждый seed-брекет (все непустые).
-    key, count = received[0]
+    assert len(received) == len(
-    assert key == f"room_2_komn:0:{_AVITO_MAX_PRICE}", f"Неверный bucket_key: {key!r}"
+        _AVITO_PRICE_SEED_BRACKETS
-    assert count == 5
+    ), f"Ожидалось {len(_AVITO_PRICE_SEED_BRACKETS)} вызовов on_bucket, получено {len(received)}"
+    keys = [k for (k, _c) in received]
+    # Закрытые брекеты: hi = br_hi - 1. Открытый (последний): ':open'.
+    first_lo, first_hi = _AVITO_PRICE_SEED_BRACKETS[0]
+    assert first_hi is not None
+    assert keys[0] == f"room_2_komn:{first_lo}:{first_hi - 1}", f"Неверный ключ: {keys[0]!r}"
+    last_lo, last_hi = _AVITO_PRICE_SEED_BRACKETS[-1]
+    assert last_hi is None
+    assert keys[-1] == f"room_2_komn:{last_lo}:open", f"Открытый ключ неверен: {keys[-1]!r}"
+    assert all(count == 5 for (_k, count) in received)
 
 
 @pytest.mark.asyncio
 async def test_skip_buckets_skips_pagination(scraper: AvitoScraper) -> None:
-    """skip_buckets: probe выполняется, но пагинация и on_bucket пропускаются."""
+    """skip_buckets: для skip-бакета probe выполняется, пагинация и on_bucket — нет."""
-    fetch_calls: list[int] = []
+    fetch_calls: list[tuple[int, int | None, int | None]] = []
     on_bucket_calls: list[str] = []
 
     async def fake_fetch(
         room_slug: str, page: int, min_price: int | None, max_price: int | None
     ) -> str:
-        fetch_calls.append(page)
+        fetch_calls.append((page, min_price, max_price))
-        return f"<html>page={page}</html>"
+        return f"<html>page={page} min={min_price} max={max_price}</html>"
 
     def fake_total(html: str) -> int | None:
         return 5
 
     def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
-        return [_make_lot("lot_1")]
+        import re
+
+        min_m = re.search(r"min=(\w+)", html)
+        max_m = re.search(r"max=(\w+)", html)
+        return [_make_lot(f"lot_{min_m.group(1) if min_m else 0}_{max_m.group(1) if max_m else 0}")]
 
     def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None:
         on_bucket_calls.append(bucket_key)
 
-    skip_set = {f"room_1_komn:0:{_AVITO_MAX_PRICE}"}
+    # Skip ВСЕ брекеты → ни одной on_bucket-нотификации, но каждый брекет
+    # всё равно делает probe (page=1).
+    first_lo, first_hi = _AVITO_PRICE_SEED_BRACKETS[0]
+    assert first_hi is not None
+    first_key = f"room_1_komn:{first_lo}:{first_hi - 1}"
+    skip_all: set[str] = set()
+    for br_lo, br_hi in _AVITO_PRICE_SEED_BRACKETS:
+        hi_repr = "open" if br_hi is None else str(br_hi - 1)
+        skip_all.add(f"room_1_komn:{br_lo}:{hi_repr}")
 
     with (
         patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
@@ -364,11 +405,18 @@ async def test_skip_buckets_skips_pagination(scraper: AvitoScraper) -> None:
             rooms_buckets=[("1-комн.", "1-komnatnye-X")],
             price_cap_per_bucket=1400,
             on_bucket=fake_on_bucket,
-            skip_buckets=skip_set,
+            skip_buckets=skip_all,
         )
 
-    assert fetch_calls == [1], f"Ожидался ровно 1 вызов (probe), got {fetch_calls}"
+    # Ровно один probe (page=1) на каждый seed-брекет, без доп. страниц.
-    assert on_bucket_calls == [], "on_bucket не должен вызываться для skip-бакета"
+    assert first_key in skip_all
+    assert all(
+        p == 1 for (p, _mn, _mx) in fetch_calls
+    ), f"Ожидались только probe-вызовы: {fetch_calls}"
+    assert len(fetch_calls) == len(
+        _AVITO_PRICE_SEED_BRACKETS
+    ), f"Ожидался 1 probe на брекет ({len(_AVITO_PRICE_SEED_BRACKETS)}), got {len(fetch_calls)}"
+    assert on_bucket_calls == [], "on_bucket не должен вызываться для skip-бакетов"
     assert lots == []
 
 
@@ -465,12 +513,18 @@ async def test_probe_browser_500_on_wide_bucket_splits_not_crash(
     browser HTTP 500. Раньше это роняло весь full-load с 0 uniq. Теперь широкий
     bucket принудительно делится пополам; под-бакеты пагинируются нормально.
     """
-    probe_calls: list[tuple[int | None, int]] = []
+    # Первый закрытый seed-брекет (0, 4_000_000) → walk_hi=3_999_999.
+    _br_lo, br_hi_raw = _AVITO_PRICE_SEED_BRACKETS[0]
+    assert br_hi_raw is not None
+    br_hi = br_hi_raw - 1
+    probe_calls: list[tuple[int | None, int | None]] = []
 
     async def fake_fetch(
         room_slug: str, page: int, min_price: int | None, max_price: int | None
     ) -> str:
-        if page == 1 and (min_price in (None, 0)) and max_price == _AVITO_MAX_PRICE:
+        # Probe верхней границы ШИРОКОГО первого брекета падает (имитация HTTP 500
+        # от tradein-browser на тяжёлой выдаче) → должен произойти split.
+        if page == 1 and (min_price in (None, 0)) and max_price == br_hi:
             probe_calls.append((min_price, max_price))
             raise _http_500_error()
         return f"<html>page={page} min={min_price} max={max_price}</html>"
@@ -483,10 +537,12 @@ async def test_probe_browser_500_on_wide_bucket_splits_not_crash(
 
         page_m = re.search(r"page=(\d+)", html)
         page = int(page_m.group(1)) if page_m else 1
+        min_m = re.search(r"min=(\w+)", html)
         max_m = re.search(r"max=(\w+)", html)
+        min_p = min_m.group(1) if min_m else "0"
         max_p = max_m.group(1) if max_m else "MAX"
         if page == 1:
-            return [_make_lot(f"lot_{max_p}_{i}") for i in range(3)]
+            return [_make_lot(f"lot_{min_p}_{max_p}_{i}") for i in range(3)]
         return []
 
     with (
@@ -500,7 +556,7 @@ async def test_probe_browser_500_on_wide_bucket_splits_not_crash(
             price_cap_per_bucket=1400,
         )
 
-    assert probe_calls, "Probe полного диапазона должен был вызваться и упасть"
+    assert probe_calls, "Probe первого (широкого) брекета должен был вызваться и упасть"
     assert len(lots) > 0, f"Ожидались лоты из под-бакетов после split, got {len(lots)}"
 
 
@@ -509,16 +565,19 @@ async def test_probe_timeout_on_wide_bucket_splits_not_crash(
     scraper: AvitoScraper,
 ) -> None:
     """httpx.TimeoutException на широком bucket обрабатывается как HTTP 500 → split."""
+    _br_lo, br_hi_raw = _AVITO_PRICE_SEED_BRACKETS[0]
+    assert br_hi_raw is not None
+    br_hi = br_hi_raw - 1
     probe_failed = False
 
     async def fake_fetch(
         room_slug: str, page: int, min_price: int | None, max_price: int | None
     ) -> str:
         nonlocal probe_failed
-        if page == 1 and (min_price in (None, 0)) and max_price == _AVITO_MAX_PRICE:
+        if page == 1 and (min_price in (None, 0)) and max_price == br_hi:
             probe_failed = True
             raise httpx.TimeoutException("Page.goto timeout")
-        return f"<html>page={page} max={max_price}</html>"
+        return f"<html>page={page} min={min_price} max={max_price}</html>"
 
     def fake_total(html: str) -> int | None:
         return 40
@@ -528,9 +587,11 @@ async def test_probe_timeout_on_wide_bucket_splits_not_crash(
 
         page_m = re.search(r"page=(\d+)", html)
         page = int(page_m.group(1)) if page_m else 1
+        min_m = re.search(r"min=(\w+)", html)
         max_m = re.search(r"max=(\w+)", html)
+        min_p = min_m.group(1) if min_m else "0"
         max_p = max_m.group(1) if max_m else "MAX"
-        return [_make_lot(f"lot_{max_p}_{i}") for i in range(2)] if page == 1 else []
+        return [_make_lot(f"lot_{min_p}_{max_p}_{i}") for i in range(2)] if page == 1 else []
 
     with (
         patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
@@ -634,3 +695,197 @@ async def test_probe_avito_blocked_propagates_not_swallowed(
                 rooms_buckets=[("1-комн.", "1-komnatnye-X")],
                 price_cap_per_bucket=1400,
             )
+
+
+# ── Seed price-brackets entry point (perf: open top bucket) ───────────────────
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_secondary_walks_seed_brackets_per_room(
+    scraper: AvitoScraper,
+) -> None:
+    """fetch_all_secondary зовёт _walk_price_range РОВНО len(seed) раз на комнатность.
+
+    Аргументы lo/hi совпадают с seed-брекетами: закрытые передаются с hi=br_hi-1
+    (эксклюзивный верх, без overlap на стыках), последний (открытый) — hi=None.
+    """
+    walk_calls: list[tuple[int, int | None]] = []
+
+    async def spy_walk(*args: object, **kwargs: object) -> None:
+        walk_calls.append((kwargs["lo"], kwargs["hi"]))  # type: ignore[arg-type]
+        # Не уходим в реальную рекурсию — фиксируем только верхнеуровневые входы.
+        return None
+
+    with patch.object(scraper, "_walk_price_range", side_effect=spy_walk):
+        await scraper.fetch_all_secondary(
+            rooms_buckets=[("1-комн.", "1-komnatnye-X")],
+            price_cap_per_bucket=2000,
+        )
+
+    expected = [
+        (br_lo, (br_hi - 1) if br_hi is not None else None)
+        for (br_lo, br_hi) in _AVITO_PRICE_SEED_BRACKETS
+    ]
+    assert walk_calls == expected, f"Неверные входы _walk_price_range: {walk_calls}"
+    # Последний — открытый (hi=None).
+    assert walk_calls[-1][1] is None, "Последний seed-брекет должен быть открытым (hi=None)"
+    # Ровно один вызов на seed-брекет (1 комнатность).
+    assert len(walk_calls) == len(_AVITO_PRICE_SEED_BRACKETS)
+
+
+@pytest.mark.asyncio
+async def test_open_bracket_probes_pmax_none_and_paginates_without_split(
+    scraper: AvitoScraper,
+) -> None:
+    """Открытый брекет (hi=None): probe с pmax=None, пагинация БЕЗ попытки split.
+
+    Даже если total > cap (мокнуто) — НЕ должно быть рекурсивного деления:
+    при hi=None mid=(lo+hi)//2 невозможен, поэтому leaf пагинируется напрямую.
+    Проверяем: (1) probe идёт с max_price=None (pmax не ставится), (2) НЕТ
+    под-вызовов с пересчитанным mid (никаких max_price != None в fetch'ах).
+    """
+    open_lo = _AVITO_PRICE_SEED_BRACKETS[-1][0]
+    assert _AVITO_PRICE_SEED_BRACKETS[-1][1] is None
+
+    fetch_args: list[tuple[int, int | None, int | None]] = []
+
+    async def fake_fetch(
+        room_slug: str, page: int, min_price: int | None, max_price: int | None
+    ) -> str:
+        fetch_args.append((page, min_price, max_price))
+        return f"<html>page={page} min={min_price} max={max_price}</html>"
+
+    # total > cap для ОТКРЫТОГО брекета — провоцируем split-соблазн. Закрытые
+    # брекеты возвращают 0 (не пагинируются), чтобы изолировать открытый.
+    def fake_total(html: str) -> int | None:
+        import re
+
+        max_m = re.search(r"max=(\w+)", html)
+        max_raw = max_m.group(1) if max_m else "None"
+        if max_raw == "None":  # открытый брекет (pmax отсутствует)
+            return 5000  # > cap=2000 — но split НЕ должен случиться
+        return 0
+
+    def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+        import re
+
+        page_m = re.search(r"page=(\d+)", html)
+        page = int(page_m.group(1)) if page_m else 1
+        # Только открытый брекет (max=None) отдаёт лоты.
+        if "max=None" not in html:
+            return []
+        return [_make_lot(f"open_lot_p{page}_{i}") for i in range(3)]
+
+    with (
+        patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+        patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+        patch.object(scraper, "_parse_html", side_effect=fake_parse),
+        patch.object(scraper, "_rotate_ip", return_value=False),
+    ):
+        lots = await scraper.fetch_all_secondary(
+            rooms_buckets=[("1-комн.", "1-komnatnye-X")],
+            price_cap_per_bucket=2000,
+            max_pages_per_bucket=100,
+        )
+
+    # Probe открытого брекета: min_price=open_lo, max_price=None.
+    open_probes = [
+        (mn, mx) for (p, mn, mx) in fetch_args if p == 1 and mn == open_lo and mx is None
+    ]
+    assert open_probes, f"Ожидался probe открытого брекета с pmax=None: {fetch_args}"
+    # Все fetch'и открытого брекета (min=open_lo) идут с max_price=None — НЕТ
+    # под-брекетов с вычисленным mid (что означало бы split).
+    open_fetches = [(mn, mx) for (p, mn, mx) in fetch_args if mn == open_lo]
+    assert all(
+        mx is None for (_mn, mx) in open_fetches
+    ), f"Открытый брекет НЕ должен порождать closed-под-брекеты (split): {open_fetches}"
+    # Открытый брекет всё же собрал лоты (пагинация прошла, total=5000 → cap pages).
+    assert len(lots) > 0, "Открытый брекет должен пагинироваться и собрать лоты"
+
+
+@pytest.mark.asyncio
+async def test_closed_bracket_under_cap_single_probe_then_paginate(
+    scraper: AvitoScraper,
+) -> None:
+    """Закрытый брекет total<cap → 1 probe + прямая пагинация (без split)."""
+    # Изолируем второй закрытый брекет (4_000_000, 5_000_000) → walk_hi=4_999_999.
+    target_lo, target_hi_raw = _AVITO_PRICE_SEED_BRACKETS[1]
+    assert target_hi_raw is not None
+    target_hi = target_hi_raw - 1
+
+    fetch_args: list[tuple[int, int | None, int | None]] = []
+
+    async def fake_fetch(
+        room_slug: str, page: int, min_price: int | None, max_price: int | None
+    ) -> str:
+        fetch_args.append((page, min_price, max_price))
+        return f"<html>page={page} min={min_price} max={max_price}</html>"
+
+    def fake_total(html: str) -> int | None:
+        import re
+
+        min_m = re.search(r"min=(\w+)", html)
+        max_m = re.search(r"max=(\w+)", html)
+        min_p = min_m.group(1) if min_m else "0"
+        max_p = max_m.group(1) if max_m else "None"
+        if min_p == str(target_lo) and max_p == str(target_hi):
+            return 60  # ceil(60/50)=2 страницы, < cap → без split
+        return 0
+
+    def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+        import re
+
+        if f"min={target_lo} max={target_hi}" not in html:
+            return []
+        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"t_lot_p{page}_{i}") for i in range(30)]
+        return []
+
+    with (
+        patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+        patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+        patch.object(scraper, "_parse_html", side_effect=fake_parse),
+        patch.object(scraper, "_rotate_ip", return_value=False),
+    ):
+        lots = await scraper.fetch_all_secondary(
+            rooms_buckets=[("1-комн.", "1-komnatnye-X")],
+            price_cap_per_bucket=2000,
+        )
+
+    # Целевой брекет: probe (page=1) + страница 2, без под-брекетов (split).
+    target_pages = sorted(p for (p, mn, mx) in fetch_args if mn == target_lo and mx == target_hi)
+    assert target_pages == [1, 2], f"Ожидались probe+page2 без split: {target_pages}"
+    assert len(lots) == 60, f"Ожидалось 60 лотов (2 страницы × 30), got {len(lots)}"
+
+
+@pytest.mark.asyncio
+async def test_dedup_source_id_across_brackets(scraper: AvitoScraper) -> None:
+    """Дедуп по source_id работает между разными seed-брекетами (общий seen)."""
+
+    async def fake_fetch(
+        room_slug: str, page: int, min_price: int | None, max_price: int | None
+    ) -> str:
+        return f"<html>page={page} min={min_price} max={max_price}</html>"
+
+    def fake_total(html: str) -> int | None:
+        return 3  # 1 страница на каждый брекет
+
+    def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+        # ВСЕ брекеты отдают ОДИН и тот же source_id → должен остаться 1 уникальный.
+        return [_make_lot("shared_dup_id")]
+
+    with (
+        patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+        patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+        patch.object(scraper, "_parse_html", side_effect=fake_parse),
+        patch.object(scraper, "_rotate_ip", return_value=False),
+    ):
+        lots = await scraper.fetch_all_secondary(
+            rooms_buckets=[("1-комн.", "1-komnatnye-X")],
+            price_cap_per_bucket=2000,
+        )
+
+    assert len(lots) == 1, f"Дедуп между брекетами должен дать 1 уникальный лот, got {len(lots)}"
+    assert lots[0].source_id == "shared_dup_id"