"""Cian.ru scraper — state-based SERP parser (137 fields per offer). Стратегия (Stage 3 рефактор): - Cian React micro-frontends хранят state в window._cianConfig['frontend-serp'].push(). - URL pattern: https://ekb.cian.ru/cat.php?deal_type=sale&offer_type=flat&engine_version=2 ekb.cian.ru — city-specific subdomain для ЕКБ (per Schema_Cian_SERP_Inventory sec 13). ВАЖНО: Циан блокирует httpx/curl по TLS+fingerprint и сыплет Google reCAPTCHA на датацентр-IP. Транспорт SERP переведён на BrowserFetcher (camoufox real-browser fingerprint через tradein-browser сервис, #1806 / epic #883 Phase 2). Браузер отдаёт полный SSR-HTML с Redux initialState — парсинг state не меняется. Мобильный прокси (ha.mobileproxy.space) применяется server-side в браузер-контейнере и ротирует IP сам, поэтому code-side _rotate_ip (changeip) больше не нужен. НЕ используем 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'ы дают одну выдачу). Полный сбор = ОДИН проход с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление диапазона). """ from __future__ import annotations import asyncio import hashlib import inspect import logging import math import re from collections.abc import Callable from datetime import UTC, datetime from typing import TYPE_CHECKING, Any from urllib.parse import urlencode from scraper_kit.base import BaseScraper, ScrapedLot from scraper_kit.browser_fetcher import BrowserFetcher from scraper_kit.cian_state_parser import extract_state from scraper_kit.house_type_normalizer import normalize_house_type from scraper_kit.price_brackets import get_price_seed_brackets from scraper_kit.pricing import BisectionConfig, ProbeFailPolicy, ProbeResult, walk_price_range from scraper_kit.repair_state_normalizer import ( infer_repair_state_from_text, normalize_repair_state, ) if TYPE_CHECKING: from scraper_kit.contracts import ProxyProvider, ScraperConfig logger = logging.getLogger(__name__) # Регион 4743 = Свердловская область (Cian internal region ID) CIAN_EKB_REGION_ID = 4743 # SERP MFE name и state key (per Schema sec 1.2) _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 # Конфиг единого движка бисекции (scraper_kit.pricing) для cian. cap задаётся # per-call (price_cap_per_bucket) — здесь только структурные константы/политики. # probe_fail=SKIP: totalOffers недоступен после ретрая → бакет пропускается. # open_split_ceiling=None: открытый брекет пагинируется напрямую (не делится). def _cian_bisection_config(cap: int) -> BisectionConfig: return BisectionConfig( cap=cap, min_bracket=_MIN_BRACKET, probe_fail_policy=ProbeFailPolicy.SKIP, ) # Комнатности по умолчанию для 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. Тянет SSR-HTML через BrowserFetcher (camoufox). Извлекает 137 полей на offer из Redux initialState через cian_state_parser. Координаты точные — НЕТ anchor jitter (offer.geo.coordinates). Транспорт: одна BrowserFetcher(source="cian") сессия открывается в __aenter__ и переиспользуется на всех page-fetch'ах (#1806). camoufox-fingerprint + мобильный прокси server-side обходят Cian reCAPTCHA; IP ротирует сам прокси. """ name = "cian" # ekb.cian.ru — city-specific subdomain для ЕКБ (per Schema sec 13, closed Q4) base_url = "https://ekb.cian.ru" # Класс-дефолт; реальное значение загружается из scraper_settings при создании экземпляра. request_delay_sec = 5.0 # консервативно: Cian менее агрессивен чем Avito, но 5s безопасно def __init__( self, config: ScraperConfig, *, delay_provider: Callable[[str], float] | None = None, proxy_provider: ProxyProvider | None = None, ) -> None: super().__init__() # Strangler-инжекция (#2133): конфиг и провайдер задержки приходят снаружи # вместо прямого импорта app.core.config.settings / # app.services.scraper_settings.get_scraper_delay. Продуктовая сторона # передаёт RealScraperConfig + get_scraper_delay; kit не знает про app / БД. self._config = config if delay_provider is not None: self.request_delay_sec = delay_provider(self.name) # #2160: пул прокси для camoufox-браузера (за флагом use_proxy_pool_browser). # None → BrowserFetcher без пула = env-прокси браузера (ship-dark, как сейчас). self._proxy_provider = proxy_provider self._browser: BrowserFetcher | None = None async def __aenter__(self) -> CianScraper: await super().__aenter__() # Одна BrowserFetcher(source="cian") сессия на весь scrape — camoufox real-browser # fingerprint через tradein-browser сервис (#1806). Сервер роутит /fetch на # отдельный браузер+мобильный прокси для cian. Прокси ротирует IP сам, поэтому # code-side warm-up cookies и changeip-ротация больше не нужны. self._browser = BrowserFetcher( source="cian", endpoint=self._config.browser_http_endpoint, proxy_provider=self._proxy_provider, use_pool=self._config.use_proxy_pool_browser, ) 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) # ── Anti-block: IP rotation ─────────────────────────────────────────────── async def _rotate_ip(self) -> bool: """No-op после миграции на BrowserFetcher (#1806). Мобильный прокси в браузер-контейнере ротирует IP сам, code-side changeip больше не нужен. Метод сохранён (returns False) для обратной совместимости с вызовом в _walk_price_range и тестами, которые его мокают. """ return False async def fetch_around( self, lat: float, lon: float, radius_m: int = 1000, rooms: tuple[int, ...] | None = None, page: int = 1, ) -> list[ScrapedLot]: """Найти объявления Циан по ЕКБ (lat/lon для reference; Cian не поддерживает bbox). Cian SERP отдаёт весь ЕКБ — фильтрация по радиусу происходит в postgres через ST_DWithin после сохранения (точные coords из state). rooms — список из 1,2,3,4 (Cian's room codes). Если None — все. page — страница выдачи (Cian ~28 объявлений на страницу). """ url = self._build_url(rooms, page) try: assert self._browser is not None html = await self._browser.fetch(url) except Exception: logger.exception("cian BrowserFetcher fetch failed for url=%s", url) return [] lots = self._parse_serp_html(html) logger.info( "cian: %d lots fetched rooms=%s page=%d url=%s", len(lots), rooms, page, url, ) await self.sleep_between_requests() return lots async def fetch_around_multi_room( self, lat: float, lon: float, radius_m: int = 1000, pages: int = 8, ) -> list[ScrapedLot]: """Скрейп Циан по 1к/2к/3к/4+ × N страниц — расширяет выборку. 4 комнаты × N страниц × ~28 = до ~330+ лотов до дедупликации. """ seen: dict[str, ScrapedLot] = {} for rooms in ((1,), (2,), (3,), (4,)): for page in range(1, pages + 1): try: lots = await self.fetch_around(lat, lon, radius_m, rooms=rooms, page=page) except Exception: logger.exception("cian multi-room fetch failed rooms=%s page=%d", rooms, page) continue if not lots: break # пустая страница → дальше смысла нет for lot in lots: key = lot.source_id or lot.source_url if key and key not in seen: seen[key] = lot logger.info( "cian multi-room: %d unique lots (lat=%.4f lon=%.4f radius=%dm)", len(seen), lat, lon, radius_m, ) return list(seen.values()) def _build_url( 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"), ("engine_version", "2"), ("offer_type", "flat"), ("region", CIAN_EKB_REGION_ID), ("sort", "creation_date_desc"), ] 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-страницы через BrowserFetcher, возвращает HTML или None при ошибке.""" assert self._browser is not None url = self._build_url(rooms, page, min_price, max_price) try: return await self._browser.fetch(url) except Exception: logger.exception("cian: BrowserFetcher fetch failed url=%s", url) return None 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, concurrency: int = 5, secondary_only: bool = True, on_bucket: Callable[..., Any] | None = None, on_progress: Callable[[int], None] | None = None, skip_buckets: set[str] | None = None, ) -> list[ScrapedLot]: """Exhaustive-загрузка Cian ЕКБ вторички через партиционирование КОМНАТНОСТЬ × ЦЕНА. Обходит Cian SERP-cap (~54 стр/запрос ≈ 1500 результатов на запрос). Вместо одной корень-бисекции [0, _MAX_PRICE] на комнатность (холостые probe-спуски через пустой верх) — стартуем с общей data-driven seed-сетки ценовых брекетов (get_price_seed_brackets() — shared EKB grid, общая с avito/yandex). Каждый (комнатность × seed-брекет) дальше дробится _walk_price_range если totalOffers > price_cap_per_bucket. Последний seed-брекет ОТКРЫТ (hi=None → maxprice не ставится) — ловит весь хвост люкса без потолка. Страницы внутри leaf-бакета запрашиваются параллельно (asyncio.gather + Semaphore). Дедуп по source_id (dict seen), ОБЩИЙ по всем брекетам и комнатностям. Параметры: rooms_buckets: список room-кодов Cian для перебора (default: 1-6). price_cap_per_bucket: максимум офферов в бакете перед делением (< 1500). max_pages_per_bucket: Cian hard cap ~54; не превышать. concurrency: максимум параллельных page-фетчей в leaf-бакете (default=5). secondary_only: если True (default) — отбрасывает новостройки (listing_segment=="novostroyki") после парсинга, до save/on_bucket. object_type=1 SERP-параметр Cian ненадёжен (~5% выдачи), поэтому фильтруем по authoritative listing_segment из offer.newbuilding.id. on_bucket: опциональный callback(bucket_key, list[ScrapedLot]) после каждого leaf-бакета. Может быть async или sync. Если кидает исключение — прерывает прогон. on_progress: опциональный callback(unique_count) для heartbeat (per room-bucket). skip_buckets: множество ключей «room_label:lo:hi» уже завершённых бакетов — пагинация и on_bucket для них пропускаются. Probe-запросы (для split-решения) всё равно выполняются (их мало по сравнению с пагинацией). Возвращает 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 concurrency=%d secondary_only=%s)", room_label, price_cap_per_bucket, concurrency, secondary_only, ) before = len(seen) # Перебираем seed-брекеты: для ЗАКРЫТЫХ передаём hi = br_hi - 1 # (maxprice эксклюзивный сверху → соседние брекеты не пересекаются), для # ОТКРЫТОГО (br_hi is None) — hi=None (без потолка). Дедуп по source_id # в общем `seen` страхует на стыках. for br_lo, br_hi in get_price_seed_brackets(): walk_hi = br_hi - 1 if br_hi is not None else None await self._walk_price_range( rooms=rooms, lo=br_lo, hi=walk_hi, seen=seen, price_cap_per_bucket=price_cap_per_bucket, max_pages_per_bucket=max_pages_per_bucket, concurrency=concurrency, secondary_only=secondary_only, on_bucket=on_bucket, skip_buckets=skip_buckets, ) 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, concurrency: int = 5, secondary_only: bool = True, on_bucket: Callable[..., Any] | None = None, skip_buckets: set[str] | None = None, _depth: int = 0, ) -> None: """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi]. Алгоритм для ЗАКРЫТОГО брекета (hi is not None): 1. Запросить page=1 с min_price=lo, max_price=hi → totalOffers. 2. Если totalOffers <= cap → пагинировать leaf-бакет параллельно (concurrency). 3. Если totalOffers > cap → разбить бакет пополам (рекурсия). Guard: hi - lo < _MIN_BRACKET → пагинировать как есть (логируем WARNING). ОТКРЫТЫЙ брекет (hi is None) — верхний seed-брекет без потолка (maxprice не ставится, Cian отдаёт всё ≥lo). Делить нельзя (mid=(lo+hi)//2 невозможен) — probe + пагинируем leaf напрямую. Хвост 250М+ крошечный; за cap обычно не выходит, если вышел — пагинируем до max_pages + WARNING (tail-loss accepted). После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots) если задан. bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый) — checkpoint-ключ для resume. on_bucket может быть async или sync. Исключение в on_bucket прерывает прогон. skip_buckets: если bucket_key в skip_buckets — пагинацию и on_bucket пропускаем. secondary_only: если True — новостройки (listing_segment=="novostroyki") отбрасываются после сбора bucket_lots, до дедупа в seen и вызова on_bucket. Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Cian-spec probe (fetch page 1 + retry) и leaf-пагинация переданы callback'ами; пороги (cap, _MIN_BRACKET) и политика probe-fail (SKIP) — в BisectionConfig. """ room_label = f"room{'_'.join(str(r) for r in rooms)}" async def _probe(plo: int | None, phi: int | None, depth: int) -> ProbeResult: lo_param = plo if plo and plo > 0 else None hi_repr = "open" if phi is None else str(phi) # ── Шаг 1: probe page 1 ─────────────────────────────────────────── html = await self._fetch_page_html(rooms, 1, lo_param, phi) await self.sleep_between_requests() total = self._extract_total_offers(html) if html is not None else None # Ретрай на captcha/ошибку: 1 повторный fetch. Мобильный прокси в браузер- # контейнере ротирует IP сам (#1806), поэтому отдельный changeip-вызов # (_rotate_ip) больше не нужен — просто повторяем запрос на чистом IP. if total is None: logger.warning( "cian: totalOffers=None for %s [%s, %s] depth=%d — retry " "(proxy auto-rotates IP)", room_label, plo, hi_repr, depth, ) await self._rotate_ip() # no-op (back-compat); прокси ротирует IP сам html = await self._fetch_page_html(rooms, 1, lo_param, phi) 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 [%s, %s] — totalOffers unavailable after retry", room_label, plo, hi_repr, ) else: logger.info( "cian: %s [%s, %s] totalOffers=%d depth=%d", room_label, plo, hi_repr, total, depth, ) return ProbeResult(count=total, payload=html) async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None: total = result.count assert total is not None # SKIP-политика: leaf не вызывается при None if phi is None and total > price_cap_per_bucket: logger.warning( "cian: OPEN bucket %s [%s, open] totalOffers=%d > cap=%d — paginating " "without split (tail-loss accepted, lux tail tiny)", room_label, plo, total, price_cap_per_bucket, ) await self._paginate_leaf_bucket( rooms=rooms, room_label=room_label, lo=plo if plo is not None else 0, hi=phi, html=result.payload, total=total, seen=seen, max_pages_per_bucket=max_pages_per_bucket, concurrency=concurrency, secondary_only=secondary_only, on_bucket=on_bucket, skip_buckets=skip_buckets, ) await walk_price_range( lo=lo, hi=hi, config=_cian_bisection_config(price_cap_per_bucket), probe=_probe, on_leaf=_leaf, depth=_depth, ) async def _paginate_leaf_bucket( self, *, rooms: tuple[int, ...], room_label: str, lo: int, hi: int | None, html: str | None, total: int, seen: dict[str, ScrapedLot], max_pages_per_bucket: int, concurrency: int, secondary_only: bool, on_bucket: Callable[..., Any] | None, skip_buckets: set[str] | None, ) -> None: """Параллельная пагинация leaf-бакета [lo, hi] (закрытого или открытого). ОТКРЫТЫЙ (hi is None): maxprice не ставится (_build_url отдаёт всё ≥lo), bucket_key = "room_label:lo:open". Страница 1 берётся из probe-`html`. """ _lo_param = lo if lo > 0 else None _hi_repr = "open" if hi is None else str(hi) bucket_key = f"{room_label}:{lo}:{_hi_repr}" # Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket. if skip_buckets and bucket_key in skip_buckets: logger.info( "cian: skip bucket %s — already done (resume)", bucket_key, ) return pages_needed = math.ceil(total / _CIAN_OFFERS_PER_PAGE) max_pages = min(pages_needed, max_pages_per_bucket) if pages_needed > max_pages_per_bucket: # Cian hard-cap срабатывает: часть офферов недостижима через пагинацию. # Это происходит когда bracket < _MIN_BRACKET но totalOffers всё равно > cap. tail_loss = total - max_pages_per_bucket * _CIAN_OFFERS_PER_PAGE logger.warning( "cian: %s [%d, %s] totalOffers=%d exceeds page cap " "(max_pages=%d × %d=%d offers) — tail loss ~%d offers, " "consider narrowing price_cap_per_bucket or _MIN_BRACKET", room_label, lo, _hi_repr, total, max_pages_per_bucket, _CIAN_OFFERS_PER_PAGE, max_pages_per_bucket * _CIAN_OFFERS_PER_PAGE, tail_loss, ) # Страница 1 уже есть (html из probe выше); остальные — параллельно. sem = asyncio.Semaphore(concurrency) async def _one_page(p: int) -> list[ScrapedLot]: if p == 1: # Используем уже полученный HTML от probe return self._parse_serp_html(html) 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: logger.warning( "cian: page_html=None %s [%d, %s] page=%d — skipping page", room_label, lo, _hi_repr, p, ) return [] return self._parse_serp_html(page_html) page_results = await asyncio.gather( *[_one_page(p) for p in range(1, max_pages + 1)], return_exceptions=True, ) bucket_lots: list[ScrapedLot] = [] for p_idx, res in enumerate(page_results, start=1): if isinstance(res, BaseException): logger.warning( "cian: page exception %s [%d, %s] page=%d — %r", room_label, lo, _hi_repr, p_idx, res, ) else: bucket_lots.extend(res) collected_this_bucket = len(bucket_lots) # ── Фильтр новостроек (secondary_only) ──────────────────────────────── dropped_nb = 0 if secondary_only: filtered = [lot for lot in bucket_lots if lot.listing_segment != "novostroyki"] dropped_nb = collected_this_bucket - len(filtered) bucket_lots = filtered # Дедуп в общий seen for lot in bucket_lots: key = lot.source_id or lot.source_url if key: seen[key] = lot logger.info( "cian: %s [%d, %s] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d", room_label, lo, _hi_repr, max_pages, collected_this_bucket, dropped_nb, len(seen), ) # ── on_bucket callback: инкрементальный save ────────────────────────── if on_bucket is not None and bucket_lots: res_cb = on_bucket(bucket_key, bucket_lots) if inspect.isawaitable(res_cb): await res_cb def _parse_serp_html(self, html: str) -> list[ScrapedLot]: """Извлечь offers из Cian Redux state. Использует cian_state_parser.extract_state() — общая утилита Stage 2. Возвращает пустой список если state не найден. """ state = extract_state(html, mfe=_MFE_SERP, key=_STATE_KEY) if state is None: logger.warning( "cian SERP state extraction failed (mfe=%s key=%s) — " "возможно Cian изменил структуру или вернул captcha", _MFE_SERP, _STATE_KEY, ) return [] offers_data: list[dict[str, Any]] = state.get("results", {}).get("offers", []) if not offers_data: logger.warning( "cian state found but results.offers пуст (totalOffers=%s)", state.get("results", {}).get("totalOffers", "?"), ) return [] logger.info( "cian SERP state ok: %d offers (totalOffers=%s)", len(offers_data), state.get("results", {}).get("totalOffers", "?"), ) lots: list[ScrapedLot] = [] for offer in offers_data: lot = self._offer_to_lot(offer) if lot is not None: lots.append(lot) raw_count = len(offers_data) saved_count = len(lots) # Охрана от silent-failure: если получили offers, но ни один не прошёл парсинг # — likely schema regression (как descriptionMinhash str→list[int] 2026-05-23). if raw_count > 0 and saved_count == 0: logger.error( "cian SERP: 0/%d offers прошли _offer_to_lot — возможна schema regression", raw_count, ) try: if self._config.glitchtip_dsn: # Ленивый импорт: sentry_sdk — app-side error-reporting, не dep # scraper_kit. Только когда glitchtip настроен И случилась # schema-regression. ImportError глотается общим except ниже. import sentry_sdk sentry_sdk.capture_message( f"cian SERP: {raw_count}/{raw_count} offers failed _offer_to_lot" " — possible schema regression", level="error", ) except Exception: pass # sentry_sdk not installed/initialised in dev return lots def _offer_to_lot(self, offer: dict[str, Any]) -> ScrapedLot | None: """Парсинг одного Cian offer (137 fields) → ScrapedLot. Маппинг полей per Schema_Cian_SERP_Inventory sec 3, sec 8, sec 11. Без anchor jitter: координаты из offer.geo.coordinates (точные). """ try: # ── Identity ───────────────────────────────────────────────────── offer_id = offer.get("cianId") or offer.get("id") if not offer_id: logger.debug("cian offer пропущен: нет cianId/id") return None source_id = str(offer_id) url = offer.get("fullUrl") if not url: url = f"{self.base_url}/sale/flat/{offer_id}/" # ── Цена ───────────────────────────────────────────────────────── bargain: dict[str, Any] = offer.get("bargainTerms") or {} price = bargain.get("priceRur") or bargain.get("price") if not price: logger.debug("cian offer %s пропущен: нет цены", offer_id) return None try: price_rub = int(price) except (TypeError, ValueError): logger.debug("cian offer %s: не удалось распарсить цену %r", offer_id, price) return None if price_rub <= 0: return None # ── Параметры квартиры ─────────────────────────────────────────── rooms: int | None = offer.get("roomsCount") # Студия: flatType == 'studio' → 0 комнат if offer.get("flatType") == "studio" and rooms is None: rooms = 0 area_raw = offer.get("totalArea") area_m2: float | None = None if area_raw is not None: try: area_m2 = float(area_raw) except (TypeError, ValueError): pass living_area_raw = offer.get("livingArea") living_area_m2: float | None = None if living_area_raw is not None: try: living_area_m2 = float(living_area_raw) except (TypeError, ValueError): pass # kitchen_area: сырое (str) в raw_payload + промоутим float в ScrapedLot # колонку kitchen_area_m2 (#2008, Class B — дашборд/matching, не valuation). kitchen_area_raw = offer.get("kitchenArea") kitchen_area_m2: float | None = None if kitchen_area_raw is not None: try: kitchen_area_m2 = float(kitchen_area_raw) except (TypeError, ValueError): pass floor: int | None = offer.get("floorNumber") # ── Здание ─────────────────────────────────────────────────────── building: dict[str, Any] = offer.get("building") or {} total_floors: int | None = building.get("floorsCount") year_built: int | None = building.get("buildYear") # house_type: cian materialType (camelCase: monolithBrick / gasSilicateBlock # / stalin / ...) НЕ равен каноничному enum → estimator soft-penalty ложно # штрафовал аналог. Нормализуем на ингесте (#2008, Class A); сырьё — в # raw_payload['raw_material_type'] для аудита. raw_material_type: str | None = building.get("materialType") house_type: str | None = normalize_house_type(raw_material_type) # Newbuilding deadline year — если нет buildYear if year_built is None: deadline = building.get("deadline") or {} year_built = deadline.get("year") # ── Кадастр ───────────────────────────────────────────────────── # cadastralNumber — кадастр КВАРТИРЫ cadastral_number: str | None = offer.get("cadastralNumber") # buildingCadastralNumber — кадастр ДОМА building_cadastral_number: str | None = offer.get("buildingCadastralNumber") # ── Геолокация (точные координаты — NO jitter) ─────────────────── geo: dict[str, Any] = offer.get("geo") or {} coords: dict[str, Any] = geo.get("coordinates") or {} lat: float | None = coords.get("lat") lon: float | None = coords.get("lng") # Если нет coords → сохраняем без lat/lon (geocode-missing обработает позже) if lat is None or lon is None: logger.debug( "cian offer %s: нет точных координат в state (geo=%s)", offer_id, coords, ) lat = lon = None # ── Адрес из geo.address[] ─────────────────────────────────────── # newbuilding нужен ДО _format_address: для ЖК geo.address[] часто без # части {type:"house"} → дом-номер восстанавливаем из building/кадастра/ # названия ЖК (#1773), иначе все корпуса схлопываются в house-less # catch-all и same-building anchor рвётся. address_parts = geo.get("address") or [] newbuilding: dict[str, Any] = offer.get("newbuilding") or {} nb_id = newbuilding.get("id") is_newbuilding = bool(nb_id) and int(nb_id) > 0 address = _format_address( address_parts, building=building, building_cadastral_number=building_cadastral_number, newbuilding=newbuilding, is_newbuilding=is_newbuilding, ) # ── Metro stations ─────────────────────────────────────────────── undergrounds: list[dict[str, Any]] = geo.get("undergrounds") or [] metro_stations = [ { "name": u.get("name"), "time": u.get("time"), "mode": u.get("transportType"), "line_color": u.get("lineColor"), "line_id": u.get("lineId"), "is_default": u.get("isDefault", False), } for u in undergrounds if u.get("name") ] # ── Newbuilding link ───────────────────────────────────────────── # newbuilding / nb_id / is_newbuilding уже вычислены выше (до адреса). # id == 0 — sentinel для вторички (per Schema sec 20.5) house_source: str | None = None house_ext_id: str | None = None listing_segment: str | None = None if is_newbuilding: # Предпочитаем кадастр ДОМА как same-building anchor (#1773): nb_id # группирует по ЖК целиком, а один ЖК — это несколько корпусов # (6679/11110/7019). buildingCadastralNumber стабильно идентифицирует # ФИЗИЧЕСКИЙ дом → Tier 0 cadastr_exact в match_or_create_house # схлопывает раздробленные корпуса. nb_id оставляем fallback'ом. if building_cadastral_number: house_source = "cian_building_cad" house_ext_id = building_cadastral_number else: house_source = "cian_newbuilding" house_ext_id = str(nb_id) listing_segment = "novostroyki" else: house_source = "cian" listing_segment = "vtorichka" # ── Дополнительные характеристики ──────────────────────────────── balconies_count: int | None = offer.get("balconiesCount") loggias_count: int | None = offer.get("loggiasCount") bedrooms_count: int | None = offer.get("bedroomsCount") has_balcony: bool | None = None if balconies_count is not None: has_balcony = balconies_count > 0 elif loggias_count is not None: has_balcony = loggias_count > 0 # Мебель и repair_state # hasFurniture (мебель вообще) и isSoldFurnished (продаётся с мебелью) — # разная семантика, нельзя OR-склеивать: явный False у hasFurniture # затирался бы коалесингом. Берём hasFurniture; только если он None, # падаем на isSoldFurnished как fallback (#1388). _has_furniture = offer.get("hasFurniture") has_furniture: bool | None = ( _has_furniture if _has_furniture is not None else offer.get("isSoldFurnished") ) # decoration — для новостроек (отделка); repair_state — для вторички # Cian SERP: offer.decoration содержит raw значения (without/cosmetic/euro/design/...) # Нормализуем до enum needs_repair/standard/good/excellent при инgesте. repair_state: str | None = normalize_repair_state(offer.get("decoration")) # ── Продавец ───────────────────────────────────────────────────── phones: list[dict[str, Any]] = offer.get("phones") or [] is_homeowner: bool | None = offer.get("isByHomeowner") is_pro_seller: bool | None = offer.get("isPro") # ── Сделка ─────────────────────────────────────────────────────── sale_type: str | None = bargain.get("saleType") bargain_allowed: bool | None = bargain.get("bargainAllowed") # ── Class B promote (#2008): ипотека / апартаменты / проверка ЕГРН ── # Раньше лежали только в raw_payload; промоутим в колонки listings для # дашборда покрытия и matching (НЕ для valuation — estimator их не читает). mortgage_available: bool | None = bargain.get("mortgageAllowed") is_apartments: bool | None = offer.get("isApartments") is_rosreestr_checked: bool | None = offer.get("isRosreestrChecked") # ── Description + minhash ───────────────────────────────────────── description: str | None = offer.get("description") # Cian предоставляет готовый minhash для dedup и cross-source matching. # С ~2026-05-23 Cian изменил тип descriptionMinhash: str → list[int]. # Нормализуем оба варианта; НЕ sorted() — порядок band-ов значим для LSH. _raw_minhash = offer.get("descriptionMinhash") if isinstance(_raw_minhash, list): description_minhash: str | None = ",".join(str(x) for x in _raw_minhash) or None elif isinstance(_raw_minhash, str): description_minhash = _raw_minhash or None else: description_minhash = None # Если Cian не дал minhash — вычисляем простой SHA1 из description if not description_minhash and description: description_minhash = hashlib.sha1( description.lower().encode("utf-8", errors="replace") ).hexdigest()[:32] # Fallback: repair_state из описания, если decoration отсутствует (#622) if repair_state is None and description: repair_state = infer_repair_state_from_text(description) # ── Фото ───────────────────────────────────────────────────────── photo_urls: list[str] = [ p["fullUrl"] for p in (offer.get("photos") or []) if isinstance(p, dict) and p.get("fullUrl") ] # ── Дата публикации ─────────────────────────────────────────────── added_ts = offer.get("addedTimestamp") listing_date = None if added_ts: try: listing_date = datetime.fromtimestamp(int(added_ts), tz=UTC).date() except (TypeError, ValueError, OSError): pass # ── Raw payload (полный offer для enrichment в будущем) ─────────── raw_payload: dict[str, Any] = { "cian_id": offer_id, "flat_type": offer.get("flatType"), "is_apartments": offer.get("isApartments"), "offer_type": offer.get("offerType"), "category": offer.get("category"), "has_furniture": has_furniture, # сырой kitchenArea (str) — колонка kitchen_area_m2 (float) промоучена выше "kitchen_area_m2": kitchen_area_raw, # сырой materialType (camelCase) — house_type нормализован выше (#2008) "raw_material_type": raw_material_type, "mortgage_allowed": bargain.get("mortgageAllowed"), "sale_type": sale_type, "newbuilding_name": newbuilding.get("name"), "is_from_developer": ( offer.get("fromDeveloper") or newbuilding.get("isFromDeveloper") ), "builders_ids": offer.get("buildersIds"), "is_rosreestr_checked": offer.get("isRosreestrChecked"), "is_layout_approved": offer.get("isLayoutApproved"), "user_id": offer.get("userId"), "published_user_id": offer.get("publishedUserId"), "is_by_commercial_owner": offer.get("isByCommercialOwner"), "is_cian_partner": offer.get("isCianPartner"), "description_words_highlighted": offer.get("descriptionWordsHighlighted"), "building_parking": building.get("parking"), "building_total_area": building.get("totalArea"), } return ScrapedLot( source="cian", source_url=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, house_type=house_type, repair_state=repair_state, has_balcony=has_balcony, kitchen_area_m2=kitchen_area_m2, mortgage_available=mortgage_available, is_apartments=is_apartments, is_rosreestr_checked=is_rosreestr_checked, kadastr_num=cadastral_number, house_source=house_source, house_ext_id=house_ext_id, listing_segment=listing_segment, price_rub=price_rub, price_per_m2=None, # compute_price_per_m2() вычислит listing_date=listing_date, photo_urls=photo_urls, raw_payload=raw_payload, # Cian-specific (Stage 2 fields) living_area_m2=living_area_m2, bedrooms_count=bedrooms_count, balconies_count=balconies_count, loggias_count=loggias_count, description_minhash=description_minhash, cadastral_number=cadastral_number, building_cadastral_number=building_cadastral_number, phones=phones, is_homeowner=is_homeowner, is_pro_seller=is_pro_seller, bargain_allowed=bargain_allowed, sale_type=sale_type, metro_stations=metro_stations, ) except Exception: _oid = offer.get("cianId") or offer.get("id") logger.exception("cian _offer_to_lot failed for offer_id=%s", _oid) return None # ── Address formatter ──────────────────────────────────────────────────────── # Дом-токен в хвосте строки: «29», «29А», «12 к1», «5/2», «10 стр.3», «1С7». # Якорим в конец строки (\Z) — берём именно завершающий номер, не цифры из # названия ЖК («Суходольский квартал»). Не матчим «р-н», «мкр» и пр. слова. _HOUSE_TOKEN_RE = re.compile( r"(\d+[А-Яа-яA-Za-z]?" # базовый номер: 29 / 29А / 1С r"(?:\s*(?:к|корп\.?|стр\.?|с)\s*\d+[А-Яа-яA-Za-z]?)?" # корпус/строение r"(?:\s*/\s*\d+[А-Яа-яA-Za-z]?)?)" # дробь 5/2 r"\s*\Z" ) def _recover_newbuilding_house( *, building: dict[str, Any] | None, building_cadastral_number: str | None, newbuilding: dict[str, Any] | None, ) -> str | None: """Восстановить дом-токен для новостройки, когда geo.address[] без {type:"house"}. Приоритет (#1773): (a) building.houseNumber / building.address — если SERP-state их отдал; (b) building_cadastral_number — стабильный same-building anchor: если ни номер, ни название не дали дом, добавляем «кад. 66:41:...:11396» как ключ-якорь, чтобы раздробленные корпуса схлопывались по кадастру; (c) хвостовой дом-токен из newbuilding.name («… 29» / «… 29А к1»). НЕ выдумываем catch-all номер: если ничего из (a)/(b)/(c) не нашли — None. Кадастр (b) предпочтительнее как house anchor (см. _offer_to_lot). """ building = building or {} newbuilding = newbuilding or {} # (a) building.houseNumber / building.address — прямые поля SERP-state house_number = building.get("houseNumber") if house_number is not None: token = str(house_number).strip() if token: return token b_addr = (building.get("address") or "").strip() if b_addr: m = _HOUSE_TOKEN_RE.search(b_addr) if m: return m.group(1).strip() # (c) хвостовой дом-токен из названия ЖК (до кадастра — он читабельнее) nb_name = (newbuilding.get("name") or "").strip() if nb_name: m = _HOUSE_TOKEN_RE.search(nb_name) if m: return m.group(1).strip() # (b) кадастр дома как anchor-ключ (последний приоритет в строке адреса, # но самый надёжный для матчинга — Tier 0 cadastr_exact) if building_cadastral_number: cad = building_cadastral_number.strip() if cad: return f"кад. {cad}" return None def _format_address( address_parts: list[dict[str, Any]], *, building: dict[str, Any] | None = None, building_cadastral_number: str | None = None, newbuilding: dict[str, Any] | None = None, is_newbuilding: bool = False, ) -> str: """Сформировать читаемый адрес из geo.address[] Cian. Пропускаем location (страна/регион) и metro — берём раион/улицу/дом. Пример: [location=Москва, raion=Пресненский, street=..., house=1С7, metro=Москва-Сити] → 'Пресненский, улица ..., 1С7' #1773: для новостроек (is_newbuilding) geo.address[] часто без {type:"house"}. Если части house нет — восстанавливаем номер дома через _recover_newbuilding_house (building → название ЖК → кадастр) и дописываем его в конец строки, чтобы same-building anchor (fingerprint/кадастр) не рвался по house-less улице. """ if not address_parts: return "Екатеринбург (Cian)" skip_types = {"location", "metro"} parts: list[str] = [] has_house = False for part in address_parts: if not isinstance(part, dict): continue ptype = part.get("type", "") if ptype == "house": has_house = True if ptype in skip_types: continue name = (part.get("fullName") or part.get("name") or "").strip() if name: parts.append(name) # Восстановление дома только для новостроек без части house — вторичка с # house-less адресом не трогается (не выдумываем номер). if is_newbuilding and not has_house: recovered = _recover_newbuilding_house( building=building, building_cadastral_number=building_cadastral_number, newbuilding=newbuilding, ) if recovered: parts.append(recovered) return ", ".join(parts) if parts else "Екатеринбург (Cian)"