From 3f85fa59ff8a8dc8621a01dd62a3d7ae176709af Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 18 Jun 2026 17:29:31 +0300 Subject: [PATCH] =?UTF-8?q?perf(tradein):=20avito=20seed=20price-brackets?= =?UTF-8?q?=20with=20open=20top=20bucket=20(=E2=89=A42k=20blocks,=20no=20p?= =?UTF-8?q?rice=20ceiling)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../backend/app/services/scrapers/avito.py | 244 +++++++++++--- .../tests/scrapers/test_avito_exhaustive.py | 315 ++++++++++++++++-- 2 files changed, 480 insertions(+), 79 deletions(-) diff --git a/tradein-mvp/backend/app/services/scrapers/avito.py b/tradein-mvp/backend/app/services/scrapers/avito.py index 72460f2b..fed43837 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito.py +++ b/tradein-mvp/backend/app/services/scrapers/avito.py @@ -229,10 +229,32 @@ NOVOSTROYKA_SLUG = "novostroyka-ASgBAgICAkSSA8YQ5geOUg" # ── Exhaustive full-load (room×price bisection) ─────────────────────────────── # Верхняя граница цены при первом рекурсивном делении (нет явного hi). # 200 млн ₽ заведомо выше ТОПа ЕКБ — дальние бакеты дадут 0 результатов. +# Больше НЕ корень бисекции (см. _AVITO_PRICE_SEED_BRACKETS) — оставлен для +# совместимости (используется в тестах probe-устойчивости как sentinel). _AVITO_MAX_PRICE = 200_000_000 # Минимальный ценовой диапазон для рекурсии; при hi - lo < MIN_BRACKET прекращаем # деление и пагинируем как есть (принимаем возможный «хвост»). _AVITO_MIN_BRACKET = 50_000 + +# Seed price-brackets (руб) — data-driven по прод-распределению вторички ЕКБ +# (пик 4-8М, bulk 3-16М; самая плотная комнатность — студии 5783, единственный +# >2к блок был студии 4-6М → пик дроблю по 1М). Каждый (комнатность × брекет) +# дальше дробится _walk_price_range если total > cap (=2000 → блоки ≤2к). +# Последний брекет ОТКРЫТ (hi=None → pmax не ставится) — ловит весь хвост люкса +# без потолка; он крошечный (avito >50М ≈ 73), пагинируется напрямую без бисекции. +_AVITO_PRICE_SEED_BRACKETS: list[tuple[int, int | None]] = [ + (0, 4_000_000), + (4_000_000, 5_000_000), + (5_000_000, 6_000_000), + (6_000_000, 7_000_000), + (7_000_000, 8_000_000), + (8_000_000, 10_000_000), + (10_000_000, 13_000_000), + (13_000_000, 17_000_000), + (17_000_000, 25_000_000), + (25_000_000, 250_000_000), + (250_000_000, None), # 250М+ открытый catch-all (без потолка) +] # Avito SERP показывает ~50 карточек на страницу. _AVITO_OFFERS_PER_PAGE = 50 @@ -639,7 +661,7 @@ class AvitoScraper(BaseScraper): return await self._fetch_serp_html(url, page) async def _probe_total( - self, room_slug: str, min_price: int | None, max_price: int + self, room_slug: str, min_price: int | None, max_price: int | None ) -> tuple[str | None, int | None]: """Probe page=1 бакета и извлекает total, устойчиво к browser-ошибкам. @@ -661,7 +683,7 @@ class AvitoScraper(BaseScraper): raise except (httpx.HTTPStatusError, httpx.TimeoutException, httpx.HTTPError) as exc: logger.warning( - "avito: probe browser error %s for %s [%s, %d] — treating as total=None", + "avito: probe browser error %s for %s [%s, %s] — treating as total=None", type(exc).__name__, room_slug, min_price, @@ -670,7 +692,7 @@ class AvitoScraper(BaseScraper): return None, None except Exception: logger.warning( - "avito: probe unexpected error for %s [%s, %d] — treating as total=None", + "avito: probe unexpected error for %s [%s, %s] — treating as total=None", room_slug, min_price, max_price, @@ -695,11 +717,16 @@ class AvitoScraper(BaseScraper): ) -> list[ScrapedLot]: """Exhaustive-загрузка Avito ЕКБ вторички через КОМНАТНОСТЬ × ЦЕНА. - Зеркало CianScraper.fetch_all_secondary. Обходит Avito SERP-cap (~5000 - результатов на запрос): внутри каждой комнатности адаптивно бьёт ценовой - диапазон на бакеты так, чтобы в каждом total < price_cap_per_bucket → - пагинирует бакет полностью. Страницы leaf-бакета — параллельно (Semaphore). - Дедуп по source_id (dict seen). + Обходит Avito SERP-cap (~5000 результатов на запрос). Вместо корень-бисекции + [0, _AVITO_MAX_PRICE] на каждую комнатность (≈5 холостых probe-спусков через + пустой верх, т.к. вторичка ЕКБ почти вся <16М) — стартуем с data-driven + seed-сетки ценовых брекетов (_AVITO_PRICE_SEED_BRACKETS) по прод-распределению. + Каждый (комнатность × seed-брекет) дальше дробится _walk_price_range если + total > price_cap_per_bucket. Последний seed-брекет ОТКРЫТ (hi=None → pmax + не ставится) — ловит весь хвост люкса без потолка. + + Страницы leaf-бакета — параллельно (Semaphore). Дедуп по source_id (dict seen), + ОБЩИЙ по всем брекетам и комнатностям. Параметры: rooms_buckets: список (name, slug) комнатностей (default: все ROOM_SLUGS, @@ -731,19 +758,25 @@ class AvitoScraper(BaseScraper): secondary_only, ) before = len(seen) - await self._walk_price_range( - room_slug=slug, - room_label=room_label, - lo=0, - hi=_AVITO_MAX_PRICE, - seen=seen, - price_cap_per_bucket=price_cap_per_bucket, - max_pages_per_bucket=max_pages_per_bucket, - concurrency=concurrency, - secondary_only=secondary_only, - on_bucket=on_bucket, - skip_buckets=skip_buckets, - ) + # Перебираем seed-брекеты: для ЗАКРЫТЫХ передаём hi = br_hi - 1 + # (pmax эксклюзивный сверху → соседние брекеты не пересекаются), для + # ОТКРЫТОГО (br_hi is None) — hi=None (без потолка). Дедуп по source_id + # в общем `seen` страхует на стыках. + for br_lo, br_hi in _AVITO_PRICE_SEED_BRACKETS: + walk_hi = br_hi - 1 if br_hi is not None else None + await self._walk_price_range( + room_slug=slug, + room_label=room_label, + lo=br_lo, + hi=walk_hi, + seen=seen, + price_cap_per_bucket=price_cap_per_bucket, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + ) room_collected = len(seen) - before logger.info( "avito exhaustive: %s done — collected %d (total unique=%d)", @@ -763,7 +796,7 @@ class AvitoScraper(BaseScraper): room_slug: str, room_label: str, lo: int, - hi: int, + hi: int | None, seen: dict[str, ScrapedLot], price_cap_per_bucket: int, max_pages_per_bucket: int, @@ -775,14 +808,19 @@ class AvitoScraper(BaseScraper): ) -> None: """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi]. - Зеркало CianScraper._walk_price_range. Алгоритм: + Алгоритм для ЗАКРЫТОГО брекета (hi is not None): 1. Запросить page=1 с pmin=lo, pmax=hi → total (page-title/count). 2. Если total <= cap → пагинировать leaf-бакет параллельно (concurrency). 3. Если total > cap → разбить бакет пополам (рекурсия). Guard: hi - lo < _AVITO_MIN_BRACKET → пагинировать как есть (WARNING). + ОТКРЫТЫЙ брекет (hi is None) — верхний seed-брекет без потолка (pmax не + ставится, avito отдаёт всё ≥lo). Делить нельзя (mid=(lo+hi)//2 невозможен) — + probe + пагинируем leaf напрямую. Хвост 250М+ крошечный (≈73), за cap + обычно не выходит; если вышел — пагинируем до max_pages + WARNING (acceptable). + После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots). - bucket_key = "room_label:lo:hi" (checkpoint-ключ для resume). + bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый). on_bucket может быть async или sync. Исключение в on_bucket прерывает прогон. skip_buckets: если bucket_key в skip_buckets — пагинация и on_bucket пропускаются. secondary_only: новостройки (listing_segment=="novostroyki") отбрасываются после @@ -790,6 +828,7 @@ class AvitoScraper(BaseScraper): AvitoBlockedError/AvitoRateLimitedError из _fetch_serp_html пробрасываются наверх. """ _lo_param = lo if lo > 0 else None + _hi_repr = "open" if hi is None else str(hi) # ── Шаг 1: probe page 1 ──────────────────────────────────────────────── # _probe_total ловит browser-ошибки (HTTP 500 / timeout от tradein-browser @@ -800,10 +839,10 @@ class AvitoScraper(BaseScraper): # Ретрай на captcha/missing-count/browser-ошибке: rotate IP + 1 retry. if total is None: logger.warning( - "avito: total=None for %s [%d, %d] depth=%d — rotating IP + retry", + "avito: total=None for %s [%d, %s] depth=%d — rotating IP + retry", room_label, lo, - hi, + _hi_repr, _depth, ) rotated = await self._rotate_ip() @@ -811,6 +850,33 @@ class AvitoScraper(BaseScraper): html, total = await self._probe_total(room_slug, _lo_param, hi) if total is None: + # ОТКРЫТЫЙ брекет: делить нельзя (нет hi). Probe failed → нет total для + # ceil(); пагинируем до max_pages_per_bucket (best-effort, хвост 250М+ + # крошечный) и WARNING. НЕ роняем прогон. + if hi is None: + logger.warning( + "avito: probe failed for OPEN bucket %s [%d, open] depth=%d — " + "paginating up to max_pages (tail-loss accepted)", + room_label, + lo, + _depth, + ) + await self._paginate_leaf_bucket( + room_slug=room_slug, + room_label=room_label, + lo=lo, + hi=None, + html=None, + max_pages=max_pages_per_bucket, + seen=seen, + price_cap_per_bucket=price_cap_per_bucket, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + ) + return bracket_size = hi - lo # Probe не удался даже после ретрая. Если бакет ещё делим — НЕ теряем # его целиком: тяжёлая страница (огромная выдача) скорее всего станет @@ -867,10 +933,10 @@ class AvitoScraper(BaseScraper): return logger.info( - "avito: %s [%d, %d] total=%d depth=%d", + "avito: %s [%d, %s] total=%d depth=%d", room_label, lo, - hi, + _hi_repr, total, _depth, ) @@ -878,6 +944,39 @@ class AvitoScraper(BaseScraper): if total == 0: return + # ── ОТКРЫТЫЙ брекет (hi is None): делить нельзя — пагинируем leaf напрямую ─ + # total известен из probe → max_pages = min(ceil(total/per_page), cap). + # Если total > cap (не должно для 250М+, но guard) — пагинируем как есть до + # max_pages_per_bucket с WARNING (хвост люкса крошечный, tail-loss accepted). + if hi is None: + if total > price_cap_per_bucket: + logger.warning( + "avito: OPEN bucket %s [%d, open] total=%d > cap=%d — paginating " + "without split (tail-loss accepted, lux tail tiny)", + room_label, + lo, + total, + price_cap_per_bucket, + ) + pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE) + max_pages = min(pages_needed, max_pages_per_bucket) + await self._paginate_leaf_bucket( + room_slug=room_slug, + room_label=room_label, + lo=lo, + hi=None, + html=html, + max_pages=max_pages, + seen=seen, + price_cap_per_bucket=price_cap_per_bucket, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + ) + return + # ── Шаг 2: деление или пагинация ────────────────────────────────────── bracket_size = hi - lo need_split = total > price_cap_per_bucket @@ -932,15 +1031,7 @@ class AvitoScraper(BaseScraper): ) return - # ── Checkpoint-ключ leaf-бакета ────────────────────────────────────── - bucket_key = f"{room_label}:{lo}:{hi}" - - # Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket. - if skip_buckets and bucket_key in skip_buckets: - logger.info("avito: skip bucket %s — already done (resume)", bucket_key) - return - - # ── Параллельная пагинация leaf-бакета ──────────────────────────────── + # ── Закрытый leaf-бакет: считаем max_pages + warning, делегируем helper'у ─ pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE) max_pages = min(pages_needed, max_pages_per_bucket) @@ -959,26 +1050,81 @@ class AvitoScraper(BaseScraper): tail_loss, ) - # Страница 1 уже есть (html из probe выше); остальные — параллельно. + await self._paginate_leaf_bucket( + room_slug=room_slug, + room_label=room_label, + lo=lo, + hi=hi, + html=html, + max_pages=max_pages, + seen=seen, + price_cap_per_bucket=price_cap_per_bucket, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + ) + + async def _paginate_leaf_bucket( + self, + *, + room_slug: str, + room_label: str, + lo: int, + hi: int | None, + html: str | None, + max_pages: int, + seen: dict[str, ScrapedLot], + price_cap_per_bucket: int, + max_pages_per_bucket: int, + concurrency: int, + secondary_only: bool, + on_bucket: Callable[..., Any] | None, + skip_buckets: set[str] | None, + ) -> None: + """Параллельная пагинация одного leaf-бакета + фильтр + дедуп + on_bucket. + + Общая для ЗАКРЫТЫХ ([lo, hi]) и ОТКРЫТОГО (hi=None → pmax не ставится) + брекетов. ``html`` — HTML страницы 1 из probe (переиспользуем, не + перезапрашиваем); для открытого-без-probe передаётся None → стр.1 фетчится + наравне с остальными. ``max_pages`` уже посчитан вызывающим. + + bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый). + skip_buckets: если bucket_key в skip_buckets — пагинация и on_bucket пропускаются. + AvitoBlockedError/AvitoRateLimitedError из page-фетчей пробрасываются наверх. + """ + _lo_param = lo if lo > 0 else None + _hi_param = hi # None → _build_rooms_url не ставит pmax + _hi_repr = "open" if hi is None else str(hi) + bucket_key = f"{room_label}:{lo}:{_hi_repr}" + + # Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket. + if skip_buckets and bucket_key in skip_buckets: + logger.info("avito: skip bucket %s — already done (resume)", bucket_key) + return + + # Страница 1 — из probe-html если он есть (закрытый/открытый с probe); + # иначе фетчим её как обычную страницу (открытый брекет без probe-html). sem = asyncio.Semaphore(concurrency) - first_url = self._build_rooms_url(room_slug, 1, _lo_param, hi) + first_url = self._build_rooms_url(room_slug, 1, _lo_param, _hi_param) async def _one_page(p: int) -> list[ScrapedLot]: - if p == 1: - return self._parse_html(html, source_url_base=first_url) if html else [] + if p == 1 and html is not None: + return self._parse_html(html, source_url_base=first_url) async with sem: - page_html = await self._fetch_rooms_page_html(room_slug, p, _lo_param, hi) + page_html = await self._fetch_rooms_page_html(room_slug, p, _lo_param, _hi_param) await asyncio.sleep(self.request_delay_sec) if page_html is None: logger.warning( - "avito: page_html=None %s [%d, %d] page=%d — skipping page", + "avito: page_html=None %s [%d, %s] page=%d — skipping page", room_label, lo, - hi, + _hi_repr, p, ) return [] - page_url = self._build_rooms_url(room_slug, p, _lo_param, hi) + page_url = self._build_rooms_url(room_slug, p, _lo_param, _hi_param) return self._parse_html(page_html, source_url_base=page_url) page_results = await asyncio.gather( @@ -993,10 +1139,10 @@ class AvitoScraper(BaseScraper): if isinstance(res, AvitoBlockedError | AvitoRateLimitedError): raise res logger.warning( - "avito: page exception %s [%d, %d] page=%d — %r", + "avito: page exception %s [%d, %s] page=%d — %r", room_label, lo, - hi, + _hi_repr, p_idx, res, ) @@ -1019,10 +1165,10 @@ class AvitoScraper(BaseScraper): seen[key] = lot logger.info( - "avito: %s [%d, %d] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d", + "avito: %s [%d, %s] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d", room_label, lo, - hi, + _hi_repr, max_pages, collected_this_bucket, dropped_nb, diff --git a/tradein-mvp/backend/tests/scrapers/test_avito_exhaustive.py b/tradein-mvp/backend/tests/scrapers/test_avito_exhaustive.py index 58e4e762..fbfa8089 100644 --- a/tradein-mvp/backend/tests/scrapers/test_avito_exhaustive.py +++ b/tradein-mvp/backend/tests/scrapers/test_avito_exhaustive.py @@ -23,6 +23,7 @@ from app.services.scrapers.avito import ( _AVITO_MAX_PRICE, _AVITO_MIN_BRACKET, _AVITO_OFFERS_PER_PAGE, + _AVITO_PRICE_SEED_BRACKETS, AvitoScraper, ) from app.services.scrapers.avito_exceptions import AvitoBlockedError @@ -119,8 +120,12 @@ def test_extract_total_count_missing_returns_none(scraper: AvitoScraper) -> None @pytest.mark.asyncio async def test_fetch_all_secondary_splits_on_cap(scraper: AvitoScraper) -> None: - """total > cap → бакет делится; оба суб-бакета пагинируются.""" - mid = _AVITO_MAX_PRICE // 2 + """Внутри seed-брекета total > cap → бакет делится пополам; суб-бакеты пагинируются.""" + # Первый закрытый seed-брекет (0, 4_000_000) → walk_hi=3_999_999. + br_lo, br_hi_raw = _AVITO_PRICE_SEED_BRACKETS[0] + assert br_hi_raw is not None + br_hi = br_hi_raw - 1 # эксклюзивный верх + mid = (br_lo + br_hi) // 2 call_log: list[tuple] = [] async def fake_fetch( @@ -135,12 +140,16 @@ async def test_fetch_all_secondary_splits_on_cap(scraper: AvitoScraper) -> None: min_m = re.search(r"min=(\w+)", html) max_m = re.search(r"max=(\w+)", html) min_p = int(min_m.group(1)) if min_m and min_m.group(1) != "None" else 0 - max_p = int(max_m.group(1)) if max_m and max_m.group(1) != "None" else _AVITO_MAX_PRICE - if max_p == mid: + max_p = int(max_m.group(1)) if max_m and max_m.group(1) != "None" else None + # Внутри первого брекета [0, 3_999_999]: весь брекет > cap → split; + # половинки < cap → пагинируются. Прочие seed-брекеты — пусто (0). + if min_p == 0 and max_p == br_hi: + return 3000 # весь первый брекет — split + if min_p == 0 and max_p == mid: return 800 # левая половина — пагинировать - if min_p == mid + 1: + if min_p == mid + 1 and max_p == br_hi: return 700 # правая половина — пагинировать - return 3000 # весь диапазон — split + return 0 # остальные seed-брекеты пусты def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]: import re @@ -170,7 +179,7 @@ async def test_fetch_all_secondary_splits_on_cap(scraper: AvitoScraper) -> None: ) assert len(lots) > 0 - # Должны быть пробы суб-бакетов (после split весь диапазон не пагинируется) + # Должны быть пробы суб-бакетов (после split весь брекет не пагинируется). probed_max = [mx for (p, _mn, mx) in call_log if p == 1] assert any(mx == mid for mx in probed_max), f"Ожидался probe левой половины, got {probed_max}" @@ -292,16 +301,23 @@ async def test_fetch_all_secondary_concurrent_pages_deduped(scraper: AvitoScrape @pytest.mark.asyncio async def test_on_bucket_receives_key_and_lots(scraper: AvitoScraper) -> None: - """on_bucket получает (bucket_key, lots) — bucket_key = room_label:lo:hi.""" + """on_bucket получает (bucket_key, lots) на каждый seed-брекет. + + bucket_key закрытого брекета = "room_label:lo:hi" (hi = br_hi-1, эксклюзивный); + открытого (последнего) = "room_label:lo:open". Одна on_bucket-нотификация + на каждый непустой seed-брекет (здесь все непустые → len(brackets) вызовов). + """ received: list[tuple[str, int]] = [] def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None: received.append((bucket_key, len(lots))) + # Уникальные source_id на каждый брекет, чтобы on_bucket получал непустые + # списки во ВСЕХ брекетах (дедуп в seen не должен опустошать поздние брекеты). async def fake_fetch( room_slug: str, page: int, min_price: int | None, max_price: int | None ) -> str: - return f"page={page}" + return f"page={page} min={min_price} max={max_price}" def fake_total(html: str) -> int | None: return 5 # 1 страница @@ -309,9 +325,13 @@ async def test_on_bucket_receives_key_and_lots(scraper: AvitoScraper) -> None: def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]: import re + min_m = re.search(r"min=(\w+)", html) + max_m = re.search(r"max=(\w+)", html) + min_p = min_m.group(1) if min_m else "0" + max_p = max_m.group(1) if max_m else "open" page_m = re.search(r"page=(\d+)", html) page = int(page_m.group(1)) if page_m else 1 - return [_make_lot(f"lot_p{page}_{i}") for i in range(5)] + return [_make_lot(f"lot_{min_p}_{max_p}_p{page}_{i}") for i in range(5)] with ( patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch), @@ -325,34 +345,55 @@ async def test_on_bucket_receives_key_and_lots(scraper: AvitoScraper) -> None: on_bucket=fake_on_bucket, ) - assert len(received) == 1, f"Ожидался 1 вызов on_bucket, получено {len(received)}" - key, count = received[0] - assert key == f"room_2_komn:0:{_AVITO_MAX_PRICE}", f"Неверный bucket_key: {key!r}" - assert count == 5 + # Один on_bucket на каждый seed-брекет (все непустые). + assert len(received) == len( + _AVITO_PRICE_SEED_BRACKETS + ), f"Ожидалось {len(_AVITO_PRICE_SEED_BRACKETS)} вызовов on_bucket, получено {len(received)}" + keys = [k for (k, _c) in received] + # Закрытые брекеты: hi = br_hi - 1. Открытый (последний): ':open'. + first_lo, first_hi = _AVITO_PRICE_SEED_BRACKETS[0] + assert first_hi is not None + assert keys[0] == f"room_2_komn:{first_lo}:{first_hi - 1}", f"Неверный ключ: {keys[0]!r}" + last_lo, last_hi = _AVITO_PRICE_SEED_BRACKETS[-1] + assert last_hi is None + assert keys[-1] == f"room_2_komn:{last_lo}:open", f"Открытый ключ неверен: {keys[-1]!r}" + assert all(count == 5 for (_k, count) in received) @pytest.mark.asyncio async def test_skip_buckets_skips_pagination(scraper: AvitoScraper) -> None: - """skip_buckets: probe выполняется, но пагинация и on_bucket пропускаются.""" - fetch_calls: list[int] = [] + """skip_buckets: для skip-бакета probe выполняется, пагинация и on_bucket — нет.""" + fetch_calls: list[tuple[int, int | None, int | None]] = [] on_bucket_calls: list[str] = [] async def fake_fetch( room_slug: str, page: int, min_price: int | None, max_price: int | None ) -> str: - fetch_calls.append(page) - return f"page={page}" + fetch_calls.append((page, min_price, max_price)) + return f"page={page} min={min_price} max={max_price}" def fake_total(html: str) -> int | None: return 5 def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]: - return [_make_lot("lot_1")] + import re + + min_m = re.search(r"min=(\w+)", html) + max_m = re.search(r"max=(\w+)", html) + return [_make_lot(f"lot_{min_m.group(1) if min_m else 0}_{max_m.group(1) if max_m else 0}")] def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None: on_bucket_calls.append(bucket_key) - skip_set = {f"room_1_komn:0:{_AVITO_MAX_PRICE}"} + # Skip ВСЕ брекеты → ни одной on_bucket-нотификации, но каждый брекет + # всё равно делает probe (page=1). + first_lo, first_hi = _AVITO_PRICE_SEED_BRACKETS[0] + assert first_hi is not None + first_key = f"room_1_komn:{first_lo}:{first_hi - 1}" + skip_all: set[str] = set() + for br_lo, br_hi in _AVITO_PRICE_SEED_BRACKETS: + hi_repr = "open" if br_hi is None else str(br_hi - 1) + skip_all.add(f"room_1_komn:{br_lo}:{hi_repr}") with ( patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch), @@ -364,11 +405,18 @@ async def test_skip_buckets_skips_pagination(scraper: AvitoScraper) -> None: rooms_buckets=[("1-комн.", "1-komnatnye-X")], price_cap_per_bucket=1400, on_bucket=fake_on_bucket, - skip_buckets=skip_set, + skip_buckets=skip_all, ) - assert fetch_calls == [1], f"Ожидался ровно 1 вызов (probe), got {fetch_calls}" - assert on_bucket_calls == [], "on_bucket не должен вызываться для skip-бакета" + # Ровно один probe (page=1) на каждый seed-брекет, без доп. страниц. + assert first_key in skip_all + assert all( + p == 1 for (p, _mn, _mx) in fetch_calls + ), f"Ожидались только probe-вызовы: {fetch_calls}" + assert len(fetch_calls) == len( + _AVITO_PRICE_SEED_BRACKETS + ), f"Ожидался 1 probe на брекет ({len(_AVITO_PRICE_SEED_BRACKETS)}), got {len(fetch_calls)}" + assert on_bucket_calls == [], "on_bucket не должен вызываться для skip-бакетов" assert lots == [] @@ -465,12 +513,18 @@ async def test_probe_browser_500_on_wide_bucket_splits_not_crash( browser HTTP 500. Раньше это роняло весь full-load с 0 uniq. Теперь широкий bucket принудительно делится пополам; под-бакеты пагинируются нормально. """ - probe_calls: list[tuple[int | None, int]] = [] + # Первый закрытый seed-брекет (0, 4_000_000) → walk_hi=3_999_999. + _br_lo, br_hi_raw = _AVITO_PRICE_SEED_BRACKETS[0] + assert br_hi_raw is not None + br_hi = br_hi_raw - 1 + probe_calls: list[tuple[int | None, int | None]] = [] async def fake_fetch( room_slug: str, page: int, min_price: int | None, max_price: int | None ) -> str: - if page == 1 and (min_price in (None, 0)) and max_price == _AVITO_MAX_PRICE: + # Probe верхней границы ШИРОКОГО первого брекета падает (имитация HTTP 500 + # от tradein-browser на тяжёлой выдаче) → должен произойти split. + if page == 1 and (min_price in (None, 0)) and max_price == br_hi: probe_calls.append((min_price, max_price)) raise _http_500_error() return f"page={page} min={min_price} max={max_price}" @@ -483,10 +537,12 @@ async def test_probe_browser_500_on_wide_bucket_splits_not_crash( page_m = re.search(r"page=(\d+)", html) page = int(page_m.group(1)) if page_m else 1 + min_m = re.search(r"min=(\w+)", html) max_m = re.search(r"max=(\w+)", html) + min_p = min_m.group(1) if min_m else "0" max_p = max_m.group(1) if max_m else "MAX" if page == 1: - return [_make_lot(f"lot_{max_p}_{i}") for i in range(3)] + return [_make_lot(f"lot_{min_p}_{max_p}_{i}") for i in range(3)] return [] with ( @@ -500,7 +556,7 @@ async def test_probe_browser_500_on_wide_bucket_splits_not_crash( price_cap_per_bucket=1400, ) - assert probe_calls, "Probe полного диапазона должен был вызваться и упасть" + assert probe_calls, "Probe первого (широкого) брекета должен был вызваться и упасть" assert len(lots) > 0, f"Ожидались лоты из под-бакетов после split, got {len(lots)}" @@ -509,16 +565,19 @@ async def test_probe_timeout_on_wide_bucket_splits_not_crash( scraper: AvitoScraper, ) -> None: """httpx.TimeoutException на широком bucket обрабатывается как HTTP 500 → split.""" + _br_lo, br_hi_raw = _AVITO_PRICE_SEED_BRACKETS[0] + assert br_hi_raw is not None + br_hi = br_hi_raw - 1 probe_failed = False async def fake_fetch( room_slug: str, page: int, min_price: int | None, max_price: int | None ) -> str: nonlocal probe_failed - if page == 1 and (min_price in (None, 0)) and max_price == _AVITO_MAX_PRICE: + if page == 1 and (min_price in (None, 0)) and max_price == br_hi: probe_failed = True raise httpx.TimeoutException("Page.goto timeout") - return f"page={page} max={max_price}" + return f"page={page} min={min_price} max={max_price}" def fake_total(html: str) -> int | None: return 40 @@ -528,9 +587,11 @@ async def test_probe_timeout_on_wide_bucket_splits_not_crash( page_m = re.search(r"page=(\d+)", html) page = int(page_m.group(1)) if page_m else 1 + min_m = re.search(r"min=(\w+)", html) max_m = re.search(r"max=(\w+)", html) + min_p = min_m.group(1) if min_m else "0" max_p = max_m.group(1) if max_m else "MAX" - return [_make_lot(f"lot_{max_p}_{i}") for i in range(2)] if page == 1 else [] + return [_make_lot(f"lot_{min_p}_{max_p}_{i}") for i in range(2)] if page == 1 else [] with ( patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch), @@ -634,3 +695,197 @@ async def test_probe_avito_blocked_propagates_not_swallowed( rooms_buckets=[("1-комн.", "1-komnatnye-X")], price_cap_per_bucket=1400, ) + + +# ── Seed price-brackets entry point (perf: open top bucket) ─────────────────── + + +@pytest.mark.asyncio +async def test_fetch_all_secondary_walks_seed_brackets_per_room( + scraper: AvitoScraper, +) -> None: + """fetch_all_secondary зовёт _walk_price_range РОВНО len(seed) раз на комнатность. + + Аргументы lo/hi совпадают с seed-брекетами: закрытые передаются с hi=br_hi-1 + (эксклюзивный верх, без overlap на стыках), последний (открытый) — hi=None. + """ + walk_calls: list[tuple[int, int | None]] = [] + + async def spy_walk(*args: object, **kwargs: object) -> None: + walk_calls.append((kwargs["lo"], kwargs["hi"])) # type: ignore[arg-type] + # Не уходим в реальную рекурсию — фиксируем только верхнеуровневые входы. + return None + + with patch.object(scraper, "_walk_price_range", side_effect=spy_walk): + await scraper.fetch_all_secondary( + rooms_buckets=[("1-комн.", "1-komnatnye-X")], + price_cap_per_bucket=2000, + ) + + expected = [ + (br_lo, (br_hi - 1) if br_hi is not None else None) + for (br_lo, br_hi) in _AVITO_PRICE_SEED_BRACKETS + ] + assert walk_calls == expected, f"Неверные входы _walk_price_range: {walk_calls}" + # Последний — открытый (hi=None). + assert walk_calls[-1][1] is None, "Последний seed-брекет должен быть открытым (hi=None)" + # Ровно один вызов на seed-брекет (1 комнатность). + assert len(walk_calls) == len(_AVITO_PRICE_SEED_BRACKETS) + + +@pytest.mark.asyncio +async def test_open_bracket_probes_pmax_none_and_paginates_without_split( + scraper: AvitoScraper, +) -> None: + """Открытый брекет (hi=None): probe с pmax=None, пагинация БЕЗ попытки split. + + Даже если total > cap (мокнуто) — НЕ должно быть рекурсивного деления: + при hi=None mid=(lo+hi)//2 невозможен, поэтому leaf пагинируется напрямую. + Проверяем: (1) probe идёт с max_price=None (pmax не ставится), (2) НЕТ + под-вызовов с пересчитанным mid (никаких max_price != None в fetch'ах). + """ + open_lo = _AVITO_PRICE_SEED_BRACKETS[-1][0] + assert _AVITO_PRICE_SEED_BRACKETS[-1][1] is None + + fetch_args: list[tuple[int, int | None, int | None]] = [] + + async def fake_fetch( + room_slug: str, page: int, min_price: int | None, max_price: int | None + ) -> str: + fetch_args.append((page, min_price, max_price)) + return f"page={page} min={min_price} max={max_price}" + + # total > cap для ОТКРЫТОГО брекета — провоцируем split-соблазн. Закрытые + # брекеты возвращают 0 (не пагинируются), чтобы изолировать открытый. + def fake_total(html: str) -> int | None: + import re + + max_m = re.search(r"max=(\w+)", html) + max_raw = max_m.group(1) if max_m else "None" + if max_raw == "None": # открытый брекет (pmax отсутствует) + return 5000 # > cap=2000 — но split НЕ должен случиться + return 0 + + def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]: + import re + + page_m = re.search(r"page=(\d+)", html) + page = int(page_m.group(1)) if page_m else 1 + # Только открытый брекет (max=None) отдаёт лоты. + if "max=None" not in html: + return [] + return [_make_lot(f"open_lot_p{page}_{i}") for i in range(3)] + + with ( + patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch), + patch.object(scraper, "_extract_total_count", side_effect=fake_total), + patch.object(scraper, "_parse_html", side_effect=fake_parse), + patch.object(scraper, "_rotate_ip", return_value=False), + ): + lots = await scraper.fetch_all_secondary( + rooms_buckets=[("1-комн.", "1-komnatnye-X")], + price_cap_per_bucket=2000, + max_pages_per_bucket=100, + ) + + # Probe открытого брекета: min_price=open_lo, max_price=None. + open_probes = [ + (mn, mx) for (p, mn, mx) in fetch_args if p == 1 and mn == open_lo and mx is None + ] + assert open_probes, f"Ожидался probe открытого брекета с pmax=None: {fetch_args}" + # Все fetch'и открытого брекета (min=open_lo) идут с max_price=None — НЕТ + # под-брекетов с вычисленным mid (что означало бы split). + open_fetches = [(mn, mx) for (p, mn, mx) in fetch_args if mn == open_lo] + assert all( + mx is None for (_mn, mx) in open_fetches + ), f"Открытый брекет НЕ должен порождать closed-под-брекеты (split): {open_fetches}" + # Открытый брекет всё же собрал лоты (пагинация прошла, total=5000 → cap pages). + assert len(lots) > 0, "Открытый брекет должен пагинироваться и собрать лоты" + + +@pytest.mark.asyncio +async def test_closed_bracket_under_cap_single_probe_then_paginate( + scraper: AvitoScraper, +) -> None: + """Закрытый брекет total str: + fetch_args.append((page, min_price, max_price)) + return f"page={page} min={min_price} max={max_price}" + + def fake_total(html: str) -> int | None: + import re + + min_m = re.search(r"min=(\w+)", html) + max_m = re.search(r"max=(\w+)", html) + min_p = min_m.group(1) if min_m else "0" + max_p = max_m.group(1) if max_m else "None" + if min_p == str(target_lo) and max_p == str(target_hi): + return 60 # ceil(60/50)=2 страницы, < cap → без split + return 0 + + def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]: + import re + + if f"min={target_lo} max={target_hi}" not in html: + return [] + page_m = re.search(r"page=(\d+)", html) + page = int(page_m.group(1)) if page_m else 1 + if page <= 2: + return [_make_lot(f"t_lot_p{page}_{i}") for i in range(30)] + return [] + + with ( + patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch), + patch.object(scraper, "_extract_total_count", side_effect=fake_total), + patch.object(scraper, "_parse_html", side_effect=fake_parse), + patch.object(scraper, "_rotate_ip", return_value=False), + ): + lots = await scraper.fetch_all_secondary( + rooms_buckets=[("1-комн.", "1-komnatnye-X")], + price_cap_per_bucket=2000, + ) + + # Целевой брекет: probe (page=1) + страница 2, без под-брекетов (split). + target_pages = sorted(p for (p, mn, mx) in fetch_args if mn == target_lo and mx == target_hi) + assert target_pages == [1, 2], f"Ожидались probe+page2 без split: {target_pages}" + assert len(lots) == 60, f"Ожидалось 60 лотов (2 страницы × 30), got {len(lots)}" + + +@pytest.mark.asyncio +async def test_dedup_source_id_across_brackets(scraper: AvitoScraper) -> None: + """Дедуп по source_id работает между разными seed-брекетами (общий seen).""" + + async def fake_fetch( + room_slug: str, page: int, min_price: int | None, max_price: int | None + ) -> str: + return f"page={page} min={min_price} max={max_price}" + + def fake_total(html: str) -> int | None: + return 3 # 1 страница на каждый брекет + + def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]: + # ВСЕ брекеты отдают ОДИН и тот же source_id → должен остаться 1 уникальный. + return [_make_lot("shared_dup_id")] + + with ( + patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch), + patch.object(scraper, "_extract_total_count", side_effect=fake_total), + patch.object(scraper, "_parse_html", side_effect=fake_parse), + patch.object(scraper, "_rotate_ip", return_value=False), + ): + lots = await scraper.fetch_all_secondary( + rooms_buckets=[("1-комн.", "1-komnatnye-X")], + price_cap_per_bucket=2000, + ) + + assert len(lots) == 1, f"Дедуп между брекетами должен дать 1 уникальный лот, got {len(lots)}" + assert lots[0].source_id == "shared_dup_id"