feat(scrapers): extract + download DOM.РФ flat plan images (#2440)

domrf_kn_flat_plans было пусто (0 строк) — блокер для #299 (Potrace
production-wiring) и #300 (CubiCasa). domrf_catalog.py уже фетчит SSR-
страницы каталога квартир, где лежит планировка, но _TextCollector
сознательно дропал <img> как void-tag (#1608) — картинка никогда не
извлекалась.

parse_catalog_flat теперь возвращает plan_image_url: primary-стратегия
парсит __NEXT_DATA__ pageProps JSON (Next.js SSR-паттерн), fallback —
голый <img> по class-hint/соседству с текстом "Планировка". _TextCollector
получает аддитивный self.images-сбор ДО void-tag early-return — не трогает
существующий stack/buffer, регрессия по #1608 исключена (тест
test_img_capture_preserves_text_blocks).

download_plan_image_stub (NotImplementedError) заменён на рабочую
async-загрузку через существующий BrowserSession.download_binary
(те же cookies/TLS/throttle), Pillow для width/height, идемпотентна
(skip при уже существующем файле). Двухфазный upsert: URL пишется
первым, затем метаданные после скачивания — частичный прогресс
переживает падение загрузки.

Не заводит новый Celery beat — не подключено ни в один task (домрана
предпосылка issue #2440: scrape_one_flat/scrape_catalog_batch сейчас
нигде не вызываются). Ships inert до отдельного follow-up по wiring.

Refs #2440, #299, #300

backend/app/services/scrapers/domrf_catalog.py

@@ -33,15 +33,18 @@ Wiring (отдельный PR):
 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.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__)
@@ -87,6 +90,51 @@ _STATUS_KW_RE = re.compile(
 # не имеющим отношения к статусу продажи (#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_*.
@@ -217,8 +265,19 @@ class _TextCollector(HTMLParser):
         # Список записей: (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}
@@ -266,6 +325,134 @@ def _find_text_near(
     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.РФ.
 
@@ -277,6 +464,7 @@ def parse_catalog_flat(html: str) -> dict[str, Any]:
       - 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 обёрнуты внутри.
@@ -468,6 +656,14 @@ def parse_catalog_flat(html: str) -> dict[str, Any]:
             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()),
@@ -544,16 +740,63 @@ def upsert_catalog_data(
 # ── 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.
+    """Scrape одной квартиры: fetch HTML → parse → upsert (+ планировка).
 
-    Возвращает dict с результатом: {ods_id, success, fields_extracted, updated}.
-    Ошибки fetch/parse логируются, не бросаются — caller обрабатывает результат.
+    Помимо 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,
@@ -561,6 +804,8 @@ async def scrape_one_flat(
         "success": False,
         "fields_extracted": 0,
         "updated": False,
+        "plan_found": False,
+        "plan_downloaded": False,
         "error": None,
     }
 
@@ -590,6 +835,19 @@ async def scrape_one_flat(
 
     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 при
@@ -617,11 +875,12 @@ async def scrape_catalog_batch(
     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}.
-    Типовой источник: SELECT ods_id, catalog_url_hash FROM domrf_kn_flats
+    `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.
@@ -629,18 +888,25 @@ async def scrape_catalog_batch(
     Использует один 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}
+        {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:
@@ -648,9 +914,10 @@ async def scrape_catalog_batch(
         return stats
 
     logger.info(
-        "scrape_catalog_batch: starting %d flats region=%d",
+        "scrape_catalog_batch: starting %d flats region=%d snapshot_date=%s",
         len(flats),
         region_code,
+        snap,
     )
 
     async with BrowserSession(
@@ -660,20 +927,32 @@ async def scrape_catalog_batch(
         # auth=None — страницы каталога публичные, Basic auth не нужен
         auth=None,
     ) 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)
+            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
 
@@ -689,11 +968,14 @@ async def scrape_catalog_batch(
         raise
 
     logger.info(
-        "scrape_catalog_batch done: total=%d success=%d updated=%d failed=%d fields_total=%d",
+        "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

backend/app/services/scrapers/flat_plans.py

@@ -33,15 +33,36 @@ Issue #297 sub-task 22c.
 
 from __future__ import annotations
 
+import io
 import logging
-from datetime import date
-from typing import Any
+import re
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+from urllib.parse import urlsplit
 
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+if TYPE_CHECKING:
+    from datetime import date
+
+    from app.services.scrapers.stealth import BrowserSession
+
 logger = logging.getLogger(__name__)
 
+# Базовая директория для скачанных бинарников планировок. Зеркалит конвенцию
+# domrf_kn.py:PHOTOS_DIR_DEFAULT (data/raw/domrf_photos) — тот же raw-корень,
+# отдельная папка. Файлы: {base}/{obj_id}/{safe_ods_id}.{ext}.
+FLAT_PLANS_DIR_DEFAULT = Path("data/raw/domrf_flat_plans")
+
+# Разрешённые расширения картинок планировок. png — дефолт (DOM.РФ отдаёт планы
+# растром), svg на всякий случай (векторные схемы встречаются).
+_ALLOWED_PLAN_EXTS = (".png", ".jpg", ".jpeg", ".webp", ".svg")
+
+# Замена небезопасных для имени файла символов в ods_id (напр. "65136/1/1.4.3").
+_ODS_UNSAFE_RE = re.compile(r"[^A-Za-z0-9._-]+")
+
 
 # ── payload parsing ───────────────────────────────────────────────────────────
 
@@ -212,30 +233,131 @@ def upsert_flat_plans(
     return inserted
 
 
-# ── download stub ─────────────────────────────────────────────────────────────
+# ── image download ────────────────────────────────────────────────────────────
 
 
-def download_plan_image_stub(
+def _safe_ods_filename(ods_id: str) -> str:
+    """Превратить ods_id (напр. '65136/1/1.4.3') в безопасное имя файла.
+
+    Слэши и прочие небезопасные символы → '_'. Результат используется как
+    basename в {base_dir}/{obj_id}/{safe}.{ext}.
+    """
+    return _ODS_UNSAFE_RE.sub("_", ods_id).strip("_") or "plan"
+
+
+def _ext_from_url(url: str) -> str:
+    """Определить расширение файла по URL планировки. Default .png.
+
+    DOM.РФ отдаёт планы преимущественно PNG; расширение берём из path, если
+    оно в белом списке, иначе .png (file-API URL /api/ext/file/... часто без
+    расширения — контент-тип уточнится по факту, но png — безопасный дефолт).
+    """
+    path = urlsplit(url).path.lower()
+    for ext in _ALLOWED_PLAN_EXTS:
+        if path.endswith(ext):
+            return ".jpg" if ext == ".jpeg" else ext
+    return ".png"
+
+
+def _image_dims(data: bytes) -> tuple[int | None, int | None]:
+    """Получить (width_px, height_px) картинки через Pillow. None при ошибке.
+
+    Pillow (PIL) уже зависимость репо (pyproject: pillow>=10.4.0, используется в
+    app.services.photos.thumbs). SVG Pillow не парсит — вернёт (None, None),
+    это ОК (размеры для векторных планов необязательны).
+    """
+    try:
+        from PIL import Image
+
+        with Image.open(io.BytesIO(data)) as im:
+            return int(im.width), int(im.height)
+    except Exception as exc:
+        logger.debug("flat plan image dims parse failed: %s", exc)
+        return None, None
+
+
+async def download_plan_image(
+    session: BrowserSession,
     plan_image_url: str,
     ods_id: str,
-    dest_dir: str | None = None,
-) -> str:
-    """Заглушка для скачивания бинарника планировки.
+    obj_id: int,
+    base_dir: Path | None = None,
+) -> dict[str, Any] | None:
+    """Скачать бинарник планировки квартиры и вернуть метаданные для upsert.
 
-    Реальная реализация — отдельный Celery task (22d-track, issue #299).
-    До реализации бросает NotImplementedError, чтобы случайный вызов
-    не прошёл незаметно.
+    Реиспользует BrowserSession.download_binary — тот же паттерн что download_photos
+    в domrf_kn.py: тянет файл через browser context (те же cookies / TLS-fingerprint,
+    что прошли WAF), с лёгким throttle (jitter 200–500 мс встроен в download_binary).
+
+    Файл сохраняется в {base_dir}/{obj_id}/{safe_ods_id}.{ext}. Если файл уже есть
+    на диске — повторно НЕ качаем (idempotent, экономит WAF-бюджет), но всё равно
+    возвращаем метаданные (размеры/размер читаются с диска).
 
     Args:
-        plan_image_url: URL картинки планировки.
+        session:        Активный BrowserSession (bootstrapped + warm_up).
+        plan_image_url: Абсолютный URL картинки планировки.
         ods_id:         Идентификатор квартиры (для имени файла).
-        dest_dir:       Директория для сохранения (None = MEDIA_ROOT/flat_plans/).
+        obj_id:         ID объекта DOM.РФ (подпапка).
+        base_dir:       Корень для сохранения (None = FLAT_PLANS_DIR_DEFAULT).
 
-    Raises:
-        NotImplementedError: всегда — до реализации.
+    Returns:
+        dict с ключами local_path, width_px, height_px, size_bytes, downloaded_at —
+        для передачи в upsert_flat_plans. None при ошибке скачивания.
     """
-    raise NotImplementedError(
-        "download_plan_image_stub: скачивание планировок не реализовано. "
-        "Реальный downloader — отдельный Celery task (issue #299, 22d-track). "
-        f"URL={plan_image_url!r} ods_id={ods_id!r} dest_dir={dest_dir!r}"
+    root = base_dir or FLAT_PLANS_DIR_DEFAULT
+    obj_dir = root / str(obj_id)
+    ext = _ext_from_url(plan_image_url)
+    local = obj_dir / f"{_safe_ods_filename(ods_id)}{ext}"
+
+    # Idempotent: уже скачан — читаем метаданные с диска, не жжём WAF-запрос.
+    if local.exists() and local.stat().st_size > 0:
+        data = local.read_bytes()
+        width_px, height_px = _image_dims(data)
+        logger.debug("flat plan already on disk ods_id=%s path=%s", ods_id, local)
+        return {
+            "local_path": str(local),
+            "width_px": width_px,
+            "height_px": height_px,
+            "size_bytes": local.stat().st_size,
+            "downloaded_at": datetime.now(UTC),
+        }
+
+    try:
+        data = await session.download_binary(plan_image_url)
+    except Exception as exc:
+        logger.warning(
+            "download_plan_image failed ods_id=%s obj_id=%s url=%s: %s",
+            ods_id,
+            obj_id,
+            plan_image_url,
+            exc,
+        )
+        return None
+
+    if not data:
+        logger.warning("download_plan_image empty body ods_id=%s url=%s", ods_id, plan_image_url)
+        return None
+
+    obj_dir.mkdir(parents=True, exist_ok=True)
+    try:
+        local.write_bytes(data)
+    except OSError as exc:
+        logger.warning("download_plan_image write failed ods_id=%s path=%s: %s", ods_id, local, exc)
+        return None
+
+    width_px, height_px = _image_dims(data)
+    logger.info(
+        "download_plan_image ok ods_id=%s obj_id=%s bytes=%d dims=%sx%s",
+        ods_id,
+        obj_id,
+        len(data),
+        width_px,
+        height_px,
     )
+    return {
+        "local_path": str(local),
+        "width_px": width_px,
+        "height_px": height_px,
+        "size_bytes": len(data),
+        "downloaded_at": datetime.now(UTC),
+    }

backend/tests/services/scrapers/test_domrf_flat_plan_url.py

@@ -0,0 +1,211 @@
+"""Тесты для extraction plan_image_url из страницы квартиры DOM.РФ (issue #2440).
+
+Покрывает:
+  - __NEXT_DATA__ pageProps (primary): URL плана под plan-hint ключом.
+  - bare <img> fallback: plan-hint в атрибутах / proximity к «Планировка».
+  - нормализация URL (относительный → absolute; data:-inline / шум отбрасываются).
+  - интеграция с parse_catalog_flat: ключ plan_image_url в результате.
+  - _TextCollector НЕ ломает text-extraction при наличии <img> (регрессия #1608).
+
+Фикстуры инлайновые (минимальный реалистичный HTML), без live-fetch — DOM.РФ
+каталог-квартир это Next.js SSR (тот же стек что страница объекта в
+domrf_catalog_object.py, где план живёт в __NEXT_DATA__, а не в <img src>).
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from app.services.scrapers.domrf_catalog import (
+    _normalize_plan_url,
+    _TextCollector,
+    extract_plan_image_url,
+    parse_catalog_flat,
+)
+from app.services.scrapers.stealth import BASE_URL
+
+# Абсолютный file-API URL DOM.РФ (реальный паттерн, см. documents.py:17).
+PLAN_ABS_URL = f"{BASE_URL}/api/ext/file/plan-12345.png"
+# Относительный next/image URL (Next.js image optimizer).
+PLAN_REL_NEXT = "/_next/image?url=%2Fapi%2Fext%2Ffile%2Fplan-777.png&w=1200&q=75"
+
+
+def _html_with_next_data(page_props: dict) -> str:
+    """SSR HTML с __NEXT_DATA__ блоком, содержащим pageProps."""
+    blob = json.dumps({"props": {"pageProps": page_props}})
+    return (
+        "<!doctype html><html><head>"
+        f'<script id="__NEXT_DATA__" type="application/json">{blob}</script>'
+        "</head><body>"
+        "<h1>Квартира 2-комнатная</h1>"
+        "<div>4 500 000 ₽</div>"
+        "</body></html>"
+    )
+
+
+def _collector_for(html: str) -> _TextCollector:
+    c = _TextCollector()
+    c.feed(html)
+    return c
+
+
+# ── __NEXT_DATA__ primary path ────────────────────────────────────────────────
+
+
+def test_plan_from_next_data_plan_image_key() -> None:
+    """URL под ключом planImage в pageProps → извлекается (absolute)."""
+    html = _html_with_next_data({"planImage": PLAN_ABS_URL, "buildingClass": "Комфорт"})
+    assert extract_plan_image_url(html, _collector_for(html)) == PLAN_ABS_URL
+
+
+def test_plan_from_next_data_layout_key_nested() -> None:
+    """URL под вложенным ключом layoutImageUrl → извлекается (BFS обход)."""
+    html = _html_with_next_data(
+        {"flat": {"info": {"layoutImageUrl": PLAN_ABS_URL}}, "price": 4_500_000}
+    )
+    assert extract_plan_image_url(html, _collector_for(html)) == PLAN_ABS_URL
+
+
+def test_plan_from_next_data_relative_url_normalized() -> None:
+    """Относительный next/image URL → абсолютный через BASE_URL."""
+    html = _html_with_next_data({"floorPlanUrl": PLAN_REL_NEXT})
+    got = extract_plan_image_url(html, _collector_for(html))
+    assert got is not None
+    assert got.startswith(BASE_URL)
+    assert "plan-777" in got
+
+
+def test_plan_from_next_data_ignores_non_plan_keys() -> None:
+    """Картинка под НЕ-plan ключом (developerLogo) не должна ловиться."""
+    html = _html_with_next_data(
+        {"developerLogo": f"{BASE_URL}/api/ext/file/logo.png", "buildingClass": "Бизнес"}
+    )
+    assert extract_plan_image_url(html, _collector_for(html)) is None
+
+
+def test_plan_from_next_data_ignores_noise_url() -> None:
+    """plan-ключ, но URL содержит logo/thumb (шум) → отбрасывается."""
+    html = _html_with_next_data({"planThumb": f"{BASE_URL}/api/ext/file/plan-thumb.png"})
+    # 'thumb' в URL → _IMAGE_NOISE_RE → None
+    assert extract_plan_image_url(html, _collector_for(html)) is None
+
+
+# ── bare <img> fallback ───────────────────────────────────────────────────────
+
+
+def test_plan_from_img_class_hint() -> None:
+    """Нет __NEXT_DATA__: <img class='flat-plan__image'> → извлекается по class."""
+    html = (
+        "<!doctype html><html><body>"
+        f'<img class="flat-plan__image" alt="" src="{PLAN_ABS_URL}">'
+        "</body></html>"
+    )
+    assert extract_plan_image_url(html, _collector_for(html)) == PLAN_ABS_URL
+
+
+def test_plan_from_img_src_hint() -> None:
+    """<img> без plan-класса, но src содержит 'plan' → извлекается."""
+    html = f'<!doctype html><html><body><img src="{PLAN_ABS_URL}"></body></html>'
+    assert extract_plan_image_url(html, _collector_for(html)) == PLAN_ABS_URL
+
+
+def test_plan_from_img_proximity_to_label() -> None:
+    """<img> без plan-hint в атрибутах, но рядом с блоком «Планировка»."""
+    plain = f"{BASE_URL}/api/ext/file/imgabc.png"
+    html = (
+        "<!doctype html><html><body>"
+        "<div>Планировка</div>"
+        f'<img alt="" src="{plain}">'
+        "</body></html>"
+    )
+    got = extract_plan_image_url(html, _collector_for(html))
+    assert got == plain
+
+
+def test_plan_from_img_skips_logo_and_captcha() -> None:
+    """Лого/captcha <img> игнорируются, даже если стоят первыми."""
+    html = (
+        "<!doctype html><html><body>"
+        f'<img class="header-logo" src="{BASE_URL}/api/ext/file/logo.png">'
+        '<img class="robot" src="data:image/jpeg;base64,AAAA">'
+        "<div>Планировка квартиры</div>"
+        f'<img class="plan-view" src="{PLAN_ABS_URL}">'
+        "</body></html>"
+    )
+    assert extract_plan_image_url(html, _collector_for(html)) == PLAN_ABS_URL
+
+
+def test_no_plan_returns_none() -> None:
+    """Нет ни __NEXT_DATA__ плана, ни plan-img → None."""
+    html = (
+        "<!doctype html><html><body>"
+        f'<img class="header-logo" src="{BASE_URL}/api/ext/file/logo.png">'
+        "<p>Площадь 45 м²</p>"
+        "</body></html>"
+    )
+    assert extract_plan_image_url(html, _collector_for(html)) is None
+
+
+# ── URL normalization ─────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "raw,expected_prefix",
+    [
+        (PLAN_ABS_URL, PLAN_ABS_URL),
+        ("/api/ext/file/plan.png", f"{BASE_URL}/api/ext/file/plan.png"),
+        ("//cdn.example/plan.png", "https://cdn.example/plan.png"),
+    ],
+)
+def test_normalize_plan_url_variants(raw: str, expected_prefix: str) -> None:
+    assert _normalize_plan_url(raw) == expected_prefix
+
+
+@pytest.mark.parametrize("raw", ["", None, "data:image/png;base64,AAAA", "   "])
+def test_normalize_plan_url_rejects_junk(raw: str | None) -> None:
+    assert _normalize_plan_url(raw) is None
+
+
+# ── integration: parse_catalog_flat ───────────────────────────────────────────
+
+
+def test_parse_catalog_flat_includes_plan_image_url() -> None:
+    """parse_catalog_flat кладёт plan_image_url в результат при наличии плана."""
+    html = _html_with_next_data({"planImage": PLAN_ABS_URL, "price": 4_500_000})
+    result = parse_catalog_flat(html)
+    assert result.get("plan_image_url") == PLAN_ABS_URL
+
+
+def test_parse_catalog_flat_omits_plan_when_absent() -> None:
+    """Нет плана → ключ plan_image_url отсутствует (а не None)."""
+    html = "<!doctype html><html><body><p>Площадь 45 м²</p></body></html>"
+    result = parse_catalog_flat(html)
+    assert "plan_image_url" not in result
+
+
+# ── regression: <img> capture must not break text-extraction (#1608) ──────────
+
+
+def test_img_capture_preserves_text_blocks() -> None:
+    """Наличие <img> (void-тег) не рассинхронизирует стек — текст блоков цел."""
+    html = (
+        "<!doctype html><html><body>"
+        '<div class="price-block">4 500 000 ₽</div>'
+        f'<img src="{PLAN_ABS_URL}">'
+        '<div class="status-badge">Продана</div>'
+        "</body></html>"
+    )
+    collector = _collector_for(html)
+    texts = [t for _cls, t in collector.blocks]
+    assert any("4 500 000" in t for t in texts)
+    assert any("Продана" in t for t in texts)
+    # и картинка захвачена в отдельный список
+    assert len(collector.images) == 1
+    assert collector.images[0][0].get("src") == PLAN_ABS_URL
+
+    # parse_catalog_flat всё ещё корректно извлекает status/price при наличии img
+    result = parse_catalog_flat(html)
+    assert result.get("status") == "sold"
+    assert result.get("price_rub") == 4_500_000

backend/tests/services/scrapers/test_flat_plan_download.py

@@ -0,0 +1,136 @@
+"""Тесты для download_plan_image() из flat_plans.py (issue #2440).
+
+Покрывает:
+  - успешное скачивание: файл на диске, width/height/size_bytes через Pillow.
+  - idempotency: повторный вызов не качает заново (download_binary не дёргается).
+  - ошибка скачивания → None, файл не создаётся.
+  - _safe_ods_filename / _ext_from_url helpers.
+
+Использует фейковый BrowserSession с in-memory PNG (Pillow генерит 3x2 картинку) —
+без сети и без реального Playwright.
+"""
+
+from __future__ import annotations
+
+import io
+from pathlib import Path
+from typing import Any
+
+import pytest
+from PIL import Image
+
+from app.services.scrapers.flat_plans import (
+    _ext_from_url,
+    _safe_ods_filename,
+    download_plan_image,
+)
+
+
+def _png_bytes(w: int = 3, h: int = 2) -> bytes:
+    buf = io.BytesIO()
+    Image.new("RGB", (w, h), (10, 20, 30)).save(buf, format="PNG")
+    return buf.getvalue()
+
+
+class _FakeSession:
+    """Минимальный BrowserSession-дублёр: download_binary отдаёт заданные байты."""
+
+    def __init__(self, payload: bytes | None, raise_exc: Exception | None = None) -> None:
+        self._payload = payload
+        self._raise = raise_exc
+        self.calls = 0
+
+    async def download_binary(self, url: str) -> bytes:
+        self.calls += 1
+        if self._raise is not None:
+            raise self._raise
+        assert self._payload is not None
+        return self._payload
+
+
+# ── helpers ───────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "ods,expected",
+    [
+        ("65136/1/1.4.3", "65136_1_1.4.3"),
+        ("plain", "plain"),
+        ("a//b", "a_b"),
+        ("///", "plan"),
+    ],
+)
+def test_safe_ods_filename(ods: str, expected: str) -> None:
+    assert _safe_ods_filename(ods) == expected
+
+
+@pytest.mark.parametrize(
+    "url,expected",
+    [
+        ("https://x/plan.png", ".png"),
+        ("https://x/plan.JPG", ".jpg"),
+        ("https://x/plan.jpeg", ".jpg"),
+        ("https://x/api/ext/file/abc", ".png"),  # no ext → default png
+        ("https://x/plan.svg", ".svg"),
+    ],
+)
+def test_ext_from_url(url: str, expected: str) -> None:
+    assert _ext_from_url(url) == expected
+
+
+# ── download ──────────────────────────────────────────────────────────────────
+
+
+def _read(path: str) -> bytes:
+    """Sync file read — вынесено из async-теста, чтобы не тревожить ASYNC240."""
+    return Path(path).read_bytes()
+
+
+def _exists(path: Path) -> bool:
+    return path.exists()
+
+
+async def test_download_writes_file_and_dims(tmp_path: Path) -> None:
+    png = _png_bytes(3, 2)
+    sess: Any = _FakeSession(png)
+    meta = await download_plan_image(
+        sess, "https://x/api/ext/file/plan1.png", "65136/1/1.4.3", 65136, base_dir=tmp_path
+    )
+    assert meta is not None
+    assert meta["width_px"] == 3
+    assert meta["height_px"] == 2
+    assert meta["size_bytes"] == len(png)
+    assert meta["downloaded_at"] is not None
+    p = Path(meta["local_path"])
+    assert _read(str(p)) == png
+    # путь: {base}/{obj_id}/{safe_ods}.png
+    assert p.parent.name == "65136"
+    assert p.name == "65136_1_1.4.3.png"
+
+
+async def test_download_idempotent_skips_second_fetch(tmp_path: Path) -> None:
+    png = _png_bytes(4, 4)
+    sess: Any = _FakeSession(png)
+    m1 = await download_plan_image(sess, "https://x/plan.png", "1/2/3", 42, base_dir=tmp_path)
+    assert m1 is not None
+    assert sess.calls == 1
+    # второй вызов: файл уже на диске → download_binary НЕ дёргается
+    m2 = await download_plan_image(sess, "https://x/plan.png", "1/2/3", 42, base_dir=tmp_path)
+    assert m2 is not None
+    assert sess.calls == 1  # не увеличилось
+    assert m2["local_path"] == m1["local_path"]
+    assert m2["size_bytes"] == len(png)
+
+
+async def test_download_failure_returns_none(tmp_path: Path) -> None:
+    sess: Any = _FakeSession(None, raise_exc=RuntimeError("binary http 403"))
+    meta = await download_plan_image(sess, "https://x/plan.png", "9/9/9", 99, base_dir=tmp_path)
+    assert meta is None
+    # файл не создан
+    assert not _exists(tmp_path / "99")
+
+
+async def test_download_empty_body_returns_none(tmp_path: Path) -> None:
+    sess: Any = _FakeSession(b"")
+    meta = await download_plan_image(sess, "https://x/plan.png", "9/9/9", 99, base_dir=tmp_path)
+    assert meta is None