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

"""Единый движок адаптивной бисекции «комнатность × цена» для 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)