diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 659b39df..6cca3cfc 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -26,6 +26,7 @@ from app.services.scheduler import has_running_run from app.services.scrape_pipeline import ( run_avito_city_sweep, run_cian_city_sweep, + run_cian_full_load, run_n1_city_sweep, run_yandex_city_sweep, ) @@ -943,6 +944,94 @@ def cancel_cian_city_sweep( return {"ok": True, "run_id": run_id, "cancelled": cancelled} +# ── Cian exhaustive full load (room × price partitioning) ──────────────────── + + +class CianFullLoadRequest(BaseModel): + price_cap_per_bucket: int = Field( + default=1400, + ge=200, + le=1500, + description=( + "Максимум офферов на price-бакет. При totalOffers > cap — бакет делится пополам. " + "Ниже Cian SERP-cap ~1500; запас 100 на variance." + ), + ) + request_delay_sec: float = Field( + default=4.0, + ge=3.0, + le=15.0, + description="Задержка между SERP-запросами (сек). Увеличить при бане.", + ) + enrich_detail: bool = Field( + default=False, + description="Если True — после сбора обогатить detail-страницы (медленно).", + ) + detail_top_n: int = Field( + default=0, + ge=0, + le=50, + description="Сколько свежих листингов обогатить detail (актуально при enrich_detail=True).", + ) + + +@router.post("/scrape/cian-full-load", response_model=CitySweepStartResponse) +async def start_cian_full_load( + payload: CianFullLoadRequest, + background_tasks: BackgroundTasks, + db: Annotated[Session, Depends(get_db)], +) -> CitySweepStartResponse: + """Запустить exhaustive Cian ЕКБ full load в background. + + Обходит Cian SERP-cap (~54 стр/запрос ≈ 1500 результатов): адаптивно бьёт + цену на бакеты внутри каждой комнатности (totalOffers-driven бинарное деление), + пагинирует каждый бакет полностью, дедуп по source_id. + + Один региональный проход БЕЗ anchor'ов (Cian SERP region-wide — anchors избыточны). + IP-ротация через changeip при timeout/captcha. + + Coop cancel: POST /scrape/cian-full-load/{run_id}/cancel. + Статус прогона: смотреть через GET /scrape/cian-city-sweep/runs + (или добавить source='cian_full_load' в /admin SQL). + """ + run_id = runs_mod.create_run(db, source="cian_full_load", params=payload.model_dump()) + + async def _full_load_task() -> None: + task_db = SessionLocal() + try: + await run_cian_full_load( + task_db, + run_id=run_id, + price_cap_per_bucket=payload.price_cap_per_bucket, + request_delay_sec=payload.request_delay_sec, + enrich_detail=payload.enrich_detail, + detail_top_n=payload.detail_top_n, + ) + except Exception: + logger.exception("cian-full-load background task run_id=%d crashed", run_id) + finally: + task_db.close() + + background_tasks.add_task(_full_load_task) + logger.info("cian-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=payload.detail_top_n, + ) + + +@router.post("/scrape/cian-full-load/{run_id}/cancel") +def cancel_cian_full_load( + run_id: int, + db: Annotated[Session, Depends(get_db)], +) -> dict[str, object]: + """Отменить running Cian 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/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index f968d998..72e0ae8d 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -1526,12 +1526,186 @@ async def run_cian_city_sweep( raise +# ── Cian exhaustive full load (region-wide, без anchor'ов) ───────────────── + + +@dataclass +class CianFullLoadCounters: + """Aggregate counters для run_cian_full_load. + + Без anchor-полей — единственный региональный проход. + detail_* — опциональное обогащение detail-страниц (если enrich_detail=True). + """ + + unique_fetched: int = 0 + saved_inserted: int = 0 + saved_updated: int = 0 + detail_attempted: int = 0 + detail_enriched: int = 0 + detail_failed: 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_cian_full_load( + db: Session, + *, + run_id: int, + price_cap_per_bucket: int = 1400, + detail_top_n: int = 0, + request_delay_sec: float = 4.0, + enrich_detail: bool = False, +) -> CianFullLoadCounters: + """Exhaustive региональный сбор Cian ЕКБ вторички (БЕЗ anchor'ов). + + Cian SERP — region-wide, не geo-bbox. Anchor-loop избыточен (5 anchor'ов + возвращают одну и ту же выдачу ×5). Один проход с партиционированием по + КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление) охватывает весь ЕКБ. + + Параметры: + price_cap_per_bucket: максимум офферов на price-бакет (< Cian SERP-cap ~1500). + detail_top_n: сколько свежих cian-листингов без detail_enriched_at обогащать + (актуально только при enrich_detail=True). + request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default). + enrich_detail: включить detail-обогащение (по умолчанию отключено — тяжело). + + Cooperative cancel: scrape_runs.is_cancelled проверяется в on_progress callback + (раз в room-bucket). При cancel — возвращаем частичный результат, mark_done. + """ + from app.services.scrapers.cian import CianScraper + + counters = CianFullLoadCounters() + _cancelled = False + + def _on_progress(unique_count: int) -> None: + nonlocal _cancelled + counters.unique_fetched = unique_count + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + if scrape_runs.is_cancelled(db, run_id): + _cancelled = True + logger.info("cian-full-load run_id=%d: cancel detected in on_progress", run_id) + + try: + async with CianScraper() as scraper: + scraper.request_delay_sec = request_delay_sec + + lots = await scraper.fetch_all_secondary( + price_cap_per_bucket=price_cap_per_bucket, + on_progress=_on_progress, + ) + + counters.unique_fetched = len(lots) + logger.info( + "cian-full-load run_id=%d: fetch done — unique=%d cancelled=%s", + run_id, + len(lots), + _cancelled, + ) + + if lots: + inserted, updated = save_listings(db, lots, run_id=run_id) + counters.saved_inserted = inserted + counters.saved_updated = updated + logger.info( + "cian-full-load run_id=%d: save done — ins=%d upd=%d", + run_id, + inserted, + updated, + ) + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + + # ── Опциональный detail-энричмент ───────────────────────────────────── + if enrich_detail and detail_top_n > 0 and not _cancelled: + from sqlalchemy import text as _text + + from app.services.scrapers.cian_detail import ( + fetch_detail, + save_detail_enrichment, + ) + + priority_rows = ( + db.execute( + _text(""" + SELECT id, source_url + FROM listings + WHERE source = 'cian' + AND source_url IS NOT NULL + AND detail_enriched_at IS NULL + AND scraped_at > NOW() - INTERVAL '4 hours' + ORDER BY scraped_at DESC NULLS LAST + LIMIT :lim + """), + {"lim": detail_top_n}, + ) + .mappings() + .all() + ) + for didx, row in enumerate(priority_rows): + if scrape_runs.is_cancelled(db, run_id): + logger.info("cian-full-load run_id=%d: cancelled during detail enrich", run_id) + _cancelled = True + break + listing_id: int = row["id"] + source_url: str = row["source_url"] + counters.detail_attempted += 1 + try: + enrichment = await fetch_detail(source_url) + if enrichment is not None: + save_detail_enrichment(db, listing_id, enrichment) + counters.detail_enriched += 1 + else: + counters.detail_failed += 1 + except Exception as exc: + counters.detail_failed += 1 + counters.errors_count += 1 + logger.warning( + "cian-full-load run_id=%d: detail failed listing_id=%d: %s", + run_id, + listing_id, + exc, + ) + if didx < len(priority_rows) - 1: + await asyncio.sleep(request_delay_sec * random.uniform(0.8, 1.2)) + + logger.info( + "cian-full-load run_id=%d: detail done — enriched=%d/%d failed=%d", + run_id, + counters.detail_enriched, + counters.detail_attempted, + counters.detail_failed, + ) + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + + scrape_runs.mark_done(db, run_id, counters.to_dict()) + logger.info( + "cian-full-load run_id=%d done: unique=%d ins=%d upd=%d " + "detail=%d/%d errors=%d cancelled=%s", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + counters.detail_enriched, + counters.detail_attempted, + counters.errors_count, + _cancelled, + ) + return counters + + except Exception as exc: + logger.exception("cian-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", "CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES", "EKB_ANCHORS", "CianCitySweepCounters", + "CianFullLoadCounters", "CitySweepCounters", "PipelineCounters", "PipelineResult", @@ -1539,5 +1713,6 @@ __all__ = [ "run_avito_city_sweep", "run_avito_pipeline", "run_cian_city_sweep", + "run_cian_full_load", "run_yandex_city_sweep", ] diff --git a/tradein-mvp/backend/app/services/scrapers/cian.py b/tradein-mvp/backend/app/services/scrapers/cian.py index cbace45b..fa7eaa63 100644 --- a/tradein-mvp/backend/app/services/scrapers/cian.py +++ b/tradein-mvp/backend/app/services/scrapers/cian.py @@ -11,12 +11,21 @@ НЕ используем anchor jitter: offer.geo.coordinates.{lat,lng} — точные координаты прямо из SERP state. Jitter запрещён per implementation plan. + +Exhaustive load (fetch_all_secondary): +- Cian SERP — регион-wide (не geo-bbox). Anchor-loop в существующем city-sweep + избыточен для Cian (все anchor'ы дают одну выдачу). Полный сбор = ОДИН проход + с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление диапазона). +- IP-ротация через _rotate_ip() (changeip, зеркало avito.py). """ from __future__ import annotations +import asyncio import hashlib import logging +import math +from collections.abc import Callable from datetime import UTC, datetime from typing import Any from urllib.parse import urlencode @@ -42,6 +51,33 @@ CIAN_EKB_REGION_ID = 4743 _MFE_SERP = "frontend-serp" _STATE_KEY = "initialState" +# ── Константы exhaustive-загрузки ────────────────────────────────────────────── + +# Верхняя граница цены при первом рекурсивном делении (нет явного hi). +# 200 млн ₽ заведомо выше ТОПа ЕКБ — дальние бакеты дадут 0 офферов. +_MAX_PRICE = 200_000_000 + +# Минимальный ценовой диапазон для рекурсии; при hi - lo < MIN_BRACKET +# прекращаем деление и пагинируем как есть (принимаем возможный «хвост»). +_MIN_BRACKET = 50_000 + +# Cian SERP показывает ~28 офферов на страницу. +_CIAN_OFFERS_PER_PAGE = 28 + +# Комнатности по умолчанию для exhaustive-загрузки. +# Коды Cian: room1=1к, room2=2к, room3=3к, room4=4к, room5=5к, room6=6+. +# TODO: студии — Cian использует отдельный тип flat_type=studio, не room-param. +# Если добавить room9 — запрос может давать 0 офферов или смешивать типы. +# До верификации кода студии — студии НЕ включаем в exhaustive-загрузку. +_DEFAULT_ROOMS_BUCKETS: list[tuple[int, ...]] = [ + (1,), + (2,), + (3,), + (4,), + (5,), + (6,), +] + class CianScraper(BaseScraper): """Cian SERP scraper. Использует curl_cffi для обхода TLS fingerprint. @@ -110,6 +146,29 @@ class CianScraper(BaseScraper): await self._cffi.close() await super().__aexit__(*args) + # ── Anti-block: IP rotation (зеркало avito.py _rotate_ip) ───────────────── + + async def _rotate_ip(self) -> bool: + """Сменить мобильный IP через changeip-ссылку mobileproxy. + + Дёргается напрямую (без прокси) — это API провайдера, не Cian. Ждём ~9с: + мобильному модему нужно время поднять новый IP. Returns True при успехе. + Зеркало AvitoScraper._rotate_ip (#623). + """ + rotate_url = settings.cian_proxy_rotate_url or settings.avito_proxy_rotate_url + if not rotate_url: + return False + sep = "&" if "?" in rotate_url else "?" + try: + async with AsyncSession(timeout=30) as rot: + await rot.get(f"{rotate_url}{sep}format=json") + await asyncio.sleep(9) + logger.info("cian proxy: IP rotated via changeip") + return True + except Exception: + logger.warning("cian proxy: IP rotation failed", exc_info=True) + return False + async def fetch_around( self, lat: float, @@ -187,11 +246,14 @@ class CianScraper(BaseScraper): self, rooms: tuple[int, ...] | None = None, page: int = 1, + min_price: int | None = None, + max_price: int | None = None, ) -> str: """URL для Cian каталога вторички ЕКБ. Используем ekb.cian.ru (city-specific subdomain). Регион задаётся через region= param как fallback для reliability. + min_price / max_price — опциональные ценовые границы (руб.) для price-бакетов. """ params: list[tuple[str, Any]] = [ ("deal_type", "sale"), @@ -203,10 +265,320 @@ class CianScraper(BaseScraper): if rooms: for r in rooms: params.append((f"room{r}", "1")) + if min_price is not None: + params.append(("minprice", min_price)) + if max_price is not None: + params.append(("maxprice", max_price)) if page > 1: params.append(("p", page)) return f"{self.base_url}/cat.php?{urlencode(params)}" + def _extract_total_offers(self, html: str) -> int | None: + """Извлечь totalOffers из Redux state Cian SERP. + + Использует тот же extract_state, что и _parse_serp_html. + Возвращает None при captcha/ошибке парсинга. + """ + state = extract_state(html, mfe=_MFE_SERP, key=_STATE_KEY) + if state is None: + return None + total = state.get("results", {}).get("totalOffers") + if total is None: + return None + try: + return int(total) + except (TypeError, ValueError): + return None + + async def _fetch_page_html( + self, + rooms: tuple[int, ...] | None, + page: int, + min_price: int | None, + max_price: int | None, + ) -> str | None: + """GET одной SERP-страницы, возвращает HTML или None при ошибке.""" + assert self._cffi is not None + url = self._build_url(rooms, page, min_price, max_price) + try: + resp = await self._cffi.get(url) + except Exception: + logger.exception("cian: GET failed url=%s", url) + return None + if resp.status_code != 200: + logger.warning("cian: HTTP %d for url=%s", resp.status_code, url) + return None + return resp.text + + async def fetch_all_secondary( + self, + *, + rooms_buckets: list[tuple[int, ...]] | None = None, + price_cap_per_bucket: int = 1400, + max_pages_per_bucket: int = 54, + on_progress: Callable[[int], None] | None = None, + ) -> list[ScrapedLot]: + """Exhaustive-загрузка Cian ЕКБ вторички через партиционирование КОМНАТНОСТЬ × ЦЕНА. + + Обходит Cian SERP-cap (~54 стр/запрос ≈ 1500 результатов на запрос): + внутри каждой комнатности адаптивно бьёт диапазон цены на бакеты так, чтобы + в каждом totalOffers < price_cap_per_bucket → пагинирует бакет полностью. + Дедуп по source_id (dict seen). + + Параметры: + rooms_buckets: список room-кодов Cian для перебора (default: 1-6). + price_cap_per_bucket: максимум офферов в бакете перед делением (< 1500). + max_pages_per_bucket: Cian hard cap ~54; не превышать. + on_progress: опциональный callback(unique_count) для heartbeat. + + Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url). + """ + _buckets = rooms_buckets if rooms_buckets is not None else _DEFAULT_ROOMS_BUCKETS + seen: dict[str, ScrapedLot] = {} + + for rooms in _buckets: + room_label = f"room{'_'.join(str(r) for r in rooms)}" + logger.info( + "cian exhaustive: starting %s (price_cap=%d)", room_label, price_cap_per_bucket + ) + before = len(seen) + await self._walk_price_range( + rooms=rooms, + lo=0, + hi=None, + seen=seen, + price_cap_per_bucket=price_cap_per_bucket, + max_pages_per_bucket=max_pages_per_bucket, + ) + room_collected = len(seen) - before + logger.info( + "cian exhaustive: %s done — collected %d (total unique=%d)", + room_label, + room_collected, + len(seen), + ) + if on_progress is not None: + on_progress(len(seen)) + + logger.info("cian exhaustive: DONE — total unique=%d lots", len(seen)) + return list(seen.values()) + + async def _walk_price_range( + self, + *, + rooms: tuple[int, ...], + lo: int, + hi: int | None, + seen: dict[str, ScrapedLot], + price_cap_per_bucket: int, + max_pages_per_bucket: int, + _depth: int = 0, + ) -> None: + """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi]. + + Алгоритм: + 1. Запросить page=1 с min_price=lo, max_price=hi → totalOffers. + 2. Если totalOffers <= cap → пагинировать бакет полностью. + 3. Если totalOffers > cap → разбить бакет пополам (рекурсия). + Guard: hi - lo < _MIN_BRACKET → пагинировать как есть (логируем WARNING). + """ + # Нормализация: None-hi → _MAX_PRICE для первого уровня + effective_hi = hi if hi is not None else _MAX_PRICE + + room_label = f"room{'_'.join(str(r) for r in rooms)}" + + # ── Шаг 1: probe page 1 ──────────────────────────────────────────────── + html = await self._fetch_page_html(rooms, 1, lo if lo > 0 else None, hi) + await self.sleep_between_requests() + + total: int | None = None + if html is not None: + total = self._extract_total_offers(html) + + # Ретрай на captcha/ошибку: rotate IP + 1 retry + if total is None: + logger.warning( + "cian: totalOffers=None for %s [%d, %s] depth=%d — rotating IP + retry", + room_label, + lo, + hi, + _depth, + ) + rotated = await self._rotate_ip() + if rotated: + html = await self._fetch_page_html(rooms, 1, lo if lo > 0 else None, hi) + 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 [%d, %s] — totalOffers unavailable after retry", + room_label, + lo, + hi, + ) + return + + logger.info( + "cian: %s [%d, %s] totalOffers=%d depth=%d", + room_label, + lo, + hi, + total, + _depth, + ) + + if total == 0: + return + + # ── Шаг 2: деление или пагинация ────────────────────────────────────── + bracket_size = effective_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, %s] 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: + # При hi=None: сначала устанавливаем hi = _MAX_PRICE и рекурсируем + if hi is None: + await self._walk_price_range( + rooms=rooms, + lo=lo, + hi=_MAX_PRICE, + seen=seen, + price_cap_per_bucket=price_cap_per_bucket, + max_pages_per_bucket=max_pages_per_bucket, + _depth=_depth + 1, + ) + return + mid = (lo + effective_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, + _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, + _depth=_depth + 1, + ) + return + + # ── Пагинация бакета ─────────────────────────────────────────────────── + max_pages = min( + math.ceil(total / _CIAN_OFFERS_PER_PAGE), + max_pages_per_bucket, + ) + + # Страница 1 уже есть (html из probe выше) + collected_this_bucket = 0 + if html: + page1_lots = self._parse_serp_html(html) + for lot in page1_lots: + key = lot.source_id or lot.source_url + if key: + seen[key] = lot + collected_this_bucket += len(page1_lots) + if not page1_lots: + logger.info("cian: %s [%d, %s] page=1 empty — early stop", room_label, lo, hi) + return + + for page in range(2, max_pages + 1): + page_html: str | None = None + try: + page_html = await self._fetch_page_html(rooms, page, lo if lo > 0 else None, hi) + await self.sleep_between_requests() + except Exception: + logger.exception( + "cian: fetch failed %s [%d, %s] page=%d — rotating IP", + room_label, + lo, + hi, + page, + ) + await self._rotate_ip() + try: + page_html = await self._fetch_page_html(rooms, page, lo if lo > 0 else None, hi) + await self.sleep_between_requests() + except Exception: + logger.error( + "cian: retry failed %s [%d, %s] page=%d — breaking bucket", + room_label, + lo, + hi, + page, + ) + break + + if page_html is None: + # Повторная попытка после ротации + rotated = await self._rotate_ip() + if rotated: + try: + page_html = await self._fetch_page_html( + rooms, page, lo if lo > 0 else None, hi + ) + await self.sleep_between_requests() + except Exception: + pass + if page_html is None: + logger.error( + "cian: page_html=None %s [%d, %s] page=%d — breaking bucket", + room_label, + lo, + hi, + page, + ) + break + + page_lots = self._parse_serp_html(page_html) + if not page_lots: + logger.info( + "cian: %s [%d, %s] page=%d empty — early stop", room_label, lo, hi, page + ) + break + + for lot in page_lots: + key = lot.source_id or lot.source_url + if key: + seen[key] = lot + collected_this_bucket += len(page_lots) + + logger.info( + "cian: %s [%d, %s] paginated=%d/%d pages collected=%d unique_total=%d", + room_label, + lo, + hi, + min(max_pages, page if "page" in locals() else 1), + max_pages, + collected_this_bucket, + len(seen), + ) + def _parse_serp_html(self, html: str) -> list[ScrapedLot]: """Извлечь offers из Cian Redux state. diff --git a/tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py b/tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py new file mode 100644 index 00000000..71ea9e89 --- /dev/null +++ b/tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py @@ -0,0 +1,206 @@ +"""Тесты exhaustive-загрузки Cian: fetch_all_secondary + _walk_price_range. + +Мокаем сетевые вызовы (_fetch_page_html, _extract_total_offers, sleep_between_requests) +для детерминированного контроля поведения без реального HTTP. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from app.services.scrapers.base import ScrapedLot +from app.services.scrapers.cian import _MAX_PRICE, _MIN_BRACKET, CianScraper + + +def _make_lot(source_id: str, price: int = 3_000_000) -> ScrapedLot: + """Минимальный ScrapedLot для теста.""" + return ScrapedLot( + source="cian", + source_url=f"https://ekb.cian.ru/sale/flat/{source_id}/", + source_id=source_id, + address="Екатеринбург", + price_rub=price, + ) + + +def _make_html_stub(total_offers: int, lot_ids: list[str]) -> str: + """Заглушка HTML — не парсится напрямую; используется через мокнутые методы.""" + return f"total={total_offers} ids={lot_ids}" + + +@pytest.fixture +def scraper() -> CianScraper: + """CianScraper без реального __aenter__ (без HTTP warm-up).""" + s = CianScraper.__new__(CianScraper) + s.name = "cian" + s.base_url = "https://ekb.cian.ru" + s.request_delay_sec = 0.0 + s._cffi = MagicMock() # не None → assert self._cffi is not None пройдёт + return s + + +@pytest.mark.asyncio +async def test_fetch_all_secondary_splits_on_cap(scraper: CianScraper) -> None: + """totalOffers > cap → бакет разбивается; в итоге оба суб-бакета пагинируются.""" + # Настройка: rooms=(1,), один бакет [0, None] + # Probe [0, None]: totalOffers=3000 > cap=1400 → split на [0, MAX_PRICE] + # Probe [0, MAX_PRICE]: totalOffers=3000 > 1400 → split на [0, mid] и [mid+1, MAX_PRICE] + # Probe [0, mid] (mid = MAX_PRICE//2 = 100_000_000): totalOffers=800 <= cap → пагинировать + # Probe [mid+1, MAX_PRICE]: totalOffers=700 <= cap → пагинировать + + mid = _MAX_PRICE // 2 + + call_log: list[tuple] = [] + + async def fake_fetch_page_html( + rooms: tuple, page: int, min_price: int | None, max_price: int | None + ) -> str: + call_log.append((rooms, page, min_price, max_price)) + return f"page={page} min={min_price} max={max_price}" + + def fake_extract_total_offers(html: str) -> int | None: + # Парсим min/max из fake html + import re + + min_m = re.search(r"min=(\w+)", html) + max_m = re.search(r"max=(\w+)", html) + min_p = int(min_m.group(1)) if min_m and min_m.group(1) != "None" else 0 + max_p = int(max_m.group(1)) if max_m and max_m.group(1) != "None" else _MAX_PRICE + + # Имитируем разделение: весь диапазон > 1400, левая и правая половины <= 1400 + if max_p is None or max_p >= _MAX_PRICE: + if min_p == 0: + return 3000 # весь диапазон — нужно split + if max_p == mid: + return 800 # левая половина — пагинировать + if min_p == mid + 1: + return 700 # правая половина — пагинировать + return 3000 # всё остальное — split + + def fake_parse_serp_html(html: str) -> list[ScrapedLot]: + import re + + page_m = re.search(r"page=(\d+)", html) + min_m = re.search(r"min=(\w+)", html) + max_m = re.search(r"max=(\w+)", html) + page = int(page_m.group(1)) if page_m else 1 + min_p = min_m.group(1) if min_m else "0" + max_p = max_m.group(1) if max_m else "MAX" + # Возвращаем по 2 лота на первой странице каждого пагинируемого бакета + if page == 1: + return [ + _make_lot(f"lot_{min_p}_{max_p}_p{page}_1"), + _make_lot(f"lot_{min_p}_{max_p}_p{page}_2"), + ] + return [] # страница 2+ — пусто → early stop + + async def fake_sleep(*_: object, **__: object) -> None: + pass + + with ( + patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), + patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), + patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), + patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), + patch.object(scraper, "_rotate_ip", return_value=False), + ): + lots = await scraper.fetch_all_secondary( + rooms_buckets=[(1,)], + price_cap_per_bucket=1400, + ) + + # Должны получить лоты из ОБОИХ суб-бакетов после split + assert len(lots) > 0, "Должны получить лоты после разделения бакета" + + # Проверяем что был probe с большим диапазоном И с sub-бакетами + probed_ranges = [(min_p, max_p) for (_, page, min_p, max_p) in call_log if page == 1] + # Должен быть probe [0, None] или [0, MAX] — начальный probe всего диапазона + initial_probes = [r for r in probed_ranges if r[0] is None or r[0] == 0] + assert ( + len(initial_probes) >= 1 + ), f"Ожидался probe всего диапазона, probed_ranges={probed_ranges}" + + +@pytest.mark.asyncio +async def test_fetch_all_secondary_paginates_when_under_cap(scraper: CianScraper) -> None: + """totalOffers <= cap → пагинирует бакет полностью (страница за страницей).""" + pages_fetched: list[int] = [] + + async def fake_fetch_page_html( + rooms: tuple, page: int, min_price: int | None, max_price: int | None + ) -> str: + pages_fetched.append(page) + return f"page={page}" + + def fake_extract_total_offers(html: str) -> int | None: + return 56 # 56 офферов → ceil(56/28) = 2 страницы + + def fake_parse_serp_html(html: str) -> list[ScrapedLot]: + import re + + page_m = re.search(r"page=(\d+)", html) + page = int(page_m.group(1)) if page_m else 1 + if page <= 2: + return [_make_lot(f"lot_p{page}_{i}") for i in range(28)] + return [] + + with ( + patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), + patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), + patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), + patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), + patch.object(scraper, "_rotate_ip", return_value=False), + ): + lots = await scraper.fetch_all_secondary( + rooms_buckets=[(2,)], + price_cap_per_bucket=1400, + ) + + # 28 лотов на стр 1 + 28 на стр 2 = 56, дедуп по source_id + assert len(lots) == 56, f"Ожидалось 56 уникальных лотов, получено {len(lots)}" + # Должно быть 2 страницы пагинации (page=1 probe + page=2) + assert 2 in pages_fetched, "Должна быть запрошена страница 2" + + +@pytest.mark.asyncio +async def test_fetch_all_secondary_min_bracket_guard(scraper: CianScraper) -> None: + """Если hi - lo < MIN_BRACKET → пагинируем как есть (не делим бесконечно).""" + call_count = 0 + + async def fake_fetch_page_html( + rooms: tuple, page: int, min_price: int | None, max_price: int | None + ) -> str: + nonlocal call_count + call_count += 1 + if call_count > 100: + raise RuntimeError("Бесконечная рекурсия — MIN_BRACKET guard не сработал") + return f"page={page}" + + def fake_extract_total_offers(html: str) -> int | None: + # Всегда > cap → без guard делилось бы бесконечно + return 5000 + + def fake_parse_serp_html(html: str) -> list[ScrapedLot]: + return [] # пустой → early stop в пагинации + + with ( + patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), + patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), + patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), + patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), + patch.object(scraper, "_rotate_ip", return_value=False), + ): + # Запускаем с очень узким диапазоном — guard должен остановить деление + await scraper._walk_price_range( + rooms=(1,), + lo=1_000_000, + hi=1_000_000 + _MIN_BRACKET - 1, # bracket < MIN_BRACKET + seen={}, + price_cap_per_bucket=1400, + max_pages_per_bucket=54, + ) + + # Тест прошёл если нет RuntimeError (guard сработал) + assert call_count <= 10, f"Слишком много вызовов ({call_count}) — guard не сработал"