fix(cian,yandex): skip fully-covered price ranges before probing on resume

Гейт should_skip живёт в общем движке (walk_price_range, #3315), но предикат из done-леджера строил и передавал только avito. У cian и yandex та же бисекция и тот же чекпоинт — на резюме дерево деления спускалось ВНУТРЬ зачтённых полос живыми probe-запросами: ключи чекпоинта суть границы ДИНАМИЧЕСКОЙ бисекции, при сдвиге рынка новый лист старому не равен даже внутри собранной территории, поэтому сравнение строк ничего не ловит. Формат ключей у провайдеров разный, и ключи не трогаем (иначе протухнут живые чекпоинты): cian пишет room_label:lo:hi / :open — как avito, парсер подходит без изменений; yandex пишет _combo_label «rooms:lo-hi» с «None» вместо открытого потолка, поэтому в done_range_skipper параметризованы range_sep и open_token (дефолты = прежнее поведение avito/cian). Один леджер на два режима: у yandex incremental-ключи несут префикс сегмента (secondary/2:…) и отсев по label их не пропускает, а смешение режимов блокирует _pick_resume (params IS NOT DISTINCT FROM); у cian incremental-режима нет вовсе. Записано в докстринге предиката. Тесты по значению на обоих провайдерах: счётчик стоит на горлышке фетча (_fetch_page_html / _fetch_page_json), резюм готовой комнатности = 0 запросов, частично покрытая полоса по-прежнему пробивается. Closes #3359
2026-09-05 23:56:46 +05:00 · 2026-09-05 23:56:46 +05:00 · 35c7ea8492
commit 35c7ea8492
parent 63dbc209b2
5 changed files with 325 additions and 11 deletions
--- a/tradein-mvp/backend/tests/test_3359_cian_exhaustive_containment_skip.py
+++ b/tradein-mvp/backend/tests/test_3359_cian_exhaustive_containment_skip.py
@ -0,0 +1,119 @@
+"""Резюм exhaustive-обхода Cian не пробивает уже пройденную территорию (#3359).
+
+Гейт `should_skip` живёт в общем движке (`pricing.walk_price_range`, #3315), но
+предикат из done-леджера строил и передавал только avito: у cian та же бисекция и
+тот же чекпоинт, поэтому на резюме дерево деления спускалось ВНУТРЬ зачтённых
+полос живыми probe-запросами. Ключи чекпоинта — границы ДИНАМИЧЕСКОЙ бисекции:
+рынок сдвинулся → totals другие → новый лист (`4062500:4124999`) ключом не равен
+старому (`4000000:4999999`) даже внутри собранной территории, так что сравнение
+строк на резюме бесполезно — нужны интервалы.
+
+Формат ключей cian — `room_label:lo:hi` / `room_label:lo:open`, совпадает с avito
+(`_paginate_leaf_bucket`), парсер по умолчанию подходит без адаптации.
+
+Сеть не нужна: горлышко фетча `_fetch_page_html` подменяется счётчиком. Проверка
+ПО ЗНАЧЕНИЮ — сколько раз обход сходил в сеть; на origin/main тесты покрытия
+красные (обход делает N > 0 запросов по территории, которая уже в чекпоинте).
+Парсинг HTML тут не при чём (он не под тестом) — счётчик стоит на ЕДИНСТВЕННОМ
+сетевом вызове, а разбор страницы заглушен.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from scraper_kit.base import ScrapedLot
+from scraper_kit.providers.cian.serp import CianScraper
+
+from app.services.scraper_adapters import RealScraperConfig
+
+_ROOMS = (1,)
+_LABEL = "room1"
+# Плотная выдача: total=2000 > cap=1400 → непокрытый диапазон обязан делиться дальше.
+_DENSE_TOTAL = 2000
+
+
+def _scraper() -> tuple[CianScraper, list[int]]:
+    """Скрапер с заглушенной сетью; в списке — счётчик реальных фетчей."""
+    s = CianScraper(RealScraperConfig())
+    calls = [0]
+
+    async def fake_fetch(
+        rooms: tuple[int, ...] | None,
+        page: int,
+        min_price: int | None,
+        max_price: int | None,
+    ) -> str:
+        calls[0] += 1
+        return "<html>dense</html>"
+
+    async def no_sleep() -> None:
+        return None
+
+    s._fetch_page_html = fake_fetch  # type: ignore[method-assign]
+    s.sleep_between_requests = no_sleep  # type: ignore[method-assign]
+    s.request_delay_sec = 0.0
+    s._extract_total_offers = lambda html: _DENSE_TOTAL  # type: ignore[method-assign]
+    s._parse_serp_html = lambda html: []  # type: ignore[method-assign,return-value]
+    return s, calls
+
+
+def _walk(lo: int, hi: int | None, skip_buckets: set[str] | None) -> int:
+    """Прогнать бисекцию полосы [lo, hi] и вернуть ЧИСЛО сетевых запросов."""
+    s, calls = _scraper()
+    seen: dict[str, ScrapedLot] = {}
+    asyncio.run(
+        s._walk_price_range(
+            rooms=_ROOMS,
+            lo=lo,
+            hi=hi,
+            seen=seen,
+            price_cap_per_bucket=1400,
+            max_pages_per_bucket=1,
+            skip_buckets=skip_buckets,
+        )
+    )
+    return calls[0]
+
+
+def test_subrange_of_done_bucket_costs_zero_requests() -> None:
+    """Головной: полоса ВНУТРИ done-корзины не делает ни одного запроса."""
+    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'а: непокрытый остаток обязан пробиваться."""
+    calls = _walk(3_500_000, 4_200_000, {f"{_LABEL}:4000000:4999999"})
+    assert calls > 0, "частично покрытая полоса пропущена целиком — потеря инвентаря"
+
+
+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, {"room2:0:open"}) == baseline
+
+
+def test_fully_done_room_resumes_with_zero_requests() -> None:
+    """Приёмка issue: резюм по полностью готовой комнатности = НОЛЬ запросов.
+
+    Открытый верхний брекет в леджере записан как `label:lo:open` — это [lo, ∞).
+    """
+    s, calls = _scraper()
+    asyncio.run(
+        s.fetch_all_secondary(
+            rooms_buckets=[_ROOMS],
+            max_pages_per_bucket=1,
+            skip_buckets={f"{_LABEL}:0:open"},
+        )
+    )
+    assert calls[0] == 0, f"резюм готовой комнатности сделал {calls[0]} запрос(ов) вместо нуля"
--- a/tradein-mvp/backend/tests/test_3359_yandex_exhaustive_containment_skip.py
+++ b/tradein-mvp/backend/tests/test_3359_yandex_exhaustive_containment_skip.py
@ -0,0 +1,123 @@
+"""Резюм exhaustive-обхода Яндекса не пробивает уже пройденную территорию (#3359).
+
+То же, что #3315 у avito: гейт `should_skip` есть в общем движке, но предикат из
+done-леджера строил только avito — у yandex дерево бисекции на резюме спускалось
+внутрь зачтённых полос живыми probe-запросами к gate-API.
+
+Формат ключей У YANDEX ДРУГОЙ: `_combo_label` даёт «rooms:lo-hi» (разделитель
+границ «-», а не «:») и «rooms:lo-None» для открытого верхнего брекета («None»
+вместо «open»). Ключи НЕ подгоняются под avito (это протухило бы живые
+чекпоинты) — параметризован парсер (`range_sep`/`open_token`).
+
+Сеть не нужна: горлышко фетча `_fetch_page_json` подменяется счётчиком; probe
+читает totalItems из подсунутого gate-payload'а. Проверка ПО ЗНАЧЕНИЮ — сколько
+раз обход сходил в сеть.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from typing import Any
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from scraper_kit.base import ScrapedLot
+from scraper_kit.providers.yandex.serp import YandexRealtyScraper
+
+from app.services.scraper_adapters import RealScraperConfig
+
+_ROOMS = "2"
+# Плотная выдача: totalItems=2000 > cap=500 → непокрытый диапазон обязан делиться.
+_DENSE_PAGE: dict[str, Any] = {
+    "response": {
+        "search": {
+            "offers": {
+                "entities": [],
+                "pager": {"totalItems": 2000, "totalPages": 100, "page": 0},
+            }
+        }
+    }
+}
+
+
+def _scraper() -> tuple[YandexRealtyScraper, list[int]]:
+    """Скрапер с заглушенной сетью; в списке — счётчик реальных фетчей."""
+    s = YandexRealtyScraper(RealScraperConfig())
+    calls = [0]
+
+    async def fake_fetch(
+        rooms: str | None,
+        page: int,
+        price_min: int | None,
+        price_max: int | None,
+        new_flat: str = "NO",
+    ) -> dict[str, Any]:
+        calls[0] += 1
+        return _DENSE_PAGE
+
+    s._fetch_page_json = fake_fetch  # type: ignore[method-assign]
+    s.request_delay_sec = 0.0
+    return s, calls
+
+
+def _walk(lo: int, hi: int | None, skip_buckets: set[str] | None) -> int:
+    """Прогнать бисекцию полосы [lo, hi] и вернуть ЧИСЛО сетевых запросов."""
+    s, calls = _scraper()
+    seen: dict[str, ScrapedLot] = {}
+    asyncio.run(
+        s._walk_price_range(
+            rooms=_ROOMS,
+            lo=lo,
+            hi=hi,
+            seen=seen,
+            price_cap_per_bucket=500,
+            max_pages_per_bucket=1,
+            skip_buckets=skip_buckets,
+        )
+    )
+    return calls[0]
+
+
+def test_subrange_of_done_bucket_costs_zero_requests() -> None:
+    """Головной: полоса ВНУТРИ done-корзины не делает ни одного запроса."""
+    calls = _walk(4_062_500, 4_124_999, {f"{_ROOMS}:4000000-4999999"})
+    assert calls == 0, (
+        f"полоса [4062500, 4124999] целиком внутри зачтённой [4000000, 4999999], "
+        f"а обход сходил в сеть {calls} раз(а) — бан-бюджет горит на готовой территории"
+    )
+
+
+def test_partially_covered_range_still_probes() -> None:
+    """Контроль честности skip'а: непокрытый остаток обязан пробиваться."""
+    calls = _walk(3_500_000, 4_200_000, {f"{_ROOMS}:4000000-4999999"})
+    assert calls > 0, "частично покрытая полоса пропущена целиком — потеря инвентаря"
+
+
+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, {"3:0-None"}) == baseline
+    # Ключ incremental-режима (gate-combo с префиксом сегмента) покрытия не даёт:
+    # он не проходит отсев по префиксу label, а смешение режимов дополнительно
+    # блокирует _pick_resume (params IS NOT DISTINCT FROM).
+    assert _walk(4_000_000, 4_999_999, {f"secondary/{_ROOMS}:0-None"}) == baseline
+
+
+def test_fully_done_room_resumes_with_zero_requests() -> None:
+    """Приёмка issue: резюм по полностью готовой комнатности = НОЛЬ запросов.
+
+    Открытый верхний брекет в леджере yandex'а записан как `rooms:lo-None`.
+    """
+    s, calls = _scraper()
+    asyncio.run(
+        s.fetch_all_secondary(
+            rooms_buckets=[_ROOMS],
+            max_pages_per_bucket=1,
+            skip_buckets={f"{_ROOMS}:0-None"},
+        )
+    )
+    assert calls[0] == 0, f"резюм готовой комнатности сделал {calls[0]} запрос(ов) вместо нуля"
--- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/bisection.py
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/bisection.py
@ -127,16 +127,40 @@ 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:
+def done_range_skipper(
+    bucket_keys: Iterable[str] | None,
+    label: str,
+    *,
+    range_sep: str = ":",
+    open_token: str = "open",
+) -> SkipFn | None:
    """Предикат «диапазон уже пройден прошлым прогоном» из done-леджера чекпоинта.

-    Ключи чекпоинта — «label:lo:hi» (hi=«open» для верхнего брекета без потолка),
-    т.е. границы ДИНАМИЧЕСКОЙ бисекции: рынок сдвинулся → totals другие → дерево
-    делится иначе, и новый лист (`4062500:4124999`) ключом не равен старому
-    (`4000000:4999999`) даже внутри уже пройденной территории (#3315). Поэтому
-    сравнение ключей строкой на резюме бесполезно — сравниваем ИНТЕРВАЛЫ:
+    Ключи чекпоинта — «label<sep>lo<sep>hi» (hi=``open_token`` для верхнего брекета
+    без потолка), т.е. границы ДИНАМИЧЕСКОЙ бисекции: рынок сдвинулся → totals
+    другие → дерево делится иначе, и новый лист (`4062500:4124999`) ключом не равен
+    старому (`4000000:4999999`) даже внутри уже пройденной территории (#3315).
+    Поэтому сравнение ключей строкой на резюме бесполезно — сравниваем ИНТЕРВАЛЫ:
    ключи парсятся в отрезки, пересекающиеся/смежные сливаются, и диапазон
-    пропускается, если целиком лежит внутри объединения.
+    пропускается, если целиком лежит внутри объединения. ``hi`` во всех форматах
+    ВКЛЮЧИТЕЛЬНА (движок делит на [lo, mid] + [mid+1, hi]), поэтому смежность
+    сливается по `lo <= prev_hi + 1`.
+
+    Форматы ключей у провайдеров (парсер параметризован, ключи НЕ трогаем —
+    иначе протухнут живые чекпоинты, #3359):
+      * avito / cian — ``room_studii:4000000:4999999`` / ``…:0:open``
+        (дефолты: ``range_sep=":"``, ``open_token="open"``);
+      * yandex — ``_combo_label``: ``2:4000000-4999999`` / ``2:20000000-None``
+        (``range_sep="-"``, ``open_token="None"``; label = ``rooms or 'any'``).
+
+    Один леджер на два режима (incremental/exhaustive): предикат сливает ключи в
+    ИНТЕРВАЛЫ, поэтому incremental-ключ «целый seed-брекет» покрыл бы в exhaustive
+    всю комнатность разом. Смешение блокируется выше — ``_pick_resume`` подхватывает
+    чекпоинт только при ``params IS NOT DISTINCT FROM`` (иначе
+    ``resume_reason="params_changed"``), а режимы идут с разными params. У yandex
+    есть и вторая, независимая преграда: gate-ключи incremental'а имеют префикс
+    сегмента (``secondary/2:…``) и отсев по ``startswith(f"{label}:")`` их не
+    пропускает. У cian incremental-режима нет вовсе — один exhaustive.

    Возвращает ``None``, если по этому label в леджере нет ни одного валидного
    ключа (вызывающий тогда не ставит гейт вовсе — поведение прежнее).
@ -148,10 +172,10 @@ def done_range_skipper(bucket_keys: Iterable[str] | None, label: str) -> SkipFn
    for key in bucket_keys:
        if not key.startswith(prefix):
            continue
-        lo_raw, _, hi_raw = key[len(prefix) :].partition(":")
+        lo_raw, _, hi_raw = key[len(prefix) :].partition(range_sep)
        try:
            lo = int(lo_raw)
-            hi = float("inf") if hi_raw == "open" else float(int(hi_raw))
+            hi = float("inf") if hi_raw == open_token else float(int(hi_raw))
        except ValueError:
            # Чужой/битый ключ в леджере не должен ронять обход — просто не покрывает.
            continue
--- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/serp.py
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/serp.py
@ -38,7 +38,13 @@ from scraper_kit.base import BaseScraper, ScrapedLot
 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.pricing import (
+    BisectionConfig,
+    ProbeFailPolicy,
+    ProbeResult,
+    done_range_skipper,
+    walk_price_range,
+)
 from scraper_kit.providers._base import build_browser_fetcher
 from scraper_kit.repair_state_normalizer import (
    infer_repair_state_from_text,
@ -563,12 +569,29 @@ class CianScraper(BaseScraper):
                skip_buckets=skip_buckets,
            )

+        # #3359: гейт ПЕРЕД probe (как у avito, #3315). Без него резюм спускался
+        # бисекцией внутрь уже зачтённых полос живыми probe-запросами: ключи
+        # чекпоинта — границы ДИНАМИЧЕСКОЙ бисекции, новый лист старому не равен.
+        _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(
+                "cian: 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=_cian_bisection_config(price_cap_per_bucket),
            probe=_probe,
            on_leaf=_leaf,
+            should_skip=_skip_done,
            depth=_depth,
        )

--- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py
@ -52,7 +52,13 @@ from scraper_kit.browser_fetcher import BrowserFetcher
 from scraper_kit.ceiling_height import plausible_ceiling_m
 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.pricing import (
+    BisectionConfig,
+    ProbeFailPolicy,
+    ProbeResult,
+    done_range_skipper,
+    walk_price_range,
+)
 from scraper_kit.providers._base import build_browser_fetcher

 if TYPE_CHECKING:
@ -1249,6 +1255,24 @@ class YandexRealtyScraper(BaseScraper):
            if on_bucket is not None:
                on_bucket(bucket_key, len(seen))

+        # #3359: гейт ПЕРЕД probe (как у avito, #3315). Ключи yandex'а — `_combo_label`,
+        # т.е. «rooms:lo-hi» с «None» вместо открытого потолка: другой разделитель и
+        # другой open-токен, чем у avito/cian, поэтому парсер параметризуется, а ключи
+        # остаются как есть (живые чекпоинты не ломаем).
+        _covered = done_range_skipper(skip_buckets, rooms or "any", range_sep="-", open_token="None")
+
+        def _skip_done(plo: int | None, phi: int | None) -> bool:
+            if _covered is None or not _covered(plo, phi):
+                return False
+            logger.info(
+                "yandex gate: skip probe rooms=%s [%s, %s] — range already covered "
+                "by done buckets (resume)",
+                rooms,
+                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,
@ -1262,6 +1286,7 @@ class YandexRealtyScraper(BaseScraper):
            probe=_probe,
            on_leaf=_leaf,
            on_degraded=_degraded,
+            should_skip=_skip_done,
            depth=_depth,
        )