fix(avito): skip fully-covered price ranges before probing on resume

Резюм exhaustive-обхода пере-пробивал уже зачтённую территорию: skip проверялся в листе, ПОСЛЕ probe, поэтому дерево бисекции спускалось в поддиапазоны done-корзин живыми запросами (прогон 5718: 21 минута внутри room_studii:4000000:4999999, ноль новых корзин). На пуле из 1-2 нод это сжигает весь бан-бюджет до первой НОВОЙ работы. Ключи чекпоинта — границы ДИНАМИЧЕСКОЙ бисекции: при сдвиге рынка новый лист ключом не равен старому даже внутри покрытого диапазона, поэтому сравнение строк бесполезно. done_range_skipper парсит ключи room:lo:hi (hi=open → бесконечность) в отрезки, сливает пересекающиеся и смежные и отдаёт предикат покрытия; walk_price_range проверяет его на входе в узел, ДО probe, и обрезает готовые поддеревья без единого запроса. Closes #3315
2026-09-05 23:06:26 +05:00 · 2026-09-05 23:06:26 +05:00 · c048e8f2c9
commit c048e8f2c9
parent 99f112db0b
4 changed files with 248 additions and 6 deletions
--- a/tradein-mvp/backend/tests/test_3315_avito_exhaustive_containment_skip.py
+++ b/tradein-mvp/backend/tests/test_3315_avito_exhaustive_containment_skip.py
@ -0,0 +1,142 @@
 """Резюм exhaustive-обхода Авито не пробивает уже пройденную территорию (#3315).
 Прогон 5718 (01.09.2026) 21 минуту жёг бан-бюджет обеих живых нод внутри ЗАЧТЁННОЙ
 территории: `room_studii:4000000:4999999` лежит в done-леджере, а лог показывает
 probe поддиапазонов `[4062500, 4124999]`, `[4093750, 4124999]`. Причина — skip
 проверялся в ЛИСТЕ, после probe: дерево деления пере-пробивалось живыми запросами
 каждый прогон. Вторая грань: ключи чекпоинта — границы ДИНАМИЧЕСКОЙ бисекции, при
 сдвиге рынка новые листья ключом не равны старым даже внутри покрытого диапазона,
 поэтому сравнение ключей строкой на резюме бесполезно — нужны интервалы.
 Сеть не нужна: `_fetch_serp_html` подменяется счётчиком. Проверка ПО ЗНАЧЕНИЮ —
 сколько раз обход сходил в сеть; на origin/main первые три теста красные (обход
 делает N > 0 запросов по территории, которая уже в чекпоинте).
 """
 from __future__ import annotations
 import asyncio
 import os
 from types import SimpleNamespace
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 from scraper_kit.providers.avito.serp import ROOM_SLUGS, AvitoScraper
 _STUDII = ROOM_SLUGS[0]  # ("студии", "studii-…")
 _LABEL = "room_studii"
 # Плотная выдача: total=2000 > cap=1400 → непокрытый диапазон обязан делиться дальше.
 # Карточка обязательна: total>0 при 0 распознанных карточек = DOM-drift guard (#audit-scrapers).
 _CARD = (
    '<div data-marker="item" data-item-id="1">'
    '<a data-marker="item-title" href="/ekaterinburg/kvartiry/studiya_25_m_5_9et_1">'
    "Квартира-студия, 25 м², 5/9 эт.</a>"
    '<meta itemprop="price" content="4100000"></div>'
 )
 _DENSE_SERP = f'<html><span data-marker="page-title/count">2000</span>{_CARD}</html>'
 def _walk(lo: int, hi: int | None, skip_buckets: set[str] | None) -> int:
    """Прогнать бисекцию полосы [lo, hi] и вернуть ЧИСЛО сетевых запросов."""
    s = AvitoScraper(SimpleNamespace(avito_serp_ekb_only=True, scraper_fetch_mode="http"))  # type: ignore[arg-type]
    calls = 0
    async def fake_fetch(url: str, page: int) -> str:
        nonlocal calls
        calls += 1
        return _DENSE_SERP
    s._fetch_serp_html = fake_fetch  # type: ignore[method-assign]
    asyncio.run(
        s._walk_price_range(
            room_slug=_STUDII[1],
            room_label=_LABEL,
            lo=lo,
            hi=hi,
            seen={},
            price_cap_per_bucket=1400,
            max_pages_per_bucket=1,
            skip_buckets=skip_buckets,
        )
    )
    return calls
 def test_subrange_of_done_bucket_costs_zero_requests() -> None:
    """Головной: полоса ВНУТРИ done-корзины не делает ни одного запроса.
    Ровно случай из лога 5718: ключ `room_studii:4000000:4999999` зачтён, дерево
    спустилось в `[4062500, 4124999]`.
    """
    calls = _walk(4_062_500, 4_124_999, {f"{_LABEL}:4000000:4999999"})
    assert calls == 0, (
        f"полоса [4062500, 4124999] целиком внутри зачтённой [4000000, 4999999], "
        f"а обход сходил в сеть {calls} раз(а) — бан-бюджет горит на готовой территории"
    )
 def test_partially_covered_range_still_probes() -> None:
    """Контроль честности skip'а: непокрытый остаток обязан пробиваться.
    `[3500000, 4200000]` покрыта done-корзиной только сверху — нижние 500k никто
    не собирал, пропустить их = молча потерять инвентарь.
    """
    calls = _walk(3_500_000, 4_200_000, {f"{_LABEL}:4000000:4999999"})
    assert calls > 0, "частично покрытая полоса пропущена целиком — потеря инвентаря"
 def test_adjacent_done_keys_merge_across_the_seam() -> None:
    """Смежные корзины [4М,5М) + [5М,6М) покрывают полосу через их стык.
    Бисекция делит по `mid`, и половины ложатся встык (hi=mid, lo=mid+1): без
    слияния смежных отрезков любая полоса, пересекающая шов, пробивалась бы заново.
    """
    ledger = {f"{_LABEL}:4000000:4999999", f"{_LABEL}:5000000:5999999"}
    calls = _walk(4_500_000, 5_500_000, ledger)
    assert calls == 0, (
        f"полоса [4.5М, 5.5М] лежит в объединении двух смежных done-корзин, "
        f"а обход сходил в сеть {calls} раз(а) — стык не сшит"
    )
 def test_empty_ledger_keeps_previous_behaviour() -> None:
    """Регресс-контроль: без леджера обход прежний — пробивает и делит.
    Первый прогон (чекпоинта ещё нет) не должен ничего пропускать.
    """
    baseline = _walk(4_000_000, 4_999_999, None)
    assert baseline > 1, f"обход без леджера деградировал: {baseline} запрос(ов)"
    assert _walk(4_000_000, 4_999_999, set()) == baseline
    # Чужая комнатность в леджере не покрывает нашу.
    assert _walk(4_000_000, 4_999_999, {"room_1_komn:0:open"}) == baseline
 def test_fully_done_room_resumes_with_zero_requests() -> None:
    """Приёмка issue: резюм по полностью готовой комнатности = НОЛЬ запросов.
    Открытый верхний брекет в леджере записан как `label:lo:open` — это [lo, ∞).
    """
    s = AvitoScraper(SimpleNamespace(avito_serp_ekb_only=True, scraper_fetch_mode="http"))  # type: ignore[arg-type]
    calls = 0
    async def fake_fetch(url: str, page: int) -> str:
        nonlocal calls
        calls += 1
        return _DENSE_SERP
    s._fetch_serp_html = fake_fetch  # type: ignore[method-assign]
    asyncio.run(
        s.fetch_all_secondary(
            rooms_buckets=[_STUDII],
            max_pages_per_bucket=1,
            skip_buckets={f"{_LABEL}:0:open"},
        )
    )
    assert calls == 0, f"резюм готовой комнатности сделал {calls} запрос(ов) вместо нуля"
 def test_junk_key_in_ledger_neither_crashes_nor_covers() -> None:
    """Битый ключ рядом с валидным: обход не падает, покрытие даёт только валидный."""
    ledger = {f"{_LABEL}:мусор:5", f"{_LABEL}:1000000:2000000"}
    assert _walk(1_200_000, 1_300_000, ledger) == 0, "валидный ключ перестал покрывать"
    assert _walk(999_999, 1_300_000, ledger) > 0, "битый ключ дорисовал несобранное покрытие"
--- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/init.py
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/init.py
@ -10,6 +10,7 @@ from scraper_kit.pricing.bisection import (
    BisectionConfig,
    ProbeFailPolicy,
    ProbeResult,
    done_range_skipper,
    walk_price_range,
 )
@ -17,5 +18,6 @@ __all__ = [
    "BisectionConfig",
    "ProbeFailPolicy",
    "ProbeResult",
    "done_range_skipper",
    "walk_price_range",
 ]
--- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/bisection.py
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/bisection.py
@ -47,7 +47,7 @@ provider-callback `probe` + `ProbeResult.payload`, движку прозрачн
 from __future__ import annotations
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Iterable
 from dataclasses import dataclass
 from enum import Enum, auto
 from typing import Any
@ -124,6 +124,62 @@ class BisectionConfig:
 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]]
 SkipFn = Callable[[int | None, int | None], bool]
 def done_range_skipper(bucket_keys: Iterable[str] | None, label: str) -> SkipFn | None:
    """Предикат «диапазон уже пройден прошлым прогоном» из done-леджера чекпоинта.
    Ключи чекпоинта — «label:lo:hi» (hi=«open» для верхнего брекета без потолка),
    т.е. границы ДИНАМИЧЕСКОЙ бисекции: рынок сдвинулся → totals другие → дерево
    делится иначе, и новый лист (`4062500:4124999`) ключом не равен старому
    (`4000000:4999999`) даже внутри уже пройденной территории (#3315). Поэтому
    сравнение ключей строкой на резюме бесполезно — сравниваем ИНТЕРВАЛЫ:
    ключи парсятся в отрезки, пересекающиеся/смежные сливаются, и диапазон
    пропускается, если целиком лежит внутри объединения.
    Возвращает ``None``, если по этому label в леджере нет ни одного валидного
    ключа (вызывающий тогда не ставит гейт вовсе — поведение прежнее).
    """
    if not bucket_keys:
        return None
    prefix = f"{label}:"
    parsed: list[tuple[int, float]] = []
    for key in bucket_keys:
        if not key.startswith(prefix):
            continue
        lo_raw, _, hi_raw = key[len(prefix) :].partition(":")
        try:
            lo = int(lo_raw)
            hi = float("inf") if hi_raw == "open" else float(int(hi_raw))
        except ValueError:
            # Чужой/битый ключ в леджере не должен ронять обход — просто не покрывает.
            continue
        if hi >= lo:
            parsed.append((lo, hi))
    if not parsed:
        return None
    merged: list[tuple[int, float]] = []
    for lo, hi in sorted(parsed):
        if merged and lo <= merged[-1][1] + 1:  # пересечение ИЛИ смежность ([4М,5М)+[5М,6М))
            prev_lo, prev_hi = merged[-1]
            merged[-1] = (prev_lo, max(prev_hi, hi))
        else:
            merged.append((lo, hi))
    def _covered(lo: int | None, hi: int | None) -> bool:
        # None-границы: lo=None ≡ 0 (провайдеры так и строят bucket_key), hi=None ≡ ∞.
        q_lo = lo if lo is not None else 0
        q_hi: float = hi if hi is not None else float("inf")
        for r_lo, r_hi in merged:
            if r_lo > q_lo:
                break  # отрезки непересекающиеся и несмежные — дальше только правее
            if q_hi <= r_hi:
                return True
        return False
    return _covered
 def _can_split_span(span: int, config: BisectionConfig) -> bool:
@ -141,11 +197,13 @@ async def walk_price_range(
    probe: ProbeFn,
    on_leaf: LeafFn,
    on_degraded: DegradedFn | None = None,
    should_skip: SkipFn | None = None,
    depth: int = 0,
 ) -> None:
    """Рекурсивно обойти ценовой диапазон [lo, hi] адаптивной бисекцией.
    Скелет (общий для всех провайдеров):
      0. ``should_skip(lo, hi)`` → return БЕЗ единого запроса (см. ниже).
      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``).
@ -162,8 +220,16 @@ async def walk_price_range(
            (fetch/parse/дедуп/on_bucket целиком на стороне провайдера).
        on_degraded: async ``(lo, hi) -> None`` — обязателен для
            ``ProbeFailPolicy.DEGRADE`` (yandex «пагинировать-до-пустоты»).
        should_skip: sync ``(lo, hi) -> bool`` — гейт ПЕРЕД probe: True = поддерево
            целиком пройдено прошлым прогоном, обрезаем его без запросов
            (:func:`done_range_skipper`, #3315). Проверяется на КАЖДОМ узле
            рекурсии: у частично покрытого диапазона probe остаётся, а его уже
            готовые половины отсекаются на входе, а не в листе после probe.
        depth: текущая глубина рекурсии (для ``max_depth``).
    """
    if should_skip is not None and should_skip(lo, hi):
        return
    result = await probe(lo, hi, depth)
    total = result.count
@ -191,6 +257,7 @@ async def walk_price_range(
                probe=probe,
                on_leaf=on_leaf,
                on_degraded=on_degraded,
                should_skip=should_skip,
                depth=depth + 1,
            )
            await walk_price_range(
@ -200,6 +267,7 @@ async def walk_price_range(
                probe=probe,
                on_leaf=on_leaf,
                on_degraded=on_degraded,
                should_skip=should_skip,
                depth=depth + 1,
            )
        return
@ -236,6 +304,7 @@ async def walk_price_range(
            probe=probe,
            on_leaf=on_leaf,
            on_degraded=on_degraded,
            should_skip=should_skip,
            depth=depth + 1,
        )
        await walk_price_range(
@ -245,6 +314,7 @@ async def walk_price_range(
            probe=probe,
            on_leaf=on_leaf,
            on_degraded=on_degraded,
            should_skip=should_skip,
            depth=depth + 1,
        )
        return
--- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/serp.py
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/serp.py
@ -48,7 +48,13 @@ from scraper_kit.avito_exceptions import (
 from scraper_kit.base import BaseScraper, ScrapedLot
 from scraper_kit.browser_fetcher import BrowserFetcher, SidecarBanPageError
 from scraper_kit.price_brackets import get_price_seed_brackets
-from scraper_kit.pricing import BisectionConfig, ProbeFailPolicy, ProbeResult, walk_price_range
+from scraper_kit.pricing import (
    BisectionConfig,
    ProbeFailPolicy,
    ProbeResult,
    done_range_skipper,
    walk_price_range,
 )
 from scraper_kit.providers._base import build_browser_fetcher, build_document_session
 from scraper_kit.providers.avito.shared import _unix_to_date
@ -1197,9 +1203,11 @@ class AvitoScraper(BaseScraper):
            leaf-бакета. Может быть async или sync. Исключение прерывает прогон.
          on_progress: опциональный callback(unique_count) для heartbeat (per room-bucket).
          skip_buckets: множество ключей «room_label:lo:hi» уже завершённых бакетов —
-            пагинация и on_bucket для них пропускаются. В exhaustive-режиме probe-запросы
+            пагинация и on_bucket для них пропускаются. В exhaustive-режиме ключи
-            всё равно выполняются (skip проверяется уже в листе, после probe), в
+            дополнительно сливаются в интервалы, и диапазон, целиком покрытый ими,
-            инкрементальном probe'а нет — там пропускается весь бакет целиком.
+            обрезается ДО probe (#3315) — резюм по готовой комнатности не делает ни
            одного запроса; probe остаётся только там, где есть непокрытый остаток.
            В инкрементальном probe'а нет — там пропускается весь бакет целиком.
          since: если None (default) — EXHAUSTIVE bisection-обход (поведение без
            изменений). Если задана date — INCREMENTAL: на каждый (комнатность ×
            seed-брекет) последовательная пагинация newest-first с ранней остановкой,
@ -1372,7 +1380,9 @@ class AvitoScraper(BaseScraper):
        После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots).
        bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый).
        on_bucket может быть async или sync. Исключение в on_bucket прерывает прогон.
-        skip_buckets: если bucket_key в skip_buckets — пагинация и on_bucket пропускаются.
+        skip_buckets: ключи завершённых бакетов. Диапазон, целиком покрытый ОБЪЕДИНЕНИЕМ
          done-интервалов этой комнатности, обрезается на входе — без probe (#3315,
          `done_range_skipper`); точное совпадение ключа дополнительно скипает лист.
        secondary_only: новостройки (listing_segment=="novostroyki") отбрасываются после
          сбора bucket_lots, до дедупа в seen и вызова on_bucket.
        AvitoBlockedError/AvitoRateLimitedError из _fetch_serp_html пробрасываются наверх.
@ -1471,12 +1481,30 @@ class AvitoScraper(BaseScraper):
                expected_total=total,
            )
        # #3315: гейт ПЕРЕД probe. Раньше skip проверялся в листе, ПОСЛЕ probe, и резюм
        # пере-пробивал уже пройденную территорию живыми запросами (ключи чекпоинта —
        # границы динамической бисекции, при сдвиге рынка новые листья старым не равны).
        # На пуле из 1-2 нод бан-бюджет сгорал на пробах готовых полос до первой НОВОЙ.
        _covered = done_range_skipper(skip_buckets, room_label)
        def _skip_done(plo: int | None, phi: int | None) -> bool:
            if _covered is None or not _covered(plo, phi):
                return False
            logger.info(
                "avito: skip probe %s [%s, %s] — range already covered by done buckets (resume)",
                room_label,
                plo if plo is not None else 0,
                "open" if phi is None else phi,
            )
            return True
        await walk_price_range(
            lo=lo,
            hi=hi,
            config=_avito_bisection_config(price_cap_per_bucket),
            probe=_probe,
            on_leaf=_leaf,
            should_skip=_skip_done,
            depth=_depth,
        )