From c048e8f2c90644863b57e389c6afcf01e35314e8 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 5 Sep 2026 23:06:26 +0500 Subject: [PATCH] fix(avito): skip fully-covered price ranges before probing on resume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Резюм 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 --- ..._3315_avito_exhaustive_containment_skip.py | 142 ++++++++++++++++++ .../src/scraper_kit/pricing/__init__.py | 2 + .../src/scraper_kit/pricing/bisection.py | 72 ++++++++- .../src/scraper_kit/providers/avito/serp.py | 38 ++++- 4 files changed, 248 insertions(+), 6 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_3315_avito_exhaustive_containment_skip.py diff --git a/tradein-mvp/backend/tests/test_3315_avito_exhaustive_containment_skip.py b/tradein-mvp/backend/tests/test_3315_avito_exhaustive_containment_skip.py new file mode 100644 index 00000000..da9f5d6f --- /dev/null +++ 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 = ( + '
' + '' + "Квартира-студия, 25 м², 5/9 эт." + '
' +) +_DENSE_SERP = f'2000{_CARD}' + + +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, "битый ключ дорисовал несобранное покрытие" diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/__init__.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/__init__.py index 555e3d81..ad705018 100644 --- 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", ] diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/bisection.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/bisection.py index 017b89b4..7983d5c7 100644 --- 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 diff --git 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 index e22d1e94..c31e9c7a 100644 --- 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-запросы - всё равно выполняются (skip проверяется уже в листе, после probe), в - инкрементальном probe'а нет — там пропускается весь бакет целиком. + пагинация и on_bucket для них пропускаются. В exhaustive-режиме ключи + дополнительно сливаются в интервалы, и диапазон, целиком покрытый ими, + обрезается ДО 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, )