feat(tradein): exhaustive Yandex realty load — adaptive room×price + concurrency
This commit is contained in:
parent
5e94d558b5
commit
830657c1ab
6 changed files with 854 additions and 2 deletions
|
|
@ -29,6 +29,7 @@ from app.services.scrape_pipeline import (
|
|||
run_cian_full_load,
|
||||
run_n1_city_sweep,
|
||||
run_yandex_city_sweep,
|
||||
run_yandex_full_load,
|
||||
)
|
||||
from app.services.scraper_settings import invalidate_cache
|
||||
from app.services.scrapers.avito import AvitoScraper
|
||||
|
|
@ -1042,6 +1043,93 @@ def cancel_cian_full_load(
|
|||
return {"ok": True, "run_id": run_id, "cancelled": cancelled}
|
||||
|
||||
|
||||
# ── Yandex exhaustive full load (room × price partitioning) ──────────────────
|
||||
|
||||
|
||||
class YandexFullLoadRequest(BaseModel):
|
||||
price_cap_per_bucket: int = Field(
|
||||
default=500,
|
||||
ge=100,
|
||||
le=575,
|
||||
description=(
|
||||
"Максимум офферов на price-бакет. При totalItems > cap — бакет делится пополам. "
|
||||
"Ниже Yandex SERP-cap ~575; запас на variance. "
|
||||
"При недоступности totalItems — fallback на paginate-until-empty."
|
||||
),
|
||||
)
|
||||
request_delay_sec: float = Field(
|
||||
default=2.0,
|
||||
ge=1.0,
|
||||
le=15.0,
|
||||
description=(
|
||||
"Задержка между SERP-запросами (сек). "
|
||||
"Рекомендуется ≥ 2.0 — Yandex агрессивнее Cian по captcha-детекции."
|
||||
),
|
||||
)
|
||||
concurrency: int = Field(
|
||||
default=4,
|
||||
ge=1,
|
||||
le=8,
|
||||
description="Параллельных page-фетчей внутри одного leaf-бакета.",
|
||||
)
|
||||
|
||||
|
||||
@router.post("/scrape/yandex-full-load", response_model=CitySweepStartResponse)
|
||||
async def start_yandex_full_load(
|
||||
payload: YandexFullLoadRequest,
|
||||
background_tasks: BackgroundTasks,
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
) -> CitySweepStartResponse:
|
||||
"""Запустить exhaustive Yandex ЕКБ full load в background.
|
||||
|
||||
Обходит Yandex SERP-cap (~575 карточек/запрос): адаптивно бьёт цену на бакеты
|
||||
внутри каждой комнатности, пагинирует каждый бакет полностью, дедуп по source_id.
|
||||
При недоступности totalItems из state — автоматический fallback на paginate-until-empty.
|
||||
|
||||
Один региональный проход БЕЗ anchor'ов (Yandex SERP path-based по городу — anchors
|
||||
избыточны). IP-ротация через changeip при captcha.
|
||||
|
||||
Coop cancel: POST /scrape/yandex-full-load/{run_id}/cancel.
|
||||
Статус прогона: смотреть через GET /scrape/cian-city-sweep/runs
|
||||
(source='yandex_full_load').
|
||||
"""
|
||||
run_id = runs_mod.create_run(db, source="yandex_full_load", params=payload.model_dump())
|
||||
|
||||
async def _full_load_task() -> None:
|
||||
task_db = SessionLocal()
|
||||
try:
|
||||
await run_yandex_full_load(
|
||||
task_db,
|
||||
run_id=run_id,
|
||||
price_cap_per_bucket=payload.price_cap_per_bucket,
|
||||
request_delay_sec=payload.request_delay_sec,
|
||||
concurrency=payload.concurrency,
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("yandex-full-load background task run_id=%d crashed", run_id)
|
||||
finally:
|
||||
task_db.close()
|
||||
|
||||
background_tasks.add_task(_full_load_task)
|
||||
logger.info("yandex-full-load queued run_id=%d params=%s", run_id, payload.model_dump())
|
||||
return CitySweepStartResponse(
|
||||
run_id=run_id,
|
||||
status="running",
|
||||
pages_per_anchor=0, # нет anchor'ов в full load
|
||||
detail_top_n=0,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/scrape/yandex-full-load/{run_id}/cancel")
|
||||
def cancel_yandex_full_load(
|
||||
run_id: int,
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
) -> dict[str, object]:
|
||||
"""Отменить running Yandex full load. Cooperative: проверяется каждый room-bucket."""
|
||||
cancelled = runs_mod.mark_cancelled(db, run_id)
|
||||
return {"ok": True, "run_id": run_id, "cancelled": cancelled}
|
||||
|
||||
|
||||
# ── Yandex city sweep (#861) — on-demand trigger + runs list ─────────────────
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -213,6 +213,20 @@ class Settings(BaseSettings):
|
|||
"""Прокси для Cian-скраперов. CIAN_PROXY_URL > scraper_proxy_url (fallback)."""
|
||||
return self.cian_proxy_url_env or self.scraper_proxy_url
|
||||
|
||||
# ── Yandex dedicated mobile proxy (separate egress from Avito/Cian) ────────
|
||||
# Отдельный прокси для Yandex устраняет contention при параллельных прогонах.
|
||||
# Если не задан — fallback на общий scraper_proxy_url (backward-compat).
|
||||
# ENV: YANDEX_PROXY_URL.
|
||||
yandex_proxy_url_env: str | None = Field(default=None, validation_alias="YANDEX_PROXY_URL")
|
||||
# changeip-ссылка для Yandex-прокси (ротация IP при капче/таймауте). Если не задан —
|
||||
# fallback на avito_proxy_rotate_url. ENV: YANDEX_PROXY_ROTATE_URL.
|
||||
yandex_proxy_rotate_url: str | None = None
|
||||
|
||||
@property
|
||||
def yandex_proxy_url(self) -> str | None:
|
||||
"""Прокси для Yandex-скраперов. YANDEX_PROXY_URL > scraper_proxy_url (fallback)."""
|
||||
return self.yandex_proxy_url_env or self.scraper_proxy_url
|
||||
|
||||
# ── #759: deactivate stale avito listings ───────────────────────────────
|
||||
# Avito-объявления, не виденные scraper'ом более avito_stale_ttl_days дней,
|
||||
# помечаются is_active=false ночной задачей deactivate_stale_avito_listings.
|
||||
|
|
|
|||
|
|
@ -879,7 +879,7 @@ async def run_yandex_city_sweep(
|
|||
await asyncio.sleep(enrich_delay)
|
||||
|
||||
logger.info(
|
||||
"yandex-sweep run_id=%d anchor %s: " "address enrich=%d/%d failed=%d",
|
||||
"yandex-sweep run_id=%d anchor %s: address enrich=%d/%d failed=%d",
|
||||
run_id,
|
||||
_a_name,
|
||||
counters.address_enriched,
|
||||
|
|
@ -1722,6 +1722,129 @@ async def run_cian_full_load(
|
|||
raise
|
||||
|
||||
|
||||
# ── Yandex exhaustive full load (region-wide, без anchor'ов) ──────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class YandexFullLoadCounters:
|
||||
"""Aggregate counters для run_yandex_full_load.
|
||||
|
||||
Без detail_* полей — Yandex full load не делает detail-обогащение.
|
||||
"""
|
||||
|
||||
unique_fetched: int = 0
|
||||
saved_inserted: int = 0
|
||||
saved_updated: int = 0
|
||||
errors_count: int = 0
|
||||
|
||||
def to_dict(self) -> dict[str, int]:
|
||||
return {f.name: getattr(self, f.name) for f in fields(self)}
|
||||
|
||||
|
||||
async def run_yandex_full_load(
|
||||
db: Session,
|
||||
*,
|
||||
run_id: int,
|
||||
price_cap_per_bucket: int = 500,
|
||||
request_delay_sec: float = 2.0,
|
||||
concurrency: int = 4,
|
||||
) -> YandexFullLoadCounters:
|
||||
"""Exhaustive региональный сбор Yandex ЕКБ вторички (БЕЗ anchor'ов).
|
||||
|
||||
Yandex SERP — path-based по городу, не geo-bbox. Anchor-loop избыточен
|
||||
(все anchor'ы дают одну и ту же городскую выдачу). Один проход с
|
||||
партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление)
|
||||
охватывает весь ЕКБ.
|
||||
|
||||
Параметры:
|
||||
price_cap_per_bucket: максимум офферов на price-бакет (< Yandex SERP-cap ~575).
|
||||
request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default).
|
||||
concurrency: параллельных page-фетчей в leaf-бакете.
|
||||
|
||||
Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора.
|
||||
Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket.
|
||||
"""
|
||||
from app.services.scrapers.yandex_realty import YandexRealtyScraper
|
||||
|
||||
counters = YandexFullLoadCounters()
|
||||
|
||||
def _on_bucket(lots: list) -> None: # type: ignore[type-arg]
|
||||
"""Инкрементальный save после каждого leaf-бакета."""
|
||||
if scrape_runs.is_cancelled(db, run_id):
|
||||
logger.info("yandex-full-load run_id=%d: cancel detected in on_bucket", run_id)
|
||||
raise RuntimeError("cancelled")
|
||||
if not lots:
|
||||
return
|
||||
inserted, updated = save_listings(db, lots, run_id=run_id)
|
||||
# save_listings вызывает db.commit() внутри — данные в БД сразу
|
||||
counters.saved_inserted += inserted
|
||||
counters.saved_updated += updated
|
||||
counters.unique_fetched += len(lots)
|
||||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||||
logger.info(
|
||||
"yandex-full-load run_id=%d: bucket saved ins=%d upd=%d total_unique=%d",
|
||||
run_id,
|
||||
inserted,
|
||||
updated,
|
||||
counters.unique_fetched,
|
||||
)
|
||||
|
||||
def _on_progress(unique_count: int) -> None:
|
||||
"""Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов)."""
|
||||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||||
|
||||
try:
|
||||
async with YandexRealtyScraper() as scraper:
|
||||
scraper.request_delay_sec = request_delay_sec
|
||||
|
||||
await scraper.fetch_all_secondary(
|
||||
price_cap_per_bucket=price_cap_per_bucket,
|
||||
concurrency=concurrency,
|
||||
on_bucket=_on_bucket,
|
||||
on_progress=_on_progress,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"yandex-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d",
|
||||
run_id,
|
||||
counters.unique_fetched,
|
||||
counters.saved_inserted,
|
||||
counters.saved_updated,
|
||||
)
|
||||
scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
|
||||
scrape_runs.mark_done(db, run_id, counters.to_dict())
|
||||
logger.info(
|
||||
"yandex-full-load run_id=%d done: unique=%d ins=%d upd=%d errors=%d",
|
||||
run_id,
|
||||
counters.unique_fetched,
|
||||
counters.saved_inserted,
|
||||
counters.saved_updated,
|
||||
counters.errors_count,
|
||||
)
|
||||
return counters
|
||||
|
||||
except RuntimeError as exc:
|
||||
# on_bucket кидает RuntimeError("cancelled") при cooperative cancel
|
||||
if str(exc) == "cancelled":
|
||||
logger.info(
|
||||
"yandex-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d",
|
||||
run_id,
|
||||
counters.unique_fetched,
|
||||
counters.saved_inserted,
|
||||
counters.saved_updated,
|
||||
)
|
||||
scrape_runs.mark_done(db, run_id, counters.to_dict())
|
||||
return counters
|
||||
logger.exception("yandex-full-load run_id=%d: fatal error", run_id)
|
||||
scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
|
||||
raise
|
||||
|
||||
except Exception as exc:
|
||||
logger.exception("yandex-full-load run_id=%d: fatal error", run_id)
|
||||
scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
|
||||
raise
|
||||
|
||||
|
||||
# ── Public re-exports ───────────────────────────────────────────
|
||||
__all__ = [
|
||||
"ANCHOR_TIMEOUT_SEC",
|
||||
|
|
@ -1733,9 +1856,11 @@ __all__ = [
|
|||
"PipelineCounters",
|
||||
"PipelineResult",
|
||||
"YandexCitySweepCounters",
|
||||
"YandexFullLoadCounters",
|
||||
"run_avito_city_sweep",
|
||||
"run_avito_pipeline",
|
||||
"run_cian_city_sweep",
|
||||
"run_cian_full_load",
|
||||
"run_yandex_city_sweep",
|
||||
"run_yandex_full_load",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -21,8 +21,11 @@ URL pattern (path-based, no geo radius):
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
import math
|
||||
import re
|
||||
from datetime import date # noqa: F401 (used in type hints / future helpers)
|
||||
from pathlib import Path
|
||||
|
|
@ -94,6 +97,34 @@ DEFAULT_PRICE_RANGES: list[tuple[int | None, int | None]] = [
|
|||
(25_000_000, None),
|
||||
]
|
||||
|
||||
# ── Константы exhaustive-загрузки ──────────────────────────────────────────────
|
||||
|
||||
# Верхняя граница цены при первом рекурсивном делении (нет явного hi).
|
||||
# 200 млн ₽ заведомо выше ТОПа ЕКБ — дальние бакеты дадут 0 офферов.
|
||||
_YANDEX_MAX_PRICE = 200_000_000
|
||||
|
||||
# Минимальный ценовой диапазон для рекурсии; при hi - lo < MIN_BRACKET
|
||||
# прекращаем деление и пагинируем как есть.
|
||||
_YANDEX_MIN_BRACKET = 50_000
|
||||
|
||||
# Оценка офферов на страницу Yandex SERP.
|
||||
# Yandex показывает ~20-30 карточек на страницу (зависит от типа выдачи).
|
||||
# Консервативно 30 — гарантирует, что мы не срежем страницы в leaf-bucket.
|
||||
_YANDEX_OFFERS_PER_PAGE = 30
|
||||
|
||||
# Пути в JSON-state для поиска totalItems (порядок = приоритет).
|
||||
_TOTAL_COUNT_PATHS: list[list[str]] = [
|
||||
["pageData", "pager", "totalItems"],
|
||||
["pageData", "pager", "total"],
|
||||
["searchResults", "pager", "totalItems"],
|
||||
["pager", "totalItems"],
|
||||
["pager", "total"],
|
||||
["pageData", "searchQuery", "total"],
|
||||
]
|
||||
|
||||
# Regex для fallback-поиска количества офферов в тексте DOM.
|
||||
_RE_TOTAL_COUNT = re.compile(r"(\d[\d ]{0,8})\s+(?:объявлен|квартир|предложен)", re.IGNORECASE)
|
||||
|
||||
|
||||
def _load_cookies_from_file(path: str | None) -> dict[str, str]:
|
||||
"""Load Yandex cookies from JSON file path (Netscape/browser export format).
|
||||
|
|
@ -244,8 +275,9 @@ class YandexRealtyScraper(BaseScraper):
|
|||
self._cookies = _load_cookies_from_file(settings.yandex_cookies_file)
|
||||
# Mobile proxy wiring (#806 follow-up): Yandex blocks datacenter IPs with
|
||||
# captcha/shell HTML even with Chrome TLS fingerprint. Route via mobile proxy.
|
||||
# Dedicated YANDEX_PROXY_URL > общий scraper_proxy_url (backward-compat).
|
||||
# proxy=None → curl_cffi ходит напрямую (dev без прокси — не падаем).
|
||||
_proxy_url = settings.scraper_proxy_url
|
||||
_proxy_url = settings.yandex_proxy_url
|
||||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||||
self._cffi_session = _CurlCffiSession(
|
||||
impersonate="chrome120",
|
||||
|
|
@ -278,6 +310,404 @@ class YandexRealtyScraper(BaseScraper):
|
|||
kwargs.setdefault("timeout", 30)
|
||||
return await self._cffi_session.get(url, **kwargs)
|
||||
|
||||
# ── Anti-block: IP rotation (зеркало cian.py _rotate_ip) ──────────────────
|
||||
|
||||
async def _rotate_ip(self) -> bool:
|
||||
"""Сменить мобильный IP через changeip-ссылку mobileproxy.
|
||||
|
||||
Дёргается напрямую (без прокси) — это API провайдера, не Yandex. Ждём ~9с:
|
||||
мобильному модему нужно время поднять новый IP. Returns True при успехе.
|
||||
Зеркало CianScraper._rotate_ip.
|
||||
"""
|
||||
from app.core.config import settings as _settings
|
||||
|
||||
rotate_url = _settings.yandex_proxy_rotate_url or _settings.avito_proxy_rotate_url
|
||||
if not rotate_url:
|
||||
return False
|
||||
sep = "&" if "?" in rotate_url else "?"
|
||||
try:
|
||||
async with _CurlCffiSession(timeout=30) as rot:
|
||||
await rot.get(f"{rotate_url}{sep}format=json")
|
||||
await asyncio.sleep(9)
|
||||
logger.info("yandex proxy: IP rotated via changeip")
|
||||
return True
|
||||
except Exception:
|
||||
logger.warning("yandex proxy: IP rotation failed", exc_info=True)
|
||||
return False
|
||||
|
||||
# ── Total-count extraction (UNVERIFIED — see fallback note) ────────────────
|
||||
|
||||
def _extract_total_count(self, html: str) -> int | None:
|
||||
"""Извлечь totalItems из embedded JSON state Yandex SERP.
|
||||
|
||||
CAVEAT: Yandex state is chunked + evaporates after hydration (as of 2026-05).
|
||||
This is a best-effort extraction — the fallback (paginate-until-empty) is the
|
||||
safety net when total is not reliably available. Field is UNVERIFIED against
|
||||
live production SERP; paths are taken from known Yandex state shapes.
|
||||
|
||||
Пути кандидаты (проверяются в порядке приоритета):
|
||||
pageData.pager.totalItems
|
||||
pageData.pager.total
|
||||
searchResults.pager.totalItems
|
||||
pager.totalItems
|
||||
pager.total
|
||||
pageData.searchQuery.total
|
||||
"""
|
||||
tree = HTMLParser(html)
|
||||
candidates: list[str] = []
|
||||
for selector in _STATE_SCRIPT_SELECTORS:
|
||||
for node in tree.css(selector):
|
||||
text = node.text() or ""
|
||||
text = text.strip()
|
||||
if text:
|
||||
candidates.append(text)
|
||||
|
||||
for blob in candidates:
|
||||
try:
|
||||
state = json.loads(blob)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
continue
|
||||
if not isinstance(state, dict):
|
||||
continue
|
||||
for path in _TOTAL_COUNT_PATHS:
|
||||
node_val: Any = state
|
||||
for key in path:
|
||||
if not isinstance(node_val, dict):
|
||||
node_val = None
|
||||
break
|
||||
node_val = node_val.get(key)
|
||||
if node_val is not None:
|
||||
try:
|
||||
result = int(node_val)
|
||||
if result >= 0:
|
||||
return result
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
|
||||
# DOM / regex fallback: try to find a Russian results-count header.
|
||||
# Best-effort only — Yandex uses dynamically generated class names.
|
||||
# NBSP already normalised to space by caller.
|
||||
m = _RE_TOTAL_COUNT.search(html[:20_000])
|
||||
if m:
|
||||
try:
|
||||
return int(m.group(1).replace(" ", ""))
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
# ── Leaf-bucket fetch helper ────────────────────────────────────────────────
|
||||
|
||||
async def _fetch_page_html(
|
||||
self,
|
||||
rooms: str | None,
|
||||
page: int,
|
||||
price_min: int | None,
|
||||
price_max: int | None,
|
||||
) -> str | None:
|
||||
"""GET одной SERP-страницы, возвращает HTML или None при ошибке."""
|
||||
url = self._build_url(page=page, rooms=rooms, price_min=price_min, price_max=price_max)
|
||||
try:
|
||||
resp = await self._http_get(url)
|
||||
except Exception:
|
||||
logger.exception("yandex: GET failed url=%s", url)
|
||||
return None
|
||||
if resp.status_code != 200: # type: ignore[union-attr]
|
||||
logger.warning("yandex: HTTP %d for url=%s", resp.status_code, url) # type: ignore[union-attr]
|
||||
return None
|
||||
html = resp.text.replace("\xa0", " ") # type: ignore[union-attr]
|
||||
return html
|
||||
|
||||
# ── Exhaustive full load ────────────────────────────────────────────────────
|
||||
|
||||
async def fetch_all_secondary(
|
||||
self,
|
||||
*,
|
||||
rooms_buckets: list[str] | None = None,
|
||||
price_cap_per_bucket: int = 500,
|
||||
max_pages_per_bucket: int = 100,
|
||||
concurrency: int = 4,
|
||||
on_bucket: Any = None,
|
||||
on_progress: Any = None,
|
||||
) -> list[ScrapedLot]:
|
||||
"""Exhaustive-загрузка Yandex ЕКБ вторички через партиционирование КОМНАТНОСТЬ × ЦЕНА.
|
||||
|
||||
Обходит Yandex SERP-cap (~575 карточек/запрос): внутри каждой комнатности
|
||||
адаптивно бьёт ценовой диапазон на бакеты так, чтобы totalItems < price_cap_per_bucket.
|
||||
При недоступности totalItems (см. _extract_total_count CAVEAT) переходит на
|
||||
paginate-until-empty fallback.
|
||||
|
||||
Параметры:
|
||||
rooms_buckets: список ключей ROOM_PATH для перебора (default: все 5 комнатностей).
|
||||
price_cap_per_bucket: максимум офферов в бакете перед делением (< Yandex ~575 cap).
|
||||
max_pages_per_bucket: жёсткий cap страниц; 100 × 30 ≈ 3000 > 575.
|
||||
concurrency: максимум параллельных page-фетчей в leaf-бакете.
|
||||
on_bucket: опциональный callback(list[ScrapedLot]) после каждого leaf-бакета.
|
||||
Может быть async или sync. Исключение прерывает прогон.
|
||||
on_progress: опциональный callback(unique_count) для heartbeat (per room-bucket).
|
||||
|
||||
Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url).
|
||||
"""
|
||||
_buckets = rooms_buckets if rooms_buckets is not None else list(ROOM_PATH.keys())
|
||||
seen: dict[str, ScrapedLot] = {}
|
||||
|
||||
for rooms_key in _buckets:
|
||||
logger.info(
|
||||
"yandex exhaustive: starting rooms=%s (price_cap=%d concurrency=%d)",
|
||||
rooms_key,
|
||||
price_cap_per_bucket,
|
||||
concurrency,
|
||||
)
|
||||
before = len(seen)
|
||||
await self._walk_price_range(
|
||||
rooms=rooms_key,
|
||||
lo=0,
|
||||
hi=_YANDEX_MAX_PRICE,
|
||||
seen=seen,
|
||||
price_cap_per_bucket=price_cap_per_bucket,
|
||||
max_pages_per_bucket=max_pages_per_bucket,
|
||||
concurrency=concurrency,
|
||||
on_bucket=on_bucket,
|
||||
)
|
||||
room_collected = len(seen) - before
|
||||
logger.info(
|
||||
"yandex exhaustive: rooms=%s done — collected %d (total unique=%d)",
|
||||
rooms_key,
|
||||
room_collected,
|
||||
len(seen),
|
||||
)
|
||||
if on_progress is not None:
|
||||
on_progress(len(seen))
|
||||
|
||||
logger.info("yandex exhaustive: DONE — total unique=%d lots", len(seen))
|
||||
return list(seen.values())
|
||||
|
||||
async def _walk_price_range(
|
||||
self,
|
||||
*,
|
||||
rooms: str | None,
|
||||
lo: int,
|
||||
hi: int,
|
||||
seen: dict[str, ScrapedLot],
|
||||
price_cap_per_bucket: int,
|
||||
max_pages_per_bucket: int,
|
||||
concurrency: int = 4,
|
||||
on_bucket: Any = None,
|
||||
_depth: int = 0,
|
||||
) -> None:
|
||||
"""Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi].
|
||||
|
||||
Алгоритм:
|
||||
1. Запросить page=0 (probe) с price_min=lo, price_max=hi → totalItems.
|
||||
2. Если totalItems <= cap → пагинировать leaf-бакет параллельно (concurrency).
|
||||
3. Если totalItems > cap → разбить бакет пополам (рекурсия).
|
||||
Guard: hi - lo < _YANDEX_MIN_BRACKET → пагинировать как есть.
|
||||
Fallback: если totalItems=None → paginate-until-empty (безопасная деградация).
|
||||
|
||||
# TODO(yandex-total): totalItems not reliably extractable from hydrated DOM
|
||||
# — using paginate-until-empty fallback when state extraction fails.
|
||||
"""
|
||||
_lo_param = lo if lo > 0 else None
|
||||
|
||||
# ── Шаг 1: probe page=0 ─────────────────────────────────────────────────
|
||||
html = await self._fetch_page_html(rooms, 0, _lo_param, hi)
|
||||
await asyncio.sleep(self.request_delay_sec)
|
||||
|
||||
total: int | None = None
|
||||
|
||||
# CAPTCHA-aware probe: rotate IP + 1 retry
|
||||
if html is None or _is_captcha(html):
|
||||
logger.warning(
|
||||
"yandex: probe captcha/None for rooms=%s [%d, %d] depth=%d — rotating IP + retry",
|
||||
rooms,
|
||||
lo,
|
||||
hi,
|
||||
_depth,
|
||||
)
|
||||
rotated = await self._rotate_ip()
|
||||
if rotated:
|
||||
html = await self._fetch_page_html(rooms, 0, _lo_param, hi)
|
||||
await asyncio.sleep(self.request_delay_sec)
|
||||
|
||||
if html is not None and not _is_captcha(html):
|
||||
total = self._extract_total_count(html)
|
||||
|
||||
# ── Fallback: totalItems not available ──────────────────────────────────
|
||||
# TODO(yandex-total): totalItems not reliably extractable from hydrated DOM
|
||||
# — using paginate-until-empty fallback when state extraction fails.
|
||||
if total is None:
|
||||
logger.warning(
|
||||
"yandex: totalItems=None for rooms=%s [%d, %d] depth=%d "
|
||||
"— using paginate-until-empty fallback",
|
||||
rooms,
|
||||
lo,
|
||||
hi,
|
||||
_depth,
|
||||
)
|
||||
bucket_lots: list[ScrapedLot] = []
|
||||
# page=0 probe already fetched — reuse its HTML if available
|
||||
if html is not None and not _is_captcha(html):
|
||||
probe_lots = self._parse_html(html, page=0)
|
||||
bucket_lots.extend(probe_lots)
|
||||
start_page = 1
|
||||
else:
|
||||
start_page = 0
|
||||
|
||||
for p in range(start_page, max_pages_per_bucket):
|
||||
if p == 0:
|
||||
page_lots: list[ScrapedLot] = []
|
||||
else:
|
||||
page_html = await self._fetch_page_html(rooms, p, _lo_param, hi)
|
||||
await asyncio.sleep(self.request_delay_sec)
|
||||
if page_html is None or _is_captcha(page_html):
|
||||
logger.warning(
|
||||
"yandex: fallback page=%d captcha/None rooms=%s [%d, %d] — stopping",
|
||||
p,
|
||||
rooms,
|
||||
lo,
|
||||
hi,
|
||||
)
|
||||
break
|
||||
page_lots = self._parse_html(page_html, page=p)
|
||||
if not page_lots and p > start_page:
|
||||
break
|
||||
bucket_lots.extend(page_lots)
|
||||
|
||||
for lot in bucket_lots:
|
||||
key = lot.source_id or lot.source_url
|
||||
if key:
|
||||
seen[key] = lot
|
||||
|
||||
if on_bucket and bucket_lots:
|
||||
res = on_bucket(bucket_lots)
|
||||
if inspect.isawaitable(res):
|
||||
await res
|
||||
return
|
||||
|
||||
logger.info(
|
||||
"yandex: rooms=%s [%d, %d] totalItems=%d depth=%d",
|
||||
rooms,
|
||||
lo,
|
||||
hi,
|
||||
total,
|
||||
_depth,
|
||||
)
|
||||
|
||||
if total == 0:
|
||||
return
|
||||
|
||||
# ── Шаг 2: деление или пагинация ────────────────────────────────────────
|
||||
bracket_size = hi - lo
|
||||
need_split = total > price_cap_per_bucket
|
||||
too_narrow = bracket_size < _YANDEX_MIN_BRACKET
|
||||
|
||||
if need_split and too_narrow:
|
||||
logger.warning(
|
||||
"yandex: rooms=%s [%d, %d] totalItems=%d > cap=%d but bracket=%d < "
|
||||
"MIN_BRACKET=%d — paginating as-is (tail loss ~%d)",
|
||||
rooms,
|
||||
lo,
|
||||
hi,
|
||||
total,
|
||||
price_cap_per_bucket,
|
||||
bracket_size,
|
||||
_YANDEX_MIN_BRACKET,
|
||||
max(0, total - price_cap_per_bucket),
|
||||
)
|
||||
need_split = False
|
||||
|
||||
if need_split:
|
||||
mid = (lo + hi) // 2
|
||||
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,
|
||||
on_bucket=on_bucket,
|
||||
_depth=_depth + 1,
|
||||
)
|
||||
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,
|
||||
on_bucket=on_bucket,
|
||||
_depth=_depth + 1,
|
||||
)
|
||||
return
|
||||
|
||||
# ── Параллельная пагинация leaf-бакета ─────────────────────────────────
|
||||
# Yandex SERP pages are 0-indexed (unlike Cian which is 1-indexed)
|
||||
max_pages = min(
|
||||
math.ceil(total / _YANDEX_OFFERS_PER_PAGE),
|
||||
max_pages_per_bucket,
|
||||
)
|
||||
|
||||
sem = asyncio.Semaphore(concurrency)
|
||||
|
||||
async def _one_page(p: int) -> list[ScrapedLot]:
|
||||
if p == 0:
|
||||
# Reuse probe HTML
|
||||
return self._parse_html(html, page=0) if html else []
|
||||
async with sem:
|
||||
page_html = await self._fetch_page_html(rooms, p, _lo_param, hi)
|
||||
await asyncio.sleep(self.request_delay_sec)
|
||||
if page_html is None or _is_captcha(page_html):
|
||||
logger.warning(
|
||||
"yandex: page_html=None/captcha rooms=%s [%d, %d] page=%d — skipping",
|
||||
rooms,
|
||||
lo,
|
||||
hi,
|
||||
p,
|
||||
)
|
||||
return []
|
||||
return self._parse_html(page_html, page=p)
|
||||
|
||||
page_results = await asyncio.gather(
|
||||
*[_one_page(p) for p in range(0, max_pages)],
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
bucket_lots_leaf: list[ScrapedLot] = []
|
||||
for p_idx, res in enumerate(page_results):
|
||||
if isinstance(res, BaseException):
|
||||
logger.warning(
|
||||
"yandex: page exception rooms=%s [%d, %d] page=%d — %r",
|
||||
rooms,
|
||||
lo,
|
||||
hi,
|
||||
p_idx,
|
||||
res,
|
||||
)
|
||||
else:
|
||||
bucket_lots_leaf.extend(res)
|
||||
|
||||
for lot in bucket_lots_leaf:
|
||||
key = lot.source_id or lot.source_url
|
||||
if key:
|
||||
seen[key] = lot
|
||||
|
||||
logger.info(
|
||||
"yandex: rooms=%s [%d, %d] paginated=%d pages collected=%d unique_total=%d",
|
||||
rooms,
|
||||
lo,
|
||||
hi,
|
||||
max_pages,
|
||||
len(bucket_lots_leaf),
|
||||
len(seen),
|
||||
)
|
||||
|
||||
if on_bucket and bucket_lots_leaf:
|
||||
res_cb = on_bucket(bucket_lots_leaf)
|
||||
if inspect.isawaitable(res_cb):
|
||||
await res_cb
|
||||
|
||||
async def fetch_around(
|
||||
self,
|
||||
lat: float,
|
||||
|
|
|
|||
|
|
@ -32,6 +32,9 @@ def _mock_settings(
|
|||
# scraper_proxy_url fallback). Mirror scraper_proxy_url here so the existing
|
||||
# cian proxy-wiring assertions keep passing (#923 follow-up).
|
||||
cian_proxy_url=scraper_proxy_url,
|
||||
# Yandex scrapers read settings.yandex_proxy_url (property: YANDEX_PROXY_URL or
|
||||
# scraper_proxy_url fallback). Mirror scraper_proxy_url here for test compat.
|
||||
yandex_proxy_url=scraper_proxy_url,
|
||||
yandex_cookies_file=yandex_cookies_file,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -634,3 +634,195 @@ async def test_fetch_around_multi_room_legacy_no_combos():
|
|||
# Should call with rooms=None, price_min=None (citywide)
|
||||
assert all(c["rooms"] is None for c in calls)
|
||||
assert all(c["price_min"] is None for c in calls)
|
||||
|
||||
|
||||
# ── PART E: exhaustive load — fetch_all_secondary + _walk_price_range ────────
|
||||
|
||||
|
||||
def _make_lots(n: int, prefix: str = "lot") -> list[MagicMock]:
|
||||
"""Create n mock ScrapedLot objects with unique source_id."""
|
||||
lots = []
|
||||
for i in range(n):
|
||||
lot = MagicMock(spec=ScrapedLot)
|
||||
lot.source_id = f"{prefix}_{i}"
|
||||
lot.source_url = f"https://realty.yandex.ru/offer/{prefix}_{i}/"
|
||||
lots.append(lot)
|
||||
return lots
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_walk_price_range_splits_when_total_exceeds_cap():
|
||||
"""When totalItems > price_cap_per_bucket and bracket >= MIN_BRACKET,
|
||||
_walk_price_range recurses and subdivides — each leaf has ≤ cap items."""
|
||||
s = YandexRealtyScraper()
|
||||
s._cffi_session = AsyncMock()
|
||||
|
||||
call_ranges: list[tuple[int | None, int]] = []
|
||||
|
||||
async def fake_fetch_page_html(
|
||||
rooms: str | None, page: int, price_min: int | None, price_max: int
|
||||
) -> str:
|
||||
call_ranges.append((price_min, price_max))
|
||||
# Return captcha-free HTML with 0 cards (parsing gives no lots, which is fine —
|
||||
# we only care that the range is split, not about actual lot count in this test)
|
||||
return "<html><body>empty</body></html>"
|
||||
|
||||
# total_count returns 600 for the first (wide) probe, then 0 for sub-ranges
|
||||
# so recursion terminates after the first split.
|
||||
call_counts: list[int] = [0]
|
||||
|
||||
def fake_extract_total(html: str) -> int | None:
|
||||
call_counts[0] += 1
|
||||
# First call (wide range) returns 600 > cap=500, triggering a split
|
||||
# Subsequent calls (sub-ranges) return 0 → no further recursion
|
||||
if call_counts[0] == 1:
|
||||
return 600
|
||||
return 0
|
||||
|
||||
seen: dict = {}
|
||||
with (
|
||||
patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
|
||||
patch.object(s, "_extract_total_count", side_effect=fake_extract_total),
|
||||
patch.object(s, "_is_captcha_html", return_value=False, create=True),
|
||||
):
|
||||
# Patch asyncio.sleep to be instant
|
||||
with patch("asyncio.sleep", new_callable=AsyncMock):
|
||||
await s._walk_price_range(
|
||||
rooms="1",
|
||||
lo=0,
|
||||
hi=10_000_000,
|
||||
seen=seen,
|
||||
price_cap_per_bucket=500,
|
||||
max_pages_per_bucket=100,
|
||||
concurrency=2,
|
||||
)
|
||||
|
||||
# Should have probed at least the initial range + 2 sub-ranges
|
||||
assert len(call_ranges) >= 3, f"Expected >= 3 probes (wide + 2 sub-ranges), got {call_ranges}"
|
||||
# Initial probe covers the full range
|
||||
assert call_ranges[0] == (None, 10_000_000) # lo=0 → _lo_param=None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_walk_price_range_fires_on_bucket_per_leaf():
|
||||
"""on_bucket callback is called once per leaf bucket with the bucket's lots."""
|
||||
s = YandexRealtyScraper()
|
||||
s._cffi_session = AsyncMock()
|
||||
|
||||
bucket_calls: list[list] = []
|
||||
|
||||
def fake_on_bucket(lots: list) -> None:
|
||||
bucket_calls.append(list(lots))
|
||||
|
||||
# total = 10 (< cap=500) → paginate leaf immediately, 1 page (page=0)
|
||||
async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str:
|
||||
# Return HTML with 1 parseable card for page=0, empty for others
|
||||
if page == 0:
|
||||
return SINGLE_CARD_HTML
|
||||
return EMPTY_PAGE_HTML
|
||||
|
||||
seen: dict = {}
|
||||
with (
|
||||
patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
|
||||
patch.object(s, "_extract_total_count", return_value=10),
|
||||
patch("asyncio.sleep", new_callable=AsyncMock),
|
||||
):
|
||||
await s._walk_price_range(
|
||||
rooms="1",
|
||||
lo=0,
|
||||
hi=20_000_000,
|
||||
seen=seen,
|
||||
price_cap_per_bucket=500,
|
||||
max_pages_per_bucket=5,
|
||||
concurrency=2,
|
||||
on_bucket=fake_on_bucket,
|
||||
)
|
||||
|
||||
# on_bucket must have been called at least once
|
||||
assert len(bucket_calls) >= 1
|
||||
# All calls contain ScrapedLot-like objects (the actual card from SINGLE_CARD_HTML)
|
||||
for call_lots in bucket_calls:
|
||||
assert len(call_lots) >= 0 # may be 0 if card parse yields empty — still fires
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_walk_price_range_fallback_when_total_none():
|
||||
"""When totalItems=None (state unavailable), falls back to paginate-until-empty
|
||||
and still fires on_bucket with whatever was collected."""
|
||||
s = YandexRealtyScraper()
|
||||
s._cffi_session = AsyncMock()
|
||||
|
||||
bucket_calls: list[list] = []
|
||||
|
||||
def fake_on_bucket(lots: list) -> None:
|
||||
bucket_calls.append(list(lots))
|
||||
|
||||
page_htmls = {
|
||||
0: SINGLE_CARD_HTML, # page=0: 1 card
|
||||
1: EMPTY_PAGE_HTML, # page=1: empty → stop
|
||||
}
|
||||
|
||||
async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str | None:
|
||||
return page_htmls.get(page, EMPTY_PAGE_HTML)
|
||||
|
||||
seen: dict = {}
|
||||
with (
|
||||
patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
|
||||
patch.object(s, "_extract_total_count", return_value=None),
|
||||
patch("asyncio.sleep", new_callable=AsyncMock),
|
||||
):
|
||||
await s._walk_price_range(
|
||||
rooms="2",
|
||||
lo=0,
|
||||
hi=15_000_000,
|
||||
seen=seen,
|
||||
price_cap_per_bucket=500,
|
||||
max_pages_per_bucket=10,
|
||||
concurrency=2,
|
||||
on_bucket=fake_on_bucket,
|
||||
)
|
||||
|
||||
# Fallback path must fire on_bucket with the collected lots
|
||||
assert len(bucket_calls) >= 1, "on_bucket must be called in fallback path"
|
||||
# The card from SINGLE_CARD_HTML should have been collected
|
||||
all_lots = [lot for call in bucket_calls for lot in call]
|
||||
assert len(all_lots) >= 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_all_secondary_deduplicates_across_rooms():
|
||||
"""fetch_all_secondary deduplicates offers that appear in multiple room buckets."""
|
||||
s = YandexRealtyScraper()
|
||||
|
||||
shared_lot = MagicMock(spec=ScrapedLot)
|
||||
shared_lot.source_id = "shared_999"
|
||||
shared_lot.source_url = "https://realty.yandex.ru/offer/shared_999/"
|
||||
|
||||
unique_lot_1 = MagicMock(spec=ScrapedLot)
|
||||
unique_lot_1.source_id = "unique_001"
|
||||
unique_lot_1.source_url = "https://realty.yandex.ru/offer/unique_001/"
|
||||
|
||||
unique_lot_2 = MagicMock(spec=ScrapedLot)
|
||||
unique_lot_2.source_id = "unique_002"
|
||||
unique_lot_2.source_url = "https://realty.yandex.ru/offer/unique_002/"
|
||||
|
||||
call_count = 0
|
||||
|
||||
async def fake_walk(*, rooms, lo, hi, seen, **kwargs): # type: ignore[override]
|
||||
nonlocal call_count
|
||||
# Both room buckets yield the shared lot + one unique each
|
||||
seen["shared_999"] = shared_lot
|
||||
if call_count == 0:
|
||||
seen["unique_001"] = unique_lot_1
|
||||
else:
|
||||
seen["unique_002"] = unique_lot_2
|
||||
call_count += 1
|
||||
|
||||
with patch.object(s, "_walk_price_range", side_effect=fake_walk):
|
||||
result = await s.fetch_all_secondary(rooms_buckets=["1", "2"])
|
||||
|
||||
ids = {lot.source_id for lot in result}
|
||||
assert "shared_999" in ids
|
||||
assert "unique_001" in ids
|
||||
assert "unique_002" in ids
|
||||
assert len(ids) == 3 # shared deduplicated — only 3 unique
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue