refactor(scraper-kit): unify price bisection engine across providers + dedup (#2134) #2168
8 changed files with 1069 additions and 610 deletions
455
tradein-mvp/backend/tests/test_scraper_kit_bisection_parity.py
Normal file
455
tradein-mvp/backend/tests/test_scraper_kit_bisection_parity.py
Normal file
|
|
@ -0,0 +1,455 @@
|
|||
"""Golden-parity бисекции: старый боевой обход `app.services.scrapers.<provider>`
|
||||
≡ новый единый движок `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
|
||||
|
|
@ -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",
|
||||
]
|
||||
|
|
@ -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)
|
||||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue