From 1ebefcdb985d9196b498b180d4fc03bf219c72ac Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 17 May 2026 18:34:10 +0300 Subject: [PATCH] =?UTF-8?q?feat(22d):=20domrf=5Fcatalog.py=20=E2=80=94=20p?= =?UTF-8?q?er-flat=20price/status=20from=20SSR=20HTML=20catalog=20page?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds new scraper module for DOM.RF catalog-квартир SSR pages (issue #297 Phase 5). kn-API returns price for only 0.3% of EKB flats; prices + finishing/ceiling/section data live on per-flat HTML pages at /сервисы/каталог-квартир/квартира/{hash}. - fetch_catalog_html(): in-browser fetch via BrowserSession (same WAF bypass pattern as get_json, but returns text/html instead of JSON) - parse_catalog_flat(): 2-level extraction via stdlib html.parser + regex on raw HTML (no new deps; beautifulsoup4 noted as TODO) - upsert_catalog_data(): UPDATE domrf_kn_flats SET catalog-only cols with COALESCE guards; SAVEPOINT per row; does not touch kn-API metadata - scrape_catalog_batch(): full batch orchestrator over one BrowserSession Wiring (Celery task + beat schedule) deferred to a follow-up PR. Depends on migration 22b (section_no, finishing_type, ceiling_height_m, catalog_updated_at, catalog_url_hash columns) — also separate PR. Closes #297 (partial — 22d code delivery) --- .../app/services/scrapers/domrf_catalog.py | 563 ++++++++++++++++++ 1 file changed, 563 insertions(+) create mode 100644 backend/app/services/scrapers/domrf_catalog.py diff --git a/backend/app/services/scrapers/domrf_catalog.py b/backend/app/services/scrapers/domrf_catalog.py new file mode 100644 index 00000000..b0cfd052 --- /dev/null +++ b/backend/app/services/scrapers/domrf_catalog.py @@ -0,0 +1,563 @@ +"""DOM.РФ catalog-квартир HTML scraper (issue #297 22d). + +kn-API не возвращает цену для большинства квартир (91.5% NULL). Цены живут на +отдельной странице каталога: + https://наш.дом.рф/сервисы/каталог-квартир/квартира/{catalog_url_hash} + +Этот модуль: +1. Строит URL каталога по `catalog_url_hash` (колонка появляется после миграции 22b). +2. Получает SSR-HTML через BrowserSession (Playwright, anti-bot — тот же паттерн + что и get_json, но возвращает HTML text вместо JSON). +3. Извлекает price_rub, status, finishing_type, ceiling_height_m, section_no, + catalog_updated_at из HTML с помощью stdlib `html.parser` + regex. +4. Пишет только catalog-derived поля через UPDATE ... WHERE ods_id = :ods_id — + НЕ перетирает kn-API метаданные (total_area, rooms и т.д.). + +Зависимости: нет новых. Использует `html.parser` из stdlib + `re`. +NOTE: beautifulsoup4 НЕ установлен (нет в pyproject.toml). Если потребуется + структурированный парсинг — добавить `beautifulsoup4>=4.12` в pyproject.toml + и заменить _HtmlTextExtractor на `BeautifulSoup(html, "html.parser")`. + +Контекст Roadmap: + - Phase 5 (22d) — catalog scraper для цен + - Согласно update 2026-05-17 (Objective goldmine): Objective уже содержит 81.4% + цен. Для `domrf_kn_flats` этот scraper остаётся полезен для полей: + finishing_type, ceiling_height_m, section_no, catalog_updated_at, catalog_url_hash. + Price coverage через Objective (OBJ-3) — приоритетнее. + +Wiring (отдельный PR): + - Celery task: `backend/app/workers/tasks/scrape_catalog.py` + - Beat schedule: кварть + `catalog_updated_at < NOW() - INTERVAL '30 days'` +""" + +from __future__ import annotations + +import asyncio +import logging +import re +from datetime import date +from html.parser import HTMLParser +from typing import Any + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.services.scrapers.stealth import BASE_URL, BrowserSession, jitter_sleep + +logger = logging.getLogger(__name__) + +# Per-flat catalog page URL template (IDN encoded — same as what browsers send). +# Человекочитаемый вид: https://наш.дом.рф/сервисы/каталог-квартир/квартира/{hash} +CATALOG_FLAT_PATH = "/сервисы/каталог-квартир/квартира/{catalog_url_hash}" + +# JS snippet: выполняется внутри живой Playwright-страницы. +# Возвращает HTML текст страницы (text/html). +# Это аналог _FETCH_JS из stealth.py, но для text/html вместо application/json. +_FETCH_HTML_JS = """ +async ({url}) => { + try { + const r = await fetch(url, {credentials: 'include'}); + const ctype = r.headers.get('content-type') || ''; + const body = await r.text(); + return {ok: r.ok, status: r.status, body, contentType: ctype}; + } catch (e) { + return {ok: false, status: 0, body: String(e), contentType: ''}; + } +} +""" + +# Нормализованные значения статуса продажи. +STATUS_FREE = "free" +STATUS_SOLD = "sold" +STATUS_RESERVED = "reserved" + + +# ── HTML fetching ───────────────────────────────────────────────────────────── + + +async def fetch_catalog_html(session: BrowserSession, catalog_url_hash: str) -> str: + """Получить SSR-HTML страницы квартиры в каталоге DOM.РФ. + + Использует тот же паттерн что get_json(): fetch() внутри живой Playwright-страницы. + Так WAF-fingerprint идентичен браузеру, cookies проброшены автоматически. + + Raises: + RuntimeError: при транзиентной ошибке после 5 попыток. + WafBlockedError: (из stealth) если вернулся JS-challenge вместо HTML. + """ + if session._page is None: + raise RuntimeError("BrowserSession not bootstrapped") + + url = BASE_URL + CATALOG_FLAT_PATH.format(catalog_url_hash=catalog_url_hash) + last_err: Exception | None = None + + for attempt in range(5): + async with session._sem: + await jitter_sleep() + try: + session._request_count += 1 + result = await session._page.evaluate(_FETCH_HTML_JS, {"url": url}) + except Exception as exc: + last_err = exc + logger.warning( + "catalog html evaluate err attempt=%d hash=%s: %r", + attempt, + catalog_url_hash, + exc, + ) + await asyncio.sleep(2**attempt) + continue + + status: int = result.get("status", 0) + body: str = result.get("body", "") + ctype: str = result.get("contentType", "") + + if status in (429,) or status >= 500 or status == 0: + last_err = RuntimeError(f"transient status={status}") + logger.warning( + "catalog html transient status=%d attempt=%d hash=%s, backing off", + status, + attempt, + catalog_url_hash, + ) + await asyncio.sleep(2**attempt) + continue + + if status == 404: + raise RuntimeError(f"catalog 404 for hash={catalog_url_hash}") + + if status != 200: + raise RuntimeError(f"catalog http {status}: {body[:200]} hash={catalog_url_hash}") + + # Успех — но нужно проверить что не пришёл WAF JS-challenge (нет text/html) + # Страница каталога — SSR, всегда text/html. Если что-то другое — WAF. + if body and "text/html" not in ctype and " None: + super().__init__() + self._stack: list[tuple[str, dict[str, str]]] = [] + # Список записей: (class_hint, full_text) + self.blocks: list[tuple[str, str]] = [] + self._buf: list[str] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attr_dict = {k: (v or "") for k, v in attrs} + self._stack.append((tag, attr_dict)) + self._buf.append("") # начало нового буфера для этого тега + + def handle_endtag(self, _tag: str) -> None: + if not self._stack: + return + tag, attr_dict = self._stack.pop() + text = (self._buf.pop() if self._buf else "").strip() + cls = attr_dict.get("class", "") + if text and (cls or tag in ("h1", "h2", "h3", "p", "span", "div", "li")): + self.blocks.append((cls, text)) + # Propagate accumulated text up to parent buffer + if self._buf: + self._buf[-1] += " " + text + + def handle_data(self, data: str) -> None: + if self._buf: + self._buf[-1] += data + + +def _find_text_near( + blocks: list[tuple[str, str]], label_pattern: str, value_pattern: str | None = None +) -> str | None: + """Найти текстовое значение рядом с блоком, matching label_pattern. + + Стратегия: ищем блок где text матчит label_pattern — берём следующий блок + как значение (value_pattern если задан). + """ + label_re = re.compile(label_pattern, re.IGNORECASE) + for i, (_cls, block_text) in enumerate(blocks): + if label_re.search(block_text): + # Попробовать next block + for j in range(i + 1, min(i + 4, len(blocks))): + candidate = blocks[j][1].strip() + if candidate: + if value_pattern is None: + return candidate + if re.search(value_pattern, candidate, re.IGNORECASE): + return candidate + return None + + +def parse_catalog_flat(html: str) -> dict[str, Any]: + """Извлечь поля из SSR-HTML страницы квартиры DOM.РФ. + + Возвращаемые поля (None если не найдено): + - price_rub (int) Цена квартиры в рублях + - price_per_m2 (float) Цена за м² (если указана отдельно) + - status (str) 'free' | 'sold' | 'reserved' + - finishing_type (str) Тип отделки (Предчистовая, Чистовая, Без отделки, ...) + - ceiling_height_m (float) Высота потолков в метрах + - section_no (int) Номер подъезда / секции + - catalog_updated_at (date) Дата обновления информации на странице + + Парсинг хрупкий по природе (SSR HTML DOM.РФ меняется без уведомлений). + Все extraction best-effort — KeyError/AttributeError обёрнуты внутри. + """ + result: dict[str, Any] = {} + + # ── Шаг 1: собрать все текстовые блоки через HTMLParser ────────────────── + collector = _TextCollector() + try: + collector.feed(html) + except Exception as exc: + logger.warning("html parse error (non-fatal): %s", exc) + + blocks = collector.blocks + + # ── Шаг 2: regex-extraction из полного HTML текста ──────────────────────── + # Страница DOM.РФ SSR встраивает данные и в мета-тегах и в JSON-LD. + # Ищем в сыром HTML — надёжнее чем DOM-обход для хрупкой структуры. + + # Price: "7 890 000 ₽" или "7 890 000 руб" + price_match = re.search( + r"([\d][\d\s]{3,12}[\d])\s*(?:₽|руб)", + html, + re.UNICODE, + ) + if price_match: + raw_price = re.sub(r"\s+", "", price_match.group(1)) + try: + price_val = int(raw_price) + # Санity: цена квартиры в ЕКБ от 1 до 500 млн + if 1_000_000 <= price_val <= 500_000_000: + result["price_rub"] = price_val + except ValueError: + pass + + # Price per m²: "217 835 ₽/м²" + ppm2_match = re.search( + r"([\d][\d\s]{2,9}[\d])\s*(?:₽|руб)[/⁄](?:м²|кв\.?\s*м)", + html, + re.UNICODE, + ) + if ppm2_match: + raw_ppm2 = re.sub(r"\s+", "", ppm2_match.group(1)) + try: + result["price_per_m2"] = float(raw_ppm2) + except ValueError: + pass + + # Status: ищем характерные слова рядом с "статус" или в badge + status_match = re.search( + r"(в\s*продаже|свободна|free|продано|sold|забронирована|бронь|reserved)", + html, + re.IGNORECASE | re.UNICODE, + ) + if status_match: + s = status_match.group(1).lower() + if any(kw in s for kw in ("продаже", "свободна", "free")): + result["status"] = STATUS_FREE + elif any(kw in s for kw in ("продано", "sold")): + result["status"] = STATUS_SOLD + elif any(kw in s for kw in ("бронь", "забронирована", "reserved")): + result["status"] = STATUS_RESERVED + + # Finishing type: "Предчистовая", "Чистовая", "Без отделки", "Под ключ" + finishing_match = re.search( + r"(предчистовая|чистовая|без\s+отделки|под\s+ключ|white\s+box|whitebox)", + html, + re.IGNORECASE | re.UNICODE, + ) + if finishing_match: + result["finishing_type"] = finishing_match.group(1).strip().capitalize() + + # Ceiling height: "2,7 м" или "2.7 м" или "высота потолков 2,7" + ceiling_match = re.search( + r"(?:высота\s*потолков?|потолки?)\D{0,20}?([\d][,.][\d])\s*м", + html, + re.IGNORECASE | re.UNICODE, + ) + if not ceiling_match: + # Fallback: просто "2,7 м" в характеристиках квартиры (диапазон 2.0–4.5 м) + ceiling_match = re.search( + r"\b([2-4][,.][\d])\s*м\b", + html, + re.UNICODE, + ) + if ceiling_match: + raw_ceil = ceiling_match.group(1).replace(",", ".") + try: + ceil_val = float(raw_ceil) + if 2.0 <= ceil_val <= 6.0: + result["ceiling_height_m"] = ceil_val + except ValueError: + pass + + # Section (подъезд): "Подъезд 1", "Секция 3", "Подъезд №2" + section_match = re.search( + r"(?:подъезд|секция)\s*[№#]?\s*(\d+)", + html, + re.IGNORECASE | re.UNICODE, + ) + if section_match: + try: + result["section_no"] = int(section_match.group(1)) + except ValueError: + pass + + # catalog_updated_at: "Информация обновлена 17.04.2026" или "Обновлено 17.04.2026" + updated_match = re.search( + r"(?:информация\s+обновлена|обновлено|обновлён?а?)\D{0,10}?(\d{1,2}[./]\d{1,2}[./]\d{4})", + html, + re.IGNORECASE | re.UNICODE, + ) + if updated_match: + raw_dt = updated_match.group(1).replace("/", ".") + try: + parts = raw_dt.split(".") + if len(parts) == 3: + result["catalog_updated_at"] = date(int(parts[2]), int(parts[1]), int(parts[0])) + except (ValueError, IndexError): + pass + + # ── Шаг 3: блочный fallback для ceiling / section (если regex не нашёл) ── + if "ceiling_height_m" not in result: + candidate = _find_text_near(blocks, r"потолк|высота", r"[23][,.][\d]") + if candidate: + m = re.search(r"([23][,.][\d])", candidate) + if m: + try: + v = float(m.group(1).replace(",", ".")) + if 2.0 <= v <= 6.0: + result["ceiling_height_m"] = v + except ValueError: + pass + + if "section_no" not in result: + candidate = _find_text_near(blocks, r"подъезд|секция", r"^\d+$") + if candidate: + try: + result["section_no"] = int(candidate.strip()) + except ValueError: + pass + + logger.debug( + "parse_catalog_flat: extracted fields=%s", + list(result.keys()), + ) + return result + + +# ── DB writes ───────────────────────────────────────────────────────────────── + + +def upsert_catalog_data( + db: Session, ods_id: str, catalog_url_hash: str, data: dict[str, Any] +) -> bool: + """UPDATE catalog-derived поля в domrf_kn_flats для конкретной квартиры. + + Обновляет ТОЛЬКО catalog-only колонки: + price_rub, price_per_m2, status, finishing_type, ceiling_height_m, + section_no, catalog_updated_at, catalog_url_hash. + + НЕ трогает: total_area, rooms, floor, num_floors, flat_type, obj_id и + другие kn-API метаданные. + + Использует COALESCE: если новое значение NULL — старое сохраняется. + Это позволяет повторно запускать scraper не затирая частично заполненные поля. + + ВАЖНО: колонки section_no, finishing_type, ceiling_height_m, + catalog_updated_at, catalog_url_hash должны существовать в таблице. + Они появляются после миграции 22b. Если таблица старая — UPDATE упадёт + с 'column does not exist'. Решение: сначала выполнить data/sql/NN_22b_flats_cols.sql. + + Возвращает True если строка найдена и обновлена, False если ods_id не найден. + """ + params: dict[str, Any] = { + "ods_id": ods_id, + "catalog_url_hash": catalog_url_hash, + "price_rub": data.get("price_rub"), + "price_per_m2": data.get("price_per_m2"), + "status": data.get("status"), + "finishing_type": data.get("finishing_type"), + "ceiling_height_m": data.get("ceiling_height_m"), + "section_no": data.get("section_no"), + "catalog_updated_at": data.get("catalog_updated_at"), + } + + try: + with db.begin_nested(): + result = db.execute( + text( + """ + UPDATE domrf_kn_flats SET + catalog_url_hash = :catalog_url_hash, + price_rub = COALESCE(:price_rub, price_rub), + price_per_m2 = COALESCE(:price_per_m2, price_per_m2), + status = COALESCE(:status, status), + finishing_type = COALESCE(:finishing_type, finishing_type), + ceiling_height_m = COALESCE(:ceiling_height_m, ceiling_height_m), + section_no = COALESCE(:section_no, section_no), + catalog_updated_at = COALESCE(:catalog_updated_at, catalog_updated_at) + WHERE ods_id = :ods_id + """ + ), + params, + ) + except Exception as exc: + logger.warning("upsert_catalog_data ods_id=%s failed: %s", ods_id, exc) + return False + + rows_affected: int = result.rowcount or 0 + if rows_affected == 0: + logger.warning("upsert_catalog_data: ods_id=%s not found in domrf_kn_flats", ods_id) + return rows_affected > 0 + + +# ── Per-flat scrape orchestration ───────────────────────────────────────────── + + +async def scrape_one_flat( + session: BrowserSession, + db: Session, + ods_id: str, + catalog_url_hash: str, +) -> dict[str, Any]: + """Scrape одной квартиры: fetch HTML → parse → upsert. + + Возвращает dict с результатом: {ods_id, success, fields_extracted, updated}. + Ошибки fetch/parse логируются, не бросаются — caller обрабатывает результат. + """ + outcome: dict[str, Any] = { + "ods_id": ods_id, + "catalog_url_hash": catalog_url_hash, + "success": False, + "fields_extracted": 0, + "updated": False, + "error": None, + } + + try: + html = await fetch_catalog_html(session, catalog_url_hash) + except Exception as exc: + logger.warning( + "catalog fetch failed ods_id=%s hash=%s: %s", + ods_id, + catalog_url_hash, + exc, + ) + outcome["error"] = str(exc)[:500] + return outcome + + try: + data = parse_catalog_flat(html) + except Exception as exc: + logger.warning( + "catalog parse failed ods_id=%s hash=%s: %s", + ods_id, + catalog_url_hash, + exc, + ) + outcome["error"] = f"parse: {exc!s}"[:500] + return outcome + + outcome["fields_extracted"] = len([v for v in data.values() if v is not None]) + outcome["updated"] = upsert_catalog_data(db, ods_id, catalog_url_hash, data) + outcome["success"] = True + logger.info( + "catalog scrape ods_id=%s: fields=%d updated=%s", + ods_id, + outcome["fields_extracted"], + outcome["updated"], + ) + return outcome + + +async def scrape_catalog_batch( + db: Session, + flats: list[dict[str, Any]], + region_code: int = 66, + headed: bool = False, + load_state: str | None = None, +) -> dict[str, Any]: + """Scrape пачки квартир каталога DOM.РФ. + + `flats` — список dict'ов с ключами {ods_id, catalog_url_hash}. + Типовой источник: SELECT ods_id, catalog_url_hash FROM domrf_kn_flats + WHERE catalog_url_hash IS NOT NULL + AND (catalog_updated_at IS NULL OR catalog_updated_at < NOW() - INTERVAL '30 days') + LIMIT :batch_size. + + Использует один BrowserSession на весь пакет (bootstrapped 1 раз). + jitter_sleep между запросами встроен в fetch_catalog_html (через BrowserSession._sem). + + Returns: + {total, success, failed, fields_total} + """ + stats: dict[str, Any] = { + "total": len(flats), + "success": 0, + "failed": 0, + "fields_total": 0, + } + + if not flats: + logger.info("scrape_catalog_batch: empty batch, nothing to do") + return stats + + logger.info( + "scrape_catalog_batch: starting %d flats region=%d", + len(flats), + region_code, + ) + + async with BrowserSession( + region_code=region_code, + headed=headed, + load_state=load_state, + # auth=None — страницы каталога публичные, Basic auth не нужен + auth=None, + ) as session: + for flat in flats: + ods_id = flat.get("ods_id", "") + catalog_url_hash = flat.get("catalog_url_hash", "") + if not ods_id or not catalog_url_hash: + logger.warning("scrape_catalog_batch: skip flat with missing ods_id/hash: %r", flat) + stats["failed"] += 1 + continue + + outcome = await scrape_one_flat(session, db, ods_id, catalog_url_hash) + if outcome["success"]: + stats["success"] += 1 + stats["fields_total"] += outcome["fields_extracted"] + else: + stats["failed"] += 1 + + logger.info( + "scrape_catalog_batch done: total=%d success=%d failed=%d fields_total=%d", + stats["total"], + stats["success"], + stats["failed"], + stats["fields_total"], + ) + return stats