bot-backend
fa2bd20ac6
Some checks failed
Deploy Trade-In / build-backend (push) Blocked by required conditions
Deploy Trade-In / deploy (push) Blocked by required conditions
Deploy Trade-In / changes (push) Successful in 4s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / test (push) Has been cancelled
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
366 lines
14 KiB
Python
366 lines
14 KiB
Python
"""Yandex Realty detail page scraper.
|
||
|
||
Fetches /offer/<id>/ and extracts Product JSON-LD + DOM sections into
|
||
a DetailEnrichment Pydantic model. Used by enrichment pipeline
|
||
(Wave 5+ matching / Wave 6 estimator).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
from datetime import date
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel, Field
|
||
from selectolax.parser import HTMLParser, Node
|
||
|
||
from app.services.scraper_settings import get_scraper_delay
|
||
from app.services.scrapers.base import BaseScraper
|
||
from app.services.scrapers.repair_state_normalizer import infer_repair_state_from_text
|
||
from app.services.scrapers.yandex_helpers import (
|
||
RE_AGENCY_FOUNDED,
|
||
RE_AGENCY_OBJECTS,
|
||
RE_METRO_WALK,
|
||
RE_VIEWS,
|
||
RE_YEAR,
|
||
find_ld_by_type,
|
||
parse_house_type,
|
||
parse_ru_date,
|
||
parse_rub,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ── Pydantic models ───────────────────────────────────────────────────────────
|
||
|
||
|
||
class MetroStation(BaseModel):
|
||
name: str
|
||
walk_min: int | None = None
|
||
|
||
|
||
class DetailEnrichment(BaseModel):
|
||
"""Enrichment payload from a Yandex detail page."""
|
||
|
||
offer_id: str
|
||
source_url: str
|
||
|
||
# Pricing — Product JSON-LD `offers.price` is the exact int
|
||
price_rub: int | None = None
|
||
price_per_m2: int | None = None
|
||
|
||
# Title + basic params
|
||
title: str | None = None
|
||
rooms: int | None = None
|
||
area_m2: float | None = None
|
||
floor: int | None = None
|
||
total_floors: int | None = None
|
||
|
||
# Address (full)
|
||
address: str | None = None
|
||
|
||
# Description (full text)
|
||
description: str | None = None
|
||
|
||
# Repair state — enum, inferred from description text (#622).
|
||
# Yandex не отдаёт структурного поля ремонта, поэтому только инференс.
|
||
repair_state: str | None = None
|
||
|
||
# Stats
|
||
views_total: int | None = None
|
||
publish_date: date | None = None
|
||
publish_date_relative: str | None = None
|
||
|
||
# Agency block (OfferCardAuthorInfo)
|
||
agency_name: str | None = None
|
||
agency_founded_year: int | None = None
|
||
agency_objects_count: int | None = None
|
||
seller_name: str | None = None # last text line before "Агентство «...»"
|
||
|
||
# Metro stations from "Расположение" section
|
||
metro_stations: list[MetroStation] = Field(default_factory=list)
|
||
|
||
# Photos — 8 sizes from Product.image[]
|
||
photo_urls: list[str] = Field(default_factory=list)
|
||
|
||
# Newbuilding linkage
|
||
newbuilding_url: str | None = None
|
||
newbuilding_id: str | None = None
|
||
|
||
# NLP from description (best-effort)
|
||
house_type_nlp: str | None = None
|
||
year_built_hint: int | None = None
|
||
metro_walk_min: int | None = None
|
||
|
||
# Raw payload (trimmed)
|
||
raw_payload: dict[str, Any] | None = None
|
||
|
||
|
||
# ── Scraper ───────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class YandexDetailScraper(BaseScraper):
|
||
"""Detail page scraper for realty.yandex.ru."""
|
||
|
||
name = "yandex_detail"
|
||
base_url = "https://realty.yandex.ru"
|
||
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)
|
||
|
||
# BaseScraper requires fetch_around — detail isn't geo-based, raise NotImplementedError
|
||
async def fetch_around(self, lat: float, lon: float, radius_m: int = 1000) -> list: # type: ignore[override]
|
||
raise NotImplementedError(
|
||
"YandexDetailScraper is offer-id-based; use fetch_detail(offer_url) instead."
|
||
)
|
||
|
||
async def fetch_detail(self, offer_url: str) -> DetailEnrichment | None:
|
||
try:
|
||
response = await self._http_get(offer_url)
|
||
except Exception:
|
||
logger.exception("yandex detail fetch failed: %s", offer_url)
|
||
return None
|
||
if response.status_code != 200:
|
||
logger.warning("yandex detail returned %d for %s", response.status_code, offer_url)
|
||
return None
|
||
result = self.parse(response.text, offer_url=offer_url)
|
||
await self.sleep_between_requests()
|
||
return result
|
||
|
||
def parse(self, html: str, offer_url: str) -> DetailEnrichment | None:
|
||
offer_id_match = re.search(r"/offer/(\d+)/?", offer_url)
|
||
if not offer_id_match:
|
||
logger.warning("offer_url has no /offer/<id>/: %s", offer_url)
|
||
return None
|
||
offer_id = offer_id_match.group(1)
|
||
|
||
tree = HTMLParser(html)
|
||
|
||
# --- Product JSON-LD (authoritative price) ---
|
||
product = find_ld_by_type(html, "Product") or {}
|
||
offers_ld = product.get("offers") or {}
|
||
if isinstance(offers_ld, list) and offers_ld:
|
||
offers_ld = offers_ld[0]
|
||
price_ld = offers_ld.get("price") if isinstance(offers_ld, dict) else None
|
||
try:
|
||
price_rub = int(price_ld) if price_ld else None
|
||
except (TypeError, ValueError):
|
||
price_rub = None
|
||
|
||
# Photos from JSON-LD image[] (typically 8 size variants)
|
||
images = product.get("image") or []
|
||
if isinstance(images, str):
|
||
images = [images]
|
||
photo_urls = [u for u in images if isinstance(u, str)]
|
||
|
||
# --- Title + summary ---
|
||
title_node = tree.css_first("h1")
|
||
title = title_node.text(strip=True) if title_node else None
|
||
|
||
rooms, area_m2, floor, total_floors = _parse_title(title or "")
|
||
|
||
# --- OfferCardSummary text block ---
|
||
summary_node = tree.css_first('[data-test="OfferCardSummary"]')
|
||
summary_text = summary_node.text(strip=True) if summary_node else ""
|
||
|
||
# Views + relative publish date from summary text
|
||
views_match = RE_VIEWS.search(summary_text)
|
||
views_total = int(views_match.group(1)) if views_match else None
|
||
publish_date = parse_ru_date(summary_text)
|
||
publish_date_relative = _extract_relative_date(summary_text)
|
||
|
||
# price_per_m2 — from summary text if absent in LD
|
||
price_per_m2: int | None = None
|
||
ppm2_match = re.search(r"(\d[\d\s]+)\s*₽\s*за\s*м²", summary_text)
|
||
if ppm2_match:
|
||
price_per_m2 = parse_rub(ppm2_match.group(1))
|
||
|
||
# --- OfferCardAuthorInfo (agency block) ---
|
||
author_node = tree.css_first('[data-test="OfferCardAuthorInfo"]')
|
||
agency_name: str | None = None
|
||
agency_founded_year: int | None = None
|
||
agency_objects_count: int | None = None
|
||
seller_name: str | None = None
|
||
if author_node is not None:
|
||
author_text = author_node.text(strip=True)
|
||
agency_h2 = author_node.css_first("h2")
|
||
agency_name = agency_h2.text(strip=True) if agency_h2 else None
|
||
founded_m = RE_AGENCY_FOUNDED.search(author_text)
|
||
if founded_m:
|
||
agency_founded_year = int(founded_m.group(1))
|
||
objects_m = RE_AGENCY_OBJECTS.search(author_text)
|
||
if objects_m:
|
||
agency_objects_count = int(objects_m.group(1))
|
||
# seller_name — last text line before "Агентство"
|
||
seller_name = _extract_seller_name(summary_text, agency_name)
|
||
|
||
# --- Description section (after H2 "Описание") ---
|
||
description = _find_section_text(tree, "Описание")
|
||
|
||
# --- Repair state: инференс из описания (#622), Yandex без структурного поля ---
|
||
repair_state = infer_repair_state_from_text(description or summary_text)
|
||
|
||
# --- Address ---
|
||
address = _extract_address(summary_text)
|
||
|
||
# --- Metro stations from "Расположение" section ---
|
||
location_text = _find_section_text(tree, "Расположение") or ""
|
||
metro_stations = _parse_metro_stations(location_text)
|
||
|
||
# --- Newbuilding link ---
|
||
nb_url: str | None = None
|
||
nb_id: str | None = None
|
||
nb_link = tree.css_first('a[href*="/kupit/novostrojka/"]')
|
||
if nb_link is not None:
|
||
nb_href = nb_link.attributes.get("href", "")
|
||
nb_match = re.search(r"/novostrojka/[\w-]+?-(\d+)/?", nb_href)
|
||
if nb_match:
|
||
nb_id = nb_match.group(1)
|
||
nb_url = (
|
||
nb_href if nb_href.startswith("http") else f"https://realty.yandex.ru{nb_href}"
|
||
)
|
||
|
||
# --- NLP best-effort from description ---
|
||
nlp_text = description or summary_text
|
||
house_type_nlp = parse_house_type(nlp_text)
|
||
year_hint_m = RE_YEAR.search(nlp_text or "")
|
||
year_built_hint = int(year_hint_m.group(1)) if year_hint_m else None
|
||
walk_m = RE_METRO_WALK.search(nlp_text or "")
|
||
metro_walk_min = int(walk_m.group(1)) if walk_m else None
|
||
|
||
return DetailEnrichment(
|
||
offer_id=offer_id,
|
||
source_url=offer_url,
|
||
price_rub=price_rub,
|
||
price_per_m2=price_per_m2,
|
||
title=title,
|
||
rooms=rooms,
|
||
area_m2=area_m2,
|
||
floor=floor,
|
||
total_floors=total_floors,
|
||
address=address,
|
||
description=description,
|
||
repair_state=repair_state,
|
||
views_total=views_total,
|
||
publish_date=publish_date,
|
||
publish_date_relative=publish_date_relative,
|
||
agency_name=agency_name,
|
||
agency_founded_year=agency_founded_year,
|
||
agency_objects_count=agency_objects_count,
|
||
seller_name=seller_name,
|
||
metro_stations=metro_stations,
|
||
photo_urls=photo_urls,
|
||
newbuilding_url=nb_url,
|
||
newbuilding_id=nb_id,
|
||
house_type_nlp=house_type_nlp,
|
||
year_built_hint=year_built_hint,
|
||
metro_walk_min=metro_walk_min,
|
||
raw_payload={
|
||
"summary_text": summary_text[:1000],
|
||
"description_len": len(description) if description else 0,
|
||
"photo_count": len(photo_urls),
|
||
},
|
||
)
|
||
|
||
|
||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _parse_title(title: str) -> tuple[int | None, float | None, int | None, int | None]:
|
||
"""Extract (rooms, area_m2, floor, total_floors) from h1 text."""
|
||
rooms: int | None = None
|
||
area_m2: float | None = None
|
||
floor: int | None = None
|
||
total_floors: int | None = None
|
||
|
||
area_m = re.search(r"(\d+[.,]?\d*)\s*м²", title)
|
||
if area_m:
|
||
area_m2 = float(area_m.group(1).replace(",", "."))
|
||
|
||
if re.search(r"студи[яюй]", title, re.IGNORECASE):
|
||
rooms = 0
|
||
else:
|
||
rooms_m = re.search(r"(\d+)\s*-?\s*комнатн", title, re.IGNORECASE)
|
||
if rooms_m:
|
||
try:
|
||
rooms = int(rooms_m.group(1))
|
||
except ValueError:
|
||
pass
|
||
|
||
floor_m = re.search(r"(\d+)\s+этаж\s+из\s+(\d+)", title, re.IGNORECASE)
|
||
if floor_m:
|
||
floor = int(floor_m.group(1))
|
||
total_floors = int(floor_m.group(2))
|
||
|
||
return rooms, area_m2, floor, total_floors
|
||
|
||
|
||
def _find_section_text(tree: HTMLParser, heading: str) -> str | None:
|
||
"""Find the text content of a <section>/<div> whose preceding h2/h3 matches heading.
|
||
|
||
Yandex page structure varies; this scans h2/h3 nodes, then returns the
|
||
concatenated text of subsequent sibling blocks until the next heading.
|
||
"""
|
||
for h in tree.css("h2, h3"):
|
||
if heading.lower() in (h.text(strip=True) or "").lower():
|
||
# collect subsequent siblings until the next h2/h3
|
||
parts: list[str] = []
|
||
node: Node | None = h.next
|
||
while node is not None:
|
||
tag = (node.tag or "").lower()
|
||
if tag in {"h2", "h3"}:
|
||
break
|
||
txt = node.text(strip=True) if hasattr(node, "text") else ""
|
||
if txt:
|
||
parts.append(txt)
|
||
node = node.next
|
||
return " ".join(parts).strip() or None
|
||
return None
|
||
|
||
|
||
def _extract_address(summary_text: str) -> str | None:
|
||
"""Best-effort address extraction from summary block."""
|
||
# Pattern: "Россия, Свердловская область, Екатеринбург, улица Х, д. N"
|
||
m = re.search(r"(Россия[^•]+?)(?:•|\d+\s+просмотр|$)", summary_text)
|
||
if m:
|
||
addr = m.group(1).strip().rstrip(",").strip()
|
||
return addr if len(addr) > 10 else None
|
||
return None
|
||
|
||
|
||
def _parse_metro_stations(location_text: str) -> list[MetroStation]:
|
||
"""Parse 'Уральская 11 мин. Динамо 16 мин.' → list of MetroStation."""
|
||
stations: list[MetroStation] = []
|
||
# name (1+ Cyrillic words) + space + N + space + мин(.|у|ут)
|
||
for m in re.finditer(r"([А-ЯЁ][А-Яа-яё\s-]+?)\s+(\d+)\s*мин", location_text):
|
||
name = m.group(1).strip()
|
||
if 2 <= len(name) <= 40:
|
||
stations.append(MetroStation(name=name, walk_min=int(m.group(2))))
|
||
if len(stations) >= 5:
|
||
break
|
||
return stations
|
||
|
||
|
||
def _extract_relative_date(summary_text: str) -> str | None:
|
||
"""Capture phrases like '6 часов назад' / 'вчера' / '3 дня назад'."""
|
||
m = re.search(
|
||
r"(\d+\s+(?:минут|час|часов|часа|день|дня|дней|недел[ьюи])\s+назад"
|
||
r"|вчера|сегодня|позавчера)",
|
||
summary_text,
|
||
re.IGNORECASE,
|
||
)
|
||
return m.group(1).strip() if m else None
|
||
|
||
|
||
def _extract_seller_name(summary_text: str, agency_name: str | None) -> str | None:
|
||
"""Heuristic: line right before 'Агентство ...' in summary text."""
|
||
if not agency_name or agency_name not in summary_text:
|
||
return None
|
||
head = summary_text.split(agency_name, 1)[0]
|
||
# last short token sequence (likely "Имя Фамилия")
|
||
m = re.findall(r"([А-ЯЁ][а-яё]+(?:\s+[А-ЯЁ][а-яё]+){1,2})", head)
|
||
return m[-1] if m else None
|