"""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.scraper_settings import get_scraper_delay 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 = 5.0 # class default; instance value loaded from scraper_settings def __init__(self) -> None: super().__init__() self.request_delay_sec = get_scraper_delay(self.name) 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]