237 lines
8.1 KiB
Python
237 lines
8.1 KiB
Python
"""Yandex.Недвижимость scraper (realty.yandex.ru) — DOM-based SERP parser.
|
|
|
|
History: Yandex SSR used to embed full state in `<script id="initial_state_script">`.
|
|
As of 2026-05, the state is chunked + evaporates after hydration. This version
|
|
scrapes the rendered DOM via selectolax + helpers from yandex_helpers.py.
|
|
|
|
URL pattern (path-based, no geo radius):
|
|
https://realty.yandex.ru/{city}/kupit/kvartira/vtorichniy-rynok/?page=N
|
|
|
|
Typical: 23 [data-test="OffersSerpItem"] cards per page.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import date # noqa: F401 (used in type hints / future helpers)
|
|
from typing import Any
|
|
|
|
from selectolax.parser import HTMLParser, Node
|
|
|
|
from app.services.scraper_settings import get_scraper_delay
|
|
from app.services.scrapers.base import BaseScraper, ScrapedLot
|
|
from app.services.scrapers.yandex_helpers import (
|
|
RE_FLOOR,
|
|
RE_JK_ID,
|
|
RE_OFFER_ID,
|
|
RE_PPM2,
|
|
RE_PRICE,
|
|
RE_TITLE_AREA,
|
|
RE_TITLE_ROOMS,
|
|
parse_ru_date,
|
|
parse_rub,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_CITY = "ekaterinburg"
|
|
MAX_PAGES = 3
|
|
PHOTO_DOMAIN = "avatars.mds.yandex"
|
|
PHOTO_SIZE_FROM = "app_snippet_small"
|
|
PHOTO_SIZE_TO = "main"
|
|
|
|
|
|
class YandexRealtyScraper(BaseScraper):
|
|
name = "yandex"
|
|
base_url = "https://realty.yandex.ru"
|
|
request_delay_sec = 5.0
|
|
|
|
def __init__(self, city: str = DEFAULT_CITY) -> None:
|
|
super().__init__()
|
|
self.city = city
|
|
# Load global Yandex delay from DB at instantiation
|
|
self.request_delay_sec = get_scraper_delay(self.name)
|
|
|
|
async def fetch_around(
|
|
self,
|
|
lat: float,
|
|
lon: float,
|
|
radius_m: int = 1000,
|
|
page: int = 0,
|
|
) -> list[ScrapedLot]:
|
|
"""Fetch ONE page of SERP cards. lat/lon/radius_m ignored (Yandex uses
|
|
path-based vtorichka URL); kept for BaseScraper compat + logging.
|
|
"""
|
|
url = self._build_url(page=page)
|
|
try:
|
|
response = await self._http_get(url)
|
|
except Exception:
|
|
logger.exception("yandex serp fetch failed: %s", url)
|
|
return []
|
|
if response.status_code != 200:
|
|
logger.warning("yandex serp returned %d for %s", response.status_code, url)
|
|
return []
|
|
lots = self._parse_html(response.text, page=page)
|
|
logger.info(
|
|
"yandex serp page=%d city=%s: %d cards (anchor=(%.4f,%.4f) ignored)",
|
|
page, self.city, len(lots), lat, lon,
|
|
)
|
|
await self.sleep_between_requests()
|
|
return lots
|
|
|
|
async def fetch_around_multi_room(
|
|
self,
|
|
lat: float,
|
|
lon: float,
|
|
radius_m: int = 1000,
|
|
max_pages: int = MAX_PAGES,
|
|
**_legacy_kwargs: Any, # swallow rooms/sorts/pages from old callers
|
|
) -> list[ScrapedLot]:
|
|
"""Fetch up to max_pages of SERP cards, dedup by offer_id (source_id)."""
|
|
seen: dict[str, ScrapedLot] = {}
|
|
for page in range(max_pages):
|
|
lots = await self.fetch_around(lat, lon, radius_m, page=page)
|
|
if not lots:
|
|
break # empty page → no more results
|
|
for lot in lots:
|
|
key = lot.source_id or lot.source_url
|
|
if key and key not in seen:
|
|
seen[key] = lot
|
|
logger.info(
|
|
"yandex serp aggregate: %d unique lots over %d pages (city=%s)",
|
|
len(seen), max_pages, self.city,
|
|
)
|
|
return list(seen.values())
|
|
|
|
def _build_url(self, page: int = 0) -> str:
|
|
# Yandex paginates 1-based; `page=0` → no param (first page)
|
|
base = f"{self.base_url}/{self.city}/kupit/kvartira/vtorichniy-rynok/"
|
|
if page > 0:
|
|
return f"{base}?page={page}"
|
|
return base
|
|
|
|
def _parse_html(self, html: str, page: int = 0) -> list[ScrapedLot]:
|
|
tree = HTMLParser(html)
|
|
cards = tree.css('[data-test="OffersSerpItem"]')
|
|
lots: list[ScrapedLot] = []
|
|
for card in cards:
|
|
lot = self._card_to_lot(card, page=page)
|
|
if lot is not None:
|
|
lots.append(lot)
|
|
return lots
|
|
|
|
def _card_to_lot(self, card: Node, page: int = 0) -> ScrapedLot | None:
|
|
try:
|
|
# offer_id — required
|
|
offer_link = card.css_first('a[href^="/offer/"]')
|
|
if offer_link is None:
|
|
return None
|
|
href = offer_link.attributes.get("href", "")
|
|
id_match = RE_OFFER_ID.search(href)
|
|
if not id_match:
|
|
return None
|
|
offer_id = id_match.group(1)
|
|
|
|
text = card.text(strip=True)
|
|
|
|
# Price — required (ScrapedLot.price_rub > 0)
|
|
price_m = RE_PRICE.search(text)
|
|
price_rub = parse_rub(price_m.group(1)) if price_m else None
|
|
if not price_rub or price_rub <= 0:
|
|
return None
|
|
|
|
# Optional fields
|
|
area_m = RE_TITLE_AREA.search(text)
|
|
area_m2 = (
|
|
float(area_m.group(1).replace(",", ".")) if area_m else None
|
|
)
|
|
|
|
rooms = self._parse_rooms(text)
|
|
|
|
floor_m = RE_FLOOR.search(text)
|
|
floor = int(floor_m.group(1)) if floor_m else None
|
|
total_floors = int(floor_m.group(2)) if floor_m else None
|
|
|
|
ppm2_m = RE_PPM2.search(text)
|
|
price_per_m2 = parse_rub(ppm2_m.group(1)) if ppm2_m else None
|
|
|
|
bargain = "торг" in text.lower()
|
|
listing_date = parse_ru_date(text)
|
|
|
|
# Address — street aggregator link
|
|
street_link = card.css_first('a[href*="/kupit/kvartira/st-"]')
|
|
address = street_link.text(strip=True) if street_link else None
|
|
|
|
# Photos
|
|
photo_urls = self._extract_photos(card)
|
|
|
|
# Newbuilding linkage
|
|
house_source: str | None = None
|
|
house_ext_id: str | None = None
|
|
house_url: str | None = None
|
|
nb_link = card.css_first('a[href*="/kupit/novostrojka/"]')
|
|
if nb_link is not None:
|
|
nb_href = nb_link.attributes.get("href", "")
|
|
nb_match = RE_JK_ID.search(nb_href)
|
|
if nb_match:
|
|
house_source = "yandex_realty_nb"
|
|
house_ext_id = nb_match.group(2)
|
|
house_url = (
|
|
nb_href if nb_href.startswith("http")
|
|
else f"https://realty.yandex.ru{nb_href}"
|
|
)
|
|
|
|
return ScrapedLot(
|
|
source=self.name,
|
|
source_url=f"https://realty.yandex.ru/offer/{offer_id}/",
|
|
source_id=offer_id,
|
|
address=address,
|
|
lat=None,
|
|
lon=None,
|
|
rooms=rooms,
|
|
area_m2=area_m2,
|
|
floor=floor,
|
|
total_floors=total_floors,
|
|
price_rub=price_rub,
|
|
price_per_m2=price_per_m2,
|
|
bargain_allowed=bargain,
|
|
listing_date=listing_date,
|
|
photo_urls=photo_urls,
|
|
house_source=house_source,
|
|
house_ext_id=house_ext_id,
|
|
house_url=house_url,
|
|
listing_segment="vtorichka",
|
|
raw_payload={
|
|
"card_text": text[:1000],
|
|
"page": page,
|
|
"city": self.city,
|
|
},
|
|
)
|
|
except Exception:
|
|
logger.exception("yandex card parse failed (page=%d)", page)
|
|
return None
|
|
|
|
@staticmethod
|
|
def _parse_rooms(text: str) -> int | None:
|
|
m = RE_TITLE_ROOMS.search(text)
|
|
if not m:
|
|
return None
|
|
if m.group(2): # studio
|
|
return 0
|
|
if m.group(1): # numbered
|
|
try:
|
|
return int(m.group(1))
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
@staticmethod
|
|
def _extract_photos(card: Node) -> list[str]:
|
|
urls: list[str] = []
|
|
for img in card.css("img"):
|
|
src = img.attributes.get("src", "") or ""
|
|
if PHOTO_DOMAIN in src:
|
|
urls.append(src.replace(PHOTO_SIZE_FROM, PHOTO_SIZE_TO))
|
|
if len(urls) >= 5:
|
|
break
|
|
return urls
|