feat(tradein): exhaustive Cian EKB secondary load — room×price partitioning #926
4 changed files with 842 additions and 0 deletions
|
|
@ -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 ─────────────────
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
206
tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py
Normal file
206
tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py
Normal file
|
|
@ -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"<html>total={total_offers} ids={lot_ids}</html>"
|
||||
|
||||
|
||||
@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"<html>page={page} min={min_price} max={max_price}</html>"
|
||||
|
||||
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"<html>page={page}</html>"
|
||||
|
||||
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"<html>page={page}</html>"
|
||||
|
||||
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 не сработал"
|
||||
Loading…
Add table
Reference in a new issue