From fda9ae4dc11c5b0a00b5fa7c195a9147c8d10553 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sat, 23 May 2026 16:40:23 +0300 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20yandex=5Fvaluation.py=20?= =?UTF-8?q?=E2=80=94=20anonymous=20house-history=20scraper?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 6 of YandexRealtyScraper v1 (Wave 4 / Worker C — parallel with yandex_detail.py and yandex_newbuilding.py). Module: app/services/scrapers/yandex_valuation.py - YandexValuationScraper(BaseScraper) — fetch_house_history(address, ...) URL: GET /otsenka-kvartiry-po-adresu-onlayn/?address=...&offerCategory=... &offerType=...&page=N ANONYMOUS — no cookies, no auth (vs Cian Calculator). - ValuationHouseMeta (year_built, total_floors, house_type, ceiling_height, has_lift, total_objects, has_panorama) - ValuationHistoryItem (area_m2, rooms, floor, start/last price + ppm2, publish_date DMY, exposure_days, status) - YandexValuationResult — house + history_items[] + pagination metadata - Two-strategy history extraction: data-test container (if present), else text-chunk-around-DMY-date fallback with dedup by (date, area, floor). - _RE_PRICE_TOKEN uses negative lookahead (?!\s*за\s*м²) to exclude ppm2 tokens from total-price extraction. Tests: 16 unit tests against fixture HTML. All pass. Ruff clean. --- .../app/services/scrapers/yandex_valuation.py | 331 ++++++++++++++++++ .../backend/tests/test_yandex_valuation.py | 242 +++++++++++++ 2 files changed, 573 insertions(+) create mode 100644 tradein-mvp/backend/app/services/scrapers/yandex_valuation.py create mode 100644 tradein-mvp/backend/tests/test_yandex_valuation.py diff --git a/tradein-mvp/backend/app/services/scrapers/yandex_valuation.py b/tradein-mvp/backend/app/services/scrapers/yandex_valuation.py new file mode 100644 index 00000000..daa3606c --- /dev/null +++ b/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] diff --git a/tradein-mvp/backend/tests/test_yandex_valuation.py b/tradein-mvp/backend/tests/test_yandex_valuation.py new file mode 100644 index 00000000..2de8afd8 --- /dev/null +++ b/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"{body_content}" + + +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