lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(tradein): yandex_valuation.py — anonymous house-history scraper

Browse source 
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.
This commit is contained in:
lekss361 2026-05-23 16:40:23 +03:00
parent b5d206bfa8
commit fda9ae4dc1
2 changed files with 573 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

331
tradein-mvp/backend/app/services/scrapers/yandex_valuation.py Normal file
View file

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

242
tradein-mvp/backend/tests/test_yandex_valuation.py Normal file
View file

@ -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"<html><body>{body_content}</body></html>"
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