02267d0306
- FastAPI backend: PostGIS estimator + 3 scrapers (Avito/Cian/Yandex)
- Next.js 15 frontend: tradein.html mockup design, basePath=/trade-in
- WeasyPrint PDF (Брусника-style 4-page report)
- Address autocomplete с typo-tolerance + 6 EKB presets
- Изолированный docker stack gendesign-tradein (отдельная postgres БД)
- Caddy inline routes: gendsgn.ru/trade-in/* и /trade-in/api/v1/*
- Forgejo Actions: .forgejo/workflows/deploy-tradein.yml (shell-based GHCR login)
- Триггер только по paths: tradein-mvp/** (не пересекается с deploy.yml)
- Образы: ghcr.io/lekss361/gendesign-tradein-{backend,frontend}:latest
Первый запуск на сервере (вручную, один раз):
- создать /opt/gendesign/tradein-mvp/.env.runtime (postgres pwd, contact email)
- docker network create gendesign_shared (если нет)
- docker compose -p gendesign-tradein up -d
- docker compose -p gendesign exec caddy caddy reload
290 lines
10 KiB
Python
290 lines
10 KiB
Python
"""Yandex.Недвижимость scraper (realty.yandex.ru).
|
||
|
||
Я.Недвижимость отдаёт SSR HTML с `__INITIAL_STATE__` JSON в `<script>` теге.
|
||
Структура offers: state.search.offers — list of offers.
|
||
|
||
URL pattern:
|
||
https://realty.yandex.ru/ekaterinburg/kupit/kvartira/?
|
||
rgid=240970 # ЕКБ region
|
||
&offerType=SELL
|
||
&category=APARTMENT
|
||
&geoLat=56.83
|
||
&geoLon=60.59
|
||
&geoRadius=1000 # метры
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import random
|
||
import re
|
||
from typing import Any
|
||
from urllib.parse import urlencode, urljoin
|
||
|
||
from selectolax.parser import HTMLParser
|
||
|
||
from app.services.scrapers.base import BaseScraper, ScrapedLot
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
YANDEX_EKB_RGID = 240970
|
||
|
||
|
||
class YandexRealtyScraper(BaseScraper):
|
||
"""Yandex.Недвижимость parser."""
|
||
|
||
name = "yandex"
|
||
base_url = "https://realty.yandex.ru"
|
||
request_delay_sec = 5.0
|
||
|
||
async def fetch_around(
|
||
self,
|
||
lat: float,
|
||
lon: float,
|
||
radius_m: int = 1000,
|
||
rooms: str | None = None,
|
||
sort: str = "DATE_DESC",
|
||
page: int = 0,
|
||
) -> list[ScrapedLot]:
|
||
self._anchor_lat = lat
|
||
self._anchor_lon = lon
|
||
self._anchor_radius_m = radius_m
|
||
|
||
url = self._build_url(lat, lon, radius_m, rooms, sort, page)
|
||
try:
|
||
response = await self._http_get(url)
|
||
except Exception:
|
||
logger.exception("yandex realty fetch failed for %s", url)
|
||
return []
|
||
|
||
if response.status_code != 200:
|
||
logger.warning("yandex realty returned %d", response.status_code)
|
||
return []
|
||
|
||
lots = self._parse_html(response.text)
|
||
logger.info(
|
||
"yandex realty: %d lots around (%.4f, %.4f) rooms=%s sort=%s page=%d",
|
||
len(lots), lat, lon, rooms or "ANY", sort, page,
|
||
)
|
||
await self.sleep_between_requests()
|
||
return lots
|
||
|
||
async def fetch_around_multi_room(
|
||
self, lat: float, lon: float, radius_m: int = 1000,
|
||
sorts: tuple[str, ...] = ("DATE_DESC",),
|
||
pages: tuple[int, ...] = (0,),
|
||
) -> list[ScrapedLot]:
|
||
"""Скрейп Yandex по сегментам STUDIO/1k/2k/3k/4k+ и опционально по
|
||
нескольким sort/page вариантам.
|
||
|
||
Каждая выдача = 20 лотов. Полный обход rooms×sorts×pages даст
|
||
5 × len(sorts) × len(pages) × 20 потенциальных лотов до дедупликации.
|
||
"""
|
||
seen: dict[str, ScrapedLot] = {}
|
||
for rooms in ("STUDIO", "1", "2", "3", "PLUS_4"):
|
||
for sort in sorts:
|
||
for page in pages:
|
||
try:
|
||
lots = await self.fetch_around(
|
||
lat, lon, radius_m, rooms=rooms, sort=sort, page=page,
|
||
)
|
||
except Exception:
|
||
logger.exception(
|
||
"yandex realty multi-room fetch failed: rooms=%s sort=%s page=%d",
|
||
rooms, sort, page,
|
||
)
|
||
continue
|
||
for lot in lots:
|
||
key = lot.source_id or lot.source_url
|
||
if key and key not in seen:
|
||
seen[key] = lot
|
||
logger.info(
|
||
"yandex realty multi-room: %d unique lots around (%.4f, %.4f)",
|
||
len(seen), lat, lon,
|
||
)
|
||
return list(seen.values())
|
||
|
||
def _build_url(
|
||
self,
|
||
lat: float,
|
||
lon: float,
|
||
radius_m: int,
|
||
rooms: str | None = None,
|
||
sort: str = "DATE_DESC",
|
||
page: int = 0,
|
||
) -> str:
|
||
params: dict[str, Any] = {
|
||
"rgid": YANDEX_EKB_RGID,
|
||
"offerType": "SELL",
|
||
"category": "APARTMENT",
|
||
"geoLat": f"{lat:.4f}",
|
||
"geoLon": f"{lon:.4f}",
|
||
"geoRadius": radius_m,
|
||
"sort": sort,
|
||
}
|
||
if rooms:
|
||
# Yandex Realty принимает roomsTotal=STUDIO|1|2|3|PLUS_4
|
||
params["roomsTotal"] = rooms
|
||
if page > 0:
|
||
params["page"] = page
|
||
return f"{self.base_url}/ekaterinburg/kupit/kvartira/?{urlencode(params)}"
|
||
|
||
def _parse_html(self, html: str) -> list[ScrapedLot]:
|
||
# Поиск __INITIAL_STATE__ или __PRELOADED_STATE__
|
||
state = _extract_yandex_state(html)
|
||
if not state:
|
||
return []
|
||
offers = _items_from_state(state)
|
||
return [
|
||
lot for lot in (self._item_to_lot(o) for o in offers if isinstance(o, dict))
|
||
if lot is not None
|
||
]
|
||
|
||
def _item_to_lot(self, offer: dict[str, Any]) -> ScrapedLot | None:
|
||
try:
|
||
# Цена: offer.price.value (там целое число рублей)
|
||
price_obj = offer.get("price") or {}
|
||
price = price_obj.get("value")
|
||
if not price:
|
||
pf = price_obj.get("priceForWhole") or {}
|
||
price = pf.get("value") if isinstance(pf, dict) else pf
|
||
if not price:
|
||
return None
|
||
|
||
# Площадь
|
||
area_obj = offer.get("area") or {}
|
||
area = area_obj.get("value")
|
||
if area:
|
||
area = float(area)
|
||
|
||
rooms = offer.get("roomsTotal") or offer.get("rooms")
|
||
|
||
# Этажи — offer.floorsOffered = [N], building.floorsTotal
|
||
floors_offered = offer.get("floorsOffered") or []
|
||
floor = floors_offered[0] if floors_offered else offer.get("floor")
|
||
building = offer.get("building") or {}
|
||
total_floors = building.get("floorsTotal") or offer.get("floorsTotal")
|
||
year_built = building.get("builtYear")
|
||
wall_type = building.get("buildingType")
|
||
|
||
# Локация
|
||
loc = offer.get("location") or {}
|
||
address = (
|
||
loc.get("address")
|
||
or loc.get("geocoderAddress")
|
||
or "Екатеринбург (Я.Недвижимость)"
|
||
)
|
||
point = loc.get("point") or {}
|
||
lat = point.get("latitude")
|
||
lon = point.get("longitude")
|
||
|
||
# Anchor fallback (если EXACT нет — но обычно есть)
|
||
if (lat is None or lon is None) and hasattr(self, "_anchor_lat"):
|
||
jitter_deg = (self._anchor_radius_m / 1000) / 111
|
||
lat = self._anchor_lat + (random.random() - 0.5) * jitter_deg
|
||
lon = self._anchor_lon + (random.random() - 0.5) * jitter_deg / 2
|
||
|
||
# URL
|
||
offer_id = offer.get("offerId") or offer.get("id")
|
||
internal_url = offer.get("unsignedInternalUrl") or ""
|
||
if internal_url.startswith("//"):
|
||
full_url = "https:" + internal_url
|
||
elif offer_id:
|
||
full_url = f"https://realty.yandex.ru/offer/{offer_id}/"
|
||
else:
|
||
return None
|
||
|
||
# Photos
|
||
photos = []
|
||
for img_field in ("appLargeImages", "extImages", "alikeImages"):
|
||
for img in (offer.get(img_field) or [])[:5]:
|
||
if isinstance(img, dict):
|
||
url = img.get("xl") or img.get("large") or img.get("main", "")
|
||
if url:
|
||
photos.append(url)
|
||
elif isinstance(img, str):
|
||
photos.append(img)
|
||
if photos:
|
||
break
|
||
|
||
return ScrapedLot(
|
||
source="yandex",
|
||
source_url=full_url,
|
||
source_id=str(offer_id) if offer_id else None,
|
||
address=address,
|
||
lat=lat,
|
||
lon=lon,
|
||
rooms=rooms,
|
||
area_m2=area,
|
||
floor=floor,
|
||
total_floors=total_floors,
|
||
year_built=year_built,
|
||
house_type=_map_wall_type(wall_type),
|
||
price_rub=int(price),
|
||
photo_urls=photos[:5],
|
||
raw_payload={
|
||
# Compact dump (full offer 50KB+, тяжеловато для БД)
|
||
"offerId": offer_id,
|
||
"creationDate": offer.get("creationDate"),
|
||
"dealStatus": offer.get("dealStatus"),
|
||
"description": (offer.get("description") or "")[:500],
|
||
},
|
||
)
|
||
except Exception:
|
||
logger.exception("yandex item parse failed: %s", offer.get("offerId"))
|
||
return None
|
||
|
||
|
||
# ── Helpers ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
_RE_STATE = re.compile(
|
||
r'<script id="initial_state_script"[^>]*>(.+?)</script>',
|
||
re.DOTALL,
|
||
)
|
||
|
||
|
||
def _extract_yandex_state(html: str) -> dict[str, Any] | None:
|
||
m = _RE_STATE.search(html)
|
||
if not m:
|
||
return None
|
||
raw = m.group(1)
|
||
# Может начинаться с `window.__INITIAL_STATE__=` префикса
|
||
raw = re.sub(r'^\s*window\.[\w_]+\s*=\s*', '', raw).rstrip(';').strip()
|
||
try:
|
||
return json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
return None
|
||
|
||
|
||
def _items_from_state(state: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Yandex Realty: state.map.offers.points содержит список объявлений (десктоп desktop SSR)."""
|
||
for path in [
|
||
("map", "offers", "points"), # верифицировано на проде
|
||
("search", "offers"),
|
||
("offers", "points"),
|
||
("offers", "items"),
|
||
]:
|
||
cursor: Any = state
|
||
for key in path:
|
||
if not isinstance(cursor, dict):
|
||
cursor = None
|
||
break
|
||
cursor = cursor.get(key)
|
||
if isinstance(cursor, list) and cursor:
|
||
return cursor
|
||
return []
|
||
|
||
|
||
def _map_wall_type(yandex_type: str | None) -> str | None:
|
||
if not yandex_type:
|
||
return None
|
||
mapping = {
|
||
"PANEL": "panel",
|
||
"BRICK": "brick",
|
||
"MONOLITH": "monolith",
|
||
"MONOLITH_BRICK": "monolith_brick",
|
||
"BLOCK": "panel",
|
||
}
|
||
return mapping.get(yandex_type.upper(), "other")
|