From 5dc830fb26962bad312f740140afedd964ae81ab Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 2 Jul 2026 19:39:25 +0300 Subject: [PATCH] refactor(scraper-kit): unify price bisection engine across providers + dedup (#2134) --- .../test_scraper_kit_bisection_parity.py | 455 ++++++++++++++++++ .../src/scraper_kit/pricing/__init__.py | 21 + .../src/scraper_kit/pricing/bisection.py | 253 ++++++++++ .../src/scraper_kit/providers/avito/serp.py | 315 ++++-------- .../src/scraper_kit/providers/cian/serp.py | 230 +++------ .../scraper_kit/providers/domclick/serp.py | 81 ++-- .../src/scraper_kit/providers/yandex/serp.py | 226 ++++----- .../scraper_kit/providers/yandex/shared.py | 98 +--- 8 files changed, 1069 insertions(+), 610 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_scraper_kit_bisection_parity.py create mode 100644 tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/__init__.py create mode 100644 tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/bisection.py diff --git a/tradein-mvp/backend/tests/test_scraper_kit_bisection_parity.py b/tradein-mvp/backend/tests/test_scraper_kit_bisection_parity.py new file mode 100644 index 00000000..011c8d70 --- /dev/null +++ b/tradein-mvp/backend/tests/test_scraper_kit_bisection_parity.py @@ -0,0 +1,455 @@ +"""Golden-parity бисекции: старый боевой обход `app.services.scrapers.` +≡ новый единый движок `scraper_kit.pricing.walk_price_range` (#2134). + +Четыре независимые копии адаптивной бисекции «комнатность × цена» были +консолидированы в один движок (`scraper_kit/pricing/bisection.py`). Провайдеры +(avito/cian/yandex/domclick) теперь передают в движок свои probe/paginate + свой +`BisectionConfig`. Этот тест доказывает, что unification НЕ изменил ОБХОД: для +каждого провайдера при одинаковом входе (детерминированный мок счётчика по +диапазону) старая боевая реализация и новая (движок) посещают ИДЕНТИЧНУЮ +последовательность бакетов — probe-запросы, leaf-пагинации, деления, ранние +возвраты на нуле, probe-fail-ветки. + +Механика: на ОБЕ реализации (старую и новую) ставятся ОДИНАКОВЫЕ моки на +низкоуровневые методы (probe-fetch + leaf-пагинатор), которые (а) возвращают +счётчик из общего оракула `count(lo, hi)` и (б) пишут событие в общий список. +Совпадение списков событий = идентичный обход. +""" + +from __future__ import annotations + +import os +from collections.abc import Callable +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from scraper_kit.price_brackets import get_price_seed_brackets +from scraper_kit.providers.avito.serp import AvitoScraper as NewAvito +from scraper_kit.providers.cian.serp import CianScraper as NewCian +from scraper_kit.providers.domclick.serp import DomClickScraper as NewDomClick +from scraper_kit.providers.yandex.serp import YandexRealtyScraper as NewYandex + +from app.services.scrapers.avito import AvitoScraper as OldAvito +from app.services.scrapers.cian import CianScraper as OldCian +from app.services.scrapers.domclick import DomClickScraper as OldDomClick +from app.services.scrapers.yandex_realty import YandexRealtyScraper as OldYandex + +# ── Оракул счётчиков ───────────────────────────────────────────────────────── + +Oracle = Callable[[Any, Any], int | None] + + +def make_oracle( + *, + divisor: int, + open_ceiling: int, + zeros: frozenset[tuple[Any, Any]] = frozenset(), + nones: frozenset[tuple[Any, Any]] = frozenset(), +) -> Oracle: + """Детерминированный счётчик от (lo, hi): count = span // divisor. + + span считается по клампнутым границам (None-lo → 0, None-hi → open_ceiling), + чтобы широкие брекеты делились, а узкие уходили в leaf. `zeros` форсит 0 (ранний + возврат / leaf-на-нуле у yandex), `nones` форсит None (probe-fail). + """ + + def oracle(lo: Any, hi: Any) -> int | None: + if (lo, hi) in nones: + return None + if (lo, hi) in zeros: + return 0 + eff_lo = lo if lo is not None else 0 + eff_hi = hi if hi is not None else open_ceiling + span = eff_hi - eff_lo + return max(0, span // divisor) + + return oracle + + +# ── Провайдер-специфичные «сборщики событий» ───────────────────────────────── +# +# Каждый ставит одинаковые моки на старый/новый экземпляр и возвращает список +# событий обхода. `("probe", lo, hi)` — запрос счётчика бакета; `("leaf", ...)` — +# пагинация leaf-бакета. + + +def _patch_avito(scraper: Any, oracle: Oracle, events: list[Any]) -> None: + def probe(room_slug: str, min_price: Any, max_price: Any) -> tuple[str | None, int | None]: + events.append(("probe", min_price, max_price)) + return "H", oracle(min_price, max_price) + + def leaf(**kw: Any) -> None: + events.append(("leaf", kw["lo"], kw["hi"], kw["max_pages"])) + + scraper._probe_total = AsyncMock(side_effect=probe) + scraper._paginate_leaf_bucket = AsyncMock(side_effect=leaf) + scraper._rotate_ip = AsyncMock(return_value=False) + + +def _patch_cian(scraper: Any, oracle: Oracle, events: list[Any]) -> None: + holder: dict[str, tuple[Any, Any]] = {} + + def fph(rooms: Any, page: int, min_price: Any, max_price: Any) -> str: + events.append(("probe", min_price, max_price)) + holder["b"] = (min_price, max_price) + return "H" + + def eto(html: str) -> int | None: + return oracle(*holder["b"]) + + def leaf(**kw: Any) -> None: + events.append(("leaf", kw["lo"], kw["hi"], kw["total"])) + + scraper._fetch_page_html = AsyncMock(side_effect=fph) + scraper._extract_total_offers = MagicMock(side_effect=eto) + scraper._paginate_leaf_bucket = AsyncMock(side_effect=leaf) + scraper._rotate_ip = AsyncMock(return_value=False) + scraper.sleep_between_requests = AsyncMock() + + +def _patch_domclick(scraper: Any, oracle: Oracle, events: list[Any]) -> None: + def count(fetcher: Any, rooms: Any, gte: Any, lte: Any) -> int: + events.append(("probe", gte, lte)) + val = oracle(gte, lte) + return val if val is not None else 0 + + def paginate( + fetcher: Any, rooms: Any, gte: Any, lte: Any, seen_ids: Any, out_lots: Any, pages: Any + ) -> None: + events.append(("leaf", gte, lte)) + + scraper._count = AsyncMock(side_effect=count) + scraper._paginate = AsyncMock(side_effect=paginate) + + +def _patch_yandex( + scraper: Any, oracle: Oracle, events: list[Any], *, record_pages: bool = False +) -> None: + """Yandex: leaf-пагинация встроена в walk (нет отдельного метода-хука). + + Стабим `_fetch_page_json` (page=1 отдаёт payload с totalItems из оракула; + page>1 — пустой payload) и передаём `on_bucket`, который пишет завершение + leaf/degraded бакета. probe = page-1-fetch; leaf-страницы 2..N идут через + asyncio.gather (порядок недетерминирован) — их НЕ пишем в норм-режиме; факт + leaf-обхода фиксируется через on_bucket. Для degraded-теста (пагинация + последовательная) `record_pages=True` пишет все страницы. + """ + + def gate(total: int | None, entities: list[Any] | None = None) -> dict[str, Any]: + pager: dict[str, Any] = {} + if total is not None: + pager["totalItems"] = total + return {"response": {"search": {"offers": {"entities": entities or [], "pager": pager}}}} + + def fpj(rooms: Any, page: int, price_min: Any, price_max: Any, new_flat: str = "NO") -> Any: + if page == 1: + events.append(("probe", price_min, price_max)) + return gate(oracle(price_min, price_max)) + if record_pages: + events.append(("page", price_min, price_max, page)) + return gate(0, entities=[]) + + scraper._fetch_page_json = AsyncMock(side_effect=fpj) + scraper._rotate_ip = AsyncMock(return_value=False) + scraper.request_delay_sec = 0.0 + + +def _yandex_on_bucket(events: list[Any]) -> Callable[[str, int], None]: + def on_bucket(bucket_key: str, count: int) -> None: + events.append(("bucket", bucket_key)) + + return on_bucket + + +# ── Конструкторы старых/новых экземпляров (offline, без DB) ─────────────────── + +_KIT_CONFIG = SimpleNamespace( + scraper_proxy_url=None, + yandex_proxy_rotate_url=None, + avito_proxy_rotate_url=None, + avito_proxy_max_rotations=0, + scraper_fetch_mode="browser", + browser_http_endpoint="http://tradein-browser:9000", + glitchtip_dsn=None, + avito_serp_ekb_only=False, +) + + +def _old_avito() -> Any: + with patch("app.services.scrapers.avito.get_scraper_delay", return_value=7.0): + return OldAvito() + + +def _old_cian() -> Any: + with patch("app.services.scrapers.cian.get_scraper_delay", return_value=5.0): + return OldCian() + + +def _old_yandex() -> Any: + with patch("app.services.scrapers.yandex_realty.get_scraper_delay", return_value=5.0): + return OldYandex() + + +def _old_domclick() -> Any: + with patch("app.services.scrapers.domclick.get_scraper_delay", return_value=8.0): + return OldDomClick() + + +# ── avito / cian: одинаковый скелет (probe → split → leaf) ──────────────────── + + +# Провайдер-специфичные kwargs для _walk_price_range (avito идентифицирует бакет +# room_slug/room_label; cian — кортежем rooms). +_AVITO_WALK_EXTRA = {"room_slug": "room1", "room_label": "1к"} +_CIAN_WALK_EXTRA = {"rooms": (1,)} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "old_factory, new_factory, patch_fn, walk_extra", + [ + (_old_avito, lambda: NewAvito(config=_KIT_CONFIG), _patch_avito, _AVITO_WALK_EXTRA), + (_old_cian, lambda: NewCian(config=_KIT_CONFIG), _patch_cian, _CIAN_WALK_EXTRA), + ], +) +async def test_bisection_parity_avito_cian( + old_factory: Any, new_factory: Any, patch_fn: Any, walk_extra: dict[str, Any] +) -> None: + """avito/cian: split-дерево + leaf + ранний возврат на нуле — обход идентичен. + + Оракул: divisor=1000, cap=1400 → брекет 8М делится 3 уровня (8М→4М→2М→1М-leaf); + один под-брекет форсим в 0 (ранний возврат без leaf). + """ + # После первого split (0,7_999_999): mid=3_999_999 → дети (None,3_999_999) и + # (4_000_000, 7_999_999). Форсим 0 на (4_000_000, 5_999_999) — под-брекет после + # второго split правой ветки. + oracle = make_oracle( + divisor=1000, + open_ceiling=30_000_000, + zeros=frozenset({(4_000_000, 5_999_999)}), + ) + brackets: list[tuple[int, int | None]] = [(0, 7_999_999), (25_000_000, None)] + + async def run(scraper: Any) -> list[Any]: + events: list[Any] = [] + patch_fn(scraper, oracle, events) + for lo, hi in brackets: + await scraper._walk_price_range( + lo=lo, + hi=hi, + seen={}, + price_cap_per_bucket=1400, + max_pages_per_bucket=100, + concurrency=5, + secondary_only=True, + on_bucket=None, + skip_buckets=None, + **walk_extra, + ) + return events + + old_events = await run(old_factory()) + new_events = await run(new_factory()) + assert old_events == new_events + # Санити: обход реально нетривиален (несколько probe + несколько leaf). + assert sum(1 for e in old_events if e[0] == "leaf") >= 4 + assert sum(1 for e in old_events if e[0] == "probe") >= 6 + + +@pytest.mark.asyncio +async def test_bisection_parity_avito_probe_fail() -> None: + """avito probe-fail (total=None): SPLIT_OR_SKIP — принудительный split делимого + закрытого брекета, best-effort leaf открытого — идентичен старому.""" + # (0, 7_999_999): probe None → делим (bracket>=MIN). Дети (None,4M),(4000001,...): + # им дадим реальные счётчики. Открытый (25M, None) probe None → best-effort leaf. + oracle = make_oracle( + divisor=1000, + open_ceiling=30_000_000, + nones=frozenset({(None, 7_999_999), (25_000_000, None)}), + ) + brackets: list[tuple[int, int | None]] = [(0, 7_999_999), (25_000_000, None)] + + async def run(scraper: Any) -> list[Any]: + events: list[Any] = [] + _patch_avito(scraper, oracle, events) + for lo, hi in brackets: + await scraper._walk_price_range( + room_slug="room1", + room_label="1к", + lo=lo, + hi=hi, + seen={}, + price_cap_per_bucket=1400, + max_pages_per_bucket=100, + concurrency=5, + secondary_only=True, + on_bucket=None, + skip_buckets=None, + ) + return events + + old_events = await run(_old_avito()) + new_events = await run(NewAvito(config=_KIT_CONFIG)) + assert old_events == new_events + # Открытый брекет с probe-fail → ровно один best-effort leaf на (25M, None). + assert ("leaf", 25_000_000, None, 100) in old_events + + +# ── domclick: count-endpoint + None-both-ends + split открытого корня ───────── + + +@pytest.mark.asyncio +async def test_bisection_parity_domclick() -> None: + """domclick: unbounded-корень (None,None) клампится к [0,1e9] и делится; + count-endpoint; span<=MIN_PRICE_SPAN строго → leaf. Обход идентичен старому.""" + oracle = make_oracle(divisor=100_000, open_ceiling=1_000_000_000) + + async def run(scraper: Any) -> list[Any]: + events: list[Any] = [] + _patch_domclick(scraper, oracle, events) + await scraper._sweep_bucket( + fetcher=None, + rooms="1", + price_gte=None, + price_lte=None, + seen_ids=set(), + out_lots=[], + pages=100, + ) + return events + + old_events = await run(_old_domclick()) + new_events = await run(NewDomClick(_make_domclick_config())) + assert old_events == new_events + assert sum(1 for e in old_events if e[0] == "leaf") >= 4 + # Корень (None,None) обязательно пробится первым. + assert old_events[0] == ("probe", None, None) + + +def _make_domclick_config() -> Any: + return SimpleNamespace(browser_http_endpoint="http://tradein-browser:9000") + + +# ── yandex: max_depth=8, min_bracket=500k, count-based on_bucket ────────────── + + +@pytest.mark.asyncio +async def test_bisection_parity_yandex_normal() -> None: + """yandex: split-дерево (порог 500k, cap 500) + leaf-on-bucket + zero-as-leaf. + + Оракул divisor=1000, cap=500 → 4М→2М→1М→500k-leaf. Один под-брекет=0 + (yandex вызывает on_bucket даже на нуле — treat_zero_as_leaf). + """ + oracle = make_oracle( + divisor=1000, + open_ceiling=30_000_000, + zeros=frozenset({(2_000_000, 2_999_999)}), + ) + brackets: list[tuple[int, int | None]] = [(0, 3_999_999), (25_000_000, None)] + + async def run(scraper: Any) -> list[Any]: + events: list[Any] = [] + _patch_yandex(scraper, oracle, events) + for lo, hi in brackets: + await scraper._walk_price_range( + rooms="1", + lo=lo, + hi=hi, + seen={}, + price_cap_per_bucket=500, + max_pages_per_bucket=50, + concurrency=4, + on_bucket=_yandex_on_bucket(events), + skip_buckets=None, + ) + return events + + old_events = await run(_old_yandex()) + new_events = await run(NewYandex(config=_KIT_CONFIG)) + assert old_events == new_events + assert sum(1 for e in old_events if e[0] == "bucket") >= 4 + assert sum(1 for e in old_events if e[0] == "probe") >= 6 + + +@pytest.mark.asyncio +async def test_bisection_parity_yandex_degraded() -> None: + """yandex probe-fail (totalItems неизвестен) → degraded paginate-until-empty. + + Оракул форсит None на бакете → probe page1 payload без totalItems → total=None + → degraded последовательно пагинирует. record_pages=True фиксирует все страницы. + """ + oracle = make_oracle( + divisor=1000, + open_ceiling=30_000_000, + nones=frozenset({(None, 3_999_999)}), + ) + + async def run(scraper: Any) -> list[Any]: + events: list[Any] = [] + _patch_yandex(scraper, oracle, events, record_pages=True) + await scraper._walk_price_range( + rooms="1", + lo=0, + hi=3_999_999, + seen={}, + price_cap_per_bucket=500, + max_pages_per_bucket=50, + concurrency=4, + on_bucket=_yandex_on_bucket(events), + skip_buckets=None, + ) + return events + + old_events = await run(_old_yandex()) + new_events = await run(NewYandex(config=_KIT_CONFIG)) + assert old_events == new_events + # degraded: probe page1 + затем последовательная пагинация страниц 1..K до пустой. + assert ("probe", None, 3_999_999) in old_events + assert any(e[0] == "bucket" for e in old_events) + + +# ── Full seed-grid обход (все seed-брекеты одной комнатности) ───────────────── + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "old_factory, new_factory, patch_fn, walk_extra", + [ + (_old_avito, lambda: NewAvito(config=_KIT_CONFIG), _patch_avito, _AVITO_WALK_EXTRA), + (_old_cian, lambda: NewCian(config=_KIT_CONFIG), _patch_cian, _CIAN_WALK_EXTRA), + ], +) +async def test_bisection_parity_full_seed_grid( + old_factory: Any, new_factory: Any, patch_fn: Any, walk_extra: dict[str, Any] +) -> None: + """Полный обход всех seed-брекетов (как в fetch_all_secondary) — включая + открытый catch-all — идентичен между старым и новым для avito/cian.""" + oracle = make_oracle(divisor=1500, open_ceiling=300_000_000) + + async def run(scraper: Any) -> list[Any]: + events: list[Any] = [] + patch_fn(scraper, oracle, events) + for br_lo, br_hi in get_price_seed_brackets(): + walk_hi = br_hi - 1 if br_hi is not None else None + await scraper._walk_price_range( + lo=br_lo, + hi=walk_hi, + seen={}, + price_cap_per_bucket=1400, + max_pages_per_bucket=100, + concurrency=5, + secondary_only=True, + on_bucket=None, + skip_buckets=None, + **walk_extra, + ) + return events + + old_events = await run(old_factory()) + new_events = await run(new_factory()) + assert old_events == new_events 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 new file mode 100644 index 00000000..555e3d81 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/__init__.py @@ -0,0 +1,21 @@ +"""scraper_kit.pricing — общие примитивы ценового партиционирования SERP-обхода. + +Единый движок адаптивной бисекции «комнатность × цена» для всех провайдеров +(avito / cian / yandex / domclick). См. `bisection.py`. +""" + +from __future__ import annotations + +from scraper_kit.pricing.bisection import ( + BisectionConfig, + ProbeFailPolicy, + ProbeResult, + walk_price_range, +) + +__all__ = [ + "BisectionConfig", + "ProbeFailPolicy", + "ProbeResult", + "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 new file mode 100644 index 00000000..017b89b4 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/bisection.py @@ -0,0 +1,253 @@ +"""Единый движок адаптивной бисекции «комнатность × цена» для SERP-обхода. + +Все четыре провайдера вторички (avito / cian / yandex / domclick) обходят +SERP-cap площадки одинаковым по СКЕЛЕТУ приёмом: для каждой комнатности берётся +ценовой диапазон [lo, hi], запрашивается «сколько всего результатов» (probe +page=1 или отдельный count-endpoint), и если результатов больше потолка бакета — +диапазон делится пополам (`mid = (lo + hi) // 2`) и обход рекурсивно спускается +в обе половины; иначе бакет пагинируется как leaf. + +Исторически это было ЧЕТЫРЕ независимые копии (`_walk_price_range` ×3 + +`_sweep_bucket`), разошедшиеся в константах и краевых ветках. Здесь скелет +рекурсии вынесен в одну функцию `walk_price_range`, а РАЗЛИЧИЯ провайдеров +переданы в `BisectionConfig` (пороги, глубина, поведение на открытом брекете и +при неудачном probe) + три provider-callback'а. Ни одно различие не потеряно — +golden-parity тесты доказывают, что обход байт-в-байт совпадает со старой +боевой реализацией для каждого провайдера. + +Provider-side (в callback'ах, движок их НЕ трогает): + * фактический HTTP-fetch + retry/rotation; + * парсинг счётчика (totalOffers / page-title-count / totalItems / snippetsCount); + * пагинация leaf-бакета, дедуп (`seen`), on_bucket-callback. + +Engine-side (здесь): + * дерево рекурсии: какие [lo, hi] пробятся / делятся / пагинируются и в каком + порядке; + * арифметика split-решения (cap, min_bracket, max_depth, clamp None-границ); + * применение probe-fail-политики и обработка открытого (hi=None) брекета. + +Семантические различия провайдеров (задокументировано в конфигах-инстансах на +стороне провайдеров), которые движок воспроизводит через `BisectionConfig`: + +| поле | avito | cian | yandex | domclick | +|-------------------------|-----------|-----------|---------------|-----------------| +| cap | 1400 | 1400 | 500 | 2000 (OFFSET) | +| min_bracket | 50_000 | 50_000 | 500_000 | 100_000 | +| min_bracket_inclusive | True (>=) | True (>=) | True (>=) | False (>) | +| max_depth | None | None | 8 | None | +| probe_fail_policy | SPLIT/SKIP| SKIP | DEGRADE | SKIP (не рвётся)| +| open_split_ceiling | None(leaf)| None(leaf)| None(leaf) | 1_000_000_000 | +| treat_zero_as_leaf | False | False | True | False | + +Различие «probe vs count»: avito/cian/yandex читают счётчик из probe page=1 и +ПЕРЕИСПОЛЬЗУЮТ HTML/лоты страницы 1 в leaf (payload); domclick дёргает отдельный +count-endpoint (payload=None, пагинация с offset=0). Это инкапсулировано в +provider-callback `probe` + `ProbeResult.payload`, движку прозрачно. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from enum import Enum, auto +from typing import Any + + +class ProbeFailPolicy(Enum): + """Что делать, когда probe не вернул счётчик (`ProbeResult.count is None`). + + * ``SKIP`` — бросить бакет целиком (cian: totalOffers недоступен → пропуск). + * ``SPLIT_OR_SKIP`` — открытый брекет пагинируем best-effort; закрытый ещё + делимый (span >= min_bracket) принудительно делим (тяжёлая страница станет + загружаемой в узком диапазоне); иначе пропуск (avito). + * ``DEGRADE`` — деградировать в «пагинировать-до-пустоты» без знания total + (yandex): вызывается ``on_degraded``. + + domclick счётчик-endpoint при ошибке возвращает 0 (а не None), поэтому его + политика формально ``SKIP``, но ветка probe-fail недостижима (count==0 → + ранний return). + """ + + SKIP = auto() + SPLIT_OR_SKIP = auto() + DEGRADE = auto() + + +@dataclass +class ProbeResult: + """Результат probe одного бакета. + + * ``count`` — сколько всего результатов в [lo, hi] (или ``None`` при + captcha/ошибке probe). ``0`` — валидная пустая выборка. + * ``payload`` — provider-specific данные страницы 1 для переиспользования в + leaf-пагинации (avito: HTML; cian: HTML; yandex: list[ScrapedLot] page 1; + domclick: None — count-endpoint без карточек). + """ + + count: int | None + payload: Any = None + + +@dataclass(frozen=True) +class BisectionConfig: + """Per-provider константы/политики бисекции (все различия провайдеров тут). + + Args: + cap: потолок результатов в бакете; ``count > cap`` — кандидат на split. + min_bracket: минимальный размах диапазона, ниже которого делить нельзя. + min_bracket_inclusive: split разрешён при ``span >= min_bracket`` (True, + avito/cian/yandex) или строго ``span > min_bracket`` (False, + domclick — исходное `span <= MIN_PRICE_SPAN → paginate`). + max_depth: жёсткий предел глубины рекурсии (yandex=8) либо ``None``. + probe_fail_policy: см. :class:`ProbeFailPolicy`. + open_split_ceiling: если ``None`` (avito/cian/yandex) — открытый брекет + (hi=None) НЕ делится, пагинируется напрямую как leaf. Если задан + (domclick=1e9) — открытые границы клампятся к + ``[open_split_floor, open_split_ceiling]`` и диапазон делится. + open_split_floor: нижний кламп для None-lo при делении (domclick=0). + treat_zero_as_leaf: если True (yandex) — при ``count == 0`` всё равно + вызвать ``on_leaf`` (yandex on_bucket-callback срабатывает и на 0); + иначе (avito/cian/domclick) — ранний return без leaf. + """ + + cap: int + min_bracket: int + max_depth: int | None = None + min_bracket_inclusive: bool = True + probe_fail_policy: ProbeFailPolicy = ProbeFailPolicy.SKIP + open_split_ceiling: int | None = None + open_split_floor: int = 0 + treat_zero_as_leaf: bool = False + + +# Provider-callback типы. +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]] + + +def _can_split_span(span: int, config: BisectionConfig) -> bool: + """Разрешает ли размах диапазона дальнейшее деление (по min_bracket).""" + if config.min_bracket_inclusive: + return span >= config.min_bracket + return span > config.min_bracket + + +async def walk_price_range( + *, + lo: int | None, + hi: int | None, + config: BisectionConfig, + probe: ProbeFn, + on_leaf: LeafFn, + on_degraded: DegradedFn | None = None, + depth: int = 0, +) -> None: + """Рекурсивно обойти ценовой диапазон [lo, hi] адаптивной бисекцией. + + Скелет (общий для всех провайдеров): + 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``). + 4. Открытый брекет (hi=None) при ``open_split_ceiling is None`` → leaf. + 5. Иначе: если ``count > cap`` и размах делим и глубина в пределах — + ``mid = (lo + hi) // 2``, рекурсия в обе половины; иначе leaf. + + Args: + lo / hi: границы диапазона (``None`` = без границы; клампится при делении + для провайдеров с ``open_split_ceiling``). + config: per-provider пороги/политики. + probe: async ``(lo, hi, depth) -> ProbeResult`` — счётчик + payload. + on_leaf: async ``(lo, hi, result) -> None`` — пагинация leaf-бакета + (fetch/parse/дедуп/on_bucket целиком на стороне провайдера). + on_degraded: async ``(lo, hi) -> None`` — обязателен для + ``ProbeFailPolicy.DEGRADE`` (yandex «пагинировать-до-пустоты»). + depth: текущая глубина рекурсии (для ``max_depth``). + """ + result = await probe(lo, hi, depth) + total = result.count + + # ── Probe-fail: счётчик недоступен ──────────────────────────────────────── + if total is None: + policy = config.probe_fail_policy + if policy is ProbeFailPolicy.SKIP: + return + if policy is ProbeFailPolicy.DEGRADE: + if on_degraded is not None: + await on_degraded(lo, hi) + return + # SPLIT_OR_SKIP (avito): открытый брекет пагинируем best-effort; закрытый + # ещё делимый — принудительно делим; иначе пропускаем (хвост теряем). + if hi is None: + await on_leaf(lo, hi, result) + return + _lo = lo if lo is not None else config.open_split_floor + if _can_split_span(hi - _lo, config): + mid = (_lo + hi) // 2 + await walk_price_range( + lo=_lo, + hi=mid, + config=config, + probe=probe, + on_leaf=on_leaf, + on_degraded=on_degraded, + depth=depth + 1, + ) + await walk_price_range( + lo=mid + 1, + hi=hi, + config=config, + probe=probe, + on_leaf=on_leaf, + on_degraded=on_degraded, + depth=depth + 1, + ) + return + + # ── Счётчик получен ─────────────────────────────────────────────────────── + if total == 0 and not config.treat_zero_as_leaf: + return + + # Открытый брекет без split-потолка (avito/cian/yandex): делить нельзя — + # пагинируем leaf напрямую (хвост люкса крошечный, tail-loss accepted). + if hi is None and config.open_split_ceiling is None: + await on_leaf(lo, hi, result) + return + + # Эффективные границы для арифметики деления: None-границы клампятся к + # [open_split_floor, open_split_ceiling] (domclick unbounded root). + eff_lo = lo if lo is not None else config.open_split_floor + eff_hi = hi if hi is not None else config.open_split_ceiling + + can_split = ( + total > config.cap + and eff_hi is not None + and _can_split_span(eff_hi - eff_lo, config) + and (config.max_depth is None or depth < config.max_depth) + ) + + if can_split: + assert eff_hi is not None # для mypy: гарантировано can_split + mid = (eff_lo + eff_hi) // 2 + await walk_price_range( + lo=eff_lo, + hi=mid, + config=config, + probe=probe, + on_leaf=on_leaf, + on_degraded=on_degraded, + depth=depth + 1, + ) + await walk_price_range( + lo=mid + 1, + hi=eff_hi, + config=config, + probe=probe, + on_leaf=on_leaf, + on_degraded=on_degraded, + depth=depth + 1, + ) + return + + # Leaf-бакет: пагинация + дедуп + on_bucket — целиком на стороне провайдера. + await on_leaf(lo, hi, result) 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 3587416d..4c0c8bf7 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 @@ -46,6 +46,7 @@ from scraper_kit.avito_exceptions import ( from scraper_kit.base import BaseScraper, ScrapedLot from scraper_kit.browser_fetcher import BrowserFetcher from scraper_kit.price_brackets import get_price_seed_brackets +from scraper_kit.pricing import BisectionConfig, ProbeFailPolicy, ProbeResult, walk_price_range from scraper_kit.providers.avito.shared import _unix_to_date if TYPE_CHECKING: @@ -250,6 +251,21 @@ _AVITO_PRICE_SEED_BRACKETS: list[tuple[int, int | None]] = get_price_seed_bracke # Avito SERP показывает ~50 карточек на страницу. _AVITO_OFFERS_PER_PAGE = 50 + +def _avito_bisection_config(cap: int) -> BisectionConfig: + """Конфиг единого движка бисекции (scraper_kit.pricing) для avito. + + probe_fail=SPLIT_OR_SKIP: probe вернул None (captcha/browser-timeout) → открытый + брекет пагинируем best-effort, закрытый ещё делимый (bracket >= MIN) принудительно + делим (тяжёлая выдача станет загружаемой в узком диапазоне), иначе пропуск. + open_split_ceiling=None: открытый брекет (hi=None) пагинируется напрямую, не делится. + """ + return BisectionConfig( + cap=cap, + min_bracket=_AVITO_MIN_BRACKET, + probe_fail_policy=ProbeFailPolicy.SPLIT_OR_SKIP, + ) + # HTTP 429 в curl_cffi-режиме через backconnect-прокси (mproxy.site) — НЕ IP-ban, а # transient «слишком много одновременных соединений» (лимит 5). Проходит на коротком # retry без ротации IP. Делаем до _AVITO_429_MAX_RETRIES коротких пауз; если они @@ -370,7 +386,9 @@ class AvitoScraper(BaseScraper): # FALLBACK — если браузер отдаёт firewall/пусто, _fetch_serp_html делает # один retry через curl_cffi. _build_cffi_session самодостаточен # (отдельный AsyncSession, не зависит от BaseScraper.__aenter__). - self._browser = BrowserFetcher(source="avito") + self._browser = BrowserFetcher( + source="avito", endpoint=self._config.browser_http_endpoint + ) await self._browser.__aenter__() self._cffi = self._build_cffi_session() logger.info( @@ -1155,145 +1173,89 @@ class AvitoScraper(BaseScraper): secondary_only: новостройки (listing_segment=="novostroyki") отбрасываются после сбора bucket_lots, до дедупа в seen и вызова on_bucket. AvitoBlockedError/AvitoRateLimitedError из _fetch_serp_html пробрасываются наверх. + + Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Avito-spec + probe (page=1 + IP-rotate retry) и leaf-пагинация переданы callback'ами; + пороги (cap, _AVITO_MIN_BRACKET) и политика probe-fail (SPLIT_OR_SKIP — + принудительный split тяжёлой выдачи / best-effort для открытого) — в + BisectionConfig. AvitoBlockedError/AvitoRateLimitedError из probe/leaf + всплывают наверх (mark_banned). """ - _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 - # на тяжёлой выдаче, напр. студии pmax=200M) → total=None вместо краха. - # AvitoBlockedError/AvitoRateLimitedError всплывают наверх (mark_banned). - html, total = await self._probe_total(room_slug, _lo_param, hi) - - # Ретрай на captcha/missing-count/browser-ошибке: rotate IP + 1 retry. - if total is None: - logger.warning( - "avito: total=None for %s [%d, %s] depth=%d — rotating IP + retry", - room_label, - lo, - _hi_repr, - _depth, - ) - rotated = await self._rotate_ip() - if rotated: - 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: + async def _probe(plo: int | None, phi: int | None, depth: int) -> ProbeResult: + lo_param = plo if plo and plo > 0 else None + hi_repr = "open" if phi is None else str(phi) + # _probe_total ловит browser-ошибки (HTTP 500 / timeout от tradein-browser + # на тяжёлой выдаче) → total=None вместо краха. Block-исключения всплывают. + html, total = await self._probe_total(room_slug, lo_param, phi) + if total is None: logger.warning( - "avito: probe failed for OPEN bucket %s [%d, open] depth=%d — " + "avito: total=None for %s [%s, %s] depth=%d — rotating IP + retry", + room_label, + plo, + hi_repr, + depth, + ) + rotated = await self._rotate_ip() + if rotated: + html, total = await self._probe_total(room_slug, lo_param, phi) + if total is not None: + logger.info( + "avito: %s [%s, %s] total=%d depth=%d", + room_label, + plo, + hi_repr, + total, + depth, + ) + return ProbeResult(count=total, payload=html) + + async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None: + _lo = plo if plo is not None else 0 + total = result.count + if total is None: + # SPLIT_OR_SKIP на открытом брекете: probe провалился, делить нельзя — + # пагинируем best-effort до max_pages_per_bucket (tail-loss accepted). + html: str | None = None + max_pages = max_pages_per_bucket + logger.warning( + "avito: probe failed for OPEN bucket %s [%d, open] — " "paginating up to max_pages (tail-loss accepted)", room_label, - lo, - _depth, + _lo, ) - 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 не удался даже после ретрая. Если бакет ещё делим — НЕ теряем - # его целиком: тяжёлая страница (огромная выдача) скорее всего станет - # загружаемой в более узком ценовом диапазоне → принудительный split. - if bracket_size >= _AVITO_MIN_BRACKET: - logger.warning( - "avito: probe failed for %s [%d, %d] depth=%d — splitting (assume heavy)", - room_label, - lo, - hi, - _depth, - ) - mid = (lo + hi) // 2 - await self._walk_price_range( - room_slug=room_slug, - room_label=room_label, - lo=lo, - hi=mid, - 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, - _depth=_depth + 1, - ) - await self._walk_price_range( - room_slug=room_slug, - room_label=room_label, - lo=mid + 1, - hi=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, - _depth=_depth + 1, - ) - return - # Бакет уже узкий (< MIN_BRACKET) — split не поможет, хвост теряем - # осознанно (как cian) и логируем. - logger.error( - "avito: skipping bucket %s [%d, %d] — total unavailable after retry " - "(bracket=%d < MIN_BRACKET=%d, tail loss accepted)", - room_label, - lo, - hi, - bracket_size, - _AVITO_MIN_BRACKET, - ) - return - - logger.info( - "avito: %s [%d, %s] total=%d depth=%d", - room_label, - lo, - _hi_repr, - total, - _depth, - ) - - 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) + else: + if phi is None and 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) + if phi is not None and pages_needed > max_pages_per_bucket: + tail_loss = total - max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE + logger.warning( + "avito: %s [%d, %d] total=%d exceeds page cap " + "(max_pages=%d × %d=%d offers) — tail loss ~%d offers", + room_label, + _lo, + phi, + total, + max_pages_per_bucket, + _AVITO_OFFERS_PER_PAGE, + max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE, + tail_loss, + ) + html = result.payload await self._paginate_leaf_bucket( room_slug=room_slug, room_label=room_label, - lo=lo, - hi=None, + lo=_lo, + hi=phi, html=html, max_pages=max_pages, seen=seen, @@ -1304,95 +1266,14 @@ class AvitoScraper(BaseScraper): on_bucket=on_bucket, skip_buckets=skip_buckets, ) - return - # ── Шаг 2: деление или пагинация ────────────────────────────────────── - bracket_size = hi - lo - need_split = total > price_cap_per_bucket - too_narrow = bracket_size < _AVITO_MIN_BRACKET - - if need_split and too_narrow: - logger.warning( - "avito: %s [%d, %d] total=%d > cap=%d but bracket=%d < MIN_BRACKET=%d " - "— paginating as-is (tail loss ~%d)", - room_label, - lo, - hi, - total, - price_cap_per_bucket, - bracket_size, - _AVITO_MIN_BRACKET, - max(0, total - price_cap_per_bucket), - ) - need_split = False # принудительно пагинируем - - if need_split: - mid = (lo + hi) // 2 - # [lo, mid] - await self._walk_price_range( - room_slug=room_slug, - room_label=room_label, - lo=lo, - hi=mid, - 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, - _depth=_depth + 1, - ) - # [mid+1, hi] - await self._walk_price_range( - room_slug=room_slug, - room_label=room_label, - lo=mid + 1, - hi=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, - _depth=_depth + 1, - ) - return - - # ── Закрытый leaf-бакет: считаем max_pages + warning, делегируем helper'у ─ - pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE) - max_pages = min(pages_needed, max_pages_per_bucket) - - if pages_needed > max_pages_per_bucket: - tail_loss = total - max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE - logger.warning( - "avito: %s [%d, %d] total=%d exceeds page cap " - "(max_pages=%d × %d=%d offers) — tail loss ~%d offers", - room_label, - lo, - hi, - total, - max_pages_per_bucket, - _AVITO_OFFERS_PER_PAGE, - max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE, - tail_loss, - ) - - await self._paginate_leaf_bucket( - room_slug=room_slug, - room_label=room_label, + await walk_price_range( 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, + config=_avito_bisection_config(price_cap_per_bucket), + probe=_probe, + on_leaf=_leaf, + depth=_depth, ) async def _paginate_leaf_bucket( diff --git 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 index 59ed1c03..5c5679e7 100644 --- 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 @@ -39,6 +39,7 @@ from scraper_kit.browser_fetcher import BrowserFetcher 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.repair_state_normalizer import ( infer_repair_state_from_text, normalize_repair_state, @@ -69,6 +70,17 @@ _MIN_BRACKET = 50_000 # Cian SERP показывает ~28 офферов на страницу. _CIAN_OFFERS_PER_PAGE = 28 +# Конфиг единого движка бисекции (scraper_kit.pricing) для cian. cap задаётся +# per-call (price_cap_per_bucket) — здесь только структурные константы/политики. +# probe_fail=SKIP: totalOffers недоступен после ретрая → бакет пропускается. +# open_split_ceiling=None: открытый брекет пагинируется напрямую (не делится). +def _cian_bisection_config(cap: int) -> BisectionConfig: + return BisectionConfig( + cap=cap, + min_bracket=_MIN_BRACKET, + probe_fail_policy=ProbeFailPolicy.SKIP, + ) + # Комнатности по умолчанию для exhaustive-загрузки. # Коды Cian: room1=1к, room2=2к, room3=3к, room4=4к, room5=5к, room6=6+. # TODO: студии — Cian использует отдельный тип flat_type=studio, не room-param. @@ -123,7 +135,9 @@ class CianScraper(BaseScraper): # fingerprint через tradein-browser сервис (#1806). Сервер роутит /fetch на # отдельный браузер+мобильный прокси для cian. Прокси ротирует IP сам, поэтому # code-side warm-up cookies и changeip-ротация больше не нужны. - self._browser = BrowserFetcher(source="cian") + self._browser = BrowserFetcher( + source="cian", endpoint=self._config.browser_http_endpoint + ) await self._browser.__aenter__() return self @@ -398,76 +412,75 @@ class CianScraper(BaseScraper): skip_buckets: если bucket_key в skip_buckets — пагинацию и on_bucket пропускаем. secondary_only: если True — новостройки (listing_segment=="novostroyki") отбрасываются после сбора bucket_lots, до дедупа в seen и вызова on_bucket. + + Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Cian-spec + probe (fetch page 1 + retry) и leaf-пагинация переданы callback'ами; пороги + (cap, _MIN_BRACKET) и политика probe-fail (SKIP) — в BisectionConfig. """ room_label = f"room{'_'.join(str(r) for r in rooms)}" - _lo_param = lo if lo > 0 else None - _hi_repr = "open" if hi is None else str(hi) - # ── Шаг 1: probe page 1 ──────────────────────────────────────────────── - html = await self._fetch_page_html(rooms, 1, _lo_param, hi) - await self.sleep_between_requests() - - total: int | None = None - if html is not None: - total = self._extract_total_offers(html) - - # Ретрай на captcha/ошибку: 1 повторный fetch. Мобильный прокси в браузер- - # контейнере ротирует IP сам (#1806), поэтому отдельный changeip-вызов - # (_rotate_ip) больше не нужен — просто повторяем запрос на чистом IP. - if total is None: - logger.warning( - "cian: totalOffers=None for %s [%d, %s] depth=%d — retry (proxy auto-rotates IP)", - room_label, - lo, - _hi_repr, - _depth, - ) - await self._rotate_ip() # no-op (back-compat); прокси ротирует IP сам - html = await self._fetch_page_html(rooms, 1, _lo_param, hi) + async def _probe(plo: int | None, phi: int | None, depth: int) -> ProbeResult: + lo_param = plo if plo and plo > 0 else None + hi_repr = "open" if phi is None else str(phi) + # ── Шаг 1: probe page 1 ─────────────────────────────────────────── + html = await self._fetch_page_html(rooms, 1, lo_param, phi) await self.sleep_between_requests() - if html is not None: - total = self._extract_total_offers(html) + total = self._extract_total_offers(html) if html is not None else None - if total is None: - logger.error( - "cian: skipping bucket %s [%d, %s] — totalOffers unavailable after retry", - room_label, - lo, - _hi_repr, - ) - return - - logger.info( - "cian: %s [%d, %s] totalOffers=%d depth=%d", - room_label, - lo, - _hi_repr, - total, - _depth, - ) - - 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: + # Ретрай на captcha/ошибку: 1 повторный fetch. Мобильный прокси в браузер- + # контейнере ротирует IP сам (#1806), поэтому отдельный changeip-вызов + # (_rotate_ip) больше не нужен — просто повторяем запрос на чистом IP. + if total is None: logger.warning( - "cian: OPEN bucket %s [%d, open] totalOffers=%d > cap=%d — paginating " + "cian: totalOffers=None for %s [%s, %s] depth=%d — retry " + "(proxy auto-rotates IP)", + room_label, + plo, + hi_repr, + depth, + ) + await self._rotate_ip() # no-op (back-compat); прокси ротирует IP сам + html = await self._fetch_page_html(rooms, 1, lo_param, phi) + await self.sleep_between_requests() + if html is not None: + total = self._extract_total_offers(html) + + if total is None: + logger.error( + "cian: skipping bucket %s [%s, %s] — totalOffers unavailable after retry", + room_label, + plo, + hi_repr, + ) + else: + logger.info( + "cian: %s [%s, %s] totalOffers=%d depth=%d", + room_label, + plo, + hi_repr, + total, + depth, + ) + return ProbeResult(count=total, payload=html) + + async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None: + total = result.count + assert total is not None # SKIP-политика: leaf не вызывается при None + if phi is None and total > price_cap_per_bucket: + logger.warning( + "cian: OPEN bucket %s [%s, open] totalOffers=%d > cap=%d — paginating " "without split (tail-loss accepted, lux tail tiny)", room_label, - lo, + plo, total, price_cap_per_bucket, ) - await self._paginate_open_bucket( + await self._paginate_leaf_bucket( rooms=rooms, room_label=room_label, - lo=lo, - html=html, + lo=plo if plo is not None else 0, + hi=phi, + html=result.payload, total=total, seen=seen, max_pages_per_bucket=max_pages_per_bucket, @@ -476,105 +489,14 @@ class CianScraper(BaseScraper): on_bucket=on_bucket, skip_buckets=skip_buckets, ) - return - # ── Шаг 2: деление или пагинация (только ЗАКРЫТЫЙ брекет, hi: int) ───── - bracket_size = hi - lo - need_split = total > price_cap_per_bucket - too_narrow = bracket_size < _MIN_BRACKET - - if need_split and too_narrow: - logger.warning( - "cian: %s [%d, %d] totalOffers=%d > cap=%d but bracket=%d < MIN_BRACKET=%d " - "— paginating as-is (tail loss ~%d)", - room_label, - lo, - hi, - total, - price_cap_per_bucket, - bracket_size, - _MIN_BRACKET, - max(0, total - price_cap_per_bucket), - ) - need_split = False # принудительно пагинируем - - if need_split: - mid = (lo + hi) // 2 - # [lo, mid] - await self._walk_price_range( - rooms=rooms, - lo=lo, - hi=mid, - 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, - _depth=_depth + 1, - ) - # [mid+1, hi] - await self._walk_price_range( - rooms=rooms, - lo=mid + 1, - hi=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, - _depth=_depth + 1, - ) - return - - # Закрытый leaf-бакет — общая пагинация (см. _paginate_leaf_bucket). - await self._paginate_leaf_bucket( - rooms=rooms, - room_label=room_label, + await walk_price_range( lo=lo, hi=hi, - html=html, - total=total, - seen=seen, - 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_open_bucket( - self, - *, - rooms: tuple[int, ...], - room_label: str, - lo: int, - html: str | None, - total: int, - seen: dict[str, ScrapedLot], - max_pages_per_bucket: int, - concurrency: int, - secondary_only: bool, - on_bucket: Callable[..., Any] | None, - skip_buckets: set[str] | None, - ) -> None: - """Пагинация ОТКРЫТОГО leaf-бакета (hi=None, без maxprice).""" - await self._paginate_leaf_bucket( - rooms=rooms, - room_label=room_label, - lo=lo, - hi=None, - html=html, - total=total, - seen=seen, - max_pages_per_bucket=max_pages_per_bucket, - concurrency=concurrency, - secondary_only=secondary_only, - on_bucket=on_bucket, - skip_buckets=skip_buckets, + config=_cian_bisection_config(price_cap_per_bucket), + probe=_probe, + on_leaf=_leaf, + depth=_depth, ) async def _paginate_leaf_bucket( diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/domclick/serp.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/domclick/serp.py index 9525d3a1..890bb8a8 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/domclick/serp.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/domclick/serp.py @@ -37,6 +37,7 @@ from urllib.parse import urlencode from scraper_kit.base import BaseScraper, ScrapedLot from scraper_kit.domclick_exceptions import DomClickBlockedError +from scraper_kit.pricing import BisectionConfig, ProbeResult, walk_price_range from scraper_kit.repair_state_normalizer import infer_repair_state_from_text if TYPE_CHECKING: @@ -61,6 +62,24 @@ PAGE_SIZE: int = 20 # items per page (жёстко задан API) LTE_MAX: int = 1_000_000_000 MIN_PRICE_SPAN: int = 100_000 # ниже этого span прекращаем делить +# Конфиг единого движка бисекции (scraper_kit.pricing) для domclick. Отличия от +# avito/cian/yandex закодированы явно: +# * cap = OFFSET_CAP (2000) — потолок offset BFF API, не «результатов бакета»; +# * count через отдельный count-endpoint (probe.payload=None, пагинация с offset=0); +# * min_bracket_inclusive=False — исходное `span <= MIN_PRICE_SPAN → paginate` +# (делим только пока span СТРОГО > MIN_PRICE_SPAN); +# * open_split_ceiling=LTE_MAX / floor=0 — unbounded-бакет (price_gte/lte=None) +# клампится к [0, LTE_MAX] и ДЕЛИТСЯ (в отличие от открытого seed-брекета +# avito/cian/yandex, который пагинируется напрямую); +# * seed-брекетов нет — каждый ROOM_BUCKET стартует с одного (None, None). +_DOMCLICK_BISECTION = BisectionConfig( + cap=OFFSET_CAP, + min_bracket=MIN_PRICE_SPAN, + min_bracket_inclusive=False, + open_split_ceiling=LTE_MAX, + open_split_floor=0, +) + # ── QRATOR block detection ──────────────────────────────────────────────────── _QRATOR_MARKERS: tuple[str, ...] = ( @@ -406,49 +425,27 @@ class DomClickScraper(BaseScraper): Raises: DomClickBlockedError: если QRATOR-блок — propagate наверх. - """ - count = await self._count(fetcher, rooms, price_gte, price_lte) - if count == 0: - return - if count > OFFSET_CAP: - _gte = price_gte if price_gte is not None else 0 - _lte = price_lte if price_lte is not None else LTE_MAX - span = _lte - _gte - if span <= MIN_PRICE_SPAN: - logger.warning( - "domklik: rooms=%r price=[%s,%s] count=%d > cap=%d " - "but span=%d <= min=%d — paginating as-is (bucket truncated at %d)", - rooms, - price_gte, - price_lte, - count, - OFFSET_CAP, - span, - MIN_PRICE_SPAN, - OFFSET_CAP, - ) - await self._paginate( - fetcher, rooms, price_gte, price_lte, seen_ids, out_lots, pages - ) - return - mid = (_gte + _lte) // 2 - logger.debug( - "domklik: rooms=%r count=%d > cap=%d — price-split [%d,%d] → [%d,%d]+[%d,%d]", - rooms, - count, - OFFSET_CAP, - _gte, - _lte, - _gte, - mid, - mid + 1, - _lte, - ) - await self._sweep_bucket(fetcher, rooms, _gte, mid, seen_ids, out_lots, pages) - await self._sweep_bucket(fetcher, rooms, mid + 1, _lte, seen_ids, out_lots, pages) - else: - await self._paginate(fetcher, rooms, price_gte, price_lte, seen_ids, out_lots, pages) + Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Domclick-spec + count-endpoint (probe) и offset-пагинация (leaf) переданы callback'ами; отличия + (OFFSET_CAP, MIN_PRICE_SPAN строго-больше, clamp unbounded к [0, LTE_MAX] с + делением) закодированы в _DOMCLICK_BISECTION. + """ + + async def _probe(plo: int | None, phi: int | None, depth: int) -> ProbeResult: + count = await self._count(fetcher, rooms, plo, phi) + return ProbeResult(count=count, payload=None) + + async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None: + await self._paginate(fetcher, rooms, plo, phi, seen_ids, out_lots, pages) + + await walk_price_range( + lo=price_gte, + hi=price_lte, + config=_DOMCLICK_BISECTION, + probe=_probe, + on_leaf=_leaf, + ) async def _paginate( self, 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 c8621c08..e93b4985 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 @@ -48,8 +48,9 @@ from curl_cffi.requests import AsyncSession as _CurlCffiSession from scraper_kit.base import BaseScraper, ScrapedLot from scraper_kit.browser_fetcher import BrowserFetcher +from scraper_kit.house_type_normalizer import normalize_house_type from scraper_kit.price_brackets import get_price_seed_brackets -from scraper_kit.providers.yandex.shared import normalize_house_type +from scraper_kit.pricing import BisectionConfig, ProbeFailPolicy, ProbeResult, walk_price_range if TYPE_CHECKING: from scraper_kit.contracts import ScraperConfig @@ -90,7 +91,15 @@ DEFAULT_PRICE_RANGES: list[tuple[int | None, int | None]] = [ ] _YANDEX_MAX_PRICE = 200_000_000 +# ВНИМАНИЕ: реальный порог split у yandex — 500_000 (см. _YANDEX_SPLIT_MIN_BRACKET), +# а НЕ 50_000. Историческая _YANDEX_MIN_BRACKET=50_000 никогда не использовалась в +# _walk_price_range (там был локальный `_min_bracket = 500_000`), оставлена для +# обратной совместимости импортов; движок бисекции использует 500_000. _YANDEX_MIN_BRACKET = 50_000 +# Минимальный размах диапазона для деления (yandex делит только пока (hi-lo) >= 500k) +# и жёсткий предел глубины рекурсии — переносим в единый движок как есть. +_YANDEX_SPLIT_MIN_BRACKET = 500_000 +_YANDEX_MAX_SPLIT_DEPTH = 8 _GATE_PAGE_SIZE = 20 _YANDEX_TARPIT_MAX_RETRIES: int = 2 @@ -478,7 +487,9 @@ class YandexRealtyScraper(BaseScraper): на один тайм-аутовый запрос: worst-case 30s×(1+retries)=90s на combo вместо 360s. """ self._browser = BrowserFetcher( - source="yandex", fetch_timeout_s=_YANDEX_BROWSER_FETCH_TIMEOUT_S + source="yandex", + fetch_timeout_s=_YANDEX_BROWSER_FETCH_TIMEOUT_S, + endpoint=self._config.browser_http_endpoint, ) await self._browser.__aenter__() return self @@ -958,55 +969,66 @@ class YandexRealtyScraper(BaseScraper): split (no mid) -> paginate leaf directly (lux tail tiny, tail-loss accepted). Fallback: totalItems=None -> paginate-until-empty (safe degradation). skip_buckets: resume checkpoint -- skip leaf buckets already done. + + Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Yandex-spec + (probe gate-JSON page=1, degraded paginate-until-empty, leaf с first-wins + дедупом и count-based on_bucket) — в callback'ах; пороги (cap, 500k min, + depth<8) и политика DEGRADE — в BisectionConfig. """ - _lo_param = lo if lo > 0 else None - _hi_repr = "open" if hi is None else str(hi) - probe_payload = await self._fetch_page_json(rooms, 1, _lo_param, hi) - await asyncio.sleep(self.request_delay_sec) + async def _probe(plo: int | None, phi: int | None, depth: int) -> ProbeResult: + lo_param = plo if plo and plo > 0 else None + hi_repr = "open" if phi is None else str(phi) + probe_payload = await self._fetch_page_json(rooms, 1, lo_param, phi) + await asyncio.sleep(self.request_delay_sec) - total: int | None = None - probe_lots: list[ScrapedLot] = [] - - if probe_payload is None: - logger.warning( - "yandex gate: probe failed rooms=%s [%d, %s] depth=%d -- rotating + retry", - rooms, - lo, - _hi_repr, - _depth, - ) - rotated = await self._rotate_ip() - if rotated: - probe_payload = await self._fetch_page_json(rooms, 1, _lo_param, hi) - await asyncio.sleep(self.request_delay_sec) - - if probe_payload is not None: - result = _extract_gate_data(probe_payload) - if result is not None: - _, pager = result - total = pager.get("totalItems") - probe_lots = _parse_gate_json(probe_payload, page_param=1) - else: + if probe_payload is None: logger.warning( - "yandex gate: probe data extract failed rooms=%s [%d, %s]", rooms, lo, _hi_repr + "yandex gate: probe failed rooms=%s [%s, %s] depth=%d -- rotating + retry", + rooms, + plo, + hi_repr, + depth, ) + rotated = await self._rotate_ip() + if rotated: + probe_payload = await self._fetch_page_json(rooms, 1, lo_param, phi) + await asyncio.sleep(self.request_delay_sec) - if total is None: - # Degraded path: paginate-until-empty without knowing total + total: int | None = None + probe_lots: list[ScrapedLot] = [] + if probe_payload is not None: + extracted = _extract_gate_data(probe_payload) + if extracted is not None: + _, pager = extracted + total = pager.get("totalItems") + probe_lots = _parse_gate_json(probe_payload, page_param=1) + else: + logger.warning( + "yandex gate: probe data extract failed rooms=%s [%s, %s]", + rooms, + plo, + hi_repr, + ) + return ProbeResult(count=total, payload=probe_lots) + + async def _degraded(plo: int | None, phi: int | None) -> None: + # totalItems неизвестен → пагинируем до пустоты, не зная total. + hi_repr = "open" if phi is None else str(phi) + lo_param = plo if plo and plo > 0 else None logger.warning( - "yandex gate: totalItems unknown rooms=%s [%d, %s] -- paginate until empty", + "yandex gate: totalItems unknown rooms=%s [%s, %s] -- paginate until empty", rooms, - lo, - _hi_repr, + plo, + hi_repr, ) - page = 1 - bucket_key = _combo_label(rooms, lo, hi) + bucket_key = _combo_label(rooms, plo, phi) if skip_buckets and bucket_key in skip_buckets: return + page = 1 pages_fetched = 0 while pages_fetched < max_pages_per_bucket: - payload = await self._fetch_page_json(rooms, page, _lo_param, hi) + payload = await self._fetch_page_json(rooms, page, lo_param, phi) await asyncio.sleep(self.request_delay_sec) if payload is None: break @@ -1020,84 +1042,72 @@ class YandexRealtyScraper(BaseScraper): pages_fetched += 1 if on_bucket is not None: on_bucket(bucket_key, len(seen)) - return - # Split or paginate. ОТКРЫТЫЙ брекет (hi is None) делить нельзя (нет mid) — - # сразу пагинируем leaf напрямую (хвост люкса крошечный, tail-loss accepted). - cap = price_cap_per_bucket - _min_bracket = 500_000 + async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None: + total = result.count + assert total is not None # DEGRADE-политика уводит None в _degraded + lo_param = plo if plo and plo > 0 else None + probe_lots: list[ScrapedLot] = result.payload or [] + bucket_key = _combo_label(rooms, plo, phi) + if skip_buckets and bucket_key in skip_buckets: + logger.debug("yandex gate: skip bucket %s (checkpoint)", bucket_key) + return - if hi is not None and total > cap and (hi - lo) >= _min_bracket and _depth < 8: - mid = (lo + hi) // 2 - await self._walk_price_range( - rooms=rooms, - lo=lo, - hi=mid, - seen=seen, - price_cap_per_bucket=cap, - max_pages_per_bucket=max_pages_per_bucket, - concurrency=concurrency, - on_bucket=on_bucket, - skip_buckets=skip_buckets, - _depth=_depth + 1, + total_pages = min( + math.ceil(total / 20), + _GATE_MAX_PAGES_CAP, + max_pages_per_bucket, ) - await self._walk_price_range( - rooms=rooms, - lo=mid + 1, - hi=hi, - seen=seen, - price_cap_per_bucket=cap, - max_pages_per_bucket=max_pages_per_bucket, - concurrency=concurrency, - on_bucket=on_bucket, - skip_buckets=skip_buckets, - _depth=_depth + 1, + logger.info( + "yandex gate: leaf bucket %s total=%d pages=%d", bucket_key, total, total_pages ) - return - # Leaf bucket: paginate - bucket_key = _combo_label(rooms, lo, hi) - if skip_buckets and bucket_key in skip_buckets: - 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, - ) - logger.info("yandex gate: leaf bucket %s total=%d pages=%d", bucket_key, total, total_pages) - - # Add probe lots (page 1 already fetched) - for lot in probe_lots: - if lot.source_id and lot.source_id not in seen: - seen[lot.source_id] = lot - - if total_pages <= 1: - if on_bucket is not None: - on_bucket(bucket_key, len(seen)) - return - - # Paginate pages 2..total_pages with concurrency - sem = asyncio.Semaphore(concurrency) - - async def _fetch_leaf_page(pg: int) -> list[ScrapedLot]: - async with sem: - payload = await self._fetch_page_json(rooms, pg, _lo_param, hi) - await asyncio.sleep(self.request_delay_sec) - if payload is None: - return [] - return _parse_gate_json(payload, page_param=pg) - - tasks = [_fetch_leaf_page(pg) for pg in range(2, total_pages + 1)] - results = await asyncio.gather(*tasks) - for page_lots in results: - for lot in page_lots: + # Add probe lots (page 1 already fetched) + for lot in probe_lots: if lot.source_id and lot.source_id not in seen: seen[lot.source_id] = lot - if on_bucket is not None: - on_bucket(bucket_key, len(seen)) + if total_pages <= 1: + if on_bucket is not None: + on_bucket(bucket_key, len(seen)) + return + + # Paginate pages 2..total_pages with concurrency + sem = asyncio.Semaphore(concurrency) + + async def _fetch_leaf_page(pg: int) -> list[ScrapedLot]: + async with sem: + payload = await self._fetch_page_json(rooms, pg, lo_param, phi) + await asyncio.sleep(self.request_delay_sec) + if payload is None: + return [] + return _parse_gate_json(payload, page_param=pg) + + tasks = [_fetch_leaf_page(pg) for pg in range(2, total_pages + 1)] + results = await asyncio.gather(*tasks) + for page_lots in results: + for lot in page_lots: + if lot.source_id and lot.source_id not in seen: + seen[lot.source_id] = lot + + if on_bucket is not None: + on_bucket(bucket_key, len(seen)) + + await walk_price_range( + lo=lo, + hi=hi, + config=BisectionConfig( + cap=price_cap_per_bucket, + min_bracket=_YANDEX_SPLIT_MIN_BRACKET, + max_depth=_YANDEX_MAX_SPLIT_DEPTH, + probe_fail_policy=ProbeFailPolicy.DEGRADE, + treat_zero_as_leaf=True, + ), + probe=_probe, + on_leaf=_leaf, + on_degraded=_degraded, + depth=_depth, + ) def _combo_label(rooms: str | None, lo: int | None, hi: int | None) -> str: diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/shared.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/shared.py index cb6e734e..d1a69efd 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/shared.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/shared.py @@ -1,95 +1,15 @@ -"""Нормализация house_type → каноничный enum для listings.house_type. +"""Re-export `normalize_house_type` из core-слоя scraper_kit (#2134 dedup). -Strangler-копия `app.services.scrapers.house_type_normalizer` (#2133). Локальная в -yandex-провайдере, потому что house_type_normalizer ещё не перенесён в core-слой -scraper_kit (это отдельный шаг консолидации E). Копия — pure-функция без `app.*` -зависимостей; идентична источнику. - -Целевой канон (сверено с estimator._IMV_HOUSE_TYPE_MAP, estimator.py:146): - panel / brick / monolith / monolith_brick / block / wood. - -Зачем (#2007): estimator применяет soft-penalty по house_type — аналог с -house_type != target штрафуется. Yandex SERP отдаёт SCREAMING-значения -(MONOLIT / BRICK / PANEL / ...), которые НИКОГДА не равны каноничным -(monolith / brick / panel / ...) → ~70% yandex-аналогов получали ложный штраф -и фактически выпадали из скоринга. Нормализация на ингесте чинит это. - -Важно про None: неизвестное / 'other' / '' → None, НЕ 'other'. В estimator -`house_type IS NULL` нейтрально (без штрафа), а любое не-канон значение всегда -!= target → ложный штраф. Поэтому unknown лучше схлопнуть в NULL. - -Источники raw-значений: -- yandex SERP: building.buildingType — SCREAMING_SNAKE (MONOLIT, MONOLIT_BRICK, ...) -- cian SERP: building.materialType — camelCase (monolith, monolithBrick, - gasSilicateBlock, ...) — переиспользуется в #2008 (cian.py:815). +Раньше здесь лежала ПОЛНАЯ копия нормализатора house_type (strangler-шаг #2133). +Теперь консолидировано: единственная реализация — `scraper_kit.house_type_normalizer` +(идентична бывшей копии, сверено картой _RAW_TO_CANON). Модуль сохранён как тонкий +re-export, чтобы не ломать существующие импорты +(`from scraper_kit.providers.yandex.shared import normalize_house_type`) и +golden-parity тест. """ from __future__ import annotations -import logging +from scraper_kit.house_type_normalizer import normalize_house_type -logger = logging.getLogger(__name__) - -# Канон listings.house_type — выровнен по estimator._IMV_HOUSE_TYPE_MAP. -_CANON: frozenset[str] = frozenset( - {"panel", "brick", "monolith", "monolith_brick", "block", "wood"} -) - -# raw-токен → канон. Ключи — точные значения вокабуляров источников; сравнение -# регистронезависимое (см. _LOOKUP ниже), но строго по полному токену енума, -# без fuzzy-матчинга. Неизвестные токены сюда НЕ попадают → normalize вернёт None. -_RAW_TO_CANON: dict[str, str] = { - # ── yandex SERP (SCREAMING_SNAKE buildingType) ─────────────────────────── - "MONOLIT": "monolith", - "BRICK": "brick", - "PANEL": "panel", - "MONOLIT_BRICK": "monolith_brick", - "BLOCK": "block", - "WOOD": "wood", - # ── cian SERP (camelCase materialType) — reuse в #2008 ─────────────────── - "monolith": "monolith", - "brick": "brick", - "panel": "panel", - "block": "block", - "wood": "wood", - "monolithBrick": "monolith_brick", - "gasSilicateBlock": "block", - "aerocreteBlock": "block", - "foamConcreteBlock": "block", - "stalin": "brick", # «сталинка» — кирпич -} - -# Регистронезависимый lookup. Лоуэркейс-ключи не коллизят между вокабулярами: -# 'MONOLIT'→'monolit' и 'monolith'→'monolith' — разные ключи. -_LOOKUP: dict[str, str] = {k.lower(): v for k, v in _RAW_TO_CANON.items()} - - -def normalize_house_type(raw: str | None) -> str | None: - """Преобразовать raw house_type в каноничный enum. - - Уже-каноничное значение возвращается as-is (идемпотентность — нужна при - ре-обработке и для backfill-миграции). Неизвестное / 'other' / пустое / - None → None (НЕ 'other': см. docstring модуля про estimator soft-penalty). - - Args: - raw: сырое значение из парсера (e.g. «MONOLIT», «monolithBrick») или None. - - Returns: - Одно из panel / brick / monolith / monolith_brick / block / wood, либо None. - """ - if raw is None: - return None - stripped = raw.strip() - if not stripped: - return None - # Pass-through: уже каноничное значение (idempotency — в т.ч. 'monolith_brick', - # которого нет среди raw-ключей карты). - if stripped in _CANON: - return stripped - # Регистронезависимый exact-token lookup по обоим вокабулярам. - canon = _LOOKUP.get(stripped) or _LOOKUP.get(stripped.lower()) - if canon is None: - # Неизвестное / 'other' — нормально (NULL нейтрально для estimator). - # debug, не warning: 'other' встречается массово, warning засорил бы лог. - logger.debug("house_type_normalizer: unmapped raw value %r — stored as NULL", raw) - return canon +__all__ = ["normalize_house_type"]