364 lines
14 KiB
Python
364 lines
14 KiB
Python
"""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
|
||
|
||
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)
|
||
|
||
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."""
|
||
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 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
|
||
|
||
# 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)
|
||
publish_date = dates_parsed[0] if dates_parsed else None
|
||
removed_date = dates_parsed[1] if len(dates_parsed) >= 2 else 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]
|