bot-backend
77ed21a479
object_type SERP-фильтр Cian не разделяет сегмент надёжно (object_type=1 ≠ вторичка), поэтому отсекаем новостройки по authoritative listing_segment (newbuilding.id>0) после пагинации, до save. fetch_all_secondary получает secondary_only=True (default). city-sweep не затронут.
956 lines
44 KiB
Python
956 lines
44 KiB
Python
"""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 по TLS fingerprint → "Обнаружен подозрительный трафик" (403).
|
||
Используем curl_cffi с impersonate='chrome120' — это libcurl-impersonate под капотом,
|
||
который воспроизводит TLS ClientHello как настоящий Chrome.
|
||
|
||
НЕ используем 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 inspect
|
||
import logging
|
||
import math
|
||
from collections.abc import Callable
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
from urllib.parse import urlencode
|
||
|
||
import sentry_sdk
|
||
from curl_cffi.requests import AsyncSession
|
||
|
||
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.cian_state_parser import extract_state
|
||
from app.services.scrapers.repair_state_normalizer import (
|
||
infer_repair_state_from_text,
|
||
normalize_repair_state,
|
||
)
|
||
|
||
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
|
||
|
||
# Комнатности по умолчанию для 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.
|
||
|
||
Извлекает 137 полей на offer из Redux initialState через cian_state_parser.
|
||
Координаты точные — НЕТ anchor jitter (offer.geo.coordinates).
|
||
"""
|
||
|
||
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) -> None:
|
||
super().__init__()
|
||
self.request_delay_sec = get_scraper_delay(self.name)
|
||
self._cffi: AsyncSession | None = None
|
||
|
||
async def __aenter__(self) -> CianScraper:
|
||
await super().__aenter__()
|
||
# curl_cffi session с Chrome 120 TLS fingerprint — КРИТИЧЕСКИ ВАЖНО для Cian.
|
||
# proxies: mobile-proxy egress (#806) — аналогично Avito (#623). Без прокси
|
||
# datacenter-IP блокируется Cian (403/captcha) несмотря на валидные cookies.
|
||
# Пусто (SCRAPER_PROXY_URL/AVITO_PROXY_URL не заданы) → прямое подключение (dev).
|
||
_proxy_url = settings.cian_proxy_url
|
||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||
self._cffi = AsyncSession(
|
||
impersonate="chrome120",
|
||
timeout=30,
|
||
proxies=_proxies,
|
||
headers={
|
||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||
"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
|
||
"Cache-Control": "max-age=0",
|
||
"Sec-Fetch-Dest": "document",
|
||
"Sec-Fetch-Mode": "navigate",
|
||
"Sec-Fetch-Site": "none",
|
||
"Sec-Fetch-User": "?1",
|
||
"Upgrade-Insecure-Requests": "1",
|
||
},
|
||
)
|
||
# Warm-up: посетить homepage чтобы получить сессионные cookies (_CIAN_GK/_yasc)
|
||
# и не выглядеть как наглый бот (первый запрос сразу на /cat.php подозрителен).
|
||
# Failure не критична — SERP-запросы продолжатся в любом случае.
|
||
try:
|
||
_warm = await self._cffi.get(f"{self.base_url}/")
|
||
logger.info(
|
||
"cian warm-up GET / → HTTP %d (cookies=%d)",
|
||
_warm.status_code,
|
||
len(list(self._cffi.cookies.jar)),
|
||
)
|
||
# Последующие SERP-запросы выглядят как in-site navigation (Referer = /)
|
||
self._cffi.headers.update(
|
||
{
|
||
"Referer": f"{self.base_url}/",
|
||
"Sec-Fetch-Site": "same-origin",
|
||
}
|
||
)
|
||
except Exception:
|
||
logger.warning("cian warm-up GET / failed — continuing without seed cookies")
|
||
return self
|
||
|
||
async def __aexit__(self, *args: Any) -> None:
|
||
if self._cffi is not None:
|
||
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,
|
||
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._cffi is not None
|
||
response = await self._cffi.get(url)
|
||
except Exception:
|
||
logger.exception("cian curl_cffi fetch failed for url=%s", url)
|
||
return []
|
||
|
||
if response.status_code != 200:
|
||
logger.warning("cian returned HTTP %d for url=%s", response.status_code, url)
|
||
return []
|
||
|
||
lots = self._parse_serp_html(response.text)
|
||
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-страницы, возвращает 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,
|
||
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 результатов на запрос):
|
||
внутри каждой комнатности адаптивно бьёт диапазон цены на бакеты так, чтобы
|
||
в каждом totalOffers < price_cap_per_bucket → пагинирует бакет полностью.
|
||
Страницы внутри 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)
|
||
await self._walk_price_range(
|
||
rooms=rooms,
|
||
lo=0,
|
||
hi=_MAX_PRICE,
|
||
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,
|
||
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].
|
||
|
||
Алгоритм:
|
||
1. Запросить page=1 с min_price=lo, max_price=hi → totalOffers.
|
||
2. Если totalOffers <= cap → пагинировать leaf-бакет параллельно (concurrency).
|
||
3. Если totalOffers > cap → разбить бакет пополам (рекурсия).
|
||
Guard: hi - lo < _MIN_BRACKET → пагинировать как есть (логируем WARNING).
|
||
|
||
После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots) если задан.
|
||
bucket_key = "room_label:lo:hi" (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.
|
||
"""
|
||
effective_hi = hi
|
||
|
||
room_label = f"room{'_'.join(str(r) for r in rooms)}"
|
||
_lo_param = lo if lo > 0 else None
|
||
|
||
# ── Шаг 1: probe page 1 ────────────────────────────────────────────────
|
||
html = await self._fetch_page_html(rooms, 1, _lo_param, 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, %d] 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_param, 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, %d] — totalOffers unavailable after retry",
|
||
room_label,
|
||
lo,
|
||
hi,
|
||
)
|
||
return
|
||
|
||
logger.info(
|
||
"cian: %s [%d, %d] 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, %d] 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:
|
||
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,
|
||
concurrency=concurrency,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
skip_buckets=skip_buckets,
|
||
_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,
|
||
concurrency=concurrency,
|
||
secondary_only=secondary_only,
|
||
on_bucket=on_bucket,
|
||
skip_buckets=skip_buckets,
|
||
_depth=_depth + 1,
|
||
)
|
||
return
|
||
|
||
# ── Checkpoint-ключ leaf-бакета ──────────────────────────────────────
|
||
bucket_key = f"{room_label}:{lo}:{hi}"
|
||
|
||
# Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket.
|
||
if skip_buckets and bucket_key in skip_buckets:
|
||
logger.info(
|
||
"cian: skip bucket %s — already done (resume)",
|
||
bucket_key,
|
||
)
|
||
return
|
||
|
||
# ── Параллельная пагинация leaf-бакета ────────────────────────────────
|
||
max_pages = min(
|
||
math.ceil(total / _CIAN_OFFERS_PER_PAGE),
|
||
max_pages_per_bucket,
|
||
)
|
||
|
||
# Страница 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, %d] page=%d — skipping page",
|
||
room_label,
|
||
lo,
|
||
hi,
|
||
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, %d] page=%d — %r",
|
||
room_label,
|
||
lo,
|
||
hi,
|
||
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, %d] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d",
|
||
room_label,
|
||
lo,
|
||
hi,
|
||
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 settings.glitchtip_dsn:
|
||
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 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_m2 сохраняем в raw_payload (поле не в ScrapedLot Stage 2)
|
||
kitchen_area_raw = offer.get("kitchenArea")
|
||
|
||
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: str | None = building.get("materialType")
|
||
|
||
# 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[] ───────────────────────────────────────
|
||
address_parts = geo.get("address") or []
|
||
address = _format_address(address_parts)
|
||
|
||
# ── 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: dict[str, Any] = offer.get("newbuilding") or {}
|
||
nb_id = newbuilding.get("id")
|
||
# 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 nb_id and int(nb_id) > 0:
|
||
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
|
||
has_furniture: bool | None = offer.get("hasFurniture") or 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")
|
||
|
||
# ── 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,
|
||
"kitchen_area_m2": kitchen_area_raw, # не в ScrapedLot — сохраняем здесь
|
||
"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,
|
||
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 ────────────────────────────────────────────────────────
|
||
|
||
|
||
def _format_address(address_parts: list[dict[str, Any]]) -> str:
|
||
"""Сформировать читаемый адрес из geo.address[] Cian.
|
||
|
||
Пропускаем location (страна/регион) и metro — берём раион/улицу/дом.
|
||
Пример: [location=Москва, raion=Пресненский, street=..., house=1С7, metro=Москва-Сити]
|
||
→ 'Пресненский, улица ..., 1С7'
|
||
"""
|
||
if not address_parts:
|
||
return "Екатеринбург (Cian)"
|
||
|
||
skip_types = {"location", "metro"}
|
||
parts: list[str] = []
|
||
for part in address_parts:
|
||
if not isinstance(part, dict):
|
||
continue
|
||
ptype = part.get("type", "")
|
||
if ptype in skip_types:
|
||
continue
|
||
name = (part.get("fullName") or part.get("name") or "").strip()
|
||
if name:
|
||
parts.append(name)
|
||
|
||
return ", ".join(parts) if parts else "Екатеринбург (Cian)"
|