tradein-mvp/backend/app/services/scrapers/yandex_realty.py

@@ -1,290 +1,234 @@
-"""Yandex.Недвижимость scraper (realty.yandex.ru).
+"""Yandex.Недвижимость scraper (realty.yandex.ru) — DOM-based SERP parser.
 
-Я.Недвижимость отдаёт SSR HTML с `__INITIAL_STATE__` JSON в `<script>` теге.
+History: Yandex SSR used to embed full state in `<script id="initial_state_script">`.
-Структура offers: state.search.offers — list of offers.
+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:
+URL pattern (path-based, no geo radius):
-  https://realty.yandex.ru/ekaterinburg/kupit/kvartira/?
+    https://realty.yandex.ru/{city}/kupit/kvartira/vtorichniy-rynok/?page=N
-    rgid=240970               # ЕКБ region
+
-    &offerType=SELL
+Typical: 23 [data-test="OffersSerpItem"] cards per page.
-    &category=APARTMENT
-    &geoLat=56.83
-    &geoLon=60.59
-    &geoRadius=1000           # метры
 """
 
 from __future__ import annotations
 
-import json
 import logging
-import random
+from datetime import date  # noqa: F401  (used in type hints / future helpers)
-import re
 from typing import Any
-from urllib.parse import urlencode, urljoin
 
-from selectolax.parser import HTMLParser
+from selectolax.parser import HTMLParser, Node
 
 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__)
 
-YANDEX_EKB_RGID = 240970
+DEFAULT_CITY = "ekaterinburg"
+MAX_PAGES = 3
+PHOTO_DOMAIN = "avatars.mds.yandex"
+PHOTO_SIZE_FROM = "app_snippet_small"
+PHOTO_SIZE_TO = "main"
 
 
 class YandexRealtyScraper(BaseScraper):
-    """Yandex.Недвижимость parser."""
-
     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
+
     async def fetch_around(
         self,
         lat: float,
         lon: float,
         radius_m: int = 1000,
-        rooms: str | None = None,
-        sort: str = "DATE_DESC",
         page: int = 0,
     ) -> list[ScrapedLot]:
-        self._anchor_lat = lat
+        """Fetch ONE page of SERP cards. lat/lon/radius_m ignored (Yandex uses
-        self._anchor_lon = lon
+        path-based vtorichka URL); kept for BaseScraper compat + logging.
-        self._anchor_radius_m = radius_m
+        """
-
+        url = self._build_url(page=page)
-        url = self._build_url(lat, lon, radius_m, rooms, sort, page)
         try:
             response = await self._http_get(url)
         except Exception:
-            logger.exception("yandex realty fetch failed for %s", url)
+            logger.exception("yandex serp fetch failed: %s", url)
             return []
-
         if response.status_code != 200:
-            logger.warning("yandex realty returned %d", response.status_code)
+            logger.warning("yandex serp returned %d for %s", response.status_code, url)
             return []
-
+        lots = self._parse_html(response.text, page=page)
-        lots = self._parse_html(response.text)
         logger.info(
-            "yandex realty: %d lots around (%.4f, %.4f) rooms=%s sort=%s page=%d",
+            "yandex serp page=%d city=%s: %d cards (anchor=(%.4f,%.4f) ignored)",
-            len(lots), lat, lon, rooms or "ANY", sort, page,
+            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,
-        sorts: tuple[str, ...] = ("DATE_DESC",),
-        pages: tuple[int, ...] = (0,),
-    ) -> list[ScrapedLot]:
-        """Скрейп Yandex по сегментам STUDIO/1k/2k/3k/4k+ и опционально по
-        нескольким sort/page вариантам.
-
-        Каждая выдача = 20 лотов. Полный обход rooms×sorts×pages даст
-        5 × len(sorts) × len(pages) × 20 потенциальных лотов до дедупликации.
-        """
-        seen: dict[str, ScrapedLot] = {}
-        for rooms in ("STUDIO", "1", "2", "3", "PLUS_4"):
-            for sort in sorts:
-                for page in pages:
-                    try:
-                        lots = await self.fetch_around(
-                            lat, lon, radius_m, rooms=rooms, sort=sort, page=page,
-                        )
-                    except Exception:
-                        logger.exception(
-                            "yandex realty multi-room fetch failed: rooms=%s sort=%s page=%d",
-                            rooms, sort, page,
-                        )
-                        continue
-                    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 realty 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,
+        radius_m: int = 1000,
-        rooms: str | None = None,
+        max_pages: int = MAX_PAGES,
-        sort: str = "DATE_DESC",
+        **_legacy_kwargs: Any,  # swallow rooms/sorts/pages from old callers
-        page: int = 0,
+    ) -> list[ScrapedLot]:
-    ) -> str:
+        """Fetch up to max_pages of SERP cards, dedup by offer_id (source_id)."""
-        params: dict[str, Any] = {
+        seen: dict[str, ScrapedLot] = {}
-            "rgid": YANDEX_EKB_RGID,
+        for page in range(max_pages):
-            "offerType": "SELL",
+            lots = await self.fetch_around(lat, lon, radius_m, page=page)
-            "category": "APARTMENT",
+            if not lots:
-            "geoLat": f"{lat:.4f}",
+                break  # empty page → no more results
-            "geoLon": f"{lon:.4f}",
+            for lot in lots:
-            "geoRadius": radius_m,
+                key = lot.source_id or lot.source_url
-            "sort": sort,
+                if key and key not in seen:
-        }
+                    seen[key] = lot
-        if rooms:
+        logger.info(
-            # Yandex Realty принимает roomsTotal=STUDIO|1|2|3|PLUS_4
+            "yandex serp aggregate: %d unique lots over %d pages (city=%s)",
-            params["roomsTotal"] = rooms
+            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:
-            params["page"] = page
+            return f"{base}?page={page}"
-        return f"{self.base_url}/ekaterinburg/kupit/kvartira/?{urlencode(params)}"
+        return base
 
-    def _parse_html(self, html: str) -> list[ScrapedLot]:
+    def _parse_html(self, html: str, page: int = 0) -> list[ScrapedLot]:
-        # Поиск __INITIAL_STATE__ или __PRELOADED_STATE__
+        tree = HTMLParser(html)
-        state = _extract_yandex_state(html)
+        cards = tree.css('[data-test="OffersSerpItem"]')
-        if not state:
+        lots: list[ScrapedLot] = []
-            return []
+        for card in cards:
-        offers = _items_from_state(state)
+            lot = self._card_to_lot(card, page=page)
-        return [
+            if lot is not None:
-            lot for lot in (self._item_to_lot(o) for o in offers if isinstance(o, dict))
+                lots.append(lot)
-            if lot is not None
+        return lots
-        ]
 
-    def _item_to_lot(self, offer: dict[str, Any]) -> ScrapedLot | None:
+    def _card_to_lot(self, card: Node, page: int = 0) -> ScrapedLot | None:
         try:
-            # Цена: offer.price.value (там целое число рублей)
+            # offer_id — required
-            price_obj = offer.get("price") or {}
+            offer_link = card.css_first('a[href^="/offer/"]')
-            price = price_obj.get("value")
+            if offer_link is None:
-            if not price:
+                return None
-                pf = price_obj.get("priceForWhole") or {}
+            href = offer_link.attributes.get("href", "")
-                price = pf.get("value") if isinstance(pf, dict) else pf
+            id_match = RE_OFFER_ID.search(href)
-            if not price:
+            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_obj = offer.get("area") or {}
+            area_m = RE_TITLE_AREA.search(text)
-            area = area_obj.get("value")
+            area_m2 = (
-            if area:
+                float(area_m.group(1).replace(",", ".")) if area_m else None
-                area = float(area)
-
-            rooms = offer.get("roomsTotal") or offer.get("rooms")
-
-            # Этажи — offer.floorsOffered = [N], building.floorsTotal
-            floors_offered = offer.get("floorsOffered") or []
-            floor = floors_offered[0] if floors_offered else offer.get("floor")
-            building = offer.get("building") or {}
-            total_floors = building.get("floorsTotal") or offer.get("floorsTotal")
-            year_built = building.get("builtYear")
-            wall_type = building.get("buildingType")
-
-            # Локация
-            loc = offer.get("location") or {}
-            address = (
-                loc.get("address")
-                or loc.get("geocoderAddress")
-                or "Екатеринбург (Я.Недвижимость)"
             )
-            point = loc.get("point") or {}
-            lat = point.get("latitude")
-            lon = point.get("longitude")
 
-            # Anchor fallback (если EXACT нет — но обычно есть)
+            rooms = self._parse_rooms(text)
-            if (lat is None or lon is None) and 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
 
-            # URL
+            floor_m = RE_FLOOR.search(text)
-            offer_id = offer.get("offerId") or offer.get("id")
+            floor = int(floor_m.group(1)) if floor_m else None
-            internal_url = offer.get("unsignedInternalUrl") or ""
+            total_floors = int(floor_m.group(2)) if floor_m else None
-            if internal_url.startswith("//"):
+
-                full_url = "https:" + internal_url
+            ppm2_m = RE_PPM2.search(text)
-            elif offer_id:
+            price_per_m2 = parse_rub(ppm2_m.group(1)) if ppm2_m else None
-                full_url = f"https://realty.yandex.ru/offer/{offer_id}/"
+
-            else:
+            bargain = "торг" in text.lower()
-                return None
+            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
-            photos = []
+            photo_urls = self._extract_photos(card)
-            for img_field in ("appLargeImages", "extImages", "alikeImages"):
+
-                for img in (offer.get(img_field) or [])[:5]:
+            # Newbuilding linkage
-                    if isinstance(img, dict):
+            house_source: str | None = None
-                        url = img.get("xl") or img.get("large") or img.get("main", "")
+            house_ext_id: str | None = None
-                        if url:
+            house_url: str | None = None
-                            photos.append(url)
+            nb_link = card.css_first('a[href*="/kupit/novostrojka/"]')
-                    elif isinstance(img, str):
+            if nb_link is not None:
-                        photos.append(img)
+                nb_href = nb_link.attributes.get("href", "")
-                if photos:
+                nb_match = RE_JK_ID.search(nb_href)
-                    break
+                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="yandex",
+                source=self.name,
-                source_url=full_url,
+                source_url=f"https://realty.yandex.ru/offer/{offer_id}/",
-                source_id=str(offer_id) if offer_id else None,
+                source_id=offer_id,
                 address=address,
-                lat=lat,
+                lat=None,
-                lon=lon,
+                lon=None,
                 rooms=rooms,
-                area_m2=area,
+                area_m2=area_m2,
                 floor=floor,
                 total_floors=total_floors,
-                year_built=year_built,
+                price_rub=price_rub,
-                house_type=_map_wall_type(wall_type),
+                price_per_m2=price_per_m2,
-                price_rub=int(price),
+                bargain_allowed=bargain,
-                photo_urls=photos[:5],
+                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={
-                    # Compact dump (full offer 50KB+, тяжеловато для БД)
+                    "card_text": text[:1000],
-                    "offerId": offer_id,
+                    "page": page,
-                    "creationDate": offer.get("creationDate"),
+                    "city": self.city,
-                    "dealStatus": offer.get("dealStatus"),
-                    "description": (offer.get("description") or "")[:500],
                 },
             )
         except Exception:
-            logger.exception("yandex item parse failed: %s", offer.get("offerId"))
+            logger.exception("yandex card parse failed (page=%d)", page)
             return None
 
-
+    @staticmethod
-# ── Helpers ──────────────────────────────────────────────────────────────────
+    def _parse_rooms(text: str) -> int | None:
-
+        m = RE_TITLE_ROOMS.search(text)
-
+        if not m:
-_RE_STATE = re.compile(
+            return None
-    r'<script id="initial_state_script"[^>]*>(.+?)</script>',
+        if m.group(2):  # studio
-    re.DOTALL,
+            return 0
-)
+        if m.group(1):  # numbered
-
+            try:
-
+                return int(m.group(1))
-def _extract_yandex_state(html: str) -> dict[str, Any] | None:
+            except ValueError:
-    m = _RE_STATE.search(html)
+                return None
-    if not m:
-        return None
-    raw = m.group(1)
-    # Может начинаться с `window.__INITIAL_STATE__=` префикса
-    raw = re.sub(r'^\s*window\.[\w_]+\s*=\s*', '', raw).rstrip(';').strip()
-    try:
-        return json.loads(raw)
-    except json.JSONDecodeError:
         return None
 
-
+    @staticmethod
-def _items_from_state(state: dict[str, Any]) -> list[dict[str, Any]]:
+    def _extract_photos(card: Node) -> list[str]:
-    """Yandex Realty: state.map.offers.points содержит список объявлений (десктоп desktop SSR)."""
+        urls: list[str] = []
-    for path in [
+        for img in card.css("img"):
-        ("map", "offers", "points"),    # верифицировано на проде
+            src = img.attributes.get("src", "") or ""
-        ("search", "offers"),
+            if PHOTO_DOMAIN in src:
-        ("offers", "points"),
+                urls.append(src.replace(PHOTO_SIZE_FROM, PHOTO_SIZE_TO))
-        ("offers", "items"),
+            if len(urls) >= 5:
-    ]:
-        cursor: Any = state
-        for key in path:
-            if not isinstance(cursor, dict):
-                cursor = None
                 break
-            cursor = cursor.get(key)
+        return urls
-        if isinstance(cursor, list) and cursor:
-            return cursor
-    return []
-
-
-def _map_wall_type(yandex_type: str | None) -> str | None:
-    if not yandex_type:
-        return None
-    mapping = {
-        "PANEL": "panel",
-        "BRICK": "brick",
-        "MONOLITH": "monolith",
-        "MONOLITH_BRICK": "monolith_brick",
-        "BLOCK": "panel",
-    }
-    return mapping.get(yandex_type.upper(), "other")

tradein-mvp/backend/tests/test_yandex_realty_serp.py

@@ -0,0 +1,136 @@
+"""Unit tests for YandexRealtyScraper DOM-based SERP parser."""
+from datetime import date
+
+import pytest
+
+from app.services.scrapers.yandex_realty import (
+    DEFAULT_CITY,
+    MAX_PAGES,
+    YandexRealtyScraper,
+)
+
+# Realistic single-card fixture (trimmed but selector-faithful).
+# Note: selectolax.text(strip=True) concatenates all text nodes without
+# added separators — so floor/price data lives in a single line div with
+# middot (·) separators, matching real Yandex SERP card layout.
+SINGLE_CARD_HTML = """
+<html><body>
+<div data-test="OffersSerpItem">
+  <a href="/offer/7567094292504417257/">Открыть</a>
+  <a href="/ekaterinburg/kupit/kvartira/st-frezerovshchikov-12345/">Фрезеровщиков</a>
+  <a href="/ekaterinburg/kupit/novostrojka/tatlin-1592987/">ЖК Татлин</a>
+  <img src="https://avatars.mds.yandex.net/get-realty/12345/app_snippet_small/test.jpg">
+  <img src="https://avatars.mds.yandex.net/get-realty/67890/app_snippet_small/test2.jpg">
+  <img src="https://other-domain.example/photo.jpg">
+  <div>36,8 м² · 1-комнатная квартира · 8 этаж из 9 · 4 399 000 ₽ · 119 566 ₽ за м²</div>
+  <div>торг возможен · опубликовано 17 февраля 2026</div>
+</div>
+</body></html>
+"""
+
+STUDIO_CARD_HTML = """
+<div data-test="OffersSerpItem">
+  <a href="/offer/9999999/">x</a>
+  <div>Студия · 28 м² · 2 этаж из 9 · 3 200 000 ₽</div>
+</div>
+"""
+
+NO_PRICE_CARD_HTML = """
+<div data-test="OffersSerpItem">
+  <a href="/offer/1111/">x</a>
+  <div>40 м² · 1-комнатная · 3 этаж из 5</div>
+</div>
+"""
+
+EMPTY_PAGE_HTML = "<html><body><div>Нет объявлений</div></body></html>"
+
+
+def test_max_pages_default():
+    assert MAX_PAGES == 3
+
+
+def test_default_city():
+    assert DEFAULT_CITY == "ekaterinburg"
+
+
+def test_build_url_first_page():
+    s = YandexRealtyScraper()
+    assert s._build_url(page=0) == (
+        "https://realty.yandex.ru/ekaterinburg/kupit/kvartira/vtorichniy-rynok/"
+    )
+
+
+def test_build_url_paginated():
+    s = YandexRealtyScraper()
+    assert s._build_url(page=2) == (
+        "https://realty.yandex.ru/ekaterinburg/kupit/kvartira/vtorichniy-rynok/?page=2"
+    )
+
+
+def test_build_url_other_city():
+    s = YandexRealtyScraper(city="moscow")
+    assert s._build_url(page=0) == (
+        "https://realty.yandex.ru/moscow/kupit/kvartira/vtorichniy-rynok/"
+    )
+
+
+def test_parse_html_extracts_card():
+    s = YandexRealtyScraper()
+    lots = s._parse_html(SINGLE_CARD_HTML, page=0)
+    assert len(lots) == 1
+    lot = lots[0]
+    assert lot.source == "yandex"
+    assert lot.source_id == "7567094292504417257"
+    assert lot.source_url == "https://realty.yandex.ru/offer/7567094292504417257/"
+    assert lot.address == "Фрезеровщиков"
+    assert lot.area_m2 == pytest.approx(36.8)
+    assert lot.rooms == 1
+    assert lot.floor == 8
+    assert lot.total_floors == 9
+    assert lot.price_rub == 4_399_000
+    assert lot.price_per_m2 == 119_566
+    assert lot.bargain_allowed is True
+    assert lot.listing_date == date(2026, 2, 17)
+    assert lot.house_source == "yandex_realty_nb"
+    assert lot.house_ext_id == "1592987"
+    assert lot.house_url == (
+        "https://realty.yandex.ru/ekaterinburg/kupit/novostrojka/tatlin-1592987/"
+    )
+    assert lot.listing_segment == "vtorichka"
+    assert lot.lat is None and lot.lon is None  # SERP has no coords
+    # Photos: 2 from yandex CDN, the third (other-domain) skipped
+    assert len(lot.photo_urls) == 2
+    assert all("avatars.mds.yandex" in u for u in lot.photo_urls)
+    # Size upgrade applied: app_snippet_small → main
+    assert all("app_snippet_small" not in u for u in lot.photo_urls)
+    assert all("main" in u for u in lot.photo_urls)
+
+
+def test_parse_html_studio_rooms_zero():
+    s = YandexRealtyScraper()
+    lots = s._parse_html(STUDIO_CARD_HTML, page=0)
+    assert len(lots) == 1
+    assert lots[0].rooms == 0  # studio convention
+    assert lots[0].area_m2 == pytest.approx(28.0)
+    assert lots[0].floor == 2 and lots[0].total_floors == 9
+
+
+def test_parse_html_no_price_skipped():
+    s = YandexRealtyScraper()
+    lots = s._parse_html(NO_PRICE_CARD_HTML, page=0)
+    assert lots == []
+
+
+def test_parse_html_empty_page():
+    s = YandexRealtyScraper()
+    lots = s._parse_html(EMPTY_PAGE_HTML, page=0)
+    assert lots == []
+
+
+def test_parse_html_multi_card():
+    multi = SINGLE_CARD_HTML + STUDIO_CARD_HTML
+    s = YandexRealtyScraper()
+    lots = s._parse_html(multi, page=1)
+    assert len(lots) == 2
+    assert {lot.source_id for lot in lots} == {"7567094292504417257", "9999999"}
+    assert all(lot.raw_payload["page"] == 1 for lot in lots)