gendesign/backend/app/services/scrapers/domrf_catalog.py

"""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 json
import logging
import re
from datetime import date
from html.parser import HTMLParser
from typing import Any
from urllib.parse import urljoin

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.services.scrapers.flat_plans import download_plan_image, upsert_flat_plans
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"

# Паттерны для извлечения статуса (issue #1609).
# Все морфоварианты: продан/продана/продано, забронирован[аоы]?, реализован[аоы]?.
_STATUS_KW_RE = re.compile(
    r"(в\s*продаже|свободн[аоы]?|free"
    r"|продан[аоы]?|реализован[аоы]?|sold"
    r"|забронирован[аоы]?|бронь|reserved)",
    re.IGNORECASE | re.UNICODE,
)
# CSS-классы, характерные для статус-бейджей в Next.js/Tailwind.
# Требуем слово «status» как отдельный токен внутри hyphen-separated CSS-класса
# (напр. «status-badge», «flat-status-tag», «object-status»).
# Предыдущий паттерн «status|badge|tag|chip|label» был слишком широк —
# срабатывал на любой generic элемент с классом «tag»/«chip»/«label»/«badge»,
# не имеющим отношения к статусу продажи (#1609 follow-up, #1686).
_STATUS_BADGE_CLS_RE = re.compile(r"(?<![a-z])status(?![a-z])", re.IGNORECASE)

# ── plan-image extraction ─────────────────────────────────────────────────────
#
# Страница квартиры каталога DOM.РФ — Next.js SSR (тот же стек что и страница
# объекта, см. domrf_catalog_object.py). План квартиры рендерится через next/image,
# поэтому реальный URL картинки живёт в JSON-блоке <script id="__NEXT_DATA__">
# внутри pageProps, а НЕ в готовом <img src>. Стратегия extraction (в порядке
# приоритета, best-effort — DOM.РФ может переименовать ключи без уведомления):
#
#   1. __NEXT_DATA__ pageProps: ищем значение под одним из кандидат-ключей
#      (planImage / planImageUrl / layoutImage / floorPlan / scheme / image / photo).
#   2. Fallback: bare <img>, чей src/alt/class намекают на планировку
#      (plan / layout / scheme / планировк), либо <img> рядом с текстом «Планировка».
#
# URL нормализуется в абсолютный через urljoin(BASE_URL, ...): DOM.РФ отдаёт файлы
# и абсолютными (https://xn--80az8a.../api/ext/file/...  — см. documents.py), и
# относительными (/_next/image?url=..., /api/ext/file/...). Мусорные картинки
# (data:-inline base64, captcha, лого застройщика, миниатюры) отсекаются.

# Ключи в pageProps, под которыми может лежать URL планировки. Проверяются по
# порядку; первое непустое строковое значение с image-like URL выигрывает.
_NEXT_PLAN_KEY_RE = re.compile(
    r"(?:plan|layout|scheme|планировк|floor)",
    re.IGNORECASE,
)

# Признак «это картинка» в значении: расширение файла ИЛИ file-API DOM.РФ.
_IMAGE_URL_RE = re.compile(
    r"(?:\.(?:png|jpe?g|webp|svg)(?:\?|$)|/api/ext/file/|/_next/image)",
    re.IGNORECASE,
)

# Токены, по которым отсекаем НЕ-планировочные картинки (лого, аватары, иконки,
# captcha, миниатюры, банеры). Проверяются по src/alt/class.
_IMAGE_NOISE_RE = re.compile(
    r"(?:logo|avatar|icon|captcha|robot|sprite|banner|placeholder|thumb|"
    r"favicon|badge|marker|pin)",
    re.IGNORECASE,
)

# Признак «это планировка» в атрибутах <img> (src / alt / class).
_IMAGE_PLAN_HINT_RE = re.compile(
    r"(?:plan|layout|scheme|планировк)",
    re.IGNORECASE,
)


def _classify_status_kw(matched_text: str) -> str | None:
    """Классифицировать совпавшее ключевое слово в STATUS_*.

    sold/reserved проверяются ПЕРВЫМИ — они специфичнее "в продаже" и никогда
    не появляются в навигационных блоках страницы, в отличие от «в продаже».
    """
    s = matched_text.lower()
    if any(kw in s for kw in ("продан", "реализован", "sold")):
        return STATUS_SOLD
    if any(kw in s for kw in ("бронь", "забронирован", "reserved")):
        return STATUS_RESERVED
    if any(kw in s for kw in ("продаже", "свободн", "free")):
        return STATUS_FREE
    return None


# ── 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 известных структур страницы каталога.
    """

    # HTML5 void-элементы: не имеют закрывающего тега → handle_endtag не вызывается.
    # Если пушить их в стек/буфер, чужой handle_endtag поп'ает чужой фрейм →
    # рассинхрон стека, текст блоков теряется (issue #1608).
    _VOID_TAGS = frozenset(
        {
            "area",
            "base",
            "br",
            "col",
            "embed",
            "hr",
            "img",
            "input",
            "link",
            "meta",
            "param",
            "source",
            "track",
            "wbr",
        }
    )

    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] = []
        # Отдельный, аддитивный сбор <img>-атрибутов (issue #2440): нужен для
        # extraction plan_image_url как fallback к __NEXT_DATA__. НЕ влияет на
        # text-extraction — img остаётся void-тегом, фрейм в стек не пушится.
        # Формат: (attrs_dict, index_в_blocks) — index привязывает картинку к
        # позиции в потоке текстовых блоков для proximity-поиска «Планировка».
        self.images: list[tuple[dict[str, str], int]] = []

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        if tag == "img":
            # Собираем атрибуты img ДО early-return (void-тег): index = текущая
            # длина blocks, т.е. картинка «привязана» к ближайшему предыдущему
            # закрытому текстовому блоку. Text-extraction ниже не меняется.
            self.images.append(({k: (v or "") for k, v in attrs}, len(self.blocks)))
        if tag in self._VOID_TAGS:
            return  # void-теги не имеют endtag — не открываем фрейм (issue #1608)
        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 _tag in self._VOID_TAGS:
            return  # void-теги фрейм не открывали — нечего поп'ать (issue #1608)
        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 _normalize_plan_url(url: str | None) -> str | None:
    """Нормализовать URL картинки планировки в абсолютный, отсеять мусор.

    - data:-inline (base64) отбрасываем (это не удалённый файл, часто captcha/robot).
    - Относительный путь ('/api/ext/file/...', '/_next/image?url=...') → absolute
      через urljoin(BASE_URL, ...).
    - Пустое/None → None.
    """
    if not url:
        return None
    u = url.strip()
    if not u or u.lower().startswith("data:"):
        return None
    if u.startswith("//"):
        # protocol-relative → https
        return "https:" + u
    if u.startswith("http://") or u.startswith("https://"):
        return u
    return urljoin(BASE_URL + "/", u.lstrip("/"))


def _extract_plan_from_next_data(html: str) -> str | None:
    """Извлечь URL планировки из <script id="__NEXT_DATA__"> pageProps.

    Страница квартиры каталога — Next.js SSR: план рендерится через next/image,
    URL картинки живёт в JSON-блоке. Рекурсивно обходим pageProps и ищем строковое
    значение под ключом, матчащим _NEXT_PLAN_KEY_RE, чьё значение выглядит как
    image-URL (_IMAGE_URL_RE) и НЕ является шумом (_IMAGE_NOISE_RE).

    Возвращает первый подходящий URL (нормализованный) или None.
    """
    match = re.search(
        r'<script\s+id=["\']__NEXT_DATA__["\'][^>]*>(.+?)</script>',
        html,
        re.DOTALL,
    )
    if not match:
        return None
    try:
        blob = json.loads(match.group(1).strip())
    except (json.JSONDecodeError, ValueError):
        return None

    page_props = blob.get("props", {}).get("pageProps")
    root: Any = page_props if isinstance(page_props, dict) else blob

    # BFS по вложенному dict/list; ключ+значение проверяем на plan-hint.
    stack: list[Any] = [root]
    seen = 0
    while stack and seen < 20_000:  # cap: защита от патологически глубокого JSON
        node = stack.pop()
        seen += 1
        if isinstance(node, dict):
            for key, value in node.items():
                if (
                    isinstance(value, str)
                    and _NEXT_PLAN_KEY_RE.search(str(key))
                    and _IMAGE_URL_RE.search(value)
                    and not _IMAGE_NOISE_RE.search(value)
                ):
                    normalized = _normalize_plan_url(value)
                    if normalized:
                        return normalized
                if isinstance(value, dict | list):
                    stack.append(value)
        elif isinstance(node, list):
            stack.extend(node)
    return None


def _extract_plan_from_imgs(
    images: list[tuple[dict[str, str], int]], blocks: list[tuple[str, str]]
) -> str | None:
    """Fallback: выбрать <img> планировки из собранных _TextCollector.images.

    Приоритет:
      1. <img>, чей src/alt/class содержит plan/layout/scheme/планировк — прямое
         совпадение, самый надёжный сигнал.
      2. <img> рядом (в потоке DOM) с текстовым блоком «Планировка».
    Шумовые картинки (лого/иконки/captcha/миниатюры) отсекаются по _IMAGE_NOISE_RE,
    inline data:-картинки и пустые src — через _normalize_plan_url.
    """
    # Уровень 1: явный plan-hint в атрибутах.
    for attrs, _idx in images:
        src = attrs.get("src", "")
        alt = attrs.get("alt", "")
        cls = attrs.get("class", "")
        haystack = f"{src} {alt} {cls}"
        if _IMAGE_NOISE_RE.search(haystack):
            continue
        if _IMAGE_PLAN_HINT_RE.search(haystack):
            normalized = _normalize_plan_url(src)
            if normalized:
                return normalized

    # Уровень 2: <img>, чей index стоит сразу после блока «Планировка».
    plan_block_indices = [
        i for i, (_cls, text) in enumerate(blocks) if re.search(r"планировк", text, re.IGNORECASE)
    ]
    if plan_block_indices:
        for attrs, idx in images:
            src = attrs.get("src", "")
            if _IMAGE_NOISE_RE.search(f"{src} {attrs.get('alt', '')} {attrs.get('class', '')}"):
                continue
            # картинка «принадлежит» блоку «Планировка», если её index в пределах
            # ±2 текстовых блоков от лейбла.
            if any(abs(idx - pbi) <= 2 for pbi in plan_block_indices):
                normalized = _normalize_plan_url(src)
                if normalized:
                    return normalized
    return None


def extract_plan_image_url(html: str, collector: _TextCollector) -> str | None:
    """Извлечь URL картинки планировки из SSR-HTML страницы квартиры.

    Two-strategy (issue #2440):
      1. __NEXT_DATA__ pageProps (primary — DOM.РФ Next.js SSR).
      2. bare <img> fallback (по plan-hint атрибутам либо proximity к «Планировка»).

    Возвращает абсолютный URL или None. Best-effort — структура DOM.РФ хрупкая.
    """
    url = _extract_plan_from_next_data(html)
    if url:
        return url
    return _extract_plan_from_imgs(collector.images, collector.blocks)


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) Дата обновления информации на странице
      - plan_image_url (str)      URL картинки планировки (issue #2440)

    Парсинг хрупкий по природе (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 руб"
    # Negative lookahead (?!\s*[/⁄]) исключает цену за м² ("217 835 ₽/м²"),
    # которая на странице обычно выше полной стоимости и иначе матчилась бы
    # первой → отбрасывалась санити-границей → price_rub=NULL (issue #1645).
    price_match = re.search(
        r"([\d][\d\s]{3,12}[\d])\s*(?:₽|руб)(?!\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 extraction (issue #1609 fix).
    #
    # Проблема: re.search по всему HTML находит "в продаже" из навигации
    # ("другие квартиры в продаже") РАНЬШЕ, чем реальный статус-бейдж →
    # проданная/забронированная квартира ошибочно маркируется как free.
    #
    # Стратегия (3 уровня, от точного к широкому):
    #
    # 1. Блоки с CSS-классом, содержащим "status"/"badge"/"tag"/"chip"/"label" —
    #    типовые имена для статус-бейджей в Next.js/Tailwind.
    # 2. Ближайший блок ПОСЛЕ лейбла "Статус" (proximity-поиск в blocks).
    # 3. Полный перебор всех blocks: sold/reserved побеждают free
    #    (они никогда не появляются в навигации, в отличие от "в продаже").
    #
    # Морфоварианты: продан / продана / продано / продан →  продан[аоы]?
    # Аналогично: забронирован[аоы]? / реализован[аоы]? (синоним для sold).

    # Уровень 1: блок с CSS-классом типа badge/status/tag
    status_from_badge: str | None = None
    for cls, block_text in blocks:
        if _STATUS_BADGE_CLS_RE.search(cls):
            m = _STATUS_KW_RE.search(block_text)
            if m:
                status_from_badge = _classify_status_kw(m.group(1))
                if status_from_badge:
                    break

    if status_from_badge:
        result["status"] = status_from_badge
    else:
        # Уровень 2: блок сразу после лейбла "Статус"
        status_label_value = _find_text_near(blocks, r"^статус$")
        status_from_label: str | None = None
        if status_label_value:
            m2 = _STATUS_KW_RE.search(status_label_value)
            if m2:
                status_from_label = _classify_status_kw(m2.group(1))

        if status_from_label:
            result["status"] = status_from_label
        else:
            # Уровень 3: перебор всех blocks — sold/reserved побеждают free.
            # Если встретили sold/reserved — сразу break (не появляются в nav).
            # Если только free — запоминаем как кандидата (может быть перебит).
            free_candidate: bool = False
            sold_reserved_found: str | None = None
            for _cls, block_text in blocks:
                m3 = _STATUS_KW_RE.search(block_text)
                if not m3:
                    continue
                classified = _classify_status_kw(m3.group(1))
                if classified in (STATUS_SOLD, STATUS_RESERVED):
                    sold_reserved_found = classified
                    break  # точнее nav-текстов, дальше не ищем
                if classified == STATUS_FREE:
                    free_candidate = True  # продолжаем — вдруг sold встретится позже
            if sold_reserved_found:
                result["status"] = sold_reserved_found
            elif free_candidate:
                result["status"] = STATUS_FREE

    # 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

    # ── Шаг 4: plan_image_url (issue #2440) ──────────────────────────────────
    # __NEXT_DATA__ pageProps (primary) → bare <img> fallback. collector.images
    # собран на Шаге 1 тем же feed(). None если план не найден (напр. проданная
    # квартира без планировки на странице).
    plan_url = extract_plan_image_url(html, collector)
    if plan_url:
        result["plan_image_url"] = plan_url

    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 _handle_plan_image(
    db: Session,
    session: BrowserSession,
    ods_id: str,
    obj_id: int,
    plan_url: str,
    snapshot_date: date,
    outcome: dict[str, Any],
) -> None:
    """Записать URL планировки в flat_plans, скачать бинарник, дописать метаданные.

    Порядок (issue #2440):
      1. upsert_flat_plans с одним только plan_image_url — URL зафиксирован даже
         если скачивание упадёт (частичный прогресс не теряется).
      2. download_plan_image — reuse BrowserSession (cookies/TLS/throttle).
      3. повторный upsert_flat_plans с local_path/размерами — COALESCE внутри
         upsert защищает уже скачанные файлы от перезаписи.

    Мутирует outcome['plan_downloaded']. Ошибки логируются внутри download/upsert,
    наружу не пробрасываются — планировка не должна ронять catalog-scrape квартиры.
    """
    plan_row: dict[str, Any] = {
        "ods_id": ods_id,
        "obj_id": obj_id,
        "plan_image_url": plan_url,
    }
    upsert_flat_plans(db, obj_id, [plan_row], snapshot_date)

    meta = await download_plan_image(session, plan_url, ods_id, obj_id)
    if meta:
        outcome["plan_downloaded"] = True
        upsert_flat_plans(db, obj_id, [{**plan_row, **meta}], snapshot_date)


async def scrape_one_flat(
    session: BrowserSession,
    db: Session,
    ods_id: str,
    catalog_url_hash: str,
    obj_id: int | None = None,
    snapshot_date: date | None = None,
) -> dict[str, Any]:
    """Scrape одной квартиры: fetch HTML → parse → upsert (+ планировка).

    Помимо catalog-полей (price/status/finishing/...) извлекает plan_image_url
    (issue #2440): пишет URL в domrf_kn_flat_plans через upsert_flat_plans, затем
    скачивает бинарник (download_plan_image) и повторным upsert'ом дописывает
    local_path/размеры (COALESCE-защита уже скачанных файлов внутри upsert).

    Скачивание — inline в рамках того же BrowserSession: reuse cookies/TLS,
    download_binary несёт свой лёгкий throttle (jitter 200–500 мс), отдельная
    beat-задача не заводится (требование #2440). obj_id/snapshot_date нужны для
    записи в flat_plans; если не переданы — plan-upsert пропускается с warning.

    Возвращает dict с результатом: {ods_id, success, fields_extracted, updated,
    plan_found, plan_downloaded}. Ошибки fetch/parse/download логируются, не
    бросаются — caller обрабатывает результат.
    """
    outcome: dict[str, Any] = {
        "ods_id": ods_id,
        "catalog_url_hash": catalog_url_hash,
        "success": False,
        "fields_extracted": 0,
        "updated": False,
        "plan_found": False,
        "plan_downloaded": 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)

    # ── plan_image_url → domrf_kn_flat_plans (issue #2440) ───────────────────
    plan_url = data.get("plan_image_url")
    if plan_url:
        outcome["plan_found"] = True
        if obj_id is not None and snapshot_date is not None:
            await _handle_plan_image(db, session, ods_id, obj_id, plan_url, snapshot_date, outcome)
        else:
            logger.warning(
                "catalog scrape ods_id=%s: plan_image_url found but obj_id/snapshot_date "
                "missing — flat_plans upsert skipped",
                ods_id,
            )
    # success отражает прохождение пайплайна fetch+parse без исключения; реально
    # ли затронута строка в БД — см. outcome['updated']. Батч-статистика считает
    # отдельный stats['updated'], чтобы не рапортовать ложно высокий success при
    # ненайденном ods_id / пустом парсе (fields_extracted==0) — issue #1610.
    outcome["success"] = True
    if not outcome["updated"]:
        logger.warning(
            "catalog scrape ods_id=%s: fetched+parsed but DB row NOT updated "
            "(fields=%d, ods_id missing or all-NULL parse)",
            ods_id,
            outcome["fields_extracted"],
        )
    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,
    snapshot_date: date | None = None,
) -> dict[str, Any]:
    """Scrape пачки квартир каталога DOM.РФ.

    `flats` — список dict'ов с ключами {ods_id, catalog_url_hash [, obj_id]}.
    Типовой источник: SELECT ods_id, obj_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).

    obj_id/snapshot_date нужны для записи планировки в domrf_kn_flat_plans (issue
    #2440). snapshot_date по умолчанию = сегодня (дата первичного обнаружения URL).

    Returns:
        {total, success, updated, failed, fields_total, plans_found, plans_downloaded}
        - success: прошли fetch+parse без исключения
        - updated: реально затронули строку в БД (issue #1610) — отражает
          фактическое число записанных квартир, в отличие от success
        - plans_found / plans_downloaded: планировки найдены / скачаны (#2440)
    """
    snap = snapshot_date or date.today()
    stats: dict[str, Any] = {
        "total": len(flats),
        "success": 0,
        "updated": 0,
        "failed": 0,
        "fields_total": 0,
        "plans_found": 0,
        "plans_downloaded": 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 snapshot_date=%s",
        len(flats),
        region_code,
        snap,
    )

    async with BrowserSession(
        region_code=region_code,
        headed=headed,
        load_state=load_state,
        # auth=None — страницы каталога публичные, Basic auth не нужен
        auth=None,
        # #2445 D2 anti-ban: тот же /сервисы/* path family, что вызвал DOM.РФ WAF
        # hard-ban 2026-05-24 (#2443). Без явного override BrowserSession тихо
        # наследует модульный дефолт stealth.py (concurrency=8, jitter 600-1500ms).
        # Переиспользуем throttled настройки, принятые scrape_kn.py после #1945
        # (settings.scrape_kn_* — общий anti-ban рычаг, НЕ специфичный для KN-sweep,
        # несмотря на имя).
        concurrency=settings.scrape_kn_browser_concurrency,
        jitter_min_ms=settings.scrape_kn_request_jitter_min_ms,
        jitter_max_ms=settings.scrape_kn_request_jitter_max_ms,
    ) as session:
        # Warm-up: WAF DOM.РФ (2026-05-24) требует session cookies для SSR-страниц.
        # Зеркалит scrape_catalog_objects. Idempotent на весь batch.
        await session.warm_up()

        for flat in flats:
            ods_id = flat.get("ods_id", "")
            catalog_url_hash = flat.get("catalog_url_hash", "")
            obj_id_raw = flat.get("obj_id")
            obj_id = int(obj_id_raw) if obj_id_raw is not None else None
            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, obj_id=obj_id, snapshot_date=snap
            )
            if outcome["success"]:
                stats["success"] += 1
                stats["fields_total"] += outcome["fields_extracted"]
                if outcome["updated"]:
                    stats["updated"] += 1
                if outcome.get("plan_found"):
                    stats["plans_found"] += 1
                if outcome.get("plan_downloaded"):
                    stats["plans_downloaded"] += 1
            else:
                stats["failed"] += 1

    # Commit outer transaction: SAVEPOINT (`begin_nested`) в upsert_catalog_data
    # лишь RELEASE'ит внутренний SAVEPOINT, но outer tx остаётся autobegin'd —
    # без commit() все UPDATE'ы (цены, отделка, catalog_updated_at) откатятся
    # при db.close() в Celery task. Зеркалит scrape_catalog_objects в
    # domrf_catalog_object.py:460. Issue #1227.
    try:
        db.commit()
    except Exception:
        db.rollback()
        raise

    logger.info(
        "scrape_catalog_batch done: total=%d success=%d updated=%d failed=%d "
        "fields_total=%d plans_found=%d plans_downloaded=%d",
        stats["total"],
        stats["success"],
        stats["updated"],
        stats["failed"],
        stats["fields_total"],
        stats["plans_found"],
        stats["plans_downloaded"],
    )
    return stats