diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 9372f918..2680df10 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -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 ───────────────── diff --git a/tradein-mvp/backend/app/core/config.py b/tradein-mvp/backend/app/core/config.py index 57b78340..39f6be33 100644 --- a/tradein-mvp/backend/app/core/config.py +++ b/tradein-mvp/backend/app/core/config.py @@ -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. diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 12b7f171..e18dffdb 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -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", ] diff --git a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py index 80531abd..52eb28fd 100644 --- a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py +++ b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py @@ -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, diff --git a/tradein-mvp/backend/tests/test_scraper_proxy.py b/tradein-mvp/backend/tests/test_scraper_proxy.py index 3bdff7ad..1cb22271 100644 --- a/tradein-mvp/backend/tests/test_scraper_proxy.py +++ b/tradein-mvp/backend/tests/test_scraper_proxy.py @@ -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, ) diff --git a/tradein-mvp/backend/tests/test_yandex_realty_serp.py b/tradein-mvp/backend/tests/test_yandex_realty_serp.py index dc66109a..3a7be582 100644 --- a/tradein-mvp/backend/tests/test_yandex_realty_serp.py +++ b/tradein-mvp/backend/tests/test_yandex_realty_serp.py @@ -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 "empty" + + # 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