refactor(scraper-kit): unify price bisection engine across providers + dedup (#2134)

tradein-mvp/backend/tests/test_scraper_kit_bisection_parity.py

@@ -0,0 +1,455 @@
+"""Golden-parity бисекции: старый боевой обход `app.services.scrapers.<provider>`
+≡ новый единый движок `scraper_kit.pricing.walk_price_range` (#2134).
+
+Четыре независимые копии адаптивной бисекции «комнатность × цена» были
+консолидированы в один движок (`scraper_kit/pricing/bisection.py`). Провайдеры
+(avito/cian/yandex/domclick) теперь передают в движок свои probe/paginate + свой
+`BisectionConfig`. Этот тест доказывает, что unification НЕ изменил ОБХОД: для
+каждого провайдера при одинаковом входе (детерминированный мок счётчика по
+диапазону) старая боевая реализация и новая (движок) посещают ИДЕНТИЧНУЮ
+последовательность бакетов — probe-запросы, leaf-пагинации, деления, ранние
+возвраты на нуле, probe-fail-ветки.
+
+Механика: на ОБЕ реализации (старую и новую) ставятся ОДИНАКОВЫЕ моки на
+низкоуровневые методы (probe-fetch + leaf-пагинатор), которые (а) возвращают
+счётчик из общего оракула `count(lo, hi)` и (б) пишут событие в общий список.
+Совпадение списков событий = идентичный обход.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Callable
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from scraper_kit.price_brackets import get_price_seed_brackets
+from scraper_kit.providers.avito.serp import AvitoScraper as NewAvito
+from scraper_kit.providers.cian.serp import CianScraper as NewCian
+from scraper_kit.providers.domclick.serp import DomClickScraper as NewDomClick
+from scraper_kit.providers.yandex.serp import YandexRealtyScraper as NewYandex
+
+from app.services.scrapers.avito import AvitoScraper as OldAvito
+from app.services.scrapers.cian import CianScraper as OldCian
+from app.services.scrapers.domclick import DomClickScraper as OldDomClick
+from app.services.scrapers.yandex_realty import YandexRealtyScraper as OldYandex
+
+# ── Оракул счётчиков ─────────────────────────────────────────────────────────
+
+Oracle = Callable[[Any, Any], int | None]
+
+
+def make_oracle(
+    *,
+    divisor: int,
+    open_ceiling: int,
+    zeros: frozenset[tuple[Any, Any]] = frozenset(),
+    nones: frozenset[tuple[Any, Any]] = frozenset(),
+) -> Oracle:
+    """Детерминированный счётчик от (lo, hi): count = span // divisor.
+
+    span считается по клампнутым границам (None-lo → 0, None-hi → open_ceiling),
+    чтобы широкие брекеты делились, а узкие уходили в leaf. `zeros` форсит 0 (ранний
+    возврат / leaf-на-нуле у yandex), `nones` форсит None (probe-fail).
+    """
+
+    def oracle(lo: Any, hi: Any) -> int | None:
+        if (lo, hi) in nones:
+            return None
+        if (lo, hi) in zeros:
+            return 0
+        eff_lo = lo if lo is not None else 0
+        eff_hi = hi if hi is not None else open_ceiling
+        span = eff_hi - eff_lo
+        return max(0, span // divisor)
+
+    return oracle
+
+
+# ── Провайдер-специфичные «сборщики событий» ─────────────────────────────────
+#
+# Каждый ставит одинаковые моки на старый/новый экземпляр и возвращает список
+# событий обхода. `("probe", lo, hi)` — запрос счётчика бакета; `("leaf", ...)` —
+# пагинация leaf-бакета.
+
+
+def _patch_avito(scraper: Any, oracle: Oracle, events: list[Any]) -> None:
+    def probe(room_slug: str, min_price: Any, max_price: Any) -> tuple[str | None, int | None]:
+        events.append(("probe", min_price, max_price))
+        return "H", oracle(min_price, max_price)
+
+    def leaf(**kw: Any) -> None:
+        events.append(("leaf", kw["lo"], kw["hi"], kw["max_pages"]))
+
+    scraper._probe_total = AsyncMock(side_effect=probe)
+    scraper._paginate_leaf_bucket = AsyncMock(side_effect=leaf)
+    scraper._rotate_ip = AsyncMock(return_value=False)
+
+
+def _patch_cian(scraper: Any, oracle: Oracle, events: list[Any]) -> None:
+    holder: dict[str, tuple[Any, Any]] = {}
+
+    def fph(rooms: Any, page: int, min_price: Any, max_price: Any) -> str:
+        events.append(("probe", min_price, max_price))
+        holder["b"] = (min_price, max_price)
+        return "H"
+
+    def eto(html: str) -> int | None:
+        return oracle(*holder["b"])
+
+    def leaf(**kw: Any) -> None:
+        events.append(("leaf", kw["lo"], kw["hi"], kw["total"]))
+
+    scraper._fetch_page_html = AsyncMock(side_effect=fph)
+    scraper._extract_total_offers = MagicMock(side_effect=eto)
+    scraper._paginate_leaf_bucket = AsyncMock(side_effect=leaf)
+    scraper._rotate_ip = AsyncMock(return_value=False)
+    scraper.sleep_between_requests = AsyncMock()
+
+
+def _patch_domclick(scraper: Any, oracle: Oracle, events: list[Any]) -> None:
+    def count(fetcher: Any, rooms: Any, gte: Any, lte: Any) -> int:
+        events.append(("probe", gte, lte))
+        val = oracle(gte, lte)
+        return val if val is not None else 0
+
+    def paginate(
+        fetcher: Any, rooms: Any, gte: Any, lte: Any, seen_ids: Any, out_lots: Any, pages: Any
+    ) -> None:
+        events.append(("leaf", gte, lte))
+
+    scraper._count = AsyncMock(side_effect=count)
+    scraper._paginate = AsyncMock(side_effect=paginate)
+
+
+def _patch_yandex(
+    scraper: Any, oracle: Oracle, events: list[Any], *, record_pages: bool = False
+) -> None:
+    """Yandex: leaf-пагинация встроена в walk (нет отдельного метода-хука).
+
+    Стабим `_fetch_page_json` (page=1 отдаёт payload с totalItems из оракула;
+    page>1 — пустой payload) и передаём `on_bucket`, который пишет завершение
+    leaf/degraded бакета. probe = page-1-fetch; leaf-страницы 2..N идут через
+    asyncio.gather (порядок недетерминирован) — их НЕ пишем в норм-режиме; факт
+    leaf-обхода фиксируется через on_bucket. Для degraded-теста (пагинация
+    последовательная) `record_pages=True` пишет все страницы.
+    """
+
+    def gate(total: int | None, entities: list[Any] | None = None) -> dict[str, Any]:
+        pager: dict[str, Any] = {}
+        if total is not None:
+            pager["totalItems"] = total
+        return {"response": {"search": {"offers": {"entities": entities or [], "pager": pager}}}}
+
+    def fpj(rooms: Any, page: int, price_min: Any, price_max: Any, new_flat: str = "NO") -> Any:
+        if page == 1:
+            events.append(("probe", price_min, price_max))
+            return gate(oracle(price_min, price_max))
+        if record_pages:
+            events.append(("page", price_min, price_max, page))
+        return gate(0, entities=[])
+
+    scraper._fetch_page_json = AsyncMock(side_effect=fpj)
+    scraper._rotate_ip = AsyncMock(return_value=False)
+    scraper.request_delay_sec = 0.0
+
+
+def _yandex_on_bucket(events: list[Any]) -> Callable[[str, int], None]:
+    def on_bucket(bucket_key: str, count: int) -> None:
+        events.append(("bucket", bucket_key))
+
+    return on_bucket
+
+
+# ── Конструкторы старых/новых экземпляров (offline, без DB) ───────────────────
+
+_KIT_CONFIG = SimpleNamespace(
+    scraper_proxy_url=None,
+    yandex_proxy_rotate_url=None,
+    avito_proxy_rotate_url=None,
+    avito_proxy_max_rotations=0,
+    scraper_fetch_mode="browser",
+    browser_http_endpoint="http://tradein-browser:9000",
+    glitchtip_dsn=None,
+    avito_serp_ekb_only=False,
+)
+
+
+def _old_avito() -> Any:
+    with patch("app.services.scrapers.avito.get_scraper_delay", return_value=7.0):
+        return OldAvito()
+
+
+def _old_cian() -> Any:
+    with patch("app.services.scrapers.cian.get_scraper_delay", return_value=5.0):
+        return OldCian()
+
+
+def _old_yandex() -> Any:
+    with patch("app.services.scrapers.yandex_realty.get_scraper_delay", return_value=5.0):
+        return OldYandex()
+
+
+def _old_domclick() -> Any:
+    with patch("app.services.scrapers.domclick.get_scraper_delay", return_value=8.0):
+        return OldDomClick()
+
+
+# ── avito / cian: одинаковый скелет (probe → split → leaf) ────────────────────
+
+
+# Провайдер-специфичные kwargs для _walk_price_range (avito идентифицирует бакет
+# room_slug/room_label; cian — кортежем rooms).
+_AVITO_WALK_EXTRA = {"room_slug": "room1", "room_label": "1к"}
+_CIAN_WALK_EXTRA = {"rooms": (1,)}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "old_factory, new_factory, patch_fn, walk_extra",
+    [
+        (_old_avito, lambda: NewAvito(config=_KIT_CONFIG), _patch_avito, _AVITO_WALK_EXTRA),
+        (_old_cian, lambda: NewCian(config=_KIT_CONFIG), _patch_cian, _CIAN_WALK_EXTRA),
+    ],
+)
+async def test_bisection_parity_avito_cian(
+    old_factory: Any, new_factory: Any, patch_fn: Any, walk_extra: dict[str, Any]
+) -> None:
+    """avito/cian: split-дерево + leaf + ранний возврат на нуле — обход идентичен.
+
+    Оракул: divisor=1000, cap=1400 → брекет 8М делится 3 уровня (8М→4М→2М→1М-leaf);
+    один под-брекет форсим в 0 (ранний возврат без leaf).
+    """
+    # После первого split (0,7_999_999): mid=3_999_999 → дети (None,3_999_999) и
+    # (4_000_000, 7_999_999). Форсим 0 на (4_000_000, 5_999_999) — под-брекет после
+    # второго split правой ветки.
+    oracle = make_oracle(
+        divisor=1000,
+        open_ceiling=30_000_000,
+        zeros=frozenset({(4_000_000, 5_999_999)}),
+    )
+    brackets: list[tuple[int, int | None]] = [(0, 7_999_999), (25_000_000, None)]
+
+    async def run(scraper: Any) -> list[Any]:
+        events: list[Any] = []
+        patch_fn(scraper, oracle, events)
+        for lo, hi in brackets:
+            await scraper._walk_price_range(
+                lo=lo,
+                hi=hi,
+                seen={},
+                price_cap_per_bucket=1400,
+                max_pages_per_bucket=100,
+                concurrency=5,
+                secondary_only=True,
+                on_bucket=None,
+                skip_buckets=None,
+                **walk_extra,
+            )
+        return events
+
+    old_events = await run(old_factory())
+    new_events = await run(new_factory())
+    assert old_events == new_events
+    # Санити: обход реально нетривиален (несколько probe + несколько leaf).
+    assert sum(1 for e in old_events if e[0] == "leaf") >= 4
+    assert sum(1 for e in old_events if e[0] == "probe") >= 6
+
+
+@pytest.mark.asyncio
+async def test_bisection_parity_avito_probe_fail() -> None:
+    """avito probe-fail (total=None): SPLIT_OR_SKIP — принудительный split делимого
+    закрытого брекета, best-effort leaf открытого — идентичен старому."""
+    # (0, 7_999_999): probe None → делим (bracket>=MIN). Дети (None,4M),(4000001,...):
+    # им дадим реальные счётчики. Открытый (25M, None) probe None → best-effort leaf.
+    oracle = make_oracle(
+        divisor=1000,
+        open_ceiling=30_000_000,
+        nones=frozenset({(None, 7_999_999), (25_000_000, None)}),
+    )
+    brackets: list[tuple[int, int | None]] = [(0, 7_999_999), (25_000_000, None)]
+
+    async def run(scraper: Any) -> list[Any]:
+        events: list[Any] = []
+        _patch_avito(scraper, oracle, events)
+        for lo, hi in brackets:
+            await scraper._walk_price_range(
+                room_slug="room1",
+                room_label="1к",
+                lo=lo,
+                hi=hi,
+                seen={},
+                price_cap_per_bucket=1400,
+                max_pages_per_bucket=100,
+                concurrency=5,
+                secondary_only=True,
+                on_bucket=None,
+                skip_buckets=None,
+            )
+        return events
+
+    old_events = await run(_old_avito())
+    new_events = await run(NewAvito(config=_KIT_CONFIG))
+    assert old_events == new_events
+    # Открытый брекет с probe-fail → ровно один best-effort leaf на (25M, None).
+    assert ("leaf", 25_000_000, None, 100) in old_events
+
+
+# ── domclick: count-endpoint + None-both-ends + split открытого корня ─────────
+
+
+@pytest.mark.asyncio
+async def test_bisection_parity_domclick() -> None:
+    """domclick: unbounded-корень (None,None) клампится к [0,1e9] и делится;
+    count-endpoint; span<=MIN_PRICE_SPAN строго → leaf. Обход идентичен старому."""
+    oracle = make_oracle(divisor=100_000, open_ceiling=1_000_000_000)
+
+    async def run(scraper: Any) -> list[Any]:
+        events: list[Any] = []
+        _patch_domclick(scraper, oracle, events)
+        await scraper._sweep_bucket(
+            fetcher=None,
+            rooms="1",
+            price_gte=None,
+            price_lte=None,
+            seen_ids=set(),
+            out_lots=[],
+            pages=100,
+        )
+        return events
+
+    old_events = await run(_old_domclick())
+    new_events = await run(NewDomClick(_make_domclick_config()))
+    assert old_events == new_events
+    assert sum(1 for e in old_events if e[0] == "leaf") >= 4
+    # Корень (None,None) обязательно пробится первым.
+    assert old_events[0] == ("probe", None, None)
+
+
+def _make_domclick_config() -> Any:
+    return SimpleNamespace(browser_http_endpoint="http://tradein-browser:9000")
+
+
+# ── yandex: max_depth=8, min_bracket=500k, count-based on_bucket ──────────────
+
+
+@pytest.mark.asyncio
+async def test_bisection_parity_yandex_normal() -> None:
+    """yandex: split-дерево (порог 500k, cap 500) + leaf-on-bucket + zero-as-leaf.
+
+    Оракул divisor=1000, cap=500 → 4М→2М→1М→500k-leaf. Один под-брекет=0
+    (yandex вызывает on_bucket даже на нуле — treat_zero_as_leaf).
+    """
+    oracle = make_oracle(
+        divisor=1000,
+        open_ceiling=30_000_000,
+        zeros=frozenset({(2_000_000, 2_999_999)}),
+    )
+    brackets: list[tuple[int, int | None]] = [(0, 3_999_999), (25_000_000, None)]
+
+    async def run(scraper: Any) -> list[Any]:
+        events: list[Any] = []
+        _patch_yandex(scraper, oracle, events)
+        for lo, hi in brackets:
+            await scraper._walk_price_range(
+                rooms="1",
+                lo=lo,
+                hi=hi,
+                seen={},
+                price_cap_per_bucket=500,
+                max_pages_per_bucket=50,
+                concurrency=4,
+                on_bucket=_yandex_on_bucket(events),
+                skip_buckets=None,
+            )
+        return events
+
+    old_events = await run(_old_yandex())
+    new_events = await run(NewYandex(config=_KIT_CONFIG))
+    assert old_events == new_events
+    assert sum(1 for e in old_events if e[0] == "bucket") >= 4
+    assert sum(1 for e in old_events if e[0] == "probe") >= 6
+
+
+@pytest.mark.asyncio
+async def test_bisection_parity_yandex_degraded() -> None:
+    """yandex probe-fail (totalItems неизвестен) → degraded paginate-until-empty.
+
+    Оракул форсит None на бакете → probe page1 payload без totalItems → total=None
+    → degraded последовательно пагинирует. record_pages=True фиксирует все страницы.
+    """
+    oracle = make_oracle(
+        divisor=1000,
+        open_ceiling=30_000_000,
+        nones=frozenset({(None, 3_999_999)}),
+    )
+
+    async def run(scraper: Any) -> list[Any]:
+        events: list[Any] = []
+        _patch_yandex(scraper, oracle, events, record_pages=True)
+        await scraper._walk_price_range(
+            rooms="1",
+            lo=0,
+            hi=3_999_999,
+            seen={},
+            price_cap_per_bucket=500,
+            max_pages_per_bucket=50,
+            concurrency=4,
+            on_bucket=_yandex_on_bucket(events),
+            skip_buckets=None,
+        )
+        return events
+
+    old_events = await run(_old_yandex())
+    new_events = await run(NewYandex(config=_KIT_CONFIG))
+    assert old_events == new_events
+    # degraded: probe page1 + затем последовательная пагинация страниц 1..K до пустой.
+    assert ("probe", None, 3_999_999) in old_events
+    assert any(e[0] == "bucket" for e in old_events)
+
+
+# ── Full seed-grid обход (все seed-брекеты одной комнатности) ─────────────────
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "old_factory, new_factory, patch_fn, walk_extra",
+    [
+        (_old_avito, lambda: NewAvito(config=_KIT_CONFIG), _patch_avito, _AVITO_WALK_EXTRA),
+        (_old_cian, lambda: NewCian(config=_KIT_CONFIG), _patch_cian, _CIAN_WALK_EXTRA),
+    ],
+)
+async def test_bisection_parity_full_seed_grid(
+    old_factory: Any, new_factory: Any, patch_fn: Any, walk_extra: dict[str, Any]
+) -> None:
+    """Полный обход всех seed-брекетов (как в fetch_all_secondary) — включая
+    открытый catch-all — идентичен между старым и новым для avito/cian."""
+    oracle = make_oracle(divisor=1500, open_ceiling=300_000_000)
+
+    async def run(scraper: Any) -> list[Any]:
+        events: list[Any] = []
+        patch_fn(scraper, oracle, events)
+        for br_lo, br_hi in get_price_seed_brackets():
+            walk_hi = br_hi - 1 if br_hi is not None else None
+            await scraper._walk_price_range(
+                lo=br_lo,
+                hi=walk_hi,
+                seen={},
+                price_cap_per_bucket=1400,
+                max_pages_per_bucket=100,
+                concurrency=5,
+                secondary_only=True,
+                on_bucket=None,
+                skip_buckets=None,
+                **walk_extra,
+            )
+        return events
+
+    old_events = await run(old_factory())
+    new_events = await run(new_factory())
+    assert old_events == new_events

tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/__init__.py

@@ -0,0 +1,21 @@
+"""scraper_kit.pricing — общие примитивы ценового партиционирования SERP-обхода.
+
+Единый движок адаптивной бисекции «комнатность × цена» для всех провайдеров
+(avito / cian / yandex / domclick). См. `bisection.py`.
+"""
+
+from __future__ import annotations
+
+from scraper_kit.pricing.bisection import (
+    BisectionConfig,
+    ProbeFailPolicy,
+    ProbeResult,
+    walk_price_range,
+)
+
+__all__ = [
+    "BisectionConfig",
+    "ProbeFailPolicy",
+    "ProbeResult",
+    "walk_price_range",
+]

tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/bisection.py

@@ -0,0 +1,253 @@
+"""Единый движок адаптивной бисекции «комнатность × цена» для SERP-обхода.
+
+Все четыре провайдера вторички (avito / cian / yandex / domclick) обходят
+SERP-cap площадки одинаковым по СКЕЛЕТУ приёмом: для каждой комнатности берётся
+ценовой диапазон [lo, hi], запрашивается «сколько всего результатов» (probe
+page=1 или отдельный count-endpoint), и если результатов больше потолка бакета —
+диапазон делится пополам (`mid = (lo + hi) // 2`) и обход рекурсивно спускается
+в обе половины; иначе бакет пагинируется как leaf.
+
+Исторически это было ЧЕТЫРЕ независимые копии (`_walk_price_range` ×3 +
+`_sweep_bucket`), разошедшиеся в константах и краевых ветках. Здесь скелет
+рекурсии вынесен в одну функцию `walk_price_range`, а РАЗЛИЧИЯ провайдеров
+переданы в `BisectionConfig` (пороги, глубина, поведение на открытом брекете и
+при неудачном probe) + три provider-callback'а. Ни одно различие не потеряно —
+golden-parity тесты доказывают, что обход байт-в-байт совпадает со старой
+боевой реализацией для каждого провайдера.
+
+Provider-side (в callback'ах, движок их НЕ трогает):
+  * фактический HTTP-fetch + retry/rotation;
+  * парсинг счётчика (totalOffers / page-title-count / totalItems / snippetsCount);
+  * пагинация leaf-бакета, дедуп (`seen`), on_bucket-callback.
+
+Engine-side (здесь):
+  * дерево рекурсии: какие [lo, hi] пробятся / делятся / пагинируются и в каком
+    порядке;
+  * арифметика split-решения (cap, min_bracket, max_depth, clamp None-границ);
+  * применение probe-fail-политики и обработка открытого (hi=None) брекета.
+
+Семантические различия провайдеров (задокументировано в конфигах-инстансах на
+стороне провайдеров), которые движок воспроизводит через `BisectionConfig`:
+
+| поле                    | avito     | cian      | yandex        | domclick        |
+|-------------------------|-----------|-----------|---------------|-----------------|
+| cap                     | 1400      | 1400      | 500           | 2000 (OFFSET)   |
+| min_bracket             | 50_000    | 50_000    | 500_000       | 100_000         |
+| min_bracket_inclusive   | True (>=) | True (>=) | True (>=)     | False (>)       |
+| max_depth               | None      | None      | 8             | None            |
+| probe_fail_policy       | SPLIT/SKIP| SKIP      | DEGRADE       | SKIP (не рвётся)|
+| open_split_ceiling      | None(leaf)| None(leaf)| None(leaf)    | 1_000_000_000   |
+| treat_zero_as_leaf      | False     | False     | True          | False           |
+
+Различие «probe vs count»: avito/cian/yandex читают счётчик из probe page=1 и
+ПЕРЕИСПОЛЬЗУЮТ HTML/лоты страницы 1 в leaf (payload); domclick дёргает отдельный
+count-endpoint (payload=None, пагинация с offset=0). Это инкапсулировано в
+provider-callback `probe` + `ProbeResult.payload`, движку прозрачно.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from enum import Enum, auto
+from typing import Any
+
+
+class ProbeFailPolicy(Enum):
+    """Что делать, когда probe не вернул счётчик (`ProbeResult.count is None`).
+
+    * ``SKIP`` — бросить бакет целиком (cian: totalOffers недоступен → пропуск).
+    * ``SPLIT_OR_SKIP`` — открытый брекет пагинируем best-effort; закрытый ещё
+      делимый (span >= min_bracket) принудительно делим (тяжёлая страница станет
+      загружаемой в узком диапазоне); иначе пропуск (avito).
+    * ``DEGRADE`` — деградировать в «пагинировать-до-пустоты» без знания total
+      (yandex): вызывается ``on_degraded``.
+
+    domclick счётчик-endpoint при ошибке возвращает 0 (а не None), поэтому его
+    политика формально ``SKIP``, но ветка probe-fail недостижима (count==0 →
+    ранний return).
+    """
+
+    SKIP = auto()
+    SPLIT_OR_SKIP = auto()
+    DEGRADE = auto()
+
+
+@dataclass
+class ProbeResult:
+    """Результат probe одного бакета.
+
+    * ``count`` — сколько всего результатов в [lo, hi] (или ``None`` при
+      captcha/ошибке probe). ``0`` — валидная пустая выборка.
+    * ``payload`` — provider-specific данные страницы 1 для переиспользования в
+      leaf-пагинации (avito: HTML; cian: HTML; yandex: list[ScrapedLot] page 1;
+      domclick: None — count-endpoint без карточек).
+    """
+
+    count: int | None
+    payload: Any = None
+
+
+@dataclass(frozen=True)
+class BisectionConfig:
+    """Per-provider константы/политики бисекции (все различия провайдеров тут).
+
+    Args:
+        cap: потолок результатов в бакете; ``count > cap`` — кандидат на split.
+        min_bracket: минимальный размах диапазона, ниже которого делить нельзя.
+        min_bracket_inclusive: split разрешён при ``span >= min_bracket`` (True,
+            avito/cian/yandex) или строго ``span > min_bracket`` (False,
+            domclick — исходное `span <= MIN_PRICE_SPAN → paginate`).
+        max_depth: жёсткий предел глубины рекурсии (yandex=8) либо ``None``.
+        probe_fail_policy: см. :class:`ProbeFailPolicy`.
+        open_split_ceiling: если ``None`` (avito/cian/yandex) — открытый брекет
+            (hi=None) НЕ делится, пагинируется напрямую как leaf. Если задан
+            (domclick=1e9) — открытые границы клампятся к
+            ``[open_split_floor, open_split_ceiling]`` и диапазон делится.
+        open_split_floor: нижний кламп для None-lo при делении (domclick=0).
+        treat_zero_as_leaf: если True (yandex) — при ``count == 0`` всё равно
+            вызвать ``on_leaf`` (yandex on_bucket-callback срабатывает и на 0);
+            иначе (avito/cian/domclick) — ранний return без leaf.
+    """
+
+    cap: int
+    min_bracket: int
+    max_depth: int | None = None
+    min_bracket_inclusive: bool = True
+    probe_fail_policy: ProbeFailPolicy = ProbeFailPolicy.SKIP
+    open_split_ceiling: int | None = None
+    open_split_floor: int = 0
+    treat_zero_as_leaf: bool = False
+
+
+# Provider-callback типы.
+ProbeFn = Callable[[int | None, int | None, int], Awaitable[ProbeResult]]
+LeafFn = Callable[[int | None, int | None, ProbeResult], Awaitable[None]]
+DegradedFn = Callable[[int | None, int | None], Awaitable[None]]
+
+
+def _can_split_span(span: int, config: BisectionConfig) -> bool:
+    """Разрешает ли размах диапазона дальнейшее деление (по min_bracket)."""
+    if config.min_bracket_inclusive:
+        return span >= config.min_bracket
+    return span > config.min_bracket
+
+
+async def walk_price_range(
+    *,
+    lo: int | None,
+    hi: int | None,
+    config: BisectionConfig,
+    probe: ProbeFn,
+    on_leaf: LeafFn,
+    on_degraded: DegradedFn | None = None,
+    depth: int = 0,
+) -> None:
+    """Рекурсивно обойти ценовой диапазон [lo, hi] адаптивной бисекцией.
+
+    Скелет (общий для всех провайдеров):
+      1. ``result = probe(lo, hi, depth)`` — provider I/O (fetch + retry).
+      2. ``result.count is None`` → применить ``probe_fail_policy``.
+      3. ``count == 0`` → return (кроме ``treat_zero_as_leaf``).
+      4. Открытый брекет (hi=None) при ``open_split_ceiling is None`` → leaf.
+      5. Иначе: если ``count > cap`` и размах делим и глубина в пределах —
+         ``mid = (lo + hi) // 2``, рекурсия в обе половины; иначе leaf.
+
+    Args:
+        lo / hi: границы диапазона (``None`` = без границы; клампится при делении
+            для провайдеров с ``open_split_ceiling``).
+        config: per-provider пороги/политики.
+        probe: async ``(lo, hi, depth) -> ProbeResult`` — счётчик + payload.
+        on_leaf: async ``(lo, hi, result) -> None`` — пагинация leaf-бакета
+            (fetch/parse/дедуп/on_bucket целиком на стороне провайдера).
+        on_degraded: async ``(lo, hi) -> None`` — обязателен для
+            ``ProbeFailPolicy.DEGRADE`` (yandex «пагинировать-до-пустоты»).
+        depth: текущая глубина рекурсии (для ``max_depth``).
+    """
+    result = await probe(lo, hi, depth)
+    total = result.count
+
+    # ── Probe-fail: счётчик недоступен ────────────────────────────────────────
+    if total is None:
+        policy = config.probe_fail_policy
+        if policy is ProbeFailPolicy.SKIP:
+            return
+        if policy is ProbeFailPolicy.DEGRADE:
+            if on_degraded is not None:
+                await on_degraded(lo, hi)
+            return
+        # SPLIT_OR_SKIP (avito): открытый брекет пагинируем best-effort; закрытый
+        # ещё делимый — принудительно делим; иначе пропускаем (хвост теряем).
+        if hi is None:
+            await on_leaf(lo, hi, result)
+            return
+        _lo = lo if lo is not None else config.open_split_floor
+        if _can_split_span(hi - _lo, config):
+            mid = (_lo + hi) // 2
+            await walk_price_range(
+                lo=_lo,
+                hi=mid,
+                config=config,
+                probe=probe,
+                on_leaf=on_leaf,
+                on_degraded=on_degraded,
+                depth=depth + 1,
+            )
+            await walk_price_range(
+                lo=mid + 1,
+                hi=hi,
+                config=config,
+                probe=probe,
+                on_leaf=on_leaf,
+                on_degraded=on_degraded,
+                depth=depth + 1,
+            )
+        return
+
+    # ── Счётчик получен ───────────────────────────────────────────────────────
+    if total == 0 and not config.treat_zero_as_leaf:
+        return
+
+    # Открытый брекет без split-потолка (avito/cian/yandex): делить нельзя —
+    # пагинируем leaf напрямую (хвост люкса крошечный, tail-loss accepted).
+    if hi is None and config.open_split_ceiling is None:
+        await on_leaf(lo, hi, result)
+        return
+
+    # Эффективные границы для арифметики деления: None-границы клампятся к
+    # [open_split_floor, open_split_ceiling] (domclick unbounded root).
+    eff_lo = lo if lo is not None else config.open_split_floor
+    eff_hi = hi if hi is not None else config.open_split_ceiling
+
+    can_split = (
+        total > config.cap
+        and eff_hi is not None
+        and _can_split_span(eff_hi - eff_lo, config)
+        and (config.max_depth is None or depth < config.max_depth)
+    )
+
+    if can_split:
+        assert eff_hi is not None  # для mypy: гарантировано can_split
+        mid = (eff_lo + eff_hi) // 2
+        await walk_price_range(
+            lo=eff_lo,
+            hi=mid,
+            config=config,
+            probe=probe,
+            on_leaf=on_leaf,
+            on_degraded=on_degraded,
+            depth=depth + 1,
+        )
+        await walk_price_range(
+            lo=mid + 1,
+            hi=eff_hi,
+            config=config,
+            probe=probe,
+            on_leaf=on_leaf,
+            on_degraded=on_degraded,
+            depth=depth + 1,
+        )
+        return
+
+    # Leaf-бакет: пагинация + дедуп + on_bucket — целиком на стороне провайдера.
+    await on_leaf(lo, hi, result)

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/serp.py

@@ -46,6 +46,7 @@ from scraper_kit.avito_exceptions import (
 from scraper_kit.base import BaseScraper, ScrapedLot
 from scraper_kit.browser_fetcher import BrowserFetcher
 from scraper_kit.price_brackets import get_price_seed_brackets
+from scraper_kit.pricing import BisectionConfig, ProbeFailPolicy, ProbeResult, walk_price_range
 from scraper_kit.providers.avito.shared import _unix_to_date
 
 if TYPE_CHECKING:
@@ -250,6 +251,21 @@ _AVITO_PRICE_SEED_BRACKETS: list[tuple[int, int | None]] = get_price_seed_bracke
 # Avito SERP показывает ~50 карточек на страницу.
 _AVITO_OFFERS_PER_PAGE = 50
 
+
+def _avito_bisection_config(cap: int) -> BisectionConfig:
+    """Конфиг единого движка бисекции (scraper_kit.pricing) для avito.
+
+    probe_fail=SPLIT_OR_SKIP: probe вернул None (captcha/browser-timeout) → открытый
+    брекет пагинируем best-effort, закрытый ещё делимый (bracket >= MIN) принудительно
+    делим (тяжёлая выдача станет загружаемой в узком диапазоне), иначе пропуск.
+    open_split_ceiling=None: открытый брекет (hi=None) пагинируется напрямую, не делится.
+    """
+    return BisectionConfig(
+        cap=cap,
+        min_bracket=_AVITO_MIN_BRACKET,
+        probe_fail_policy=ProbeFailPolicy.SPLIT_OR_SKIP,
+    )
+
 # HTTP 429 в curl_cffi-режиме через backconnect-прокси (mproxy.site) — НЕ IP-ban, а
 # transient «слишком много одновременных соединений» (лимит 5). Проходит на коротком
 # retry без ротации IP. Делаем до _AVITO_429_MAX_RETRIES коротких пауз; если они
@@ -370,7 +386,9 @@ class AvitoScraper(BaseScraper):
             # FALLBACK — если браузер отдаёт firewall/пусто, _fetch_serp_html делает
             # один retry через curl_cffi. _build_cffi_session самодостаточен
             # (отдельный AsyncSession, не зависит от BaseScraper.__aenter__).
-            self._browser = BrowserFetcher(source="avito")
+            self._browser = BrowserFetcher(
+                source="avito", endpoint=self._config.browser_http_endpoint
+            )
             await self._browser.__aenter__()
             self._cffi = self._build_cffi_session()
             logger.info(
@@ -1155,145 +1173,89 @@ class AvitoScraper(BaseScraper):
         secondary_only: новостройки (listing_segment=="novostroyki") отбрасываются после
           сбора bucket_lots, до дедупа в seen и вызова on_bucket.
         AvitoBlockedError/AvitoRateLimitedError из _fetch_serp_html пробрасываются наверх.
+
+        Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Avito-spec
+        probe (page=1 + IP-rotate retry) и leaf-пагинация переданы callback'ами;
+        пороги (cap, _AVITO_MIN_BRACKET) и политика probe-fail (SPLIT_OR_SKIP —
+        принудительный split тяжёлой выдачи / best-effort для открытого) — в
+        BisectionConfig. AvitoBlockedError/AvitoRateLimitedError из probe/leaf
+        всплывают наверх (mark_banned).
         """
-        _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
-        # на тяжёлой выдаче, напр. студии pmax=200M) → total=None вместо краха.
-        # AvitoBlockedError/AvitoRateLimitedError всплывают наверх (mark_banned).
-        html, total = await self._probe_total(room_slug, _lo_param, hi)
-
-        # Ретрай на captcha/missing-count/browser-ошибке: rotate IP + 1 retry.
-        if total is None:
-            logger.warning(
-                "avito: total=None for %s [%d, %s] depth=%d — rotating IP + retry",
-                room_label,
-                lo,
-                _hi_repr,
-                _depth,
-            )
-            rotated = await self._rotate_ip()
-            if rotated:
-                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:
+        async def _probe(plo: int | None, phi: int | None, depth: int) -> ProbeResult:
+            lo_param = plo if plo and plo > 0 else None
+            hi_repr = "open" if phi is None else str(phi)
+            # _probe_total ловит browser-ошибки (HTTP 500 / timeout от tradein-browser
+            # на тяжёлой выдаче) → total=None вместо краха. Block-исключения всплывают.
+            html, total = await self._probe_total(room_slug, lo_param, phi)
+            if total is None:
                 logger.warning(
-                    "avito: probe failed for OPEN bucket %s [%d, open] depth=%d — "
+                    "avito: total=None for %s [%s, %s] depth=%d — rotating IP + retry",
+                    room_label,
+                    plo,
+                    hi_repr,
+                    depth,
+                )
+                rotated = await self._rotate_ip()
+                if rotated:
+                    html, total = await self._probe_total(room_slug, lo_param, phi)
+            if total is not None:
+                logger.info(
+                    "avito: %s [%s, %s] total=%d depth=%d",
+                    room_label,
+                    plo,
+                    hi_repr,
+                    total,
+                    depth,
+                )
+            return ProbeResult(count=total, payload=html)
+
+        async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None:
+            _lo = plo if plo is not None else 0
+            total = result.count
+            if total is None:
+                # SPLIT_OR_SKIP на открытом брекете: probe провалился, делить нельзя —
+                # пагинируем best-effort до max_pages_per_bucket (tail-loss accepted).
+                html: str | None = None
+                max_pages = max_pages_per_bucket
+                logger.warning(
+                    "avito: probe failed for OPEN bucket %s [%d, open] — "
                     "paginating up to max_pages (tail-loss accepted)",
                     room_label,
-                    lo,
-                    _depth,
+                    _lo,
                 )
-                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 не удался даже после ретрая. Если бакет ещё делим — НЕ теряем
-            # его целиком: тяжёлая страница (огромная выдача) скорее всего станет
-            # загружаемой в более узком ценовом диапазоне → принудительный split.
-            if bracket_size >= _AVITO_MIN_BRACKET:
-                logger.warning(
-                    "avito: probe failed for %s [%d, %d] depth=%d — splitting (assume heavy)",
-                    room_label,
-                    lo,
-                    hi,
-                    _depth,
-                )
-                mid = (lo + hi) // 2
-                await self._walk_price_range(
-                    room_slug=room_slug,
-                    room_label=room_label,
-                    lo=lo,
-                    hi=mid,
-                    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,
-                    _depth=_depth + 1,
-                )
-                await self._walk_price_range(
-                    room_slug=room_slug,
-                    room_label=room_label,
-                    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,
-                    secondary_only=secondary_only,
-                    on_bucket=on_bucket,
-                    skip_buckets=skip_buckets,
-                    _depth=_depth + 1,
-                )
-                return
-            # Бакет уже узкий (< MIN_BRACKET) — split не поможет, хвост теряем
-            # осознанно (как cian) и логируем.
-            logger.error(
-                "avito: skipping bucket %s [%d, %d] — total unavailable after retry "
-                "(bracket=%d < MIN_BRACKET=%d, tail loss accepted)",
-                room_label,
-                lo,
-                hi,
-                bracket_size,
-                _AVITO_MIN_BRACKET,
-            )
-            return
-
-        logger.info(
-            "avito: %s [%d, %s] total=%d depth=%d",
-            room_label,
-            lo,
-            _hi_repr,
-            total,
-            _depth,
-        )
-
-        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)
+            else:
+                if phi is None and 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)
+                if phi is not None and pages_needed > max_pages_per_bucket:
+                    tail_loss = total - max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE
+                    logger.warning(
+                        "avito: %s [%d, %d] total=%d exceeds page cap "
+                        "(max_pages=%d × %d=%d offers) — tail loss ~%d offers",
+                        room_label,
+                        _lo,
+                        phi,
+                        total,
+                        max_pages_per_bucket,
+                        _AVITO_OFFERS_PER_PAGE,
+                        max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE,
+                        tail_loss,
+                    )
+                html = result.payload
             await self._paginate_leaf_bucket(
                 room_slug=room_slug,
                 room_label=room_label,
-                lo=lo,
-                hi=None,
+                lo=_lo,
+                hi=phi,
                 html=html,
                 max_pages=max_pages,
                 seen=seen,
@@ -1304,95 +1266,14 @@ class AvitoScraper(BaseScraper):
                 on_bucket=on_bucket,
                 skip_buckets=skip_buckets,
             )
-            return
 
-        # ── Шаг 2: деление или пагинация ──────────────────────────────────────
-        bracket_size = hi - lo
-        need_split = total > price_cap_per_bucket
-        too_narrow = bracket_size < _AVITO_MIN_BRACKET
-
-        if need_split and too_narrow:
-            logger.warning(
-                "avito: %s [%d, %d] total=%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,
-                _AVITO_MIN_BRACKET,
-                max(0, total - price_cap_per_bucket),
-            )
-            need_split = False  # принудительно пагинируем
-
-        if need_split:
-            mid = (lo + hi) // 2
-            # [lo, mid]
-            await self._walk_price_range(
-                room_slug=room_slug,
-                room_label=room_label,
-                lo=lo,
-                hi=mid,
-                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,
-                _depth=_depth + 1,
-            )
-            # [mid+1, hi]
-            await self._walk_price_range(
-                room_slug=room_slug,
-                room_label=room_label,
-                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,
-                secondary_only=secondary_only,
-                on_bucket=on_bucket,
-                skip_buckets=skip_buckets,
-                _depth=_depth + 1,
-            )
-            return
-
-        # ── Закрытый leaf-бакет: считаем max_pages + warning, делегируем helper'у ─
-        pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE)
-        max_pages = min(pages_needed, max_pages_per_bucket)
-
-        if pages_needed > max_pages_per_bucket:
-            tail_loss = total - max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE
-            logger.warning(
-                "avito: %s [%d, %d] total=%d exceeds page cap "
-                "(max_pages=%d × %d=%d offers) — tail loss ~%d offers",
-                room_label,
-                lo,
-                hi,
-                total,
-                max_pages_per_bucket,
-                _AVITO_OFFERS_PER_PAGE,
-                max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE,
-                tail_loss,
-            )
-
-        await self._paginate_leaf_bucket(
-            room_slug=room_slug,
-            room_label=room_label,
+        await walk_price_range(
             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,
+            config=_avito_bisection_config(price_cap_per_bucket),
+            probe=_probe,
+            on_leaf=_leaf,
+            depth=_depth,
         )
 
     async def _paginate_leaf_bucket(

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/serp.py

@@ -39,6 +39,7 @@ from scraper_kit.browser_fetcher import BrowserFetcher
 from scraper_kit.cian_state_parser import extract_state
 from scraper_kit.house_type_normalizer import normalize_house_type
 from scraper_kit.price_brackets import get_price_seed_brackets
+from scraper_kit.pricing import BisectionConfig, ProbeFailPolicy, ProbeResult, walk_price_range
 from scraper_kit.repair_state_normalizer import (
     infer_repair_state_from_text,
     normalize_repair_state,
@@ -69,6 +70,17 @@ _MIN_BRACKET = 50_000
 # Cian SERP показывает ~28 офферов на страницу.
 _CIAN_OFFERS_PER_PAGE = 28
 
+# Конфиг единого движка бисекции (scraper_kit.pricing) для cian. cap задаётся
+# per-call (price_cap_per_bucket) — здесь только структурные константы/политики.
+# probe_fail=SKIP: totalOffers недоступен после ретрая → бакет пропускается.
+# open_split_ceiling=None: открытый брекет пагинируется напрямую (не делится).
+def _cian_bisection_config(cap: int) -> BisectionConfig:
+    return BisectionConfig(
+        cap=cap,
+        min_bracket=_MIN_BRACKET,
+        probe_fail_policy=ProbeFailPolicy.SKIP,
+    )
+
 # Комнатности по умолчанию для exhaustive-загрузки.
 # Коды Cian: room1=1к, room2=2к, room3=3к, room4=4к, room5=5к, room6=6+.
 # TODO: студии — Cian использует отдельный тип flat_type=studio, не room-param.
@@ -123,7 +135,9 @@ class CianScraper(BaseScraper):
         # fingerprint через tradein-browser сервис (#1806). Сервер роутит /fetch на
         # отдельный браузер+мобильный прокси для cian. Прокси ротирует IP сам, поэтому
         # code-side warm-up cookies и changeip-ротация больше не нужны.
-        self._browser = BrowserFetcher(source="cian")
+        self._browser = BrowserFetcher(
+            source="cian", endpoint=self._config.browser_http_endpoint
+        )
         await self._browser.__aenter__()
         return self
 
@@ -398,76 +412,75 @@ class CianScraper(BaseScraper):
         skip_buckets: если bucket_key в skip_buckets — пагинацию и on_bucket пропускаем.
         secondary_only: если True — новостройки (listing_segment=="novostroyki") отбрасываются
           после сбора bucket_lots, до дедупа в seen и вызова on_bucket.
+
+        Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Cian-spec
+        probe (fetch page 1 + retry) и leaf-пагинация переданы callback'ами; пороги
+        (cap, _MIN_BRACKET) и политика probe-fail (SKIP) — в BisectionConfig.
         """
         room_label = f"room{'_'.join(str(r) for r in rooms)}"
-        _lo_param = lo if lo > 0 else None
-        _hi_repr = "open" if hi is None else str(hi)
 
-        # ── Шаг 1: probe page 1 ────────────────────────────────────────────────
-        html = await self._fetch_page_html(rooms, 1, _lo_param, hi)
-        await self.sleep_between_requests()
-
-        total: int | None = None
-        if html is not None:
-            total = self._extract_total_offers(html)
-
-        # Ретрай на captcha/ошибку: 1 повторный fetch. Мобильный прокси в браузер-
-        # контейнере ротирует IP сам (#1806), поэтому отдельный changeip-вызов
-        # (_rotate_ip) больше не нужен — просто повторяем запрос на чистом IP.
-        if total is None:
-            logger.warning(
-                "cian: totalOffers=None for %s [%d, %s] depth=%d — retry (proxy auto-rotates IP)",
-                room_label,
-                lo,
-                _hi_repr,
-                _depth,
-            )
-            await self._rotate_ip()  # no-op (back-compat); прокси ротирует IP сам
-            html = await self._fetch_page_html(rooms, 1, _lo_param, hi)
+        async def _probe(plo: int | None, phi: int | None, depth: int) -> ProbeResult:
+            lo_param = plo if plo and plo > 0 else None
+            hi_repr = "open" if phi is None else str(phi)
+            # ── Шаг 1: probe page 1 ───────────────────────────────────────────
+            html = await self._fetch_page_html(rooms, 1, lo_param, phi)
             await self.sleep_between_requests()
-            if html is not None:
-                total = self._extract_total_offers(html)
+            total = self._extract_total_offers(html) if html is not None else None
 
-        if total is None:
-            logger.error(
-                "cian: skipping bucket %s [%d, %s] — totalOffers unavailable after retry",
-                room_label,
-                lo,
-                _hi_repr,
-            )
-            return
-
-        logger.info(
-            "cian: %s [%d, %s] totalOffers=%d depth=%d",
-            room_label,
-            lo,
-            _hi_repr,
-            total,
-            _depth,
-        )
-
-        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:
+            # Ретрай на captcha/ошибку: 1 повторный fetch. Мобильный прокси в браузер-
+            # контейнере ротирует IP сам (#1806), поэтому отдельный changeip-вызов
+            # (_rotate_ip) больше не нужен — просто повторяем запрос на чистом IP.
+            if total is None:
                 logger.warning(
-                    "cian: OPEN bucket %s [%d, open] totalOffers=%d > cap=%d — paginating "
+                    "cian: totalOffers=None for %s [%s, %s] depth=%d — retry "
+                    "(proxy auto-rotates IP)",
+                    room_label,
+                    plo,
+                    hi_repr,
+                    depth,
+                )
+                await self._rotate_ip()  # no-op (back-compat); прокси ротирует IP сам
+                html = await self._fetch_page_html(rooms, 1, lo_param, phi)
+                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 [%s, %s] — totalOffers unavailable after retry",
+                    room_label,
+                    plo,
+                    hi_repr,
+                )
+            else:
+                logger.info(
+                    "cian: %s [%s, %s] totalOffers=%d depth=%d",
+                    room_label,
+                    plo,
+                    hi_repr,
+                    total,
+                    depth,
+                )
+            return ProbeResult(count=total, payload=html)
+
+        async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None:
+            total = result.count
+            assert total is not None  # SKIP-политика: leaf не вызывается при None
+            if phi is None and total > price_cap_per_bucket:
+                logger.warning(
+                    "cian: OPEN bucket %s [%s, open] totalOffers=%d > cap=%d — paginating "
                     "without split (tail-loss accepted, lux tail tiny)",
                     room_label,
-                    lo,
+                    plo,
                     total,
                     price_cap_per_bucket,
                 )
-            await self._paginate_open_bucket(
+            await self._paginate_leaf_bucket(
                 rooms=rooms,
                 room_label=room_label,
-                lo=lo,
-                html=html,
+                lo=plo if plo is not None else 0,
+                hi=phi,
+                html=result.payload,
                 total=total,
                 seen=seen,
                 max_pages_per_bucket=max_pages_per_bucket,
@@ -476,105 +489,14 @@ class CianScraper(BaseScraper):
                 on_bucket=on_bucket,
                 skip_buckets=skip_buckets,
             )
-            return
 
-        # ── Шаг 2: деление или пагинация (только ЗАКРЫТЫЙ брекет, hi: int) ─────
-        bracket_size = 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, %d] 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:
-            mid = (lo + 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,
-                concurrency=concurrency,
-                secondary_only=secondary_only,
-                on_bucket=on_bucket,
-                skip_buckets=skip_buckets,
-                _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,
-                concurrency=concurrency,
-                secondary_only=secondary_only,
-                on_bucket=on_bucket,
-                skip_buckets=skip_buckets,
-                _depth=_depth + 1,
-            )
-            return
-
-        # Закрытый leaf-бакет — общая пагинация (см. _paginate_leaf_bucket).
-        await self._paginate_leaf_bucket(
-            rooms=rooms,
-            room_label=room_label,
+        await walk_price_range(
             lo=lo,
             hi=hi,
-            html=html,
-            total=total,
-            seen=seen,
-            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_open_bucket(
-        self,
-        *,
-        rooms: tuple[int, ...],
-        room_label: str,
-        lo: int,
-        html: str | None,
-        total: int,
-        seen: dict[str, ScrapedLot],
-        max_pages_per_bucket: int,
-        concurrency: int,
-        secondary_only: bool,
-        on_bucket: Callable[..., Any] | None,
-        skip_buckets: set[str] | None,
-    ) -> None:
-        """Пагинация ОТКРЫТОГО leaf-бакета (hi=None, без maxprice)."""
-        await self._paginate_leaf_bucket(
-            rooms=rooms,
-            room_label=room_label,
-            lo=lo,
-            hi=None,
-            html=html,
-            total=total,
-            seen=seen,
-            max_pages_per_bucket=max_pages_per_bucket,
-            concurrency=concurrency,
-            secondary_only=secondary_only,
-            on_bucket=on_bucket,
-            skip_buckets=skip_buckets,
+            config=_cian_bisection_config(price_cap_per_bucket),
+            probe=_probe,
+            on_leaf=_leaf,
+            depth=_depth,
         )
 
     async def _paginate_leaf_bucket(

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/domclick/serp.py

@@ -37,6 +37,7 @@ from urllib.parse import urlencode
 
 from scraper_kit.base import BaseScraper, ScrapedLot
 from scraper_kit.domclick_exceptions import DomClickBlockedError
+from scraper_kit.pricing import BisectionConfig, ProbeResult, walk_price_range
 from scraper_kit.repair_state_normalizer import infer_repair_state_from_text
 
 if TYPE_CHECKING:
@@ -61,6 +62,24 @@ PAGE_SIZE: int = 20  # items per page (жёстко задан API)
 LTE_MAX: int = 1_000_000_000
 MIN_PRICE_SPAN: int = 100_000  # ниже этого span прекращаем делить
 
+# Конфиг единого движка бисекции (scraper_kit.pricing) для domclick. Отличия от
+# avito/cian/yandex закодированы явно:
+#   * cap = OFFSET_CAP (2000) — потолок offset BFF API, не «результатов бакета»;
+#   * count через отдельный count-endpoint (probe.payload=None, пагинация с offset=0);
+#   * min_bracket_inclusive=False — исходное `span <= MIN_PRICE_SPAN → paginate`
+#     (делим только пока span СТРОГО > MIN_PRICE_SPAN);
+#   * open_split_ceiling=LTE_MAX / floor=0 — unbounded-бакет (price_gte/lte=None)
+#     клампится к [0, LTE_MAX] и ДЕЛИТСЯ (в отличие от открытого seed-брекета
+#     avito/cian/yandex, который пагинируется напрямую);
+#   * seed-брекетов нет — каждый ROOM_BUCKET стартует с одного (None, None).
+_DOMCLICK_BISECTION = BisectionConfig(
+    cap=OFFSET_CAP,
+    min_bracket=MIN_PRICE_SPAN,
+    min_bracket_inclusive=False,
+    open_split_ceiling=LTE_MAX,
+    open_split_floor=0,
+)
+
 # ── QRATOR block detection ────────────────────────────────────────────────────
 
 _QRATOR_MARKERS: tuple[str, ...] = (
@@ -406,49 +425,27 @@ class DomClickScraper(BaseScraper):
 
         Raises:
             DomClickBlockedError: если QRATOR-блок — propagate наверх.
-        """
-        count = await self._count(fetcher, rooms, price_gte, price_lte)
-        if count == 0:
-            return
 
-        if count > OFFSET_CAP:
-            _gte = price_gte if price_gte is not None else 0
-            _lte = price_lte if price_lte is not None else LTE_MAX
-            span = _lte - _gte
-            if span <= MIN_PRICE_SPAN:
-                logger.warning(
-                    "domklik: rooms=%r price=[%s,%s] count=%d > cap=%d "
-                    "but span=%d <= min=%d — paginating as-is (bucket truncated at %d)",
-                    rooms,
-                    price_gte,
-                    price_lte,
-                    count,
-                    OFFSET_CAP,
-                    span,
-                    MIN_PRICE_SPAN,
-                    OFFSET_CAP,
-                )
-                await self._paginate(
-                    fetcher, rooms, price_gte, price_lte, seen_ids, out_lots, pages
-                )
-                return
-            mid = (_gte + _lte) // 2
-            logger.debug(
-                "domklik: rooms=%r count=%d > cap=%d — price-split [%d,%d] → [%d,%d]+[%d,%d]",
-                rooms,
-                count,
-                OFFSET_CAP,
-                _gte,
-                _lte,
-                _gte,
-                mid,
-                mid + 1,
-                _lte,
-            )
-            await self._sweep_bucket(fetcher, rooms, _gte, mid, seen_ids, out_lots, pages)
-            await self._sweep_bucket(fetcher, rooms, mid + 1, _lte, seen_ids, out_lots, pages)
-        else:
-            await self._paginate(fetcher, rooms, price_gte, price_lte, seen_ids, out_lots, pages)
+        Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Domclick-spec
+        count-endpoint (probe) и offset-пагинация (leaf) переданы callback'ами; отличия
+        (OFFSET_CAP, MIN_PRICE_SPAN строго-больше, clamp unbounded к [0, LTE_MAX] с
+        делением) закодированы в _DOMCLICK_BISECTION.
+        """
+
+        async def _probe(plo: int | None, phi: int | None, depth: int) -> ProbeResult:
+            count = await self._count(fetcher, rooms, plo, phi)
+            return ProbeResult(count=count, payload=None)
+
+        async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None:
+            await self._paginate(fetcher, rooms, plo, phi, seen_ids, out_lots, pages)
+
+        await walk_price_range(
+            lo=price_gte,
+            hi=price_lte,
+            config=_DOMCLICK_BISECTION,
+            probe=_probe,
+            on_leaf=_leaf,
+        )
 
     async def _paginate(
         self,

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py

@@ -48,8 +48,9 @@ from curl_cffi.requests import AsyncSession as _CurlCffiSession
 
 from scraper_kit.base import BaseScraper, ScrapedLot
 from scraper_kit.browser_fetcher import BrowserFetcher
+from scraper_kit.house_type_normalizer import normalize_house_type
 from scraper_kit.price_brackets import get_price_seed_brackets
-from scraper_kit.providers.yandex.shared import normalize_house_type
+from scraper_kit.pricing import BisectionConfig, ProbeFailPolicy, ProbeResult, walk_price_range
 
 if TYPE_CHECKING:
     from scraper_kit.contracts import ScraperConfig
@@ -90,7 +91,15 @@ DEFAULT_PRICE_RANGES: list[tuple[int | None, int | None]] = [
 ]
 
 _YANDEX_MAX_PRICE = 200_000_000
+# ВНИМАНИЕ: реальный порог split у yandex — 500_000 (см. _YANDEX_SPLIT_MIN_BRACKET),
+# а НЕ 50_000. Историческая _YANDEX_MIN_BRACKET=50_000 никогда не использовалась в
+# _walk_price_range (там был локальный `_min_bracket = 500_000`), оставлена для
+# обратной совместимости импортов; движок бисекции использует 500_000.
 _YANDEX_MIN_BRACKET = 50_000
+# Минимальный размах диапазона для деления (yandex делит только пока (hi-lo) >= 500k)
+# и жёсткий предел глубины рекурсии — переносим в единый движок как есть.
+_YANDEX_SPLIT_MIN_BRACKET = 500_000
+_YANDEX_MAX_SPLIT_DEPTH = 8
 _GATE_PAGE_SIZE = 20
 _YANDEX_TARPIT_MAX_RETRIES: int = 2
 
@@ -478,7 +487,9 @@ class YandexRealtyScraper(BaseScraper):
         на один тайм-аутовый запрос: worst-case 30s×(1+retries)=90s на combo вместо 360s.
         """
         self._browser = BrowserFetcher(
-            source="yandex", fetch_timeout_s=_YANDEX_BROWSER_FETCH_TIMEOUT_S
+            source="yandex",
+            fetch_timeout_s=_YANDEX_BROWSER_FETCH_TIMEOUT_S,
+            endpoint=self._config.browser_http_endpoint,
         )
         await self._browser.__aenter__()
         return self
@@ -958,55 +969,66 @@ class YandexRealtyScraper(BaseScraper):
         split (no mid) -> paginate leaf directly (lux tail tiny, tail-loss accepted).
         Fallback: totalItems=None -> paginate-until-empty (safe degradation).
         skip_buckets: resume checkpoint -- skip leaf buckets already done.
+
+        Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Yandex-spec
+        (probe gate-JSON page=1, degraded paginate-until-empty, leaf с first-wins
+        дедупом и count-based on_bucket) — в callback'ах; пороги (cap, 500k min,
+        depth<8) и политика DEGRADE — в BisectionConfig.
         """
-        _lo_param = lo if lo > 0 else None
-        _hi_repr = "open" if hi is None else str(hi)
 
-        probe_payload = await self._fetch_page_json(rooms, 1, _lo_param, hi)
-        await asyncio.sleep(self.request_delay_sec)
+        async def _probe(plo: int | None, phi: int | None, depth: int) -> ProbeResult:
+            lo_param = plo if plo and plo > 0 else None
+            hi_repr = "open" if phi is None else str(phi)
+            probe_payload = await self._fetch_page_json(rooms, 1, lo_param, phi)
+            await asyncio.sleep(self.request_delay_sec)
 
-        total: int | None = None
-        probe_lots: list[ScrapedLot] = []
-
-        if probe_payload is None:
-            logger.warning(
-                "yandex gate: probe failed rooms=%s [%d, %s] depth=%d -- rotating + retry",
-                rooms,
-                lo,
-                _hi_repr,
-                _depth,
-            )
-            rotated = await self._rotate_ip()
-            if rotated:
-                probe_payload = await self._fetch_page_json(rooms, 1, _lo_param, hi)
-                await asyncio.sleep(self.request_delay_sec)
-
-        if probe_payload is not None:
-            result = _extract_gate_data(probe_payload)
-            if result is not None:
-                _, pager = result
-                total = pager.get("totalItems")
-                probe_lots = _parse_gate_json(probe_payload, page_param=1)
-            else:
+            if probe_payload is None:
                 logger.warning(
-                    "yandex gate: probe data extract failed rooms=%s [%d, %s]", rooms, lo, _hi_repr
+                    "yandex gate: probe failed rooms=%s [%s, %s] depth=%d -- rotating + retry",
+                    rooms,
+                    plo,
+                    hi_repr,
+                    depth,
                 )
+                rotated = await self._rotate_ip()
+                if rotated:
+                    probe_payload = await self._fetch_page_json(rooms, 1, lo_param, phi)
+                    await asyncio.sleep(self.request_delay_sec)
 
-        if total is None:
-            # Degraded path: paginate-until-empty without knowing total
+            total: int | None = None
+            probe_lots: list[ScrapedLot] = []
+            if probe_payload is not None:
+                extracted = _extract_gate_data(probe_payload)
+                if extracted is not None:
+                    _, pager = extracted
+                    total = pager.get("totalItems")
+                    probe_lots = _parse_gate_json(probe_payload, page_param=1)
+                else:
+                    logger.warning(
+                        "yandex gate: probe data extract failed rooms=%s [%s, %s]",
+                        rooms,
+                        plo,
+                        hi_repr,
+                    )
+            return ProbeResult(count=total, payload=probe_lots)
+
+        async def _degraded(plo: int | None, phi: int | None) -> None:
+            # totalItems неизвестен → пагинируем до пустоты, не зная total.
+            hi_repr = "open" if phi is None else str(phi)
+            lo_param = plo if plo and plo > 0 else None
             logger.warning(
-                "yandex gate: totalItems unknown rooms=%s [%d, %s] -- paginate until empty",
+                "yandex gate: totalItems unknown rooms=%s [%s, %s] -- paginate until empty",
                 rooms,
-                lo,
-                _hi_repr,
+                plo,
+                hi_repr,
             )
-            page = 1
-            bucket_key = _combo_label(rooms, lo, hi)
+            bucket_key = _combo_label(rooms, plo, phi)
             if skip_buckets and bucket_key in skip_buckets:
                 return
+            page = 1
             pages_fetched = 0
             while pages_fetched < max_pages_per_bucket:
-                payload = await self._fetch_page_json(rooms, page, _lo_param, hi)
+                payload = await self._fetch_page_json(rooms, page, lo_param, phi)
                 await asyncio.sleep(self.request_delay_sec)
                 if payload is None:
                     break
@@ -1020,84 +1042,72 @@ class YandexRealtyScraper(BaseScraper):
                 pages_fetched += 1
             if on_bucket is not None:
                 on_bucket(bucket_key, len(seen))
-            return
 
-        # Split or paginate. ОТКРЫТЫЙ брекет (hi is None) делить нельзя (нет mid) —
-        # сразу пагинируем leaf напрямую (хвост люкса крошечный, tail-loss accepted).
-        cap = price_cap_per_bucket
-        _min_bracket = 500_000
+        async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None:
+            total = result.count
+            assert total is not None  # DEGRADE-политика уводит None в _degraded
+            lo_param = plo if plo and plo > 0 else None
+            probe_lots: list[ScrapedLot] = result.payload or []
+            bucket_key = _combo_label(rooms, plo, phi)
+            if skip_buckets and bucket_key in skip_buckets:
+                logger.debug("yandex gate: skip bucket %s (checkpoint)", bucket_key)
+                return
 
-        if hi is not None and total > cap and (hi - lo) >= _min_bracket and _depth < 8:
-            mid = (lo + hi) // 2
-            await self._walk_price_range(
-                rooms=rooms,
-                lo=lo,
-                hi=mid,
-                seen=seen,
-                price_cap_per_bucket=cap,
-                max_pages_per_bucket=max_pages_per_bucket,
-                concurrency=concurrency,
-                on_bucket=on_bucket,
-                skip_buckets=skip_buckets,
-                _depth=_depth + 1,
+            total_pages = min(
+                math.ceil(total / 20),
+                _GATE_MAX_PAGES_CAP,
+                max_pages_per_bucket,
             )
-            await self._walk_price_range(
-                rooms=rooms,
-                lo=mid + 1,
-                hi=hi,
-                seen=seen,
-                price_cap_per_bucket=cap,
-                max_pages_per_bucket=max_pages_per_bucket,
-                concurrency=concurrency,
-                on_bucket=on_bucket,
-                skip_buckets=skip_buckets,
-                _depth=_depth + 1,
+            logger.info(
+                "yandex gate: leaf bucket %s total=%d pages=%d", bucket_key, total, total_pages
             )
-            return
 
-        # Leaf bucket: paginate
-        bucket_key = _combo_label(rooms, lo, hi)
-        if skip_buckets and bucket_key in skip_buckets:
-            logger.debug("yandex gate: skip bucket %s (checkpoint)", bucket_key)
-            return
-
-        total_pages = min(
-            math.ceil(total / 20),
-            _GATE_MAX_PAGES_CAP,
-            max_pages_per_bucket,
-        )
-        logger.info("yandex gate: leaf bucket %s total=%d pages=%d", bucket_key, total, total_pages)
-
-        # Add probe lots (page 1 already fetched)
-        for lot in probe_lots:
-            if lot.source_id and lot.source_id not in seen:
-                seen[lot.source_id] = lot
-
-        if total_pages <= 1:
-            if on_bucket is not None:
-                on_bucket(bucket_key, len(seen))
-            return
-
-        # Paginate pages 2..total_pages with concurrency
-        sem = asyncio.Semaphore(concurrency)
-
-        async def _fetch_leaf_page(pg: int) -> list[ScrapedLot]:
-            async with sem:
-                payload = await self._fetch_page_json(rooms, pg, _lo_param, hi)
-                await asyncio.sleep(self.request_delay_sec)
-                if payload is None:
-                    return []
-                return _parse_gate_json(payload, page_param=pg)
-
-        tasks = [_fetch_leaf_page(pg) for pg in range(2, total_pages + 1)]
-        results = await asyncio.gather(*tasks)
-        for page_lots in results:
-            for lot in page_lots:
+            # Add probe lots (page 1 already fetched)
+            for lot in probe_lots:
                 if lot.source_id and lot.source_id not in seen:
                     seen[lot.source_id] = lot
 
-        if on_bucket is not None:
-            on_bucket(bucket_key, len(seen))
+            if total_pages <= 1:
+                if on_bucket is not None:
+                    on_bucket(bucket_key, len(seen))
+                return
+
+            # Paginate pages 2..total_pages with concurrency
+            sem = asyncio.Semaphore(concurrency)
+
+            async def _fetch_leaf_page(pg: int) -> list[ScrapedLot]:
+                async with sem:
+                    payload = await self._fetch_page_json(rooms, pg, lo_param, phi)
+                    await asyncio.sleep(self.request_delay_sec)
+                    if payload is None:
+                        return []
+                    return _parse_gate_json(payload, page_param=pg)
+
+            tasks = [_fetch_leaf_page(pg) for pg in range(2, total_pages + 1)]
+            results = await asyncio.gather(*tasks)
+            for page_lots in results:
+                for lot in page_lots:
+                    if lot.source_id and lot.source_id not in seen:
+                        seen[lot.source_id] = lot
+
+            if on_bucket is not None:
+                on_bucket(bucket_key, len(seen))
+
+        await walk_price_range(
+            lo=lo,
+            hi=hi,
+            config=BisectionConfig(
+                cap=price_cap_per_bucket,
+                min_bracket=_YANDEX_SPLIT_MIN_BRACKET,
+                max_depth=_YANDEX_MAX_SPLIT_DEPTH,
+                probe_fail_policy=ProbeFailPolicy.DEGRADE,
+                treat_zero_as_leaf=True,
+            ),
+            probe=_probe,
+            on_leaf=_leaf,
+            on_degraded=_degraded,
+            depth=_depth,
+        )
 
 
 def _combo_label(rooms: str | None, lo: int | None, hi: int | None) -> str:

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/shared.py

@@ -1,95 +1,15 @@
-"""Нормализация house_type → каноничный enum для listings.house_type.
+"""Re-export `normalize_house_type` из core-слоя scraper_kit (#2134 dedup).
 
-Strangler-копия `app.services.scrapers.house_type_normalizer` (#2133). Локальная в
-yandex-провайдере, потому что house_type_normalizer ещё не перенесён в core-слой
-scraper_kit (это отдельный шаг консолидации E). Копия — pure-функция без `app.*`
-зависимостей; идентична источнику.
-
-Целевой канон (сверено с estimator._IMV_HOUSE_TYPE_MAP, estimator.py:146):
-  panel / brick / monolith / monolith_brick / block / wood.
-
-Зачем (#2007): estimator применяет soft-penalty по house_type — аналог с
-house_type != target штрафуется. Yandex SERP отдаёт SCREAMING-значения
-(MONOLIT / BRICK / PANEL / ...), которые НИКОГДА не равны каноничным
-(monolith / brick / panel / ...) → ~70% yandex-аналогов получали ложный штраф
-и фактически выпадали из скоринга. Нормализация на ингесте чинит это.
-
-Важно про None: неизвестное / 'other' / '' → None, НЕ 'other'. В estimator
-`house_type IS NULL` нейтрально (без штрафа), а любое не-канон значение всегда
-!= target → ложный штраф. Поэтому unknown лучше схлопнуть в NULL.
-
-Источники raw-значений:
-- yandex SERP: building.buildingType — SCREAMING_SNAKE (MONOLIT, MONOLIT_BRICK, ...)
-- cian SERP: building.materialType — camelCase (monolith, monolithBrick,
-  gasSilicateBlock, ...) — переиспользуется в #2008 (cian.py:815).
+Раньше здесь лежала ПОЛНАЯ копия нормализатора house_type (strangler-шаг #2133).
+Теперь консолидировано: единственная реализация — `scraper_kit.house_type_normalizer`
+(идентична бывшей копии, сверено картой _RAW_TO_CANON). Модуль сохранён как тонкий
+re-export, чтобы не ломать существующие импорты
+(`from scraper_kit.providers.yandex.shared import normalize_house_type`) и
+golden-parity тест.
 """
 
 from __future__ import annotations
 
-import logging
+from scraper_kit.house_type_normalizer import normalize_house_type
 
-logger = logging.getLogger(__name__)
-
-# Канон listings.house_type — выровнен по estimator._IMV_HOUSE_TYPE_MAP.
-_CANON: frozenset[str] = frozenset(
-    {"panel", "brick", "monolith", "monolith_brick", "block", "wood"}
-)
-
-# raw-токен → канон. Ключи — точные значения вокабуляров источников; сравнение
-# регистронезависимое (см. _LOOKUP ниже), но строго по полному токену енума,
-# без fuzzy-матчинга. Неизвестные токены сюда НЕ попадают → normalize вернёт None.
-_RAW_TO_CANON: dict[str, str] = {
-    # ── yandex SERP (SCREAMING_SNAKE buildingType) ───────────────────────────
-    "MONOLIT": "monolith",
-    "BRICK": "brick",
-    "PANEL": "panel",
-    "MONOLIT_BRICK": "monolith_brick",
-    "BLOCK": "block",
-    "WOOD": "wood",
-    # ── cian SERP (camelCase materialType) — reuse в #2008 ───────────────────
-    "monolith": "monolith",
-    "brick": "brick",
-    "panel": "panel",
-    "block": "block",
-    "wood": "wood",
-    "monolithBrick": "monolith_brick",
-    "gasSilicateBlock": "block",
-    "aerocreteBlock": "block",
-    "foamConcreteBlock": "block",
-    "stalin": "brick",  # «сталинка» — кирпич
-}
-
-# Регистронезависимый lookup. Лоуэркейс-ключи не коллизят между вокабулярами:
-# 'MONOLIT'→'monolit' и 'monolith'→'monolith' — разные ключи.
-_LOOKUP: dict[str, str] = {k.lower(): v for k, v in _RAW_TO_CANON.items()}
-
-
-def normalize_house_type(raw: str | None) -> str | None:
-    """Преобразовать raw house_type в каноничный enum.
-
-    Уже-каноничное значение возвращается as-is (идемпотентность — нужна при
-    ре-обработке и для backfill-миграции). Неизвестное / 'other' / пустое /
-    None → None (НЕ 'other': см. docstring модуля про estimator soft-penalty).
-
-    Args:
-        raw: сырое значение из парсера (e.g. «MONOLIT», «monolithBrick») или None.
-
-    Returns:
-        Одно из panel / brick / monolith / monolith_brick / block / wood, либо None.
-    """
-    if raw is None:
-        return None
-    stripped = raw.strip()
-    if not stripped:
-        return None
-    # Pass-through: уже каноничное значение (idempotency — в т.ч. 'monolith_brick',
-    # которого нет среди raw-ключей карты).
-    if stripped in _CANON:
-        return stripped
-    # Регистронезависимый exact-token lookup по обоим вокабулярам.
-    canon = _LOOKUP.get(stripped) or _LOOKUP.get(stripped.lower())
-    if canon is None:
-        # Неизвестное / 'other' — нормально (NULL нейтрально для estimator).
-        # debug, не warning: 'other' встречается массово, warning засорил бы лог.
-        logger.debug("house_type_normalizer: unmapped raw value %r — stored as NULL", raw)
-    return canon
+__all__ = ["normalize_house_type"]