"""DomClick.ru scraper — вторичка через bff-search-web JSON API (Layer A). Стратегия (JSON API rewrite, replaces dead HTML/camoufox scrape): GET https://bff-search-web.domclick.ru/api/offers/v1?... → list (up to 20 items) GET https://bff-search-web.domclick.ru/api/offers/count/v1?... → snippetsCount Транспорт: BrowserFetcher(source="domclick") — существующий браузерный пул (tradein-browser), роутит bff-search-web.domclick.ru → generic provider → shared RU mobile proxy. fetch() возвращает raw JSON body как строку. Bucketing: OFFSET_CAP=2000 (max 100 страниц × 20) — hardlimit DomClick API. Когда snippetsCount > 2000 — рекурсивная бинарная partition по цене. EKB вторичка ≈6400 лотов; только rooms=2 превышает cap (≈2215). Layer B (detail-backfill) и Layer C (analytics XHR) — не реализованы. """ from __future__ import annotations import json import logging from datetime import UTC, date, datetime from typing import Any from urllib.parse import urlencode, urljoin import sentry_sdk from app.core.config import settings from app.services.scraper_settings import get_scraper_delay from app.services.scrapers.base import BaseScraper, ScrapedLot from app.services.scrapers.domclick_exceptions import DomClickBlockedError logger = logging.getLogger(__name__) # ── Константы bucketing ────────────────────────────────────────────────────── # Верхняя граница первого деления открытого (hi=None) бакета (30 млн ₽) _HIGH_ANCHOR: int = 30_000_000 # Минимальная ширина ценового диапазона — ниже этого не бисектируем _MIN_BAND: int = 100_000 # Максимальная глубина рекурсии _MAX_DEPTH: int = 12 # ── Маппинг комнатности int → API token ───────────────────────────────────── _ROOMS_TOKEN: dict[int, str] = { 0: "st", 1: "1", 2: "2", 3: "3", 4: "4", 5: "5+", } # ── Гео-guard: ЕКБ bbox ───────────────────────────────────────────────────── _EKB_LAT_MIN: float = 56.7 _EKB_LAT_MAX: float = 56.95 _EKB_LON_MIN: float = 60.4 _EKB_LON_MAX: float = 60.8 def _is_ekb(offer: dict[str, Any]) -> bool: """Гео-guard: офер принадлежит ЕКБ по offerRegionName или bbox координат.""" if offer.get("offerRegionName") == "Екатеринбург": return True loc: dict[str, Any] = offer.get("location") or {} lat_raw = loc.get("lat") lon_raw = loc.get("lon") if lat_raw is not None and lon_raw is not None: try: lat = float(lat_raw) lon = float(lon_raw) return _EKB_LAT_MIN <= lat <= _EKB_LAT_MAX and _EKB_LON_MIN <= lon <= _EKB_LON_MAX except (TypeError, ValueError): pass return False def _parse_publish_date(val: Any) -> date | None: """Парсим publishedDate: epoch int/str или ISO8601 строка. None при ошибке.""" if val is None: return None try: if isinstance(val, int | float): return datetime.fromtimestamp(int(val), tz=UTC).date() s = str(val).strip() if s.isdigit(): return datetime.fromtimestamp(int(s), tz=UTC).date() # ISO8601 — берём первые 10 символов (YYYY-MM-DD) return date.fromisoformat(s[:10]) except Exception: return None # ── DomClickScraper ────────────────────────────────────────────────────────── class DomClickScraper(BaseScraper): """DomClick вторичка через bff-search-web JSON API. Источник = 'domklik'. Layer A: SERP list (offers/v1) → ScrapedLot. Layer B/C не реализованы. Транспорт: одна BrowserFetcher(source="domclick") сессия в __aenter__. fetch_around() → NotImplementedError (DomClick не поддерживает geo-radius). """ name = "domklik" source = "domklik" base_url = "https://bff-search-web.domclick.ru" request_delay_sec = 8.0 # overwritten from get_scraper_delay in __init__ EKB_REGION_GUID = "0d475b79-88de-4054-818c-37d8f9d0d440" EKB_AIDS = 20561 OFFSET_CAP = 2000 LIMIT = 20 def __init__(self) -> None: super().__init__() self.request_delay_sec = get_scraper_delay(self.name) self.parse_failures: int = 0 self._browser: Any | None = None async def __aenter__(self) -> DomClickScraper: from app.services.scrapers.browser_fetcher import BrowserFetcher await super().__aenter__() self._browser = BrowserFetcher(source="domclick") await self._browser.__aenter__() return self async def __aexit__(self, *args: Any) -> None: if self._browser is not None: await self._browser.__aexit__(*args) self._browser = None await super().__aexit__(*args) # ── fetch_around — stub ─────────────────────────────────────────────────── async def fetch_around(self, lat: float, lon: float, radius_m: int = 1000) -> list[ScrapedLot]: """DomClick не поддерживает geo-radius. Используй fetch_city().""" raise NotImplementedError( "DomClick не поддерживает geo-radius; используй fetch_city(city_id=...)" ) # ── Основной метод ──────────────────────────────────────────────────────── async def fetch_city( self, city_id: int, rooms: list[int] | None = None, pages: int = 100, ) -> list[ScrapedLot]: """Citywide sweep: rooms × recursive price-bands → дедуплицированный список. Args: city_id: числовой ID города (4 = ЕКБ). Только ЕКБ поддерживается; для других значений логируем warning, продолжаем с EKB GUID. rooms: список комнатностей (0=студия, 1, 2, 3, 4, 5=5+). None → [0,1,2,3,4,5] (все). pages: максимум страниц на leaf-bucket (default 100 = OFFSET_CAP/LIMIT). Returns: Дедуплицированный по source_id список ScrapedLot. """ if city_id != 4: logger.warning( "domklik: city_id=%d не поддерживается — используем EKB GUID/aids", city_id, ) _rooms = rooms if rooms is not None else list(_ROOMS_TOKEN.keys()) seen: dict[str, ScrapedLot] = {} total_geo_dropped: int = 0 for room_int in _rooms: room_token = _ROOMS_TOKEN.get(room_int, str(room_int)) logger.info( "domklik: room=%s (%d) sweep start (max %d pages per leaf)", room_token, room_int, pages, ) geo_dropped_ref = [0] await self._walk_price_range( room_token=room_token, lo=0, hi=None, seen=seen, pages=pages, geo_dropped_ref=geo_dropped_ref, ) total_geo_dropped += geo_dropped_ref[0] logger.info( "domklik: room=%s done — unique_total=%d geo_dropped_this_room=%d", room_token, len(seen), geo_dropped_ref[0], ) if total_geo_dropped: logger.info("domklik: fetch_city geo_dropped_total=%d", total_geo_dropped) logger.info( "domklik: fetch_city done city_id=%d total=%d parse_failures=%d", city_id, len(seen), self.parse_failures, ) return list(seen.values()) # ── Рекурсивное адаптивное деление ─────────────────────────────────────── async def _walk_price_range( self, *, room_token: str, lo: int, hi: int | None, seen: dict[str, ScrapedLot], pages: int, geo_dropped_ref: list[int], _depth: int = 0, ) -> None: """Рекурсивная бинарная partition ценового диапазона [lo, hi]. Open band (hi is None): snippetsCount ≤ OFFSET_CAP → пагинируем leaf. snippetsCount > OFFSET_CAP → разбиваем через HIGH_ANCHOR: recurse [lo, HIGH_ANCHOR] + [HIGH_ANCHOR+1, None]. Closed band (hi is int): snippetsCount ≤ OFFSET_CAP → пагинируем leaf. snippetsCount > OFFSET_CAP и (depth > MAX_DEPTH или band < MIN_BAND): пагинируем как есть (tail-loss WARNING). Иначе → mid=(lo+hi)//2, recurse [lo, mid] и [mid+1, hi]. Raises DomClickBlockedError при QRATOR-блоке (propagate к вызывающему). """ hi_repr = "open" if hi is None else str(hi) # ── 1. COUNT ───────────────────────────────────────────────────────── count_url = self._build_count_url(room_token, lo, hi) count_data = await self._fetch_json(count_url) # raises DomClickBlockedError on block snippets: int = (count_data.get("result") or {}).get("snippetsCount", 0) logger.info( "domklik: %s [%d, %s] snippetsCount=%d depth=%d", room_token, lo, hi_repr, snippets, _depth, ) if snippets == 0: return await self.sleep_between_requests() # ── 2a. Open band — нельзя бисектировать напрямую ──────────────────── if hi is None: if snippets <= self.OFFSET_CAP: await self._paginate_leaf( room_token, lo, None, snippets, seen, pages, geo_dropped_ref ) else: # Делим через HIGH_ANCHOR; хвост [HIGH_ANCHOR+1, None] обычно мал await self._walk_price_range( room_token=room_token, lo=lo, hi=_HIGH_ANCHOR, seen=seen, pages=pages, geo_dropped_ref=geo_dropped_ref, _depth=_depth + 1, ) await self._walk_price_range( room_token=room_token, lo=_HIGH_ANCHOR + 1, hi=None, seen=seen, pages=pages, geo_dropped_ref=geo_dropped_ref, _depth=_depth + 1, ) return # ── 2b. Closed band ─────────────────────────────────────────────────── if snippets <= self.OFFSET_CAP: await self._paginate_leaf(room_token, lo, hi, snippets, seen, pages, geo_dropped_ref) return # Нужно деление — проверяем ограничения band = hi - lo if _depth >= _MAX_DEPTH or band < _MIN_BAND: tail_loss = max(0, snippets - self.OFFSET_CAP) logger.warning( "domklik: %s [%d, %d] snippets=%d > cap=%d band=%d depth=%d " "— paginating as-is (tail-loss ~%d)", room_token, lo, hi, snippets, self.OFFSET_CAP, band, _depth, tail_loss, ) await self._paginate_leaf(room_token, lo, hi, snippets, seen, pages, geo_dropped_ref) return mid = (lo + hi) // 2 await self._walk_price_range( room_token=room_token, lo=lo, hi=mid, seen=seen, pages=pages, geo_dropped_ref=geo_dropped_ref, _depth=_depth + 1, ) await self._walk_price_range( room_token=room_token, lo=mid + 1, hi=hi, seen=seen, pages=pages, geo_dropped_ref=geo_dropped_ref, _depth=_depth + 1, ) # ── Пагинация leaf-bucket ───────────────────────────────────────────────── async def _paginate_leaf( self, room_token: str, lo: int, hi: int | None, snippets: int, seen: dict[str, ScrapedLot], pages: int, geo_dropped_ref: list[int], ) -> None: """Последовательная пагинация одного leaf-bucket [lo, hi] по offset.""" max_offset = min(snippets, self.OFFSET_CAP, pages * self.LIMIT) offset = 0 while offset < max_offset: list_url = self._build_list_url(room_token, lo, hi, offset) list_data = await self._fetch_json(list_url) # DomClickBlockedError propagates items: list[dict[str, Any]] = (list_data.get("result") or {}).get("items") or [] if not items: logger.debug( "domklik: empty items at %s [%s, %s] offset=%d — stopping", room_token, lo, hi or "open", offset, ) break page_lots = self._parse_page(items, room_token, geo_dropped_ref) for lot in page_lots: key = lot.source_id or lot.source_url if key: seen[key] = lot logger.debug( "domklik: %s [%s, %s] offset=%d items=%d parsed=%d total_unique=%d", room_token, lo, hi or "open", offset, len(items), len(page_lots), len(seen), ) offset += self.LIMIT if offset < max_offset: await self.sleep_between_requests() # ── Парсинг страницы офферов ────────────────────────────────────────────── def _parse_page( self, items: list[dict[str, Any]], room_token: str, geo_dropped_ref: list[int], ) -> list[ScrapedLot]: """Парсинг списка офферов → ScrapedLot с гео-guard. Эмитирует Sentry-сообщение если raw_count > 0 но 0 прошло парсинг (guard от silent schema regression). """ result: list[ScrapedLot] = [] local_geo_dropped = 0 for offer in items: if not _is_ekb(offer): local_geo_dropped += 1 geo_dropped_ref[0] += 1 continue lot = self._offer_to_lot(offer, room_token) if lot is not None: result.append(lot) raw_count = len(items) ekb_count = raw_count - local_geo_dropped saved = len(result) # Schema regression guard if raw_count > 0 and saved == 0 and ekb_count > 0: logger.error( "domklik: 0/%d offers прошли _offer_to_lot (ekb=%d) " "— возможна schema regression", raw_count, ekb_count, ) try: if settings.glitchtip_dsn: sentry_sdk.capture_message( f"domklik SERP: {raw_count} offers, 0 parsed (ekb={ekb_count})" " — possible schema regression", level="error", ) except Exception: pass # sentry_sdk не инициализирован в dev return result def _offer_to_lot(self, offer: dict[str, Any], room_token: str) -> ScrapedLot | None: """Парсинг одного DomClick bff-JSON оффера → ScrapedLot. Маппинг полей per спецификации Layer A. source='domklik', listing_segment='vtorichka' всегда. """ try: # ── Identity ───────────────────────────────────────────────────── offer_id = offer.get("id") if not offer_id: return None source_id = str(offer_id) path = offer.get("path") or "" source_url = urljoin("https://domclick.ru", path) # ── Цена ───────────────────────────────────────────────────────── price_raw = offer.get("price") if price_raw is None: return None try: price_rub = int(price_raw) except (TypeError, ValueError): return None if price_rub <= 0: return None # ── squarePrice (price_per_m2) ──────────────────────────────────── sq_price_raw = offer.get("squarePrice") price_per_m2: int | None = None if sq_price_raw is not None: try: price_per_m2 = int(sq_price_raw) except (TypeError, ValueError): pass # ── Координаты ─────────────────────────────────────────────────── location: dict[str, Any] = offer.get("location") or {} lat: float | None = None lon: float | None = None try: if location.get("lat") is not None: lat = float(location["lat"]) if location.get("lon") is not None: lon = float(location["lon"]) except (TypeError, ValueError): lat = lon = None # ── Адрес ──────────────────────────────────────────────────────── address_obj: dict[str, Any] = offer.get("address") or {} address: str | None = address_obj.get("displayName") or None # ── Параметры квартиры ──────────────────────────────────────────── obj_info: dict[str, Any] = offer.get("objectInfo") or {} area_raw = obj_info.get("area") area_m2: float | None = None if area_raw is not None: try: area_m2 = float(area_raw) except (TypeError, ValueError): pass # Комнатность: студии → 0; числовые → из objectInfo.rooms или токена if room_token == "st": rooms: int | None = 0 else: rooms_raw = obj_info.get("rooms") if rooms_raw is not None: try: rooms = int(rooms_raw) except (TypeError, ValueError): rooms = None else: # fallback: числовой токен try: rooms = int(room_token) except ValueError: rooms = None floor_raw = obj_info.get("floor") floor: int | None = None if floor_raw is not None: try: floor = int(floor_raw) except (TypeError, ValueError): pass # ── Здание ─────────────────────────────────────────────────────── house: dict[str, Any] = offer.get("house") or {} total_floors_raw = house.get("floors") total_floors: int | None = None if total_floors_raw is not None: try: total_floors = int(total_floors_raw) except (TypeError, ValueError): pass year_built_raw = house.get("buildYear") year_built: int | None = None if year_built_raw is not None: try: year_built = int(year_built_raw) except (TypeError, ValueError): pass # ── Дата публикации ─────────────────────────────────────────────── published_date = _parse_publish_date(offer.get("publishedDate")) # ── Raw payload (компактный) ─────────────────────────────────────── flat_complex: dict[str, Any] = offer.get("flatComplex") or {} raw_payload: dict[str, Any] = { "is_apartments": obj_info.get("isApartment"), "is_rosreestr_approved": offer.get("isRosreestrApproved"), "offer_region_name": offer.get("offerRegionName"), "updated_date": offer.get("updatedDate"), "last_price_history_state": offer.get("lastPriceHistoryState"), "flat_complex": ( { "id": flat_complex.get("id"), "name": flat_complex.get("name"), "slug": flat_complex.get("slug"), } if flat_complex else None ), } return ScrapedLot( source="domklik", source_url=source_url, source_id=source_id, address=address, lat=lat, lon=lon, rooms=rooms, area_m2=area_m2, floor=floor, total_floors=total_floors, year_built=year_built, price_rub=price_rub, price_per_m2=price_per_m2, listing_segment="vtorichka", listing_date=published_date, publish_date=published_date, raw_payload=raw_payload, ) except Exception: self.parse_failures += 1 logger.warning( "domklik _offer_to_lot: parse failed offer_id=%r (failures=%d)", offer.get("id"), self.parse_failures, exc_info=True, ) return None # ── URL builders ────────────────────────────────────────────────────────── def _base_params(self, room_token: str, lo: int, hi: int | None) -> list[tuple[str, Any]]: """Базовые query params (без offset/limit). Используется count и list URL.""" params: list[tuple[str, Any]] = [ ("address", self.EKB_REGION_GUID), ("aids", self.EKB_AIDS), ("deal_type", "sale"), ("category", "living"), ("offer_type", "flat"), ("rooms", room_token), ("sort", "qi"), ] if lo > 0: params.append(("sale_price__gte", lo)) if hi is not None: params.append(("sale_price__lte", hi)) return params def _build_count_url(self, room_token: str, lo: int, hi: int | None) -> str: """URL для count/v1 — без offset/limit.""" params = self._base_params(room_token, lo, hi) return f"{self.base_url}/api/offers/count/v1?{urlencode(params)}" def _build_list_url(self, room_token: str, lo: int, hi: int | None, offset: int) -> str: """URL для offers/v1 — с offset/limit.""" params = self._base_params(room_token, lo, hi) params.append(("offset", offset)) params.append(("limit", self.LIMIT)) return f"{self.base_url}/api/offers/v1?{urlencode(params)}" # ── JSON fetch helper ───────────────────────────────────────────────────── async def _fetch_json(self, url: str) -> dict[str, Any]: """Fetch URL через BrowserFetcher → parse JSON. Raises: DomClickBlockedError: ответ не JSON (QRATOR challenge HTML), или BrowserFetcher вернул HTML-страницу. """ assert self._browser is not None, "Используй DomClickScraper как async context manager" raw: str = await self._browser.fetch(url) # QRATOR block / captcha возвращает HTML — детектируем по первому символу stripped = raw.lstrip() if stripped.startswith("<"): logger.warning( "domklik: HTML challenge url=%s (первые 300 символов): %r", url, raw[:300], ) raise DomClickBlockedError("DomClick вернул HTML вместо JSON (QRATOR block)") try: return json.loads(raw) except json.JSONDecodeError as exc: logger.warning("domklik: JSONDecodeError url=%s: %s", url, exc) raise DomClickBlockedError(f"DomClick: не JSON ответ: {exc}") from exc # ── Convenience runner (для Celery tasks) ──────────────────────────────────── async def scrape_domclick_city( city_id: int, rooms: list[int] | None = None, pages: int = 100, ) -> list[ScrapedLot]: """Удобная точка входа для вызова из Celery tasks. Пример:: import asyncio lots = asyncio.run(scrape_domclick_city(city_id=4, rooms=[1, 2, 3])) """ async with DomClickScraper() as scraper: return await scraper.fetch_city(city_id=city_id, rooms=rooms, pages=pages)