"""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]