"""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 curl_cffi.requests import AsyncSession as _CurlCffiSession 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 def validate_match( self, expected_year_built: int | None = None, expected_total_floors: int | None = None, ) -> float: """Return confidence 0..1 that this house meta matches expected values. Used after fetching valuation by address to detect when Yandex returned a different house (ambiguous address geocoding). Tolerance ±1 year, ±1 floor. Both expected=None → 1.0 (no check). Mismatch on any dimension → 0.0 for it. """ score = 0.0 checks = 0 if expected_year_built is not None: checks += 1 if self.year_built is not None and abs(self.year_built - expected_year_built) <= 1: score += 1.0 if expected_total_floors is not None: checks += 1 if ( self.total_floors is not None and abs(self.total_floors - expected_total_floors) <= 1 ): score += 1.0 if checks == 0: return 1.0 return score / checks 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 removed_date: date | None = None # ← NEW 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) # Lookbehind blocks digit-only adjacency (rejects year-concat like '20244,6') # while still allowing tokens preceded by punctuation/letter. Min 2 digits # rejects sub-fragments like '2,2' that come from inside '52,2'. Max 4 digits # whole part catches obvious junk (penthouse extremes are <500 m² in practice; # sanity cap from PR #538 backs this up). RE_ITEM_AREA = re.compile(r"(? None: super().__init__() self.request_delay_sec = get_scraper_delay(self.name) self._cffi_session: _CurlCffiSession | None = None async def __aenter__(self) -> YandexValuationScraper: # type: ignore[override] # Override: Yandex valuation endpoint gates SSR data on Chrome TLS # fingerprint. Plain httpx returns shell HTML (CSR-only). # Sibling: yandex_realty.py / scripts/local-sweep-ekb-yandex.py self._cffi_session = _CurlCffiSession(impersonate="chrome120") return self async def __aexit__(self, *args: object) -> None: # type: ignore[override] if self._cffi_session is not None: await self._cffi_session.close() self._cffi_session = None async def _http_get(self, url: str, **kwargs: object) -> object: # type: ignore[override] """curl_cffi-based GET with Chrome120 impersonation. Returns curl_cffi Response (compatible API: .status_code, .text). Caller must check status_code; no automatic retry (BaseScraper.retry decorator is per-method and not inherited cleanly when overridden). """ if self._cffi_session is None: raise RuntimeError("YandexValuationScraper must be used as async context manager") kwargs.setdefault("timeout", 30) return await self._cffi_session.get(url, **kwargs) 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.""" html_normalized = html.replace("\xa0", " ") tree = HTMLParser(html_normalized) body = tree.body body_text = (body.text(strip=True) if body else "").replace("\xa0", " ") 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 historical offer items. Strategy 1 (preferred, structured): each row at .OffersArchiveSearchOffers__row, with 6 cells (.OffersArchiveSearchOffers__cell) in fixed order: [0] area+rooms, [1] floor, [2] start_price+ppm2, [3] last_price+ppm2, [4] publish_date+exposure, [5] status/removed_date. Yields per-cell parsing — no chunked-text ambiguity. Strategy 2 (last-resort fallback): legacy chunked-text scan around DD.MM.YYYY anchors. Kept for the rare case Yandex changes class names; triggers a logger.warning so we notice. """ items: list[ValuationHistoryItem] = [] for row in tree.css(".OffersArchiveSearchOffers__row"): item = self._parse_row_cells(row) if item: items.append(item) if items: return items # Fallback (only if structured selector failed) logger.warning( "yandex_valuation: .OffersArchiveSearchOffers__row matched 0 items — " "falling back to chunked-text scan" ) return self._parse_items_from_chunked_text(body_text) @classmethod def _parse_row_cells(cls, row) -> ValuationHistoryItem | None: """Parse one .OffersArchiveSearchOffers__row into a ValuationHistoryItem. Cell layout is positional (no labels). Returns None if the row doesn't match the expected 6-cell structure (defensive — Yandex may A/B-test). """ cells = row.css(".OffersArchiveSearchOffers__cell") if len(cells) < 5: return None cell_texts = [(c.text(strip=True) or "").replace("\xa0", " ") for c in cells] # [0] photo (skipped — empty cell) # [1] title — single cell contains area + rooms + floor jammed together, # e.g. "48,5 м², 1-комнатная19 этаж" (real Yandex layout, 2026-05-24) title_text = cell_texts[1] if len(cell_texts) > 1 else "" area_m = RE_ITEM_AREA.search(title_text) area_m2: float | None = float(area_m.group(1).replace(",", ".")) if area_m else None # Sanity drop (belt-and-suspenders from PR #538) if area_m2 is not None and (area_m2 > 10_000 or area_m2 <= 0): area_m2 = None rooms: int | None = None if RE_ITEM_STUDIO.search(title_text): rooms = 0 else: rooms_m = RE_ITEM_ROOMS.search(title_text) if rooms_m: rooms = int(rooms_m.group(1)) floor_m = RE_ITEM_FLOOR.search(title_text) floor: int | None = int(floor_m.group(1)) if floor_m else None # [2] start price + ppm2 — "5,1 млн ₽105 155 ₽ за м²" (no separator) start_price: int | None = None start_ppm2: int | None = None if len(cell_texts) > 2: tokens = _RE_PRICE_TOKEN.findall(cell_texts[2]) if tokens: start_price = parse_rub(tokens[0]) ppm2_tokens = _RE_PPM2_TOKEN.findall(cell_texts[2]) if ppm2_tokens: start_ppm2 = parse_rub(ppm2_tokens[0]) # [3] last price + ppm2 last_price: int | None = None last_ppm2: int | None = None if len(cell_texts) > 3: tokens = _RE_PRICE_TOKEN.findall(cell_texts[3]) if tokens: last_price = parse_rub(tokens[0]) ppm2_tokens = _RE_PPM2_TOKEN.findall(cell_texts[3]) if ppm2_tokens: last_ppm2 = parse_rub(ppm2_tokens[0]) # [4] publish_date + exposure ("23.10.2023В экспозиции 945 дней") publish_date = None exposure_days: int | None = None if len(cell_texts) > 4: publish_date = parse_dmy(cell_texts[4]) expo_m = RE_ITEM_EXPOSURE.search(cell_texts[4]) if expo_m: exposure_days = int(expo_m.group(1)) # [5] status or removed_date removed_date = None status: str | None = None if len(cell_texts) > 5: last_cell = cell_texts[5] status_m = RE_ITEM_STATUS.search(last_cell) if status_m: status = status_m.group(1) removed_date = parse_dmy(last_cell) # Sort dates chronologically (in case Yandex flips them — same fix as PR #541) if publish_date and removed_date and removed_date < publish_date: publish_date, removed_date = removed_date, publish_date # Row is valid only if we got area OR start_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, removed_date=removed_date, exposure_days=exposure_days, status=status, ) @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 | None = float(area_m.group(1).replace(",", ".")) if area_m else None # Sanity cap — Yandex page text sometimes concatenates digits across DOM # boundaries (e.g. "2025\xa0106,7 м²" → 2_025_106.7). Flats over 10_000 m² # are impossible; DB column is NUMERIC(8,2) which overflows at 10^6. # Drop the value rather than block the whole save batch. if area_m2 is not None and (area_m2 > 10_000 or area_m2 <= 0): logger.warning( "yandex_valuation: dropping nonsensical area_m2=%s from item chunk", area_m2, ) area_m2 = 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 # Extract ALL DD.MM.YYYY dates: first → publish_date, second → removed_date date_matches = list(re.finditer(r"\d{2}\.\d{2}\.\d{4}", text)) dates_parsed: list[date] = [] for m in date_matches: d = parse_dmy(m.group(0)) if d is not None: dates_parsed.append(d) # Sort dates chronologically — chunked-text scan returns them in page-text # order, but semantically publish_date is earliest and removed_date is # latest. Without this, ~1% of rows on prod show removed < publish. if dates_parsed: dates_sorted = sorted(d for d in dates_parsed if d is not None) publish_date = dates_sorted[0] if dates_sorted else None removed_date = dates_sorted[1] if len(dates_sorted) >= 2 else None else: publish_date = None removed_date = None 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, removed_date=removed_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]