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
+
+
+# ── Save helper ───────────────────────────────────────────────────────────────
+
+
+def save_detail_enrichment(db: Session, listing_id: int, e: DetailEnrichment) -> bool:
+ """Persist Yandex DetailEnrichment to listings row.
+
+ UPDATE listings SET
= COALESCE(:val,
), ..., detail_enriched_at = NOW()
+ WHERE id = listing_id.
+
+ COALESCE semantics: keeps existing non-NULL value if new value is NULL (never
+ overwrites a populated column with NULL). area_m2 from detail is more accurate
+ than SERP, but COALESCE preserves SERP value if detail returns NULL — acceptable.
+
+ Returns True if the UPDATE touched at least one row (listing_id found in DB).
+ """
+ metro_json: str | None = None
+ if e.metro_stations:
+ metro_json = json.dumps(
+ [s.model_dump(exclude_none=True) for s in e.metro_stations],
+ ensure_ascii=False,
+ )
+
+ photo_json: str | None = None
+ if e.photo_urls:
+ photo_json = json.dumps(e.photo_urls, ensure_ascii=False)
+
+ result = db.execute(
+ text("""
+ UPDATE listings SET
+ rooms = COALESCE(CAST(:rooms AS int), rooms),
+ area_m2 = COALESCE(CAST(:area_m2 AS numeric), area_m2),
+ living_area_m2 = COALESCE(
+ CAST(:living_area_m2 AS numeric),
+ living_area_m2
+ ),
+ kitchen_area_m2 = COALESCE(
+ CAST(:kitchen_area_m2 AS numeric),
+ kitchen_area_m2
+ ),
+ ceiling_height = COALESCE(
+ CAST(:ceiling_height AS numeric),
+ ceiling_height
+ ),
+ floor = COALESCE(CAST(:floor AS int), floor),
+ total_floors = COALESCE(CAST(:total_floors AS int), total_floors),
+ address = COALESCE(CAST(:address AS text), address),
+ description = COALESCE(CAST(:description AS text), description),
+ repair_state = COALESCE(CAST(:repair_state AS text),repair_state),
+ publish_date = COALESCE(CAST(:publish_date AS date),publish_date),
+ views_total_yandex = COALESCE(CAST(:views_total AS int), views_total_yandex),
+ publish_date_relative= COALESCE(
+ CAST(:pub_date_rel AS text),
+ publish_date_relative
+ ),
+ agency_name = COALESCE(CAST(:agency_name AS text), agency_name),
+ agency_founded_year = COALESCE(
+ CAST(:agency_founded_year AS int),
+ agency_founded_year
+ ),
+ agency_objects_count = COALESCE(
+ CAST(:agency_objects_count AS int),
+ agency_objects_count
+ ),
+ metro_stations = COALESCE(
+ CAST(:metro_stations AS jsonb),
+ metro_stations
+ ),
+ photo_urls = COALESCE(
+ CAST(:photo_urls AS jsonb),
+ photo_urls
+ ),
+ newbuilding_url = COALESCE(
+ CAST(:newbuilding_url AS text),
+ newbuilding_url
+ ),
+ newbuilding_id = COALESCE(
+ CAST(:newbuilding_id AS text),
+ newbuilding_id
+ ),
+ detail_enriched_at = NOW()
+ WHERE id = CAST(:listing_id AS bigint)
+ """),
+ {
+ "listing_id": listing_id,
+ "rooms": e.rooms,
+ "area_m2": e.area_m2,
+ "living_area_m2": e.living_area_m2,
+ "kitchen_area_m2": e.kitchen_area_m2,
+ "ceiling_height": e.ceiling_height,
+ "floor": e.floor,
+ "total_floors": e.total_floors,
+ "address": e.address,
+ "description": e.description,
+ "repair_state": e.repair_state,
+ "publish_date": e.publish_date,
+ "views_total": e.views_total,
+ "pub_date_rel": e.publish_date_relative,
+ "agency_name": e.agency_name,
+ "agency_founded_year": e.agency_founded_year,
+ "agency_objects_count": e.agency_objects_count,
+ "metro_stations": metro_json,
+ "photo_urls": photo_json,
+ "newbuilding_url": e.newbuilding_url,
+ "newbuilding_id": e.newbuilding_id,
+ },
+ )
+ db.commit()
+ saved = (result.rowcount or 0) > 0
+ logger.info(
+ "yandex detail enrichment saved listing_id=%s (metro=%d photos=%d saved=%s)",
+ listing_id,
+ len(e.metro_stations),
+ len(e.photo_urls),
+ saved,
+ )
+ return saved
diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/newbuilding.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/newbuilding.py
new file mode 100644
index 00000000..b36b0c94
--- /dev/null
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/newbuilding.py
@@ -0,0 +1,438 @@
+"""Yandex Realty ЖК landing page parser.
+
+URL pattern: /{city}/kupit/novostrojka/
-/
+Reference target: ЖК Татлин (id=1592987, slug=tatlin) — comfort+, June 2023,
+PRINZIP, rating 4.3, 1505 ratings, 353 text reviews, coords (56.855312, 60.576668).
+
+Fetch strategy (#974):
+ - Все network-запросы (ЖК-лендинг + SERP slug-resolve) идут через BrowserFetcher
+ (tradein-browser camoufox), а НЕ через httpx/_http_get.
+ - Yandex Realty — JS-heavy / anti-bot; curl_cffi / httpx не получают данные.
+ - BrowserFetcher.fetch(url) → str (полный HTML); вызывающий код парсит через parse().
+
+Strangler-копия `app.services.scrapers.yandex_newbuilding` (#2133). Развязка от `app.*`:
+ - `app.services.scraper_settings.get_scraper_delay` → инжектируемый `delay_provider`
+ - `app.services.scrapers.base` → `scraper_kit.base`
+ - `app.services.scrapers.browser_fetcher` → `scraper_kit.browser_fetcher`
+ - `app.services.scrapers.yandex_helpers` → `scraper_kit.yandex_helpers`
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from collections.abc import Callable
+from typing import Any
+
+from pydantic import BaseModel, Field
+from selectolax.parser import HTMLParser, Node
+
+from scraper_kit.base import BaseScraper
+from scraper_kit.browser_fetcher import BrowserFetcher
+from scraper_kit.yandex_helpers import (
+ RE_JK_ID,
+ parse_house_class,
+ parse_house_type,
+)
+
+logger = logging.getLogger(__name__)
+
+
+# ── Models ───────────────────────────────────────────────────────────────────
+
+
+class JKMetroStation(BaseModel):
+ name: str
+ walk_min: int | None = None
+
+
+class YandexNewbuildingInfo(BaseModel):
+ """ЖК landing scrape payload."""
+
+ ext_id: str # '1592987'
+ ext_slug: str # 'tatlin'
+ source_url: str
+
+ # Identification
+ name: str | None = None # 'ЖК «Татлин»'
+ address: str | None = None # 'Екатеринбург, ул. Черепанова / ул. Готвальда'
+
+ # Coords (inline in HTML — Yandex unique vs Avito/Cian)
+ lat: float | None = None
+ lon: float | None = None
+
+ # Classification + commission
+ house_class: str | None = None # 'comfort_plus' etc.
+ commission_year: int | None = None
+ commission_month: str | None = None # 'июнь' (RU)
+
+ # Footprint
+ total_floors: int | None = None # 35
+ corpus_count: int | None = None # 3
+ total_area_ha: float | None = None # 1.5
+
+ # Developer
+ developer_name: str | None = None # 'PRINZIP'
+ developer_url: str | None = None
+ developer_other_jk: list[str] = Field(default_factory=list)
+
+ # Reviews
+ rating: float | None = None # 4.3
+ ratings_count: int | None = None # 1505 — оценок
+ text_reviews_count: int | None = None # 353 — текстовых отзывов
+
+ # NLP / description
+ description: str | None = None
+
+ # Metro
+ metro_stations: list[JKMetroStation] = Field(default_factory=list)
+
+ # House type (best-effort NLP from description)
+ house_type: str | None = None
+
+ raw_payload: dict[str, Any] | None = None
+
+
+# ── Regex constants ───────────────────────────────────────────────────────────
+
+RE_FLOORS_TOWERS = re.compile(r"(\d+)-этажн\w+\s+башн", re.IGNORECASE)
+RE_AREA_HA = re.compile(r"участке\s+([\d,]+)\s*га", re.IGNORECASE)
+RE_COMMISSION = re.compile(
+ r"(?:введ[её]н|сдан)\s+в\s+эксплуатацию\s+в\s+(\w+)\s+(\d{4})", re.IGNORECASE
+)
+# Yandex drift (#974 follow-up): рейтинг рендерится без пробела перед «из»
+# ("4.3из 5" / "4,5из 5", nbsp внутри) → \s* (а не \s+) перед «из», иначе rating=NULL
+# при заполненных ratings_count/text_reviews (verified live 2026-06-15, ЖК Татлин/Рио).
+RE_RATING = re.compile(r"(\d[.,]\d)\s*из\s+5")
+RE_RATINGS_COUNT = re.compile(r"(\d+)\s+оценок", re.IGNORECASE)
+RE_TEXT_REVIEWS = re.compile(r"Смотреть\s+все\s+(\d+)\s+отзыв", re.IGNORECASE)
+RE_CORPUS_COUNT_WORD = re.compile(
+ r"(одн[ау]|две|три|четыре|пять|шесть|семь|восемь|\d+)\s+(?:[\d\s-]*)\s*-?\s*этажн",
+ re.IGNORECASE,
+)
+RE_METRO_INLINE = re.compile(
+ r"([А-ЯЁ][А-Яа-яё\s-]{2,30}?)\s+(\d+)\s*мин",
+)
+
+# Slug из href: /{city}/kupit/novostrojka/-/
+# Используется в resolve_yandex_jk_slug для извлечения slug из SERP href.
+_JK_SLUG_RE = re.compile(
+ r"/(?:ekaterinburg|moskva|spb|[a-z-]+)/kupit/novostrojka/([a-z0-9-]+)-(\d+)/"
+)
+
+# Ekb-specific coord ranges (extend per-city later)
+LAT_RANGE = (55.5, 57.5)
+LON_RANGE = (59.5, 61.5)
+
+# Word → number map
+_WORD_NUM: dict[str, int] = {
+ "одна": 1,
+ "одну": 1,
+ "одной": 1,
+ "две": 2,
+ "три": 3,
+ "четыре": 4,
+ "пять": 5,
+ "шесть": 6,
+ "семь": 7,
+ "восемь": 8,
+}
+
+
+# ── Scraper class ─────────────────────────────────────────────────────────────
+
+
+class YandexNewbuildingScraper(BaseScraper):
+ name = "yandex_realty_nb"
+ base_url = "https://realty.yandex.ru"
+ request_delay_sec = 5.0 # class default; instance value injected via delay_provider
+
+ def __init__(self, *, delay_provider: Callable[[str], float] | None = None) -> None:
+ super().__init__()
+ # Strangler-инжекция (#2133): провайдер задержки приходит снаружи вместо
+ # прямого импорта app.services.scraper_settings.get_scraper_delay.
+ if delay_provider is not None:
+ self.request_delay_sec = delay_provider(self.name)
+
+ async def fetch_around(self, lat: float, lon: float, radius_m: int = 1000) -> list: # type: ignore[override]
+ raise NotImplementedError(
+ "YandexNewbuildingScraper is JK-slug-based; use fetch_jk(slug, id) instead."
+ )
+
+ async def fetch_jk(
+ self, jk_slug: str, jk_id: str, city: str = "ekaterinburg"
+ ) -> YandexNewbuildingInfo | None:
+ """Загрузить ЖК-лендинг через BrowserFetcher и распарсить.
+
+ Yandex Realty — JS/anti-bot: curl/httpx не получают данные. Запрос идёт
+ через tradein-browser (camoufox) контейнер — единственный рабочий путь (#974).
+ """
+ url = f"{self.base_url}/{city}/kupit/novostrojka/{jk_slug}-{jk_id}/"
+ try:
+ async with BrowserFetcher(source="yandex") as fetcher:
+ html = await fetcher.fetch(url)
+ except Exception:
+ logger.exception("yandex nb browser fetch failed: %s", url)
+ return None
+ if not html or len(html) < 500:
+ logger.warning(
+ "yandex nb browser returned empty/tiny HTML (%d bytes): %s",
+ len(html) if html else 0,
+ url,
+ )
+ return None
+ result = self.parse(html, jk_slug=jk_slug, jk_id=jk_id, source_url=url)
+ return result
+
+ def parse(self, html: str, jk_slug: str, jk_id: str, source_url: str) -> YandexNewbuildingInfo:
+ tree = HTMLParser(html)
+ body = tree.body
+ body_text = body.text(strip=True) if body else ""
+
+ # Header
+ h1 = tree.css_first("h1")
+ name = h1.text(strip=True) if h1 else None
+
+ # Sections (heading-based extraction)
+ location_text = _find_section_text(tree, "Расположение") or ""
+ description_section = (
+ _find_section_text(tree, "О комплексе")
+ or _find_section_text(tree, "Расположение, транспортная доступность")
+ or location_text
+ )
+
+ # Coords — inline scan within plausible range
+ lat, lon = _extract_coords(html)
+
+ # Class
+ house_class = parse_house_class(description_section) or parse_house_class(body_text)
+
+ # Commission
+ commission_year: int | None = None
+ commission_month: str | None = None
+ c_m = RE_COMMISSION.search(description_section) or RE_COMMISSION.search(body_text)
+ if c_m:
+ commission_month = c_m.group(1).lower()
+ try:
+ commission_year = int(c_m.group(2))
+ except ValueError:
+ pass
+
+ # Floors + corpus + area
+ floors_m = RE_FLOORS_TOWERS.search(description_section) or RE_FLOORS_TOWERS.search(
+ body_text
+ )
+ total_floors = int(floors_m.group(1)) if floors_m else None
+
+ corpus_count = _parse_corpus_count(description_section or body_text)
+
+ area_m = RE_AREA_HA.search(description_section) or RE_AREA_HA.search(body_text)
+ total_area_ha = float(area_m.group(1).replace(",", ".")) if area_m else None
+
+ # Developer
+ dev_link = tree.css_first('[data-test="CARD_DEV_BADGE_DEVELOPER_LINK"]')
+ developer_name: str | None = None
+ developer_url: str | None = None
+ if dev_link is not None:
+ developer_name = dev_link.text(strip=True) or None
+ href = dev_link.attributes.get("href", "")
+ if href:
+ developer_url = (
+ href if href.startswith("http") else f"https://realty.yandex.ru{href}"
+ )
+
+ # Developer's other JKs
+ other_jk_block = tree.css_first('[data-test="CardDevSites"]')
+ developer_other_jk: list[str] = []
+ if other_jk_block is not None:
+ for jk_a in other_jk_block.css("a"):
+ t = jk_a.text(strip=True)
+ if t and t not in developer_other_jk:
+ developer_other_jk.append(t)
+
+ # Reviews
+ rating_m = RE_RATING.search(body_text)
+ rating = float(rating_m.group(1).replace(",", ".")) if rating_m else None
+ rcount_m = RE_RATINGS_COUNT.search(body_text)
+ ratings_count = int(rcount_m.group(1)) if rcount_m else None
+ treviews_m = RE_TEXT_REVIEWS.search(body_text)
+ text_reviews_count = int(treviews_m.group(1)) if treviews_m else None
+
+ # Address (header area before lat/lon block)
+ address = _extract_jk_address(body_text, name)
+
+ # Metro stations
+ metro_stations = _parse_metro(body_text)
+
+ # NLP house_type fallback
+ house_type = parse_house_type(description_section) or parse_house_type(body_text)
+
+ return YandexNewbuildingInfo(
+ ext_id=jk_id,
+ ext_slug=jk_slug,
+ source_url=source_url,
+ name=name,
+ address=address,
+ lat=lat,
+ lon=lon,
+ house_class=house_class,
+ commission_year=commission_year,
+ commission_month=commission_month,
+ total_floors=total_floors,
+ corpus_count=corpus_count,
+ total_area_ha=total_area_ha,
+ developer_name=developer_name,
+ developer_url=developer_url,
+ developer_other_jk=developer_other_jk[:10],
+ rating=rating,
+ ratings_count=ratings_count,
+ text_reviews_count=text_reviews_count,
+ description=description_section or None,
+ metro_stations=metro_stations,
+ house_type=house_type,
+ raw_payload={
+ "description_len": len(description_section or ""),
+ "body_len": len(body_text),
+ },
+ )
+
+
+# ── Slug resolution via SERP ──────────────────────────────────────────────────
+
+
+async def resolve_yandex_jk_slug(
+ jk_id: str,
+ city: str = "ekaterinburg",
+) -> str | None:
+ """Найти Yandex Realty slug для ЖК по его ext_id (jk_id) через SERP.
+
+ Стратегия (#974 — зеркало resolve_cian_zhk_url_via_search):
+ 1. Запросить поисковую страницу Yandex Realty через BrowserFetcher.
+ URL: /ekaterinburg/kupit/novostrojka/?siteId=
+ 2. В HTML найти первую ссылку вида //kupit/novostrojka/-/
+ через regex _JK_SLUG_RE.
+ 3. Вернуть slug или None при любой ошибке.
+
+ Caller несёт ответственность за anti-bot sleep (зеркало cian_newbuilding.py).
+ BrowserFetcher обязателен — Yandex Realty JS/anti-bot, httpx/curl не работают.
+
+ Returns:
+ slug (str без id-суффикса), или None при ошибке / не найден.
+ """
+ # Yandex Realty SERP: фильтр по siteId → первый результат = нужный ЖК.
+ # Альтернативный путь через Яндекс Поиск (web SERP) менее надёжен из-за
+ # вариативности разметки. Прямой realty.yandex.ru SERP — стабильнее.
+ serp_url = f"https://realty.yandex.ru/{city}/kupit/novostrojka/?siteId={jk_id}"
+ try:
+ async with BrowserFetcher(source="yandex") as fetcher:
+ html = await fetcher.fetch(serp_url)
+ except Exception as exc:
+ logger.warning("resolve_yandex_jk_slug jk_id=%s browser fetch failed: %s", jk_id, exc)
+ return None
+
+ if not html:
+ logger.warning("resolve_yandex_jk_slug jk_id=%s: empty HTML from browser", jk_id)
+ return None
+
+ # Ищем ссылку вида /{city}/kupit/novostrojka/-/
+ # Ограничиваем: id в ссылке должен совпадать с искомым jk_id.
+ for m in _JK_SLUG_RE.finditer(html):
+ if m.group(2) == str(jk_id):
+ slug = m.group(1)
+ logger.info("resolve_yandex_jk_slug jk_id=%s → slug=%s", jk_id, slug)
+ return slug
+
+ logger.warning(
+ "resolve_yandex_jk_slug jk_id=%s: no matching slug in SERP HTML (markup drift?)",
+ jk_id,
+ )
+ return None
+
+
+# ── helpers ───────────────────────────────────────────────────────────────────
+
+
+def _extract_coords(html: str) -> tuple[float | None, float | None]:
+ """Scan HTML for any `.\\d+` and `.\\d+` matching plausible city range."""
+ lats = [
+ float(m)
+ for m in re.findall(r"\b(\d{2}\.\d{4,8})\b", html)
+ if LAT_RANGE[0] <= float(m) <= LAT_RANGE[1]
+ ]
+ lons = [
+ float(m)
+ for m in re.findall(r"\b(\d{2}\.\d{4,8})\b", html)
+ if LON_RANGE[0] <= float(m) <= LON_RANGE[1]
+ ]
+ lat = lats[0] if lats else None
+ lon = lons[0] if lons else None
+ return lat, lon
+
+
+def _find_section_text(tree: HTMLParser, heading: str) -> str | None:
+ for h in tree.css("h2, h3"):
+ if heading.lower() in (h.text(strip=True) or "").lower():
+ 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 _parse_corpus_count(text: str) -> int | None:
+ m = RE_CORPUS_COUNT_WORD.search(text)
+ if not m:
+ return None
+ token = m.group(1).lower()
+ if token.isdigit():
+ return int(token)
+ return _WORD_NUM.get(token)
+
+
+def _parse_metro(text: str) -> list[JKMetroStation]:
+ stations: list[JKMetroStation] = []
+ for m in RE_METRO_INLINE.finditer(text):
+ name = m.group(1).strip()
+ # filter common noise
+ if not 2 <= len(name) <= 40 or "ходьб" in name.lower() or "минут" in name.lower():
+ continue
+ try:
+ stations.append(JKMetroStation(name=name, walk_min=int(m.group(2))))
+ except ValueError:
+ continue
+ if len(stations) >= 5:
+ break
+ return stations
+
+
+def _extract_jk_address(body_text: str, name: str | None) -> str | None:
+ """Find address line — typically right after JK name, before metro entries."""
+ if name and name in body_text:
+ after_name = body_text.split(name, 1)[1]
+ # Match 'Город, ул. X / ул. Y' until first metro mention
+ m = re.match(
+ r"\s*([А-ЯЁ][^•]{5,150}?)(?:\s+[А-ЯЁ][а-яё]+\s+\d+\s*мин|\d+\s+оценок|$)",
+ after_name,
+ )
+ if m:
+ addr = m.group(1).strip().rstrip(",").strip()
+ if 10 < len(addr) < 200:
+ return addr
+ return None
+
+
+# Expose RE_JK_ID at module level for external use (e.g. SERP link → newbuilding detail)
+__all__ = [
+ "RE_JK_ID",
+ "JKMetroStation",
+ "YandexNewbuildingInfo",
+ "YandexNewbuildingScraper",
+ "resolve_yandex_jk_slug",
+]
diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py
new file mode 100644
index 00000000..c8621c08
--- /dev/null
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py
@@ -0,0 +1,1105 @@
+"""Yandex.Nedvizhimost scraper (realty.yandex.ru) -- gate-API JSON parser.
+
+Transport: BrowserFetcher (camoufox real-browser fingerprint via tradein-browser service).
+camoufox's fingerprint is not tarpitted by yandex and does not burn the mobile-proxy IP;
+raw system-curl over SOCKS5 gets the proxy IP flagged. One BrowserFetcher context is opened
+per scraper session (__aenter__) and reused across all page fetches.
+
+camoufox returns the gate-API JSON wrapped in HTML (verified prod 2026-06-17):
+ ...{"response":...}
+JSON is extracted via _extract_json_from_content() (handles the wrapper).
+
+API path: GET https://realty.yandex.ru/gate/react-page/get/
+ _pageType=search&_providers=react-search-results-data
+ type=SELL&category=APARTMENT&newFlat=NO&rgid=559132 (EKB)
+ roomsTotal=&priceMin=...&priceMax=...&page=<1-based>
+
+Response shape: response.search.offers.{entities[], pager{page,pageSize,totalItems,totalPages}}
+
+Page param is 1-based: page=1 -> first page (pager.page=0 in response).
+page=0 -> API returns error. Paginate from page=1 to totalPages (inclusive).
+
+Studio roomsTotal:
+ Request: roomsTotal=STUDIO
+ Response entities: roomsTotal=None -> mapped to rooms=0 (existing convention).
+
+Strangler-копия `app.services.scrapers.yandex_realty` (#2133). Развязка от `app.*`:
+ - `app.core.config.settings` → инжектируемый `ScraperConfig` (yandex_proxy_rotate_url,
+ avito_proxy_rotate_url)
+ - `app.services.scraper_settings.get_scraper_delay` → инжектируемый `delay_provider`
+ - `app.services.scrapers.{base,browser_fetcher,price_brackets}` → `scraper_kit.*`
+ - `app.services.scrapers.house_type_normalizer` → `scraper_kit.providers.yandex.shared`
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import math
+import re
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, date, datetime
+from typing import TYPE_CHECKING, Any
+from urllib.parse import urlencode
+
+from curl_cffi.requests import AsyncSession as _CurlCffiSession
+
+from scraper_kit.base import BaseScraper, ScrapedLot
+from scraper_kit.browser_fetcher import BrowserFetcher
+from scraper_kit.price_brackets import get_price_seed_brackets
+from scraper_kit.providers.yandex.shared import normalize_house_type
+
+if TYPE_CHECKING:
+ from scraper_kit.contracts import ScraperConfig
+
+logger = logging.getLogger(__name__)
+
+# camoufox renders JSON endpoints as {"response":...} inside an HTML wrapper.
+_RE_PRE_JSON = re.compile("]*>(\\{.*?)
", re.DOTALL | re.IGNORECASE)
+
+DEFAULT_CITY = "ekaterinburg"
+_GATE_MAX_PAGES_CAP = 50
+_EKB_RGID = 559132
+_GATE_URL = "https://realty.yandex.ru/gate/react-page/get/"
+
+ROOM_GATE_PARAM: dict[str, str | int] = {
+ "studio": "STUDIO",
+ "1": 1,
+ "2": 2,
+ "3": 3,
+ "4+": 4,
+}
+
+ROOM_PATH: dict[str, str] = {
+ "studio": "studiya",
+ "1": "odnokomnatnaya",
+ "2": "dvuhkomnatnaya",
+ "3": "trehkomnatnaya",
+ "4+": "4-i-bolee",
+}
+
+DEFAULT_PRICE_RANGES: list[tuple[int | None, int | None]] = [
+ (None, 5_000_000),
+ (5_000_000, 7_000_000),
+ (7_000_000, 10_000_000),
+ (10_000_000, 15_000_000),
+ (15_000_000, 25_000_000),
+ (25_000_000, None),
+]
+
+_YANDEX_MAX_PRICE = 200_000_000
+_YANDEX_MIN_BRACKET = 50_000
+_GATE_PAGE_SIZE = 20
+_YANDEX_TARPIT_MAX_RETRIES: int = 2
+
+# Таймаут httpx-клиента BrowserFetcher для yandex-пути: 30s вместо глобального 120s.
+# Один проблемный combo занимает ≤30s×(1+_YANDEX_TARPIT_MAX_RETRIES)=90s вместо 360s.
+_YANDEX_BROWSER_FETCH_TIMEOUT_S: float = 30.0
+
+
+@dataclass
+class _CurlResponse:
+ status_code: int
+ text: str
+
+
+def _is_gate_error(payload: dict[str, Any]) -> bool:
+ """Return True when gate-API returned an error object instead of search results."""
+ return "error" in payload or "response" not in payload
+
+
+def _extract_gate_data(
+ payload: dict[str, Any],
+) -> tuple[list[dict[str, Any]], dict[str, Any]] | None:
+ """Extract (entities, pager) from valid gate-API response. None on missing path."""
+ try:
+ offers = payload["response"]["search"]["offers"]
+ entities: list[dict[str, Any]] = offers["entities"]
+ pager: dict[str, Any] = offers["pager"]
+ return entities, pager
+ except (KeyError, TypeError):
+ return None
+
+
+def _extract_json_from_content(content: str) -> str | None:
+ """Extract raw JSON text from camoufox HTML-wrapped content.
+
+ camoufox renders a JSON endpoint as:
+ ...{"response":...}
+
+ Strategy:
+ 1. Try ...
regex (primary -- matches prod output).
+ 2. Fallback: slice from the first '{' (handles raw-JSON response, no wrapper).
+
+ Returns the JSON string, or None if no '{' is found.
+ """
+ m = _RE_PRE_JSON.search(content)
+ if m:
+ return m.group(1)
+
+ idx = content.find("{")
+ if idx != -1:
+ return content[idx:]
+
+ return None
+
+
+def _segment_for(new_flat: str) -> str:
+ """Map a newFlat gate-API value to the listings.listing_segment value."""
+ return "novostroyki" if new_flat == "YES" else "vtorichka"
+
+
+def _parse_creation_date(raw: Any) -> tuple[date | None, int | None]:
+ """Parse gate-API creationDate ('2026-05-24T15:10:27Z') → (date, days_on_market).
+
+ days_on_market = floor((now - creationDate).days), clamped to >= 0.
+ Returns (None, None) when raw is absent or unparseable (never raises).
+ """
+ if not raw or not isinstance(raw, str):
+ return None, None
+ txt = raw.strip()
+ # datetime.fromisoformat handles 'Z' only from py3.11+, but normalise defensively.
+ if txt.endswith("Z"):
+ txt = txt[:-1] + "+00:00"
+ try:
+ dt = datetime.fromisoformat(txt)
+ except ValueError:
+ return None, None
+ if dt.tzinfo is None:
+ dt = dt.replace(tzinfo=UTC)
+ listing_date = dt.date()
+ delta_days = (datetime.now(UTC) - dt).days
+ days_on_market = max(delta_days, 0)
+ return listing_date, days_on_market
+
+
+def _author_fields(author: dict[str, Any]) -> tuple[bool | None, bool | None, str | None]:
+ """Map entity.author → (is_homeowner, is_pro_seller, agency_name).
+
+ author.category ∈ {OWNER, AGENCY, AGENT, ...}.
+ OWNER → is_homeowner=True
+ AGENCY / AGENT → is_pro_seller=True, agency_name = organization or agentName
+ Returns (None, None, None) when author/category absent.
+ """
+ category = (author.get("category") or "").upper()
+ if not category:
+ return None, None, None
+ if category == "OWNER":
+ return True, None, None
+ if category in {"AGENCY", "AGENT"}:
+ agency_name = author.get("organization") or author.get("agentName") or None
+ return None, True, agency_name
+ return None, None, None
+
+
+def _to_bigint(raw: Any) -> int | None:
+ """Coerce a gate-API numeric value (str or number) to int. None when unparseable."""
+ if raw is None:
+ return None
+ try:
+ return int(float(raw))
+ except (TypeError, ValueError):
+ return None
+
+
+def _predicted_prices(entity: dict[str, Any]) -> tuple[int | None, int | None, int | None]:
+ """Extract (value, min, max) from entity.predictions.predictedPrice.
+
+ Yandex's own per-offer valuation. The gate-API nests it under
+ ``predictions.predictedPrice`` (NOT ``predictedPrice.predictedPrice`` —
+ that key does not exist; verified live 2026-06-18 against the gate-API).
+ Fields are strings (e.g. "7554000"). Returns (None, None, None) when the
+ nested path is absent (e.g. novostroyki offers carry no valuation).
+ """
+ outer = entity.get("predictions") or {}
+ inner = outer.get("predictedPrice") or {}
+ if not isinstance(inner, dict):
+ return None, None, None
+ return (
+ _to_bigint(inner.get("value")),
+ _to_bigint(inner.get("min")),
+ _to_bigint(inner.get("max")),
+ )
+
+
+def _metro_stations(location: dict[str, Any]) -> list[dict[str, Any]]:
+ """Build metro_stations [{name, min}] from location.metroList[] (+ location.metro)."""
+ stations: list[dict[str, Any]] = []
+ raw_list = location.get("metroList")
+ if isinstance(raw_list, list):
+ for m in raw_list:
+ if isinstance(m, dict) and m.get("name"):
+ stations.append({"name": m.get("name"), "min": m.get("timeToMetro")})
+ if not stations:
+ single = location.get("metro")
+ if isinstance(single, dict) and single.get("name"):
+ stations.append({"name": single.get("name"), "min": single.get("timeToMetro")})
+ return stations
+
+
+def _parse_gate_json(
+ payload: dict[str, Any], page_param: int = 1, new_flat: str = "NO"
+) -> list[ScrapedLot]:
+ """Parse gate-API JSON payload into a list of ScrapedLot.
+
+ Verified field mapping (from .issdump/yandex_gate_api_sample.json + live prod 2026-06-17):
+ offerId -> source_id
+ url (//realty.yandex.ru/offer/) -> source_url (prepend "https:")
+ price.value -> price_rub
+ price.valuePerPart -> price_per_m2
+ area.value -> area_m2
+ livingSpace.value -> living_area_m2
+ kitchenSpace.value -> kitchen_area_m2 (+ raw_payload["kitchen_area_m2"])
+ roomsTotal (int | None) -> rooms (None -> 0 for studio)
+ floorsOffered[0] -> floor
+ floorsTotal -> total_floors
+ building.builtYear -> year_built
+ building.buildingType -> house_type (normalize_house_type: MONOLIT->monolith)
+ ceilingHeight -> ceiling_height_m (+ raw_payload["ceiling_height"])
+ location.point.latitude/longitude -> lat / lon
+ location.geocoderAddress -> address (full, with house number)
+ mainImages[] -> photo_urls (prepend "https:", up to 5)
+ building.siteId -> house_ext_id (JK linkage)
+ listing_segment -> "novostroyki" (newFlat=YES) | "vtorichka" (newFlat=NO)
+ creationDate (ISO Z) -> listing_date / publish_date + days_on_market
+ description -> description
+ author.category/organization -> is_homeowner / is_pro_seller / agency_name
+ predictedPrice.predictedPrice.* -> predicted_price_rub / _min / _max
+ price.trend / price.previous -> price_trend / price_previous_rub
+ location.metroList[] -> metro_stations [{name, min}]
+ offerId -> yandex_offer_id
+ """
+ result = _extract_gate_data(payload)
+ if result is None:
+ logger.warning("yandex gate: response missing response.search.offers path")
+ return []
+
+ entities, _pager = result
+ lots: list[ScrapedLot] = []
+ for entity in entities:
+ lot = _entity_to_lot(entity, page_param=page_param, new_flat=new_flat)
+ if lot is not None:
+ lots.append(lot)
+ return lots
+
+
+def _entity_to_lot(
+ entity: dict[str, Any], page_param: int = 1, new_flat: str = "NO"
+) -> ScrapedLot | None:
+ """Convert one gate-API entity dict to ScrapedLot. None on missing required fields."""
+ try:
+ offer_id = str(entity.get("offerId") or "")
+ if not offer_id:
+ return None
+
+ url_raw = entity.get("url") or ""
+ source_url = f"https:{url_raw}" if url_raw.startswith("//") else url_raw
+ if not source_url:
+ source_url = f"https://realty.yandex.ru/offer/{offer_id}/"
+
+ price_dict = entity.get("price") or {}
+ price_val = price_dict.get("value")
+ if not price_val or price_val <= 0:
+ return None
+ price_rub = int(price_val)
+
+ price_per_m2_raw = price_dict.get("valuePerPart")
+ price_per_m2 = int(price_per_m2_raw) if price_per_m2_raw else None
+
+ area_dict = entity.get("area") or {}
+ area_m2_raw = area_dict.get("value")
+ area_m2 = float(area_m2_raw) if area_m2_raw is not None else None
+
+ living_dict = entity.get("livingSpace") or {}
+ living_area_raw = living_dict.get("value")
+ living_area_m2 = float(living_area_raw) if living_area_raw is not None else None
+
+ kitchen_dict = entity.get("kitchenSpace") or {}
+ kitchen_area_raw = kitchen_dict.get("value")
+ kitchen_area_m2 = float(kitchen_area_raw) if kitchen_area_raw is not None else None
+
+ rooms_raw = entity.get("roomsTotal")
+ if rooms_raw is None:
+ rooms = 0 # studio
+ else:
+ try:
+ rooms = int(rooms_raw)
+ except (TypeError, ValueError):
+ rooms = None
+
+ floors_offered = entity.get("floorsOffered") or []
+ floor = int(floors_offered[0]) if floors_offered else None
+ floors_total_raw = entity.get("floorsTotal")
+ total_floors = int(floors_total_raw) if floors_total_raw is not None else None
+
+ ceiling_height = entity.get("ceilingHeight")
+ # Колонка ceiling_height — numeric(3,2), max 9.99. Yandex SERP отдаёт мусор
+ # (видели 18 м) → запись out-of-range уронила бы весь батч DataError'ом
+ # (per-lot SAVEPOINT ловит только IntegrityError). Берём только правдоподобный
+ # диапазон 2.0–6.0 м, иначе None; сырое значение остаётся в raw_payload. (#2007)
+ _ceiling_raw = float(ceiling_height) if ceiling_height is not None else None
+ ceiling_height_m = (
+ _ceiling_raw if _ceiling_raw is not None and 2.0 <= _ceiling_raw <= 6.0 else None
+ )
+
+ building = entity.get("building") or {}
+ year_built_raw = building.get("builtYear")
+ year_built = int(year_built_raw) if year_built_raw is not None else None
+ # buildingType — SCREAMING (MONOLIT/BRICK/...). Нормализуем в канон, иначе
+ # estimator soft-penalty штрафует yandex-аналоги (#2007).
+ raw_building_type = building.get("buildingType")
+ house_type = normalize_house_type(raw_building_type)
+
+ site_id = building.get("siteId")
+ site_name = building.get("siteName")
+ house_source: str | None = None
+ house_ext_id: str | None = None
+ if site_id:
+ house_source = "yandex_realty_nb"
+ house_ext_id = str(site_id)
+
+ location = entity.get("location") or {}
+ point = location.get("point") or {}
+ lat_raw = point.get("latitude")
+ lon_raw = point.get("longitude")
+ lat = float(lat_raw) if lat_raw is not None else None
+ lon = float(lon_raw) if lon_raw is not None else None
+
+ address = location.get("geocoderAddress") or location.get("streetAddress") or None
+
+ # ── Rich gate-API fields ───────────────────────────────────────────
+ listing_date, days_on_market = _parse_creation_date(entity.get("creationDate"))
+ description = entity.get("description") or None
+
+ author = entity.get("author") or {}
+ is_homeowner, is_pro_seller, agency_name = _author_fields(author)
+
+ predicted_value, predicted_min, predicted_max = _predicted_prices(entity)
+
+ price_trend = price_dict.get("trend") or None
+ price_previous_rub = _to_bigint(price_dict.get("previous"))
+
+ metro_stations = _metro_stations(location)
+
+ photo_urls: list[str] = []
+ for img in entity.get("mainImages") or []:
+ if img:
+ photo_urls.append(f"https:{img}" if img.startswith("//") else img)
+ if len(photo_urls) >= 5:
+ break
+
+ return ScrapedLot(
+ source="yandex",
+ source_url=source_url,
+ source_id=offer_id,
+ address=address,
+ lat=lat,
+ lon=lon,
+ rooms=rooms,
+ area_m2=area_m2,
+ living_area_m2=living_area_m2,
+ floor=floor,
+ total_floors=total_floors,
+ year_built=year_built,
+ house_type=house_type,
+ kitchen_area_m2=kitchen_area_m2,
+ ceiling_height_m=ceiling_height_m,
+ price_rub=price_rub,
+ price_per_m2=price_per_m2,
+ photo_urls=photo_urls,
+ house_source=house_source,
+ house_ext_id=house_ext_id,
+ house_url=None,
+ listing_segment=_segment_for(new_flat),
+ listing_date=listing_date,
+ publish_date=listing_date,
+ days_on_market=days_on_market,
+ description=description,
+ is_homeowner=is_homeowner,
+ is_pro_seller=is_pro_seller,
+ agency_name=agency_name,
+ metro_stations=metro_stations,
+ yandex_offer_id=offer_id,
+ predicted_price_rub=predicted_value,
+ predicted_price_min=predicted_min,
+ predicted_price_max=predicted_max,
+ price_trend=price_trend,
+ price_previous_rub=price_previous_rub,
+ raw_payload={
+ "page_param": page_param,
+ "ceiling_height": ceiling_height,
+ "kitchen_area_m2": kitchen_area_raw,
+ "raw_building_type": raw_building_type, # сырой SCREAMING — для отладки
+ "site_name": site_name,
+ "offer_id": offer_id,
+ },
+ )
+ except Exception:
+ logger.exception("yandex gate: entity parse failed offer_id=%s", entity.get("offerId"))
+ return None
+
+
+class YandexRealtyScraper(BaseScraper):
+ name = "yandex"
+ base_url = "https://realty.yandex.ru"
+ request_delay_sec = 5.0
+
+ def __init__(
+ self,
+ config: ScraperConfig,
+ city: str = DEFAULT_CITY,
+ *,
+ delay_provider: Callable[[str], float] | None = None,
+ ) -> None:
+ super().__init__()
+ self.city = city
+ # Strangler-инжекция (#2133): конфиг и провайдер задержки приходят снаружи
+ # вместо прямого импорта app.core.config.settings /
+ # app.services.scraper_settings.get_scraper_delay. Продуктовая сторона
+ # передаёт RealScraperConfig + get_scraper_delay; kit не знает про app / БД.
+ self._config = config
+ if delay_provider is not None:
+ self.request_delay_sec = delay_provider(self.name)
+ self._browser: BrowserFetcher | None = None
+ # _cffi_session retained only for _rotate_ip (changeip call).
+ self._cffi_session: _CurlCffiSession | None = None
+ self._cookies: dict[str, str] = {}
+
+ async def __aenter__(self) -> YandexRealtyScraper: # type: ignore[override]
+ """Open ONE BrowserFetcher (camoufox) session, reused across all page fetches.
+
+ camoufox's real-browser fingerprint avoids the mobile-proxy IP tarpit
+ escalation that raw system-curl over SOCKS5 triggers. Opening per-page is
+ expensive and unnecessary.
+
+ fetch_timeout_s=_YANDEX_BROWSER_FETCH_TIMEOUT_S (30s) ограничивает потери
+ на один тайм-аутовый запрос: worst-case 30s×(1+retries)=90s на combo вместо 360s.
+ """
+ self._browser = BrowserFetcher(
+ source="yandex", fetch_timeout_s=_YANDEX_BROWSER_FETCH_TIMEOUT_S
+ )
+ await self._browser.__aenter__()
+ return self
+
+ async def __aexit__(self, *args: object) -> None: # type: ignore[override]
+ if self._browser is not None:
+ await self._browser.__aexit__(*args)
+ self._browser = None
+ 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) -> _CurlResponse: # type: ignore[override]
+ """Fetch gate-API URL via the shared BrowserFetcher (camoufox).
+
+ camoufox returns HTML-wrapped JSON; the JSON is extracted and returned as
+ the body of a _CurlResponse so existing callers (json.loads(resp.text),
+ resp.status_code checks) keep working unchanged.
+
+ Maps outcomes onto the curl-style status contract:
+ - JSON extracted -> status_code=200, text=
+ - fetch raised / no JSON found -> status_code=0 (treated as transient
+ failure by the caller retry logic)
+ """
+ if self._browser is None:
+ raise RuntimeError("YandexRealtyScraper must be used as async context manager")
+
+ try:
+ content = await self._browser.fetch(url)
+ except Exception:
+ logger.warning("yandex gate: BrowserFetcher.fetch failed url=%s", url, exc_info=True)
+ return _CurlResponse(status_code=0, text="")
+
+ json_text = _extract_json_from_content(content)
+ if not json_text:
+ logger.warning(
+ "yandex gate: could not extract JSON from content url=%s first200=%r",
+ url,
+ content[:200],
+ )
+ return _CurlResponse(status_code=0, text="")
+
+ return _CurlResponse(status_code=200, text=json_text)
+
+ async def _rotate_ip(self) -> bool:
+ """Rotate mobile proxy IP via changeip URL. Returns True on success."""
+ rotate_url = self._config.yandex_proxy_rotate_url or self._config.avito_proxy_rotate_url
+ if not rotate_url:
+ return False
+ sep = "&" if "?" in rotate_url else "?"
+ try:
+ async with _CurlCffiSession(timeout=30) as rot:
+ await rot.get(f"{rotate_url}{sep}format=json")
+ await asyncio.sleep(9)
+ logger.info("yandex proxy: IP rotated via changeip")
+ return True
+ except Exception:
+ logger.warning("yandex proxy: IP rotation failed", exc_info=True)
+ return False
+
+ def _build_url(
+ self,
+ page: int = 1,
+ rooms: str | None = None,
+ price_min: int | None = None,
+ price_max: int | None = None,
+ new_flat: str = "NO",
+ ) -> str:
+ """Build gate-API URL. page: 1-based (page=0 -> API error).
+
+ new_flat: "NO" -> vtorichka (default, preserves legacy behavior),
+ "YES" -> novostroyki (primary-sale / reassignment segment).
+ """
+ params: dict[str, str | int] = {
+ "rgid": _EKB_RGID,
+ "type": "SELL",
+ "category": "APARTMENT",
+ "newFlat": new_flat,
+ "_pageType": "search",
+ "_providers": "react-search-results-data",
+ "page": page,
+ }
+ if rooms and rooms in ROOM_GATE_PARAM:
+ params["roomsTotal"] = ROOM_GATE_PARAM[rooms]
+ if price_min is not None:
+ params["priceMin"] = price_min
+ if price_max is not None:
+ params["priceMax"] = price_max
+ return f"{_GATE_URL}?{urlencode(params)}"
+
+ async def _fetch_page_json(
+ self,
+ rooms: str | None,
+ page: int,
+ price_min: int | None,
+ price_max: int | None,
+ new_flat: str = "NO",
+ ) -> dict[str, Any] | None:
+ """Fetch one gate-API page. Returns parsed JSON dict or None on error/invalid."""
+ url = self._build_url(
+ page=page, rooms=rooms, price_min=price_min, price_max=price_max, new_flat=new_flat
+ )
+ try:
+ resp = await self._http_get(url, timeout=60)
+ except Exception:
+ logger.exception("yandex gate: GET failed url=%s", url)
+ return None
+ if resp.status_code != 200: # type: ignore[union-attr]
+ logger.warning("yandex gate: HTTP %d url=%s", resp.status_code, url) # type: ignore[union-attr]
+ return None
+ try:
+ payload: dict[str, Any] = json.loads(resp.text) # type: ignore[union-attr]
+ except (json.JSONDecodeError, ValueError):
+ logger.warning("yandex gate: JSON parse failed url=%s", url)
+ return None
+ if _is_gate_error(payload):
+ logger.warning("yandex gate: error response url=%s keys=%s", url, list(payload.keys()))
+ return None
+ return payload
+
+ async def fetch_around(
+ self,
+ lat: float,
+ lon: float,
+ radius_m: int = 1000,
+ page: int = 1,
+ rooms: str | None = None,
+ price_min: int | None = None,
+ price_max: int | None = None,
+ new_flat: str = "NO",
+ ) -> list[ScrapedLot]:
+ """Fetch ONE page of gate-API results. lat/lon/radius_m ignored (uses rgid).
+
+ page: 1-based. Tarpit resilience: status_code==0 or JSON error
+ -> rotate IP + retry up to _YANDEX_TARPIT_MAX_RETRIES times.
+ Non-tarpit failures (4xx/5xx) are not retried.
+ new_flat: "NO" -> vtorichka, "YES" -> novostroyki.
+ """
+ url = self._build_url(
+ page=page, rooms=rooms, price_min=price_min, price_max=price_max, new_flat=new_flat
+ )
+
+ payload: dict[str, Any] | None = None
+ for attempt in range(1 + _YANDEX_TARPIT_MAX_RETRIES):
+ try:
+ response = await self._http_get(url, timeout=60)
+ except Exception:
+ logger.exception("yandex gate fetch failed: %s", url)
+ return []
+
+ status = response.status_code # type: ignore[union-attr]
+
+ if status == 0:
+ logger.warning(
+ "yandex gate: status=0 (tarpit?) rooms=%s page=%d"
+ " -- rotating IP + retry attempt %d",
+ rooms,
+ page,
+ attempt + 1,
+ )
+ await self._rotate_ip()
+ await asyncio.sleep(2)
+ continue
+
+ if status != 200:
+ logger.warning(
+ "yandex gate: HTTP %d rooms=%s page=%d url=%s", status, rooms, page, url
+ )
+ return []
+
+ try:
+ payload = json.loads(response.text) # type: ignore[union-attr]
+ except (json.JSONDecodeError, ValueError):
+ logger.warning(
+ "yandex gate: JSON parse failed rooms=%s page=%d -- rotating + retry %d",
+ rooms,
+ page,
+ attempt + 1,
+ )
+ await self._rotate_ip()
+ await asyncio.sleep(2)
+ continue
+
+ if _is_gate_error(payload):
+ logger.warning(
+ "yandex gate: error payload rooms=%s page=%d keys=%s -- rotating + retry %d",
+ rooms,
+ page,
+ list(payload.keys()),
+ attempt + 1,
+ )
+ await self._rotate_ip()
+ await asyncio.sleep(2)
+ payload = None
+ continue
+
+ break
+
+ if payload is None:
+ logger.warning(
+ "yandex gate: all retries exhausted rooms=%s page=%d -- returning empty",
+ rooms,
+ page,
+ )
+ return []
+
+ lots = _parse_gate_json(payload, page_param=page, new_flat=new_flat)
+ logger.info(
+ "yandex gate page=%d rooms=%s price=[%s-%s] newFlat=%s: %d cards",
+ page,
+ rooms or "all",
+ price_min,
+ price_max,
+ new_flat,
+ len(lots),
+ )
+ await self.sleep_between_requests()
+ return lots
+
+ async def fetch_around_multi_room(
+ self,
+ lat: float,
+ lon: float,
+ radius_m: int = 1000,
+ max_pages: int = _GATE_MAX_PAGES_CAP,
+ rooms_list: list[str] | None = None,
+ price_ranges: list[tuple[int | None, int | None]] | None = None,
+ segments: list[str] | None = None,
+ on_combo: Any = None,
+ **_legacy_kwargs: Any,
+ ) -> list[ScrapedLot]:
+ """Fetch via segment x rooms x price-range combos; paginate each combo to totalPages.
+
+ Pagination driven by pager.totalPages from first page response.
+ Each combo iterates page=1 .. min(totalPages, max_pages, _GATE_MAX_PAGES_CAP).
+ Deduplicates by source_id/source_url across ALL combos AND segments (one `seen`).
+ Legacy mode (rooms_list=None, price_ranges=None): single citywide sweep.
+ segments: list of newFlat values, default ["NO"] (vtorichka only). Pass
+ ["NO", "YES"] to sweep both vtorichka and novostroyki in one call.
+
+ on_combo: опциональный callback(combo_label: str, new_lots: list[ScrapedLot]) -> None.
+ Вызывается после каждого combo с новыми (не повторными) лотами из него.
+ Позволяет инкрементальный save: вызывающий код сохраняет new_lots сразу,
+ не дожидаясь конца sweep'а. Дубли между combo НЕ передаются повторно.
+ """
+ seen: dict[str, ScrapedLot] = {}
+ _segments = segments or ["NO"]
+ combos_skipped = 0
+
+ if rooms_list is None and price_ranges is None:
+ combos: list[tuple[str | None, int | None, int | None]] = [(None, None, None)]
+ else:
+ r_list = rooms_list or list(ROOM_PATH.keys())
+ p_ranges = price_ranges or DEFAULT_PRICE_RANGES
+ combos = [(r, lo, hi) for r in r_list for lo, hi in p_ranges]
+
+ for new_flat in _segments:
+ _seg = _segment_for(new_flat)
+ for rooms, price_min, price_max in combos:
+ combo_label = f"{_seg}/{_combo_label(rooms, price_min, price_max)}"
+ total_pages: int | None = None
+ combo_new_lots: list[ScrapedLot] = []
+ combo_skipped = False
+
+ for page in range(1, max_pages + 1):
+ if page == 1:
+ payload_p1: dict[str, Any] | None = None
+ p1_url = self._build_url(
+ page=1,
+ rooms=rooms,
+ price_min=price_min,
+ price_max=price_max,
+ new_flat=new_flat,
+ )
+ for _attempt in range(1 + _YANDEX_TARPIT_MAX_RETRIES):
+ try:
+ resp = await self._http_get(p1_url, timeout=60)
+ except Exception:
+ logger.exception(
+ "yandex gate combo fetch failed combo=%s", combo_label
+ )
+ break
+ if resp.status_code == 0: # type: ignore[union-attr]
+ logger.warning(
+ "yandex gate: tarpit combo=%s page=1 attempt=%d -- rotating",
+ combo_label,
+ _attempt + 1,
+ )
+ await self._rotate_ip()
+ await asyncio.sleep(2)
+ continue
+ if resp.status_code != 200: # type: ignore[union-attr]
+ logger.warning(
+ "yandex gate: HTTP %d combo=%s page=1",
+ resp.status_code,
+ combo_label, # type: ignore[union-attr]
+ )
+ break
+ try:
+ payload_p1 = json.loads(resp.text) # type: ignore[union-attr]
+ except (json.JSONDecodeError, ValueError):
+ logger.warning(
+ "yandex gate: JSON parse error combo=%s page=1", combo_label
+ )
+ payload_p1 = None
+ await self._rotate_ip()
+ await asyncio.sleep(2)
+ continue
+ if _is_gate_error(payload_p1):
+ logger.warning(
+ "yandex gate: error payload combo=%s page=1", combo_label
+ )
+ payload_p1 = None
+ break
+ break
+
+ if payload_p1 is None:
+ logger.warning(
+ "yandex gate combo [%s] page=1: all retries exhausted -- skipping",
+ combo_label,
+ )
+ combo_skipped = True
+ combos_skipped += 1
+ break
+
+ result = _extract_gate_data(payload_p1)
+ if result is None:
+ combo_skipped = True
+ combos_skipped += 1
+ break
+ _entities_p1, pager_p1 = result
+ total_pages = min(
+ pager_p1.get("totalPages", 1),
+ max_pages,
+ _GATE_MAX_PAGES_CAP,
+ )
+ lots_p1 = _parse_gate_json(payload_p1, page_param=1, new_flat=new_flat)
+ if not lots_p1:
+ logger.debug(
+ "yandex gate combo [%s] page=1: empty -- stopping", combo_label
+ )
+ break
+ for lot in lots_p1:
+ key = lot.source_id or lot.source_url
+ if key and key not in seen:
+ seen[key] = lot
+ combo_new_lots.append(lot)
+ await self.sleep_between_requests()
+ if total_pages <= 1:
+ break
+ continue
+
+ if total_pages is not None and page > total_pages:
+ break
+
+ lots = await self.fetch_around(
+ lat,
+ lon,
+ radius_m,
+ page=page,
+ rooms=rooms,
+ price_min=price_min,
+ price_max=price_max,
+ new_flat=new_flat,
+ )
+ if not lots:
+ logger.debug(
+ "yandex gate combo [%s] page=%d: empty -- stopping",
+ combo_label,
+ page,
+ )
+ break
+ for lot in lots:
+ key = lot.source_id or lot.source_url
+ if key and key not in seen:
+ seen[key] = lot
+ combo_new_lots.append(lot)
+
+ # Инкрементальный save: вызываем on_combo если есть новые лоты.
+ # Скипнутые combo (combo_skipped=True) не вызывают on_combo.
+ if on_combo is not None and combo_new_lots and not combo_skipped:
+ try:
+ on_combo(combo_label, combo_new_lots)
+ except Exception:
+ logger.exception(
+ "yandex gate: on_combo callback failed combo=%s", combo_label
+ )
+
+ logger.info(
+ "yandex gate aggregate: %d unique lots (%d combos x %d segments=%s, skipped=%d)",
+ len(seen),
+ len(combos),
+ len(_segments),
+ _segments,
+ combos_skipped,
+ )
+ return list(seen.values())
+
+ async def fetch_all_secondary(
+ self,
+ *,
+ rooms_buckets: list[str] | None = None,
+ price_cap_per_bucket: int = 500,
+ max_pages_per_bucket: int = _GATE_MAX_PAGES_CAP,
+ concurrency: int = 4,
+ on_bucket: Any = None,
+ on_progress: Any = None,
+ skip_buckets: set[str] | None = None,
+ ) -> list[ScrapedLot]:
+ """Exhaustive EKB vtorichka load via KOMNATNOST x TSENA partitioning.
+
+ Uses gate-API JSON (totalItems from pager) instead of DOM state extraction.
+ Стартует с общей data-driven seed-сетки ценовых брекетов
+ (get_price_seed_brackets() — shared EKB grid, общая с avito/cian) вместо
+ одной корень-бисекции [0, _YANDEX_MAX_PRICE]. Закрытые seed-брекеты передают
+ hi = br_hi - 1 (priceMax эксклюзивный сверху → нет пересечений), открытый
+ (br_hi is None) — hi=None (без priceMax, весь хвост люкса). Дедуп по
+ source_id в общем seen страхует на стыках.
+ """
+ _buckets = rooms_buckets if rooms_buckets is not None else list(ROOM_PATH.keys())
+ seen: dict[str, ScrapedLot] = {}
+
+ for rooms_key in _buckets:
+ logger.info(
+ "yandex gate exhaustive: starting rooms=%s (price_cap=%d concurrency=%d)",
+ rooms_key,
+ price_cap_per_bucket,
+ concurrency,
+ )
+ before = len(seen)
+ for br_lo, br_hi in get_price_seed_brackets():
+ walk_hi = br_hi - 1 if br_hi is not None else None
+ await self._walk_price_range(
+ rooms=rooms_key,
+ lo=br_lo,
+ hi=walk_hi,
+ seen=seen,
+ price_cap_per_bucket=price_cap_per_bucket,
+ max_pages_per_bucket=max_pages_per_bucket,
+ concurrency=concurrency,
+ on_bucket=on_bucket,
+ skip_buckets=skip_buckets,
+ )
+ room_collected = len(seen) - before
+ logger.info(
+ "yandex gate exhaustive: rooms=%s done -- collected %d (total unique=%d)",
+ rooms_key,
+ room_collected,
+ len(seen),
+ )
+ if on_progress is not None:
+ on_progress(len(seen))
+
+ logger.info("yandex gate exhaustive: DONE -- total unique=%d lots", len(seen))
+ return list(seen.values())
+
+ async def _walk_price_range(
+ self,
+ *,
+ rooms: str | None,
+ lo: int,
+ hi: int | None,
+ seen: dict[str, ScrapedLot],
+ price_cap_per_bucket: int,
+ max_pages_per_bucket: int,
+ concurrency: int = 4,
+ on_bucket: Any = None,
+ skip_buckets: set[str] | None = None,
+ _depth: int = 0,
+ ) -> None:
+ """Recursive binary price-range partitioning.
+
+ Probe page=1 -> read totalItems from pager.
+ Closed bracket (hi is not None): totalItems > cap and bracket >= MIN_BRACKET
+ -> split; else paginate leaf.
+ Open bracket (hi is None): top seed-bracket without priceMax ceiling. Cannot
+ split (no mid) -> paginate leaf directly (lux tail tiny, tail-loss accepted).
+ Fallback: totalItems=None -> paginate-until-empty (safe degradation).
+ skip_buckets: resume checkpoint -- skip leaf buckets already done.
+ """
+ _lo_param = lo if lo > 0 else None
+ _hi_repr = "open" if hi is None else str(hi)
+
+ probe_payload = await self._fetch_page_json(rooms, 1, _lo_param, hi)
+ await asyncio.sleep(self.request_delay_sec)
+
+ total: int | None = None
+ probe_lots: list[ScrapedLot] = []
+
+ if probe_payload is None:
+ logger.warning(
+ "yandex gate: probe failed rooms=%s [%d, %s] depth=%d -- rotating + retry",
+ rooms,
+ lo,
+ _hi_repr,
+ _depth,
+ )
+ rotated = await self._rotate_ip()
+ if rotated:
+ probe_payload = await self._fetch_page_json(rooms, 1, _lo_param, hi)
+ await asyncio.sleep(self.request_delay_sec)
+
+ if probe_payload is not None:
+ result = _extract_gate_data(probe_payload)
+ if result is not None:
+ _, pager = result
+ total = pager.get("totalItems")
+ probe_lots = _parse_gate_json(probe_payload, page_param=1)
+ else:
+ logger.warning(
+ "yandex gate: probe data extract failed rooms=%s [%d, %s]", rooms, lo, _hi_repr
+ )
+
+ if total is None:
+ # Degraded path: paginate-until-empty without knowing total
+ logger.warning(
+ "yandex gate: totalItems unknown rooms=%s [%d, %s] -- paginate until empty",
+ rooms,
+ lo,
+ _hi_repr,
+ )
+ page = 1
+ bucket_key = _combo_label(rooms, lo, hi)
+ if skip_buckets and bucket_key in skip_buckets:
+ return
+ pages_fetched = 0
+ while pages_fetched < max_pages_per_bucket:
+ payload = await self._fetch_page_json(rooms, page, _lo_param, hi)
+ await asyncio.sleep(self.request_delay_sec)
+ if payload is None:
+ break
+ lots = _parse_gate_json(payload, page_param=page)
+ if not lots:
+ break
+ for lot in lots:
+ if lot.source_id and lot.source_id not in seen:
+ seen[lot.source_id] = lot
+ page += 1
+ pages_fetched += 1
+ if on_bucket is not None:
+ on_bucket(bucket_key, len(seen))
+ return
+
+ # Split or paginate. ОТКРЫТЫЙ брекет (hi is None) делить нельзя (нет mid) —
+ # сразу пагинируем leaf напрямую (хвост люкса крошечный, tail-loss accepted).
+ cap = price_cap_per_bucket
+ _min_bracket = 500_000
+
+ if hi is not None and total > cap and (hi - lo) >= _min_bracket and _depth < 8:
+ mid = (lo + hi) // 2
+ await self._walk_price_range(
+ rooms=rooms,
+ lo=lo,
+ hi=mid,
+ seen=seen,
+ price_cap_per_bucket=cap,
+ max_pages_per_bucket=max_pages_per_bucket,
+ concurrency=concurrency,
+ on_bucket=on_bucket,
+ skip_buckets=skip_buckets,
+ _depth=_depth + 1,
+ )
+ await self._walk_price_range(
+ rooms=rooms,
+ lo=mid + 1,
+ hi=hi,
+ seen=seen,
+ price_cap_per_bucket=cap,
+ max_pages_per_bucket=max_pages_per_bucket,
+ concurrency=concurrency,
+ on_bucket=on_bucket,
+ skip_buckets=skip_buckets,
+ _depth=_depth + 1,
+ )
+ return
+
+ # Leaf bucket: paginate
+ bucket_key = _combo_label(rooms, lo, hi)
+ if skip_buckets and bucket_key in skip_buckets:
+ logger.debug("yandex gate: skip bucket %s (checkpoint)", bucket_key)
+ return
+
+ total_pages = min(
+ math.ceil(total / 20),
+ _GATE_MAX_PAGES_CAP,
+ max_pages_per_bucket,
+ )
+ logger.info("yandex gate: leaf bucket %s total=%d pages=%d", bucket_key, total, total_pages)
+
+ # Add probe lots (page 1 already fetched)
+ for lot in probe_lots:
+ if lot.source_id and lot.source_id not in seen:
+ seen[lot.source_id] = lot
+
+ if total_pages <= 1:
+ if on_bucket is not None:
+ on_bucket(bucket_key, len(seen))
+ return
+
+ # Paginate pages 2..total_pages with concurrency
+ sem = asyncio.Semaphore(concurrency)
+
+ async def _fetch_leaf_page(pg: int) -> list[ScrapedLot]:
+ async with sem:
+ payload = await self._fetch_page_json(rooms, pg, _lo_param, hi)
+ await asyncio.sleep(self.request_delay_sec)
+ if payload is None:
+ return []
+ return _parse_gate_json(payload, page_param=pg)
+
+ tasks = [_fetch_leaf_page(pg) for pg in range(2, total_pages + 1)]
+ results = await asyncio.gather(*tasks)
+ for page_lots in results:
+ for lot in page_lots:
+ if lot.source_id and lot.source_id not in seen:
+ seen[lot.source_id] = lot
+
+ if on_bucket is not None:
+ on_bucket(bucket_key, len(seen))
+
+
+def _combo_label(rooms: str | None, lo: int | None, hi: int | None) -> str:
+ """Уникальная метка бакета для checkpoint-логики."""
+ return f"{rooms or 'any'}:{lo}-{hi}"
diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/shared.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/shared.py
new file mode 100644
index 00000000..cb6e734e
--- /dev/null
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/shared.py
@@ -0,0 +1,95 @@
+"""Нормализация house_type → каноничный enum для listings.house_type.
+
+Strangler-копия `app.services.scrapers.house_type_normalizer` (#2133). Локальная в
+yandex-провайдере, потому что house_type_normalizer ещё не перенесён в core-слой
+scraper_kit (это отдельный шаг консолидации E). Копия — pure-функция без `app.*`
+зависимостей; идентична источнику.
+
+Целевой канон (сверено с estimator._IMV_HOUSE_TYPE_MAP, estimator.py:146):
+ panel / brick / monolith / monolith_brick / block / wood.
+
+Зачем (#2007): estimator применяет soft-penalty по house_type — аналог с
+house_type != target штрафуется. Yandex SERP отдаёт SCREAMING-значения
+(MONOLIT / BRICK / PANEL / ...), которые НИКОГДА не равны каноничным
+(monolith / brick / panel / ...) → ~70% yandex-аналогов получали ложный штраф
+и фактически выпадали из скоринга. Нормализация на ингесте чинит это.
+
+Важно про None: неизвестное / 'other' / '' → None, НЕ 'other'. В estimator
+`house_type IS NULL` нейтрально (без штрафа), а любое не-канон значение всегда
+!= target → ложный штраф. Поэтому unknown лучше схлопнуть в NULL.
+
+Источники raw-значений:
+- yandex SERP: building.buildingType — SCREAMING_SNAKE (MONOLIT, MONOLIT_BRICK, ...)
+- cian SERP: building.materialType — camelCase (monolith, monolithBrick,
+ gasSilicateBlock, ...) — переиспользуется в #2008 (cian.py:815).
+"""
+
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+# Канон listings.house_type — выровнен по estimator._IMV_HOUSE_TYPE_MAP.
+_CANON: frozenset[str] = frozenset(
+ {"panel", "brick", "monolith", "monolith_brick", "block", "wood"}
+)
+
+# raw-токен → канон. Ключи — точные значения вокабуляров источников; сравнение
+# регистронезависимое (см. _LOOKUP ниже), но строго по полному токену енума,
+# без fuzzy-матчинга. Неизвестные токены сюда НЕ попадают → normalize вернёт None.
+_RAW_TO_CANON: dict[str, str] = {
+ # ── yandex SERP (SCREAMING_SNAKE buildingType) ───────────────────────────
+ "MONOLIT": "monolith",
+ "BRICK": "brick",
+ "PANEL": "panel",
+ "MONOLIT_BRICK": "monolith_brick",
+ "BLOCK": "block",
+ "WOOD": "wood",
+ # ── cian SERP (camelCase materialType) — reuse в #2008 ───────────────────
+ "monolith": "monolith",
+ "brick": "brick",
+ "panel": "panel",
+ "block": "block",
+ "wood": "wood",
+ "monolithBrick": "monolith_brick",
+ "gasSilicateBlock": "block",
+ "aerocreteBlock": "block",
+ "foamConcreteBlock": "block",
+ "stalin": "brick", # «сталинка» — кирпич
+}
+
+# Регистронезависимый lookup. Лоуэркейс-ключи не коллизят между вокабулярами:
+# 'MONOLIT'→'monolit' и 'monolith'→'monolith' — разные ключи.
+_LOOKUP: dict[str, str] = {k.lower(): v for k, v in _RAW_TO_CANON.items()}
+
+
+def normalize_house_type(raw: str | None) -> str | None:
+ """Преобразовать raw house_type в каноничный enum.
+
+ Уже-каноничное значение возвращается as-is (идемпотентность — нужна при
+ ре-обработке и для backfill-миграции). Неизвестное / 'other' / пустое /
+ None → None (НЕ 'other': см. docstring модуля про estimator soft-penalty).
+
+ Args:
+ raw: сырое значение из парсера (e.g. «MONOLIT», «monolithBrick») или None.
+
+ Returns:
+ Одно из panel / brick / monolith / monolith_brick / block / wood, либо None.
+ """
+ if raw is None:
+ return None
+ stripped = raw.strip()
+ if not stripped:
+ return None
+ # Pass-through: уже каноничное значение (idempotency — в т.ч. 'monolith_brick',
+ # которого нет среди raw-ключей карты).
+ if stripped in _CANON:
+ return stripped
+ # Регистронезависимый exact-token lookup по обоим вокабулярам.
+ canon = _LOOKUP.get(stripped) or _LOOKUP.get(stripped.lower())
+ if canon is None:
+ # Неизвестное / 'other' — нормально (NULL нейтрально для estimator).
+ # debug, не warning: 'other' встречается массово, warning засорил бы лог.
+ logger.debug("house_type_normalizer: unmapped raw value %r — stored as NULL", raw)
+ return canon
diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/valuation.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/valuation.py
new file mode 100644
index 00000000..2d4d9825
--- /dev/null
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/valuation.py
@@ -0,0 +1,543 @@
+"""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)
+
+Strangler-копия `app.services.scrapers.yandex_valuation` (#2133). Развязка от `app.*`:
+ - `app.core.config.settings` → инжектируемый `ScraperConfig` (scraper_proxy_url)
+ - `app.services.scraper_settings.get_scraper_delay` → инжектируемый `delay_provider`
+ - `app.services.scrapers.base` → `scraper_kit.base`
+ - `app.services.scrapers.yandex_helpers` → `scraper_kit.yandex_helpers`
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from collections.abc import Callable
+from datetime import date
+from typing import TYPE_CHECKING, 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 scraper_kit.base import BaseScraper
+from scraper_kit.yandex_helpers import (
+ parse_dmy,
+ parse_house_type,
+ parse_rub,
+)
+
+if TYPE_CHECKING:
+ from scraper_kit.contracts import ScraperConfig
+
+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__()
+ # Strangler-инжекция (#2133): конфиг (прокси-egress) и провайдер задержки
+ # приходят снаружи вместо app.core.config.settings /
+ # app.services.scraper_settings.get_scraper_delay.
+ self._config = config
+ if delay_provider is not None:
+ self.request_delay_sec = delay_provider(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
+ # Mobile proxy wiring (#806 follow-up): route via mobile proxy to avoid
+ # datacenter-IP blocks. proxy=None → прямое подключение (dev).
+ _proxy_url = self._config.scraper_proxy_url
+ _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
+ self._cffi_session = _CurlCffiSession(impersonate="chrome120", proxies=_proxies)
+ 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]