diff --git a/tradein-mvp/backend/tests/test_3368_yandex_leaf_cap_complete.py b/tradein-mvp/backend/tests/test_3368_yandex_leaf_cap_complete.py new file mode 100644 index 00000000..5c1bdfdf --- /dev/null +++ b/tradein-mvp/backend/tests/test_3368_yandex_leaf_cap_complete.py @@ -0,0 +1,135 @@ +"""Leaf-бакет яндекса, обрезанный потолком страниц, не идёт в чекпоинт (#3368). + +Довесок к #3362: там честной сделали degraded-ветку (`on_bucket(..., complete=False)`), +а `_leaf` продолжал писать бакет как полный, даже когда его пагинация упиралась в +`max_pages_per_bucket`/`_GATE_MAX_PAGES_CAP`. С containment-гейтом (#3358/#3359) такой +ключ покрывает СВОЙ интервал целиком → резюм больше не заходит в полосу, чей хвост не +читали ни разу. У cian тот же случай считается честно (`complete = pages_needed <= +max_pages`, cian/serp.py). + +Переполненный leaf возможен только там, где бисекции дробить нечем: размах меньше +`_YANDEX_SPLIT_MIN_BRACKET`, открытый верхний брекет или потолок глубины. Здесь берётся +первый случай: плотная выдача (totalItems=2000 > cap=500) делится до размаха 499 999 и +дальше делиться не может. + +Сеть не нужна: `_fetch_page_json` подменяется счётчиком. Проверки ПО ЗНАЧЕНИЮ — что +попало в done-леджер и сколько запросов сделал резюм. +""" + +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" + + +def _page(total_items: int) -> dict[str, Any]: + return { + "response": { + "search": { + "offers": { + "entities": [], + "pager": { + "totalItems": total_items, + "totalPages": max(1, total_items // 20), + "page": 0, + }, + } + } + } + } + + +def _walk( + total_items: int, + max_pages_per_bucket: int, + skip_buckets: set[str] | None = None, +) -> tuple[list[tuple[str, bool]], int, int]: + """Прогнать бисекцию [4M, 5M) и вернуть (отметки бакетов, запросы, capped_buckets). + + Колбэк — той же формы, что pipeline._on_bucket: третий позиционный аргумент = + признак полноты; в done-леджер `_mark_bucket` кладёт только complete=True. + """ + s = YandexRealtyScraper(RealScraperConfig()) + s.request_delay_sec = 0.0 + 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 _page(total_items) + + s._fetch_page_json = fake_fetch # type: ignore[method-assign] + marked: list[tuple[str, bool]] = [] + + def on_bucket(key: str, _count: int, complete: bool = True) -> None: + marked.append((key, complete)) + + seen: dict[str, ScrapedLot] = {} + asyncio.run( + s._walk_price_range( + rooms=_ROOMS, + lo=4_000_000, + hi=4_999_999, + seen=seen, + price_cap_per_bucket=500, + max_pages_per_bucket=max_pages_per_bucket, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + ) + ) + # getattr, а не атрибут напрямую: без счётчика тест обязан краснеть НЕВЕРНЫМ + # ЗНАЧЕНИЕМ (ключ в леджере / ноль запросов на резюме), а не AttributeError'ом — + # «возможности нет» неотличимо от «проверка не проведена». + return marked, calls[0], getattr(s, "capped_buckets", 0) + + +def test_capped_leaf_stays_out_of_done_ledger() -> None: + """Приёмка: leaf, которому нужно больше страниц, чем потолок, — не в леджере.""" + marked, _, capped = _walk(total_items=2000, max_pages_per_bucket=1) + assert marked, "leaf не вызвал on_bucket — тест ничего не проверяет" + ledger = {key for key, complete in marked if complete} + assert not ledger, ( + f"бакеты {sorted(ledger)} прочитаны на 1 страницу из 100 (totalItems=2000), " + "но помечены complete — их интервал зачтётся containment-гейтом целиком" + ) + assert capped == len(marked), ( + f"capped_buckets={capped} при {len(marked)} обрезанных бакетах — " + "счётчик прогона не покажет, что полоса недобрана по построению" + ) + + +def test_fully_read_leaf_goes_into_done_ledger() -> None: + """Контроль честности: дочитанный до конца бакет по-прежнему чекпоинтится.""" + marked, _, capped = _walk(total_items=15, max_pages_per_bucket=1) + ledger = {key for key, complete in marked if complete} + assert ledger, ( + "бакет из одной страницы (totalItems=15) не попал в леджер — резюм будет " + "перечитывать уже собранную территорию" + ) + assert capped == 0, f"полный бакет посчитан обрезанным (capped_buckets={capped})" + + +def test_capped_band_is_rewalked_on_resume() -> None: + """По значению: полоса с обрезанным leaf'ом на резюме снова обходится.""" + marked, _, _ = _walk(total_items=2000, max_pages_per_bucket=1) + ledger = {key for key, complete in marked if complete} + _, calls, _ = _walk(total_items=2000, max_pages_per_bucket=1, skip_buckets=ledger or None) + assert calls > 0, ( + "резюм не сделал ни одного запроса по полосе, у которой прочитана 1 страница " + "из 100 — недобранный хвост потерян навсегда" + ) diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py index 8222e969..bd0448dc 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py @@ -3807,6 +3807,11 @@ class YandexFullLoadCounters: # Бакеты, собранные ЧАСТИЧНО (probe провалился → degraded-пагинация): в # чекпоинт не пишутся, следующий прогон перечитает их целиком. partial_buckets: int = 0 + # Подмножество partial_buckets: бакет неполон ПО ПОСТРОЕНИЮ (#3368) — total + # требует больше страниц, чем потолок, а бисекции делить его уже нечем. + # Отдельный счётчик, потому что лечение другое: не «повторить прогон», а + # снизить min_bracket / поднять потолок страниц. + capped_buckets: int = 0 def to_dict(self) -> dict[str, int]: return {f.name: getattr(self, f.name) for f in fields(self)} @@ -3950,6 +3955,8 @@ async def run_yandex_full_load( on_progress=_on_progress, skip_buckets=skip_set if skip_set else None, ) + # #3368: leaf'ы, обрезанные потолком страниц (в чекпоинт не попали). + counters.capped_buckets = getattr(scraper, "capped_buckets", 0) logger.info( "yandex-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d", diff --git 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 index 98533ff3..a4786cca 100644 --- 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 @@ -531,6 +531,11 @@ class YandexRealtyScraper(BaseScraper): # блокировки — см. _track_gate_result. self.gate_fetch_attempts: int = 0 self.gate_fetch_failures: int = 0 + # #3368: leaf-бакеты, чья пагинация упёрлась в потолок страниц (дробить + # бисекции уже нечем). Не пишутся в чекпоинт → читаются вызывающим для + # counters прогона. Атрибут, а не аргумент on_bucket — тот же довод, что у + # cian.last_dropped_nb: у колбэка есть внешние реализации. + self.capped_buckets: int = 0 def _track_gate_result(self, ok: bool) -> None: """Учёт исхода одного top-level gate-API запроса (#2625). @@ -1220,13 +1225,35 @@ class YandexRealtyScraper(BaseScraper): logger.debug("yandex gate: skip bucket %s (checkpoint)", bucket_key) return - total_pages = min( - math.ceil(total / 20), - _GATE_MAX_PAGES_CAP, - max_pages_per_bucket, - ) + pages_needed = math.ceil(total / 20) + max_pages = min(_GATE_MAX_PAGES_CAP, max_pages_per_bucket) + total_pages = min(pages_needed, max_pages) + # Полнота бакета (#3368): у обрезанного потолком страниц ключ ТОТ ЖЕ, что у + # собранного целиком, — как у cian (`complete = pages_needed <= max_pages`). + # complete=False → бакет не идёт в done-леджер, иначе его интервал в + # containment-гейте (#3359) склеился бы с соседними и резюм пропустил бы + # полосу, чей хвост никогда не читали. Бакет доходит сюда переполненным + # только там, где бисекции делить больше нечем (размах < min_bracket, + # открытый верхний брекет, потолок глубины) — см. walk_price_range. + complete = pages_needed <= max_pages + if not complete: + self.capped_buckets += 1 + logger.warning( + "yandex gate: leaf bucket %s НЕПОЛОН по построению — total=%d требует " + "%d страниц при потолке %d; дробить дальше нечем, в чекпоинт не пишем " + "(capped_buckets=%d)", + bucket_key, + total, + pages_needed, + max_pages, + self.capped_buckets, + ) logger.info( - "yandex gate: leaf bucket %s total=%d pages=%d", bucket_key, total, total_pages + "yandex gate: leaf bucket %s total=%d pages=%d complete=%s", + bucket_key, + total, + total_pages, + complete, ) # Add probe lots (page 1 already fetched) @@ -1236,7 +1263,7 @@ class YandexRealtyScraper(BaseScraper): if total_pages <= 1: if on_bucket is not None: - on_bucket(bucket_key, len(seen)) + on_bucket(bucket_key, len(seen), complete) return # Paginate pages 2..total_pages with concurrency @@ -1258,7 +1285,7 @@ class YandexRealtyScraper(BaseScraper): seen[lot.source_id] = lot if on_bucket is not None: - on_bucket(bucket_key, len(seen)) + on_bucket(bucket_key, len(seen), complete) # #3359: гейт ПЕРЕД probe (как у avito, #3315). Ключи yandex'а — `_combo_label`, # т.е. «rooms:lo-hi» с «None» вместо открытого потолка: другой разделитель и