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

@@ -0,0 +1,331 @@
+"""Yandex Realty anonymous valuation/history scraper.
+
+URL pattern:
+  GET /otsenka-kvartiry-po-adresu-onlayn/?address=...&offerCategory=...&offerType=...&page=N
+
+No cookies, no auth required (unlike Cian Calculator or Avito IMV).
+
+Extracts:
+- House metadata: year_built, total_floors, house_type, ceiling_height, has_lift, total_objects
+- Historical offers: area, rooms, floor, start/last price + per-m2, publish_date DMY, exposure
+
+Two-strategy history extraction:
+1. data-test container selectors (CSS, if Yandex adds them)
+2. Fallback: text chunks around DD.MM.YYYY date matches with dedup by (date, area, floor)
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from datetime import date
+from typing import Any
+from urllib.parse import urlencode
+
+from pydantic import BaseModel, Field
+from selectolax.parser import HTMLParser
+
+from app.services.scrapers.base import BaseScraper
+from app.services.scrapers.yandex_helpers import (
+    parse_dmy,
+    parse_house_type,
+    parse_rub,
+)
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Models
+# ---------------------------------------------------------------------------
+
+
+class ValuationHouseMeta(BaseModel):
+    """House metadata extracted from Yandex valuation page."""
+
+    year_built: int | None = None
+    total_floors: int | None = None
+    house_type: str | None = None  # panel/brick/monolith/...
+    ceiling_height: float | None = None  # in meters, e.g. 2.5
+    has_lift: bool | None = None
+    total_objects: int | None = None  # 'N объектов' (full archive count)
+    has_panorama: bool = False  # 'Панорама' label present
+
+
+class ValuationHistoryItem(BaseModel):
+    """One historical offer entry from the valuation page."""
+
+    area_m2: float | None = None
+    rooms: int | None = None
+    floor: int | None = None
+    start_price: int | None = None
+    start_price_per_m2: int | None = None
+    last_price: int | None = None
+    last_price_per_m2: int | None = None
+    publish_date: date | None = None
+    exposure_days: int | None = None
+    status: str | None = None  # 'В продаже' / 'Снято'
+
+
+class YandexValuationResult(BaseModel):
+    """Full result of one /otsenka-... GET."""
+
+    address: str
+    offer_category: str
+    offer_type: str
+    page: int
+    source_url: str
+    house: ValuationHouseMeta
+    history_items: list[ValuationHistoryItem] = Field(default_factory=list)
+    raw_payload: dict[str, Any] | None = None
+
+
+# ---------------------------------------------------------------------------
+# Regex constants
+# ---------------------------------------------------------------------------
+
+RE_YEAR_BUILT = re.compile(r"Дом\s+(\d{4})\s+года", re.IGNORECASE)
+RE_FLOORS = re.compile(r"(\d+)\s+этаж", re.IGNORECASE)
+RE_CEILING = re.compile(r"([\d,]+)\s*м\s+потолки", re.IGNORECASE)
+RE_TOTAL_OBJECTS = re.compile(r"(\d+)\s+объект", re.IGNORECASE)
+
+RE_ITEM_AREA = re.compile(r"(\d+[.,]?\d*)\s*м²")
+RE_ITEM_ROOMS = re.compile(r"(\d+)\s*-\s*комнатн", re.IGNORECASE)
+RE_ITEM_STUDIO = re.compile(r"студи[яюй]", re.IGNORECASE)
+RE_ITEM_FLOOR = re.compile(r"(\d+)\s*этаж", re.IGNORECASE)
+RE_ITEM_EXPOSURE = re.compile(r"экспозиции\s+(\d+)\s+дн", re.IGNORECASE)
+RE_ITEM_STATUS = re.compile(r"(В\s+продаже|Снят[оа])", re.IGNORECASE)
+
+# Matches total-price tokens (rubles) — excludes per-m2 tokens by negative lookahead
+_RE_PRICE_TOKEN = re.compile(
+    r"(?:\d[\d\s]*\d|\d)(?:[.,]\d+)?\s*(?:млн)?\s*₽(?!\s*за\s*м²)"
+)
+_RE_PPM2_TOKEN = re.compile(r"\d[\d\s]*\s*₽\s*за\s*м²")
+
+
+# ---------------------------------------------------------------------------
+# Scraper
+# ---------------------------------------------------------------------------
+
+
+class YandexValuationScraper(BaseScraper):
+    """Anonymous Yandex valuation/history scraper.
+
+    Fetches https://realty.yandex.ru/otsenka-kvartiry-po-adresu-onlayn/ by address.
+    Pagination via ?page=N. Use fetch_house_history() directly;
+    fetch_around() is not applicable to this tool.
+    """
+
+    name = "yandex_valuation"
+    base_url = "https://realty.yandex.ru"
+    valuation_path = "/otsenka-kvartiry-po-adresu-onlayn/"
+    request_delay_sec = 4.0
+
+    async def fetch_around(
+        self, lat: float, lon: float, radius_m: int = 1000
+    ) -> list:  # type: ignore[override]
+        raise NotImplementedError(
+            "YandexValuationScraper is address-based; use fetch_house_history() instead."
+        )
+
+    async def fetch_house_history(
+        self,
+        address: str,
+        offer_category: str = "APARTMENT",
+        offer_type: str = "SELL",
+        page: int = 1,
+    ) -> YandexValuationResult | None:
+        """Fetch and parse one page of house history for the given address.
+
+        Args:
+            address: Postal address string (will be URL-encoded).
+            offer_category: 'APARTMENT' / 'ROOMS' / etc.
+            offer_type: 'SELL' / 'RENT'.
+            page: 1-based pagination index.
+
+        Returns:
+            YandexValuationResult or None on HTTP error / network failure.
+        """
+        params = {
+            "address": address,
+            "offerCategory": offer_category,
+            "offerType": offer_type,
+            "page": page,
+        }
+        url = f"{self.base_url}{self.valuation_path}?{urlencode(params)}"
+        try:
+            response = await self._http_get(url)
+        except Exception:
+            logger.exception("yandex valuation fetch failed: %s", url)
+            return None
+        if response.status_code != 200:
+            logger.warning(
+                "yandex valuation returned %d for %s", response.status_code, url
+            )
+            return None
+        result = self.parse(
+            response.text,
+            address=address,
+            offer_category=offer_category,
+            offer_type=offer_type,
+            page=page,
+            source_url=url,
+        )
+        await self.sleep_between_requests()
+        return result
+
+    def parse(
+        self,
+        html: str,
+        address: str,
+        offer_category: str,
+        offer_type: str,
+        page: int,
+        source_url: str,
+    ) -> YandexValuationResult:
+        """Parse raw HTML into YandexValuationResult. Pure function — usable in unit tests."""
+        tree = HTMLParser(html)
+        body = tree.body
+        body_text = body.text(strip=True) if body else ""
+
+        house = self._parse_house_meta(body_text)
+        history_items = self._parse_history_items(tree, body_text)
+
+        return YandexValuationResult(
+            address=address,
+            offer_category=offer_category,
+            offer_type=offer_type,
+            page=page,
+            source_url=source_url,
+            house=house,
+            history_items=history_items,
+            raw_payload={
+                "body_len": len(body_text),
+                "items_count": len(history_items),
+            },
+        )
+
+    @staticmethod
+    def _parse_house_meta(body_text: str) -> ValuationHouseMeta:
+        """Extract house-level metadata from page body text."""
+        year_m = RE_YEAR_BUILT.search(body_text)
+        floors_m = RE_FLOORS.search(body_text)
+        ceiling_m = RE_CEILING.search(body_text)
+        objects_m = RE_TOTAL_OBJECTS.search(body_text)
+        return ValuationHouseMeta(
+            year_built=int(year_m.group(1)) if year_m else None,
+            total_floors=int(floors_m.group(1)) if floors_m else None,
+            house_type=parse_house_type(body_text),
+            ceiling_height=(
+                float(ceiling_m.group(1).replace(",", ".")) if ceiling_m else None
+            ),
+            has_lift="Лифт" in body_text,
+            total_objects=int(objects_m.group(1)) if objects_m else None,
+            has_panorama="Панорама" in body_text,
+        )
+
+    def _parse_history_items(
+        self, tree: HTMLParser, body_text: str
+    ) -> list[ValuationHistoryItem]:
+        """Extract list of historical offer items using best available strategy.
+
+        Strategy 1: CSS data-test containers (if Yandex exposes them).
+        Strategy 2: Fallback — split body text into chunks around DD.MM.YYYY dates.
+        """
+        items: list[ValuationHistoryItem] = []
+        # Strategy 1: explicit data-test container
+        for container in tree.css(
+            '[data-test*="HistoryItem"], [data-test*="ValuationItem"]'
+        ):
+            item = self._parse_item_text(container.text(strip=True))
+            if item:
+                items.append(item)
+        if items:
+            return items
+        # Strategy 2: text-chunk fallback
+        return self._parse_items_from_chunked_text(body_text)
+
+    @staticmethod
+    def _parse_item_text(text: str) -> ValuationHistoryItem | None:
+        """Parse a single text block into a ValuationHistoryItem.
+
+        Returns None if neither area nor start_price can be extracted
+        (item is considered invalid/noise).
+        """
+        if not text or len(text) < 20:
+            return None
+
+        # area + rooms
+        area_m = RE_ITEM_AREA.search(text)
+        area_m2 = float(area_m.group(1).replace(",", ".")) if area_m else None
+
+        rooms: int | None = None
+        if RE_ITEM_STUDIO.search(text):
+            rooms = 0
+        else:
+            rooms_m = RE_ITEM_ROOMS.search(text)
+            if rooms_m:
+                rooms = int(rooms_m.group(1))
+
+        floor_m = RE_ITEM_FLOOR.search(text)
+        floor = int(floor_m.group(1)) if floor_m else None
+
+        # Prices — find first 2 price tokens (start, last)
+        price_tokens = _RE_PRICE_TOKEN.findall(text)
+        start_price = parse_rub(price_tokens[0]) if len(price_tokens) >= 1 else None
+        last_price = parse_rub(price_tokens[1]) if len(price_tokens) >= 2 else None
+
+        ppm2_tokens = _RE_PPM2_TOKEN.findall(text)
+        start_ppm2 = parse_rub(ppm2_tokens[0]) if len(ppm2_tokens) >= 1 else None
+        last_ppm2 = parse_rub(ppm2_tokens[1]) if len(ppm2_tokens) >= 2 else None
+
+        publish_date = parse_dmy(text)
+        expo_m = RE_ITEM_EXPOSURE.search(text)
+        exposure_days = int(expo_m.group(1)) if expo_m else None
+
+        status_m = RE_ITEM_STATUS.search(text)
+        status = status_m.group(1) if status_m else None
+
+        # Item is valid only if we got at least area or price
+        if area_m2 is None and start_price is None:
+            return None
+
+        return ValuationHistoryItem(
+            area_m2=area_m2,
+            rooms=rooms,
+            floor=floor,
+            start_price=start_price,
+            start_price_per_m2=start_ppm2,
+            last_price=last_price,
+            last_price_per_m2=last_ppm2,
+            publish_date=publish_date,
+            exposure_days=exposure_days,
+            status=status,
+        )
+
+    @classmethod
+    def _parse_items_from_chunked_text(cls, body_text: str) -> list[ValuationHistoryItem]:
+        """Fallback: split body text into chunks around DD.MM.YYYY dates and parse each chunk.
+
+        Window: 200 chars before + 100 chars after each date match.
+        Deduplicates by (publish_date, area_m2, floor).
+        Caps output at 30 items per page to avoid runaway extraction.
+        """
+        items: list[ValuationHistoryItem] = []
+        for m in re.finditer(r"\d{2}\.\d{2}\.\d{4}", body_text):
+            start = max(0, m.start() - 200)
+            end = min(len(body_text), m.end() + 100)
+            chunk = body_text[start:end]
+            item = cls._parse_item_text(chunk)
+            if item:
+                items.append(item)
+
+        # Dedup by (publish_date, area_m2, floor)
+        seen: set[tuple[Any, Any, Any]] = set()
+        deduped: list[ValuationHistoryItem] = []
+        for item in items:
+            key = (item.publish_date, item.area_m2, item.floor)
+            if key not in seen:
+                seen.add(key)
+                deduped.append(item)
+        return deduped[:30]

tradein-mvp/backend/tests/test_yandex_valuation.py

@@ -0,0 +1,242 @@
+"""Unit tests for YandexValuationScraper — anonymous house-history scraper.
+
+Fixture HTML simulates the Yandex valuation page body text containing:
+- House meta block (year, floors, type, ceiling, lift, total objects, panorama)
+- 2-3 historical offer entries with full structure
+"""
+
+from __future__ import annotations
+
+from datetime import date
+
+import pytest
+
+from app.services.scrapers.yandex_valuation import (
+    YandexValuationResult,
+    YandexValuationScraper,
+)
+
+# ---------------------------------------------------------------------------
+# Fixture helpers
+# ---------------------------------------------------------------------------
+
+_HOUSE_META_BLOCK = (
+    "12 объектов  Дом 1981 года  9 этажей  Панельное здание  2,50 м потолки  Лифт"
+)
+
+_ITEM_1 = (
+    "2-комнатная квартира 45,3 м²  4 этаж  "
+    "Опубликовано 15.03.2023  "
+    "Начальная цена 3 200 000 ₽  117 000 ₽ за м²  "
+    "Последняя цена 3 100 000 ₽  114 000 ₽ за м²  "
+    "Длительность экспозиции 42 дня  В продаже"
+)
+
+_ITEM_2 = (
+    "Студия 25,0 м²  1 этаж  "
+    "Опубликовано 20.11.2022  "
+    "Начальная цена 2 100 000 ₽  84 000 ₽ за м²  "
+    "Последняя цена 1 950 000 ₽  78 000 ₽ за м²  "
+    "Длительность экспозиции 90 дней  Снято"
+)
+
+_ITEM_3 = (
+    "1-комнатная квартира 32,5 м²  7 этаж  "
+    "Опубликовано 01.06.2024  "
+    "Начальная цена 2 800 000 ₽  "
+    "Длительность экспозиции 15 дней  В продаже"
+)
+
+
+def _make_full_html(body_content: str) -> str:
+    return f"<html><body>{body_content}</body></html>"
+
+
+FULL_FIXTURE_HTML = _make_full_html(
+    f"{_HOUSE_META_BLOCK}\n{_ITEM_1}\n{_ITEM_2}\n{_ITEM_3}"
+)
+
+
+# ---------------------------------------------------------------------------
+# House meta tests
+# ---------------------------------------------------------------------------
+
+
+def test_parse_house_meta_full():
+    scraper = YandexValuationScraper()
+    meta = scraper._parse_house_meta(_HOUSE_META_BLOCK)
+    assert meta.year_built == 1981
+    assert meta.total_floors == 9
+    assert meta.house_type == "panel"
+    assert meta.ceiling_height == 2.50
+    assert meta.has_lift is True
+    assert meta.total_objects == 12
+    assert meta.has_panorama is False
+
+
+def test_parse_house_meta_no_lift():
+    text = "5 объектов  Дом 2005 года  16 этажей  Монолитное здание  3,00 м потолки"
+    meta = YandexValuationScraper._parse_house_meta(text)
+    assert meta.has_lift is False
+    assert meta.year_built == 2005
+    assert meta.total_floors == 16
+    assert meta.house_type == "monolith"
+    assert meta.ceiling_height == 3.0
+
+
+def test_parse_house_meta_with_panorama():
+    text = "7 объектов  Дом 2010 года  Панорама  Лифт  Кирпичное здание"
+    meta = YandexValuationScraper._parse_house_meta(text)
+    assert meta.has_panorama is True
+    assert meta.has_lift is True
+    assert meta.house_type == "brick"
+
+
+def test_house_meta_year_not_present_returns_none():
+    text = "8 объектов  5 этажей  Панельное здание"
+    meta = YandexValuationScraper._parse_house_meta(text)
+    assert meta.year_built is None
+    assert meta.total_floors == 5
+    assert meta.total_objects == 8
+
+
+# ---------------------------------------------------------------------------
+# History item tests
+# ---------------------------------------------------------------------------
+
+
+def test_parse_history_item_full():
+    item = YandexValuationScraper._parse_item_text(_ITEM_1)
+    assert item is not None
+    assert item.area_m2 == 45.3
+    assert item.rooms == 2
+    assert item.floor == 4
+    assert item.publish_date == date(2023, 3, 15)
+    assert item.start_price == 3_200_000
+    assert item.last_price == 3_100_000
+    assert item.exposure_days == 42
+    assert item.status is not None
+    assert "продаже" in item.status.lower() or "В" in item.status
+
+
+def test_parse_history_item_studio_rooms_zero():
+    item = YandexValuationScraper._parse_item_text(_ITEM_2)
+    assert item is not None
+    assert item.rooms == 0  # Studio
+    assert item.area_m2 == 25.0
+    assert item.floor == 1
+    assert item.publish_date == date(2022, 11, 20)
+
+
+def test_parse_history_item_status_sold():
+    item = YandexValuationScraper._parse_item_text(_ITEM_2)
+    assert item is not None
+    assert item.status is not None
+    assert "нят" in item.status.lower() or "Снят" in item.status
+
+
+def test_parse_history_item_returns_none_for_empty():
+    assert YandexValuationScraper._parse_item_text("") is None
+    assert YandexValuationScraper._parse_item_text("   ") is None
+
+
+def test_parse_history_item_returns_none_without_area_or_price():
+    # Text too short / no extractable fields
+    assert YandexValuationScraper._parse_item_text("нет данных") is None
+
+
+# ---------------------------------------------------------------------------
+# Chunked text extraction tests
+# ---------------------------------------------------------------------------
+
+
+def test_parse_items_chunked_dedup():
+    """Duplicate entry (same date + area + floor) must appear only once."""
+    duplicate_block = f"{_ITEM_1}\n{_ITEM_1}"
+    items = YandexValuationScraper._parse_items_from_chunked_text(duplicate_block)
+    # Both items reference 15.03.2023 + 45.3 m2 + floor 4 — must dedup to 1
+    matching = [
+        i for i in items if i.area_m2 == 45.3 and i.publish_date == date(2023, 3, 15)
+    ]
+    assert len(matching) == 1
+
+
+def test_parse_items_from_chunked_text_no_dates_returns_empty():
+    """Body text with no DD.MM.YYYY dates should return empty list."""
+    items = YandexValuationScraper._parse_items_from_chunked_text(
+        "Нет объявлений. Попробуйте изменить параметры поиска."
+    )
+    assert items == []
+
+
+def test_parse_items_from_chunked_text_multiple_items():
+    """Three distinct items should be extracted from combined body text."""
+    body = f"{_ITEM_1}\n{_ITEM_2}\n{_ITEM_3}"
+    items = YandexValuationScraper._parse_items_from_chunked_text(body)
+    assert len(items) >= 2  # at minimum items 1 and 2 (item 3 has only start_price)
+
+
+# ---------------------------------------------------------------------------
+# Full parse round-trip
+# ---------------------------------------------------------------------------
+
+
+def test_parse_full_result_address_propagated():
+    scraper = YandexValuationScraper()
+    result = scraper.parse(
+        FULL_FIXTURE_HTML,
+        address="Екатеринбург, ул. Ленина, 1",
+        offer_category="APARTMENT",
+        offer_type="SELL",
+        page=1,
+        source_url="https://realty.yandex.ru/otsenka-kvartiry-po-adresu-onlayn/?page=1",
+    )
+    assert isinstance(result, YandexValuationResult)
+    assert result.address == "Екатеринбург, ул. Ленина, 1"
+    assert result.offer_category == "APARTMENT"
+    assert result.offer_type == "SELL"
+    assert result.page == 1
+    assert result.house.year_built == 1981
+    assert isinstance(result.history_items, list)
+    assert result.raw_payload is not None
+    assert "body_len" in result.raw_payload
+
+
+def test_parse_full_result_house_meta_populated():
+    scraper = YandexValuationScraper()
+    result = scraper.parse(
+        FULL_FIXTURE_HTML,
+        address="test",
+        offer_category="APARTMENT",
+        offer_type="SELL",
+        page=1,
+        source_url="https://realty.yandex.ru/",
+    )
+    assert result.house.total_floors == 9
+    assert result.house.has_lift is True
+    assert result.house.house_type == "panel"
+    assert result.house.ceiling_height == 2.5
+
+
+# ---------------------------------------------------------------------------
+# fetch_around raises NotImplementedError
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_fetch_around_raises_not_implemented():
+    scraper = YandexValuationScraper()
+    with pytest.raises(NotImplementedError):
+        await scraper.fetch_around(lat=56.8, lon=60.6)
+
+
+# ---------------------------------------------------------------------------
+# Ceiling height comma parsing
+# ---------------------------------------------------------------------------
+
+
+def test_parse_ceiling_height_comma():
+    """Comma decimal separator '2,70 м потолки' must parse to 2.7."""
+    text = "Дом 1999 года  5 этажей  Кирпичное здание  2,70 м потолки  Лифт"
+    meta = YandexValuationScraper._parse_house_meta(text)
+    assert meta.ceiling_height == 2.7