gendesign/tradein-mvp/backend/app/services/backfill_block_breaker.py

"""Критерий обрыва avito_detail_backfill по доле блоков (#3184).

Модуль написан источник-агностично (см. BlockRatioBreaker ниже), но единственный
текущий вызывающий -- avito_detail_backfill. Исходная задача планировала тот же
критерий и для domclick_detail_backfill ("тот же критерий, что у avito"), но это
снято ревью (#3184 review MAJOR 1): у Домклика ОДИН выделенный residential-прокси
БЕЗ ротации (data/sql/175_scrape_schedules_seed_domclick_detail_backfill.sql,
max_consecutive_blocks=3 против avito=5 -- намеренно ниже, ранний обрыв бережёт
репутацию единственного узла), а калибровка окна/порога ниже сделана на avito-пуле
С ротацией. При устойчивой доле блоков около порога (например 0.65) Домклик по
этому критерию не обрывался бы никогда -- десятки запросов по QRATOR с одного
IP, ровно то, что max_consecutive_blocks=3 был призван предотвратить. Домклик
получит свою калибровку отдельной задачей.

Раньше avito_detail_backfill абортил по голой длине "N блоков подряд"
(max_consecutive_blocks). Замер 40 прогонов avito за 14 суток (2026-08-14..28)
показал: ВСЕ 40 закончились статусом 'banned', при этом доля блоков в них
колебалась 25-100% (47/190=25%, 39/100=39%, 33/71=46%, 20/63=32%, 5/5=100%) --
"N подряд" не отличал выгоревший прокси-пул от прогона, который честно
обогатил 125 карточек из 190. Причина: блоки автокоррелированы и идут
пачками -- при базовой доле 25-46% пачка из 5-6 подряд у независимой модели
была бы редкостью, но встречается в каждом прогоне.

Критерий заменён на долю блоков в скользящем окне последних N попыток: пачка
сама по себе больше не абортит, абортит устойчиво высокая доля (>= порога,
не строго "выше" -- окно 20 при пороге 0.7 абортит уже на 14/20, 13/20 ещё
нет). Окно короче типичной пачки бесполезно, поэтому дефолт (settings.
detail_backfill_block_ratio_window) заметно больше 5-6.

Safety-net для прогонов КОРОЧЕ окна (снапшот меньше window_size -- ratio-
критерий физически недостижим, окно никогда не заполнится) сохранён: если ВСЕ
попытки с начала прогона были блоками (ни одного успеха) и накопилось не
меньше safety_min -- абортим не дожидаясь окна. Раньше это было единственным
критерием под именем max_consecutive_blocks; имя и дефолт (5) сохранены как
safety_min, чтобы не менять поведение коротких прогонов вида 0/5 (#3184).

Условие "снапшот короче окна" обязательно, иначе safety-net гасит сам фикс:
холодный старт сессии/прокси -- самое вероятное место пачки блоков, и на
длинном прогоне (снапшот >= окна) пачка из safety_min блоков В НАЧАЛЕ
абортила бы точно так же, как до правки, до того как ratio-критерий вообще
успел бы включиться (#3184 review MAJOR 2).

Отказы-не-блоки (сеть/таймаут/парсер уровня "ответ есть, но наш") НЕ рвут
"чистоту" серии для safety-net -- это исторический инвариант
avito_detail_backfill (только успех сбрасывал consecutive_blocks,
TimeoutError/Exception его не трогали), сохранён через record_failure():
она двигает окно (значит, доля в ratio-критерии всё равно снижается), но НЕ
трогает _pure_block_run и НЕ сбрасывает серию. Из-за этого 4 таймаута + 5
блоков подряд (ни одного успеха) всё ещё дают safety-net abort -- это
осознанное поведение, а не баг: таймаут не доказывает, что тракт живой ответ
получает, только успех доказывает.
"""

from __future__ import annotations

from collections import Counter, deque
from dataclasses import dataclass, field

from scraper_kit.orchestration.runs import BAN_KIND_PLATFORM


@dataclass
class BlockRatioBreaker:
    """Стейт одного прогона detail-backfill'а: окно, safety-net, гистограмма пачек."""

    window_size: int
    ratio_threshold: float
    safety_min: int
    snapshot_size: int
    _window: deque[bool] = field(init=False, repr=False)
    _consecutive_blocks: int = field(default=0, init=False)
    _pure_block_run: bool = field(default=True, init=False)
    streak_histogram: Counter[int] = field(default_factory=Counter, init=False)

    def __post_init__(self) -> None:
        self._window = deque(maxlen=max(self.window_size, 1))

    @property
    def consecutive_blocks(self) -> int:
        """Для логов ABORT/BLOCKED -- та же цифра, что раньше выводилась в consecutive=%d."""
        return self._consecutive_blocks

    @property
    def window_blocks(self) -> int:
        """Числитель ratio-критерия: сколько блоков в текущем окне."""
        return sum(self._window)

    @property
    def window_len(self) -> int:
        """Знаменатель ratio-критерия: сколько попыток уже влезло в окно."""
        return len(self._window)

    def record_block(self, kind: str = BAN_KIND_PLATFORM) -> None:
        """Блок в числитель ratio-критерия — ТОЛЬКО отказ площадки (#3288).

        `kind` — диагноз того же исключения, что уходит в counters.ban_kinds
        (ban_kind_of_exception). Всё, что не 'platform' (в первую очередь
        BAN_KIND_INFRA: отказ нашего сайдкара, пустой пул), — отказ НАШЕЙ стороны:
        площадка его не показывала, и рвать по нему прогон значит объявить бан там,
        где площадка молчала. Прогон 5425 оборвался по доле на 41 infra из 48
        «блоков» при 41 успешно обогащённой карточке.

        Такой отказ идёт ровно туда же, куда record_failure(): в ЗНАМЕНАТЕЛЬ окна
        (доля платформенных блоков от него честно падает), мимо серии и safety-net.
        Отдельно важно, что серию он не двигает: иначе safety-net на снапшоте
        короче окна рвал бы прогон по тем же infra-отказам, только другим
        критерием. Дефолт 'platform' сохраняет поведение вызывающих, которые вид
        не считают.
        """
        if kind != BAN_KIND_PLATFORM:
            self.record_failure()
            return
        self._consecutive_blocks += 1
        self._window.append(True)

    def record_success(self) -> None:
        """Единственный исход, снимающий safety-net (#3184: пачка блоков ПОСЛЕ хотя
        бы одного успеха -- уже не "чистый с рождения прогона" burst)."""
        self._flush_streak()
        self._pure_block_run = False
        self._window.append(False)

    def record_failure(self) -> None:
        """Отказ-не-блок: в знаменатель окна идёт (тракт получил ответ, не блок),
        но серию/purity НЕ трогает -- зеркалит исторический avito-инвариант."""
        self._window.append(False)

    def record_neutral(self) -> None:
        """404/gone у avito -- не блок и не проверка тракта, не двигает ни окно, ни
        серию (тот же смысл, что "neutral to the breaker" в комментарии у
        AvitoListingGoneError). Класс написан источник-агностично на будущее (не
        только avito), но сейчас единственный вызывающий -- avito_detail_backfill
        (#3184 review MAJOR 1: применение к domclick_detail_backfill снято из этой
        задачи -- своя калибровка, свои ограничения прокси-пула)."""
        return

    def _flush_streak(self) -> None:
        if self._consecutive_blocks:
            self.streak_histogram[self._consecutive_blocks] += 1
        self._consecutive_blocks = 0

    def abort_reason(self) -> str | None:
        """Какой критерий требует обрыва прямо сейчас, или None.

        Возвращает "safety_net" / "ratio" / None. Состояние не меняет, поэтому
        вызывать можно сколько угодно раз -- в том числе повторно, ради текста лога.
        """
        # Safety-net -- ТОЛЬКО когда ratio-критерий физически недостижим (снапшот
        # короче окна), иначе пачка safety_min блоков в начале длинного прогона
        # абортила бы его так же, как до правки (#3184 review MAJOR 2).
        if (
            self.snapshot_size < self.window_size
            and self._pure_block_run
            and self._consecutive_blocks >= self.safety_min
        ):
            return "safety_net"
        if len(self._window) == self.window_size and self.window_size > 0:
            ratio = sum(self._window) / self.window_size
            if ratio >= self.ratio_threshold:
                return "ratio"
        return None

    def should_abort(self) -> bool:
        return self.abort_reason() is not None

    def abort_explanation(self) -> str:
        """Текст для лога ABORT -- ровно та величина, по которой обрыв и произошёл.

        До этого лог печатал "%d consecutive blocks" всегда, а критерий с #3184
        стал ratio: прогон 5210 (14 блоков из 20, обрыв ровно по порогу 0.7) выдал
        "ABORT -- 1 consecutive blocks", потому что в момент срабатывания текущая
        серия равнялась единице. Число верное, величина не та -- читатель лога видит
        цифру, по которой обрыва быть не могло, и идёт искать несуществующий баг.

        Пустая строка означает "рвать не по чему" -- вызывается только под
        should_abort(), так что в логи не попадает.
        """
        reason = self.abort_reason()
        if reason == "ratio":
            return (
                f"доля блоков {self.window_blocks}/{self.window_len} в окне "
                f"(порог {self.ratio_threshold * 100:.0f}%)"
            )
        if reason == "safety_net":
            return (
                f"{self._consecutive_blocks} блоков подряд без единого успеха "
                f"(снапшот {self.snapshot_size} короче окна {self.window_size})"
            )
        return ""

    def finalize(self) -> dict[str, int]:
        """Досчитать хвостовую пачку (прогон оборвался посреди серии блоков, без
        финального успеха) и отдать гистограмму для counters (jsonb) -- ключи СТРОКОЙ:
        jsonb/JSON всё равно хранит только строковые ключи, отдаём их такими сразу,
        чтобы то, что пишем ({6: 1}), совпадало с тем, что потом читаем ({"6": 1})."""
        self._flush_streak()
        return {str(streak): count for streak, count in self.streak_histogram.items()}