7cc8d8370d
После перехода на реальные координаты стало видно что данных тонко. 8 страниц × 4 комнаты ≈ 900+ лотов на источник вместо ~300.
484 lines
19 KiB
Python
484 lines
19 KiB
Python
"""Cian.ru scraper — парсинг вторички вокруг точки.
|
||
|
||
Стратегия:
|
||
- Cian.ru Next.js приложение хранит state в `__NEXT_DATA__` JSON (script tag в HTML).
|
||
- URL pattern: https://ekaterinburg.cian.ru/cat.php?deal_type=sale&offer_type=flat&engine_version=2
|
||
|
||
ВАЖНО: Циан блокирует httpx по TLS fingerprint → "Обнаружен подозрительный трафик" (403).
|
||
Используем curl_cffi с impersonate='chrome120' — это libcurl-impersonate под капотом,
|
||
который воспроизводит TLS ClientHello как настоящий Chrome.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import random
|
||
import re
|
||
from typing import Any
|
||
from urllib.parse import urlencode, urljoin
|
||
|
||
from curl_cffi.requests import AsyncSession
|
||
from selectolax.parser import HTMLParser
|
||
|
||
from app.services.scrapers.base import BaseScraper, ScrapedLot
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Регион 4743 = Свердловская область (Cian internal)
|
||
CIAN_EKB_REGION_ID = 4743
|
||
|
||
|
||
class CianScraper(BaseScraper):
|
||
"""Cian vtorichka parser. Использует curl_cffi для обхода TLS fingerprint detection."""
|
||
|
||
name = "cian"
|
||
base_url = "https://ekaterinburg.cian.ru"
|
||
request_delay_sec = 6.0
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self._cffi: AsyncSession | None = None
|
||
|
||
async def __aenter__(self) -> "CianScraper":
|
||
await super().__aenter__()
|
||
# curl_cffi session с Chrome 120 TLS fingerprint
|
||
self._cffi = AsyncSession(
|
||
impersonate="chrome120",
|
||
timeout=25,
|
||
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",
|
||
},
|
||
)
|
||
return self
|
||
|
||
async def __aexit__(self, *args: Any) -> None:
|
||
if self._cffi is not None:
|
||
await self._cffi.close()
|
||
await super().__aexit__(*args)
|
||
|
||
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) в radius_m метрах.
|
||
|
||
rooms — список из 1,2,3,4 (Cian's room codes). Если None — все.
|
||
page — страница выдачи (Cian ~28 на страницу).
|
||
"""
|
||
self._anchor_lat = lat
|
||
self._anchor_lon = lon
|
||
self._anchor_radius_m = radius_m
|
||
|
||
url = self._build_url(lat, lon, radius_m, rooms, page)
|
||
try:
|
||
assert self._cffi is not None
|
||
response = await self._cffi.get(url)
|
||
except Exception: # noqa: BLE001
|
||
logger.exception("cian curl_cffi fetch failed for %s", url)
|
||
return []
|
||
if response.status_code != 200:
|
||
logger.warning("cian returned %d for %s", response.status_code, url)
|
||
return []
|
||
|
||
lots = self._parse_html(response.text)
|
||
logger.info(
|
||
"cian: %d lots around (%.4f, %.4f) rooms=%s page=%d",
|
||
len(lots), lat, lon, rooms, page,
|
||
)
|
||
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 комнаты × 3 страницы × ~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 around (%.4f, %.4f)", len(seen), lat, lon
|
||
)
|
||
return list(seen.values())
|
||
|
||
def _build_url(
|
||
self,
|
||
lat: float,
|
||
lon: float,
|
||
radius_m: int,
|
||
rooms: tuple[int, ...] | None = None,
|
||
page: int = 1,
|
||
) -> str:
|
||
"""URL для Cian каталога с регионом ЕКБ.
|
||
|
||
Cian не имеет «радиус» в URL params, но мы фильтруем по lat/lon
|
||
в postgres через ST_DWithin позже. Сейчас тащим всё ЕКБ.
|
||
"""
|
||
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:
|
||
# Cian rooms[]=1 для 1к, rooms[]=2 для 2к, и т.д.
|
||
params.append((f"room{r}", "1"))
|
||
if page > 1:
|
||
params.append(("p", page))
|
||
return f"{self.base_url}/cat.php?{urlencode(params)}"
|
||
|
||
def _parse_html(self, html: str) -> list[ScrapedLot]:
|
||
"""Cian вставляет state в `window.__NEXT_DATA__` или подобный JSON блок."""
|
||
# Попытка 1: __NEXT_DATA__ script
|
||
items = _extract_next_data_items(html)
|
||
if items:
|
||
return [
|
||
lot for lot in (self._item_to_lot(it) for it in items) if lot is not None
|
||
]
|
||
|
||
# Попытка 2: regex-извлечение из inline JSON state
|
||
items = _extract_inline_offers(html)
|
||
if items:
|
||
return [
|
||
lot for lot in (self._item_to_lot(it) for it in items) if lot is not None
|
||
]
|
||
|
||
# Попытка 3: DOM card scrape (fallback)
|
||
tree = HTMLParser(html)
|
||
cards = tree.css('[data-name="CardComponent"]') or tree.css(
|
||
'article[data-name="GeneralCard"]'
|
||
)
|
||
return [
|
||
lot for lot in (self._card_to_lot(c) for c in cards) if lot is not None
|
||
][:50]
|
||
|
||
def _item_to_lot(self, item: dict[str, Any]) -> ScrapedLot | None:
|
||
"""Парс Cian offer JSON → ScrapedLot."""
|
||
try:
|
||
# Cian offer структура
|
||
offer_id = item.get("cianId") or item.get("id")
|
||
url = item.get("fullUrl") or item.get("url")
|
||
if not url and offer_id:
|
||
url = f"https://ekaterinburg.cian.ru/sale/flat/{offer_id}/"
|
||
if not url:
|
||
return None
|
||
|
||
# Цена
|
||
bargain = item.get("bargainTerms") or {}
|
||
price = bargain.get("price") or item.get("price")
|
||
if not price:
|
||
return None
|
||
|
||
# Параметры
|
||
rooms = item.get("roomsCount")
|
||
area = item.get("totalArea") or item.get("totalAreaSquare")
|
||
if area:
|
||
try:
|
||
area = float(area)
|
||
except (TypeError, ValueError):
|
||
area = None
|
||
floor = item.get("floorNumber")
|
||
building = item.get("building") or {}
|
||
total_floors = building.get("floorsCount")
|
||
|
||
# Адрес
|
||
geo = item.get("geo") or {}
|
||
address_parts = geo.get("userInput") or ""
|
||
if isinstance(geo.get("address"), list):
|
||
address_parts = ", ".join(
|
||
a.get("fullName", "") or a.get("name", "")
|
||
for a in geo["address"][:3]
|
||
)
|
||
|
||
# Координаты
|
||
coords = geo.get("coordinates") or {}
|
||
lat = coords.get("lat")
|
||
lon = coords.get("lng")
|
||
|
||
# Fallback anchor с jitter если нет точных
|
||
if lat is None or lon is None:
|
||
if hasattr(self, "_anchor_lat"):
|
||
jitter_deg = (self._anchor_radius_m / 1000) / 111
|
||
lat = self._anchor_lat + (random.random() - 0.5) * jitter_deg
|
||
lon = self._anchor_lon + (random.random() - 0.5) * jitter_deg / 2
|
||
|
||
# Год постройки
|
||
year_built = building.get("buildYear")
|
||
|
||
return ScrapedLot(
|
||
source="cian",
|
||
source_url=url,
|
||
source_id=str(offer_id) if offer_id else None,
|
||
address=address_parts or "Екатеринбург (Cian)",
|
||
lat=lat,
|
||
lon=lon,
|
||
rooms=rooms,
|
||
area_m2=area,
|
||
floor=floor,
|
||
total_floors=total_floors,
|
||
year_built=year_built,
|
||
price_rub=int(price),
|
||
photo_urls=[
|
||
p.get("fullUrl") or p.get("thumbnailUrl", "")
|
||
for p in (item.get("photos") or [])[:5]
|
||
if isinstance(p, dict)
|
||
],
|
||
raw_payload=item,
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
logger.exception("cian item parse failed: %s", item.get("id"))
|
||
return None
|
||
|
||
def _card_to_lot(self, card: Any) -> ScrapedLot | None:
|
||
"""DOM card-парсинг для Cian 2025+ верстки.
|
||
|
||
Использует data-mark и data-name селекторы:
|
||
- [data-mark="OfferTitle"]: "2-комн. квартира, 47,5 м², 2/9 этаж"
|
||
- [data-mark="MainPrice"]: "6 000 000 ₽"
|
||
- [data-name="GeoLabel"]: address parts (multiple, объединяем)
|
||
- [data-mark="Description"]: long text (год постройки regex'ом)
|
||
- href контейнера → source_url
|
||
"""
|
||
try:
|
||
link_el = card.css_first('a[href*="/sale/flat/"]')
|
||
if link_el is None:
|
||
return None
|
||
href = link_el.attributes.get("href", "")
|
||
url = urljoin(self.base_url, href)
|
||
|
||
# offer_id из URL pattern /sale/flat/NNN/
|
||
offer_id_m = re.search(r"/sale/flat/(\d+)/", href)
|
||
offer_id = offer_id_m.group(1) if offer_id_m else None
|
||
|
||
# Цена
|
||
price_el = card.css_first('[data-mark="MainPrice"]')
|
||
price_text = price_el.text(strip=True) if price_el else ""
|
||
price = _extract_price_rub(price_text)
|
||
if not price:
|
||
return None
|
||
|
||
# Title → rooms / area / floor / total_floors
|
||
title_el = (
|
||
card.css_first('[data-mark="OfferTitle"]')
|
||
or card.css_first('[data-name="TitleComponent"]')
|
||
or card.css_first('[data-mark="Title"]')
|
||
)
|
||
title = title_el.text(strip=True) if title_el else ""
|
||
rooms = _extract_rooms_from_text(title)
|
||
area = _extract_area_from_text(title)
|
||
floor, total = _extract_floor_from_text(title)
|
||
|
||
# Address: собираем все GeoLabel, но показываем компактно
|
||
geo_parts = [
|
||
el.text(strip=True)
|
||
for el in card.css('[data-name="GeoLabel"]')
|
||
if el.text(strip=True)
|
||
]
|
||
# Полный список сохраним в raw_payload, а display address укоротим.
|
||
# Убираем "Свердловская область", "Екатеринбург", "м. *" (метро),
|
||
# оставляем "р-н *", "мкр. *", улицу и дом.
|
||
full_geo = list(geo_parts)
|
||
display_parts: list[str] = []
|
||
street_part: str | None = None
|
||
house_part: str | None = None
|
||
district_part: str | None = None
|
||
for p in geo_parts:
|
||
if p in {"Свердловская область", "Екатеринбург"}:
|
||
continue
|
||
if p.startswith("м. "):
|
||
continue # метро не нужно в адресе
|
||
if p.startswith("р-н ") and district_part is None:
|
||
district_part = p
|
||
elif p.startswith("мкр. "):
|
||
continue # микрорайон опускаем для компактности
|
||
elif "улица" in p or "проспект" in p or "переулок" in p or "бульвар" in p or "шоссе" in p:
|
||
street_part = p
|
||
elif re.match(r"^\d", p):
|
||
house_part = p
|
||
else:
|
||
display_parts.append(p)
|
||
# Финальный display: "улица X, дом Y · район Z"
|
||
final_parts: list[str] = []
|
||
if street_part:
|
||
final_parts.append(street_part + (f", {house_part}" if house_part else ""))
|
||
if district_part:
|
||
final_parts.append(district_part)
|
||
address = " · ".join(final_parts) if final_parts else (
|
||
", ".join(p for p in geo_parts if p not in {"Свердловская область"}) or "Екатеринбург (Cian)"
|
||
)
|
||
|
||
# Description → попытка извлечь год постройки
|
||
desc_el = card.css_first('[data-mark="Description"]')
|
||
desc = desc_el.text(strip=True) if desc_el else ""
|
||
year_built = _extract_year_built(desc)
|
||
|
||
# PriceInfo: "126 316 ₽/м²" — можно посчитать area если её нет в title
|
||
if not area:
|
||
pinfo_el = card.css_first('[data-mark="PriceInfo"]')
|
||
pinfo = pinfo_el.text(strip=True) if pinfo_el else ""
|
||
m = re.search(r"([\d\s]+)\s*₽/м²", pinfo)
|
||
if m:
|
||
ppm2 = int(re.sub(r"\D", "", m.group(1)) or 0)
|
||
if ppm2 > 0:
|
||
area = round(price / ppm2, 2)
|
||
|
||
# Coords: НЕ ставим (Cian card HTML не отдаёт). Оставляем None —
|
||
# geocode-missing проставит реальные по address. Anchor+jitter тут
|
||
# был БАГ: Cian отдаёт ВСЕ объявления ЕКБ независимо от anchor, и
|
||
# jitter вокруг одного якоря давал фейковые координаты → ST_DWithin
|
||
# не находил аналоги для адресов вдали от центра.
|
||
lat = lon = None
|
||
|
||
# Photos из img.src в card (первые 5)
|
||
photo_urls: list[str] = []
|
||
for img in card.css("img.x31de4314--_18b0f--container, [data-name=Cover] img")[:5]:
|
||
src = img.attributes.get("src", "")
|
||
if src and "cian" in src and src not in photo_urls:
|
||
photo_urls.append(src)
|
||
|
||
return ScrapedLot(
|
||
source="cian",
|
||
source_url=url,
|
||
source_id=offer_id,
|
||
address=address,
|
||
lat=lat,
|
||
lon=lon,
|
||
rooms=rooms,
|
||
area_m2=area,
|
||
floor=floor,
|
||
total_floors=total,
|
||
year_built=year_built,
|
||
price_rub=price,
|
||
photo_urls=photo_urls,
|
||
raw_payload={
|
||
"title": title,
|
||
"description": desc[:500],
|
||
"geo": full_geo,
|
||
},
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
logger.exception("cian card parse failed")
|
||
return None
|
||
|
||
|
||
# ── HTML / JSON extraction helpers ──────────────────────────────────────────
|
||
|
||
|
||
_RE_NEXT_DATA = re.compile(
|
||
r'<script[^>]+id="__NEXT_DATA__"[^>]*>(.+?)</script>', re.DOTALL
|
||
)
|
||
_RE_INITIAL_STATE = re.compile(
|
||
r'window\._cianConfig\s*=\s*({.+?});</script>', re.DOTALL
|
||
)
|
||
|
||
|
||
def _extract_next_data_items(html: str) -> list[dict[str, Any]]:
|
||
"""Cian Next.js приложение хранит state в `__NEXT_DATA__` JSON."""
|
||
m = _RE_NEXT_DATA.search(html)
|
||
if not m:
|
||
return []
|
||
try:
|
||
data = json.loads(m.group(1))
|
||
except json.JSONDecodeError:
|
||
return []
|
||
# Структура у Cian примерно такая: props.pageProps.initialState.serpData.offers
|
||
cursor: Any = data
|
||
for key in ["props", "pageProps", "initialState"]:
|
||
if not isinstance(cursor, dict):
|
||
return []
|
||
cursor = cursor.get(key)
|
||
if not isinstance(cursor, dict):
|
||
return []
|
||
serp = cursor.get("results") or cursor.get("serpData") or cursor.get("data") or {}
|
||
offers = serp.get("offersSerialized") or serp.get("offers") or []
|
||
if isinstance(offers, list):
|
||
return offers
|
||
return []
|
||
|
||
|
||
def _extract_inline_offers(html: str) -> list[dict[str, Any]]:
|
||
"""Альтернативный поиск offers в любом большом JSON в HTML."""
|
||
# Найти offers: [...] inline
|
||
for match in re.finditer(r'"offers":\[(.+?)\](?=,"|\})', html):
|
||
chunk = "[" + match.group(1) + "]"
|
||
try:
|
||
offers = json.loads(chunk)
|
||
if isinstance(offers, list) and offers and isinstance(offers[0], dict):
|
||
return offers
|
||
except json.JSONDecodeError:
|
||
continue
|
||
return []
|
||
|
||
|
||
# ── Title parsers (Cian-specific format) ────────────────────────────────────
|
||
|
||
|
||
_RE_ROOMS = re.compile(r"(\d)-комн", re.IGNORECASE)
|
||
_RE_STUDIO = re.compile(r"студи", re.IGNORECASE)
|
||
_RE_AREA = re.compile(r"(\d+[.,]?\d*)\s*м²", re.IGNORECASE)
|
||
_RE_FLOOR = re.compile(r"(\d+)\s*/\s*(\d+)\s*эт", re.IGNORECASE)
|
||
_RE_PRICE = re.compile(r"([\d\s]+?)\s*₽")
|
||
|
||
|
||
def _extract_rooms_from_text(text: str) -> int | None:
|
||
if _RE_STUDIO.search(text):
|
||
return 0
|
||
m = _RE_ROOMS.search(text)
|
||
return int(m.group(1)) if m else None
|
||
|
||
|
||
def _extract_area_from_text(text: str) -> float | None:
|
||
m = _RE_AREA.search(text)
|
||
return float(m.group(1).replace(",", ".")) if m else None
|
||
|
||
|
||
def _extract_floor_from_text(text: str) -> tuple[int | None, int | None]:
|
||
m = _RE_FLOOR.search(text)
|
||
if m:
|
||
return int(m.group(1)), int(m.group(2))
|
||
return None, None
|
||
|
||
|
||
def _extract_price_rub(text: str) -> int | None:
|
||
m = _RE_PRICE.search(text)
|
||
if m:
|
||
digits = re.sub(r"\D", "", m.group(1))
|
||
return int(digits) if digits else None
|
||
return None
|
||
|
||
|
||
_RE_YEAR_BUILT = re.compile(r"(?:19[5-9]\d|20[0-2]\d)\s*(?:года?\s*постройки|г\.?\s*постройки)", re.IGNORECASE)
|
||
|
||
|
||
def _extract_year_built(text: str) -> int | None:
|
||
"""Извлечь год постройки из description: '1988 года постройки' / '2005 г. постройки'."""
|
||
m = _RE_YEAR_BUILT.search(text)
|
||
if m:
|
||
year_m = re.search(r"(19[5-9]\d|20[0-2]\d)", m.group(0))
|
||
if year_m:
|
||
return int(year_m.group(1))
|
||
return None
|