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 "<!doctype" not in body[:100].lower():
+            logger.warning(
+                "catalog unexpected content-type=%r for hash=%s, treating as HTML anyway",
+                ctype,
+                catalog_url_hash,
+            )
+
+        if not body:
+            raise RuntimeError(f"catalog empty body for hash={catalog_url_hash}")
+
+        return body
+
+    raise RuntimeError(f"catalog html max retries exhausted hash={catalog_url_hash}: {last_err!r}")
+
+
+# ── HTML parsing ──────────────────────────────────────────────────────────────
+
+
+class _TextCollector(HTMLParser):
+    """Минимальный HTMLParser: собирает текстовые блоки с сохранением атрибутов class/data-*.
+
+    Вместо полного DOM строим плоский список (tag, attrs_dict, text) для regex-поиска.
+    Это сознательный trade-off: не нужен beautifulsoup4 — stdlib достаточно для
+    extraction известных структур страницы каталога.
+    """
+
+    def __init__(self) -> 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