tradein-mvp/backend/app/services/scheduler.py

@@ -89,6 +89,17 @@ Sources:
                               city_sweep; fetch via curl_cffi chrome120 + scraper_proxy_url,
                               parse via YandexDetailScraper.parse;
                               window 12:00-15:00 UTC — после avito_detail_backfill 09-12 UTC)
+  - domclick_detail_backfill → run_domclick_detail_backfill
+                              (tasks/domclick_detail_backfill.py, Layer B #1846; nightly backfill
+                              of domclick (source='domklik') detail_enriched_at. Layer A SERP sweep
+                              collects only base fields; the card page (behind DataDome →
+                              BrowserFetcher transport ONLY, curl_cffi → 401) carries
+                              renovation/area/balconies/encumbrances/priceHistory. ONE
+                              BrowserFetcher session per run; block-abort → mark_done (transient),
+                              parse-failure → rollback+retry next run. Shipped DORMANT (seed 139,
+                              enabled=false) — enable after prod-smoke; depends on Layer A having
+                              populated source='domklik' listings. window 15:00-18:00 UTC — после
+                              avito (09-12) и yandex (12-15) detail backfills)
   - cadastral_geo_match     → run_cadastral_geo_match
                               (tasks/cadastral_geo_match.py; combined nightly job — (1) REFRESH
                               cad_buildings_local from gendesign_cad_buildings FDW (one bulk
@@ -1554,6 +1565,45 @@ async def trigger_yandex_detail_backfill_run(
     return run_id
 
 
+async def trigger_domclick_detail_backfill_run(
+    db: Session, schedule_row: dict[str, Any]
+) -> int | None:
+    """Create scrape_run + launch run_domclick_detail_backfill in asyncio.create_task.
+
+    Layer B detail-enrichment for legacy domclick listings (source='domklik'). Layer A
+    (domclick_city_sweep / #1979) collects only base SERP fields; the card page (behind
+    DataDome → BrowserFetcher only, curl_cffi → 401) carries renovation/repair_state,
+    living/kitchen area, balconies, sale_type, owners_count, encumbrances, year_built,
+    views, priceHistory.
+
+    Mirrors trigger_yandex_detail_backfill_run: claim run -> create_task ->
+    mark_done/failed delegated to tasks/domclick_detail_backfill.run_domclick_detail_backfill.
+
+    Returns run_id or None (skip -- already running).
+    """
+    run_id = _claim_run(db, schedule_row)
+    if run_id is None:
+        return None
+
+    params = schedule_row.get("default_params") or {}
+
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            from app.tasks.domclick_detail_backfill import run_domclick_detail_backfill
+
+            await run_domclick_detail_backfill(run_db, run_id=run_id, params=params)
+        except Exception:
+            logger.exception("scheduler: run_domclick_detail_backfill crashed run_id=%d", run_id)
+        finally:
+            run_db.close()
+
+    task = asyncio.create_task(_run())
+    task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+    logger.info("scheduler: triggered domclick_detail_backfill run_id=%d", run_id)
+    return run_id
+
+
 async def trigger_cadastral_geo_match_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
     """Создать scrape_runs + launch run_cadastral_geo_match в executor (sync DB-only task).
 
@@ -1795,6 +1845,8 @@ async def scheduler_loop() -> None:
                         await trigger_avito_detail_backfill_run(db, sch)
                     elif source == "yandex_detail_backfill":
                         await trigger_yandex_detail_backfill_run(db, sch)
+                    elif source == "domclick_detail_backfill":
+                        await trigger_domclick_detail_backfill_run(db, sch)
                     elif source == "cadastral_geo_match":
                         await trigger_cadastral_geo_match_run(db, sch)
                     elif source == "house_imv_backfill":

tradein-mvp/backend/app/services/scrapers/domclick_detail.py

@@ -0,0 +1,584 @@
+"""DomClick.ru Layer B — detail-card enrichment (#1846 follow-up).
+
+Layer A (domclick.py) собирает базовые поля через BFF JSON API. Layer B дочитывает
+страницу карточки (``https://ekaterinburg.domclick.ru/card/sale__flat__<id>``) и
+обогащает listings: renovation/repair_state, living/kitchen area, balconies,
+sale_type, owners_count, encumbrances, year_built, views и priceHistory.
+
+Транспорт — BrowserFetcher через QRATOR-clean прокси (source="cian"): card-хосты
+DomClick заблокированы QRATOR через domclick/avito-прокси, но чисты через cian
+(verified live 2026-06-27). curl_cffi на card-хосте отдаёт 401/стаб. fetch_detail
+принимает уже открытый browser_fetcher; жизненным циклом сессии и выбором source
+владеет оркестратор.
+
+SSR-quirk
+---------
+HTML карточки встраивает ``window.__SSR_STATE__ = {...}`` — это JS object literal,
+а НЕ чистый JSON: в value-позиции встречается bare ``undefined`` (json.loads падает).
+_extract_ssr_state делает balanced-brace scan (с пропуском скобок внутри строк) и
+заменяет bare ``undefined`` → ``null`` строго в value-позиции, не трогая строки.
+
+Форма SSR-стейта подтверждена на живой карточке 2075729321 (2026-06-27):
+productCard.objectInfo.{renovation,livingArea,kitchenArea}, productCard.priceInfo.
+priceHistory ({date ISO8601+tz, price, diff, state}), productCard.egrnData
+(snake_case owners_count/collateral/collateral_sber), productCard.legalOptions.
+saleType, productCard.viewsCount/callsCount, houseInfo.info.{wallType,floorType},
+и ТОП-УРОВНЕМ pricePrediction (DomClick AVM → raw_extra.avm).
+Schedule для Layer B ships DORMANT — включается вручную после prod-smoke.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services.scrapers.domclick_exceptions import DomClickBlockedError, DomClickParseError
+from app.services.scrapers.repair_state_normalizer import (
+    infer_repair_state_from_text,
+    normalize_repair_state,
+)
+
+if TYPE_CHECKING:
+    from app.services.scrapers.browser_fetcher import BrowserFetcher
+
+logger = logging.getLogger(__name__)
+
+# ── Маркеры anti-bot challenge (DataDome / QRATOR) ───────────────────────────
+# Зеркало идеи Layer A: если __SSR_STATE__ отсутствует И страница похожа на
+# challenge — это блок, а не parse-failure.
+_BLOCK_MARKERS: tuple[str, ...] = ("qrator", "captcha", "datadome", "access denied")
+
+# ── item_id из URL карточки: .../card/sale__flat__2075729321 ─────────────────
+_ITEM_ID_RE = re.compile(r"sale__flat__(\d+)")
+
+# ── SSR assignment marker ────────────────────────────────────────────────────
+_SSR_ASSIGN_RE = re.compile(r"window\.__SSR_STATE__\s*=\s*")
+
+# ── bare `undefined` (value-позиция) → null, не трогая строки ─────────────────
+_UNDEFINED_RE = re.compile(r"(?<=[:\[,\s])undefined(?=[\s,\]\}])")
+
+# ── renovation (RU) → каноничный repair_state enum ───────────────────────────
+# ЛОКАЛЬНАЯ карта (не трогаем shared _RAW_TO_ENUM в repair_state_normalizer).
+# Значения уже каноничны; прогоняются через normalize_repair_state для валидации.
+_RENOVATION_MAP: dict[str, str] = {
+    "евроремонт": "good",
+    "косметический": "standard",
+    "дизайнерский": "excellent",
+    "черновая": "needs_repair",
+    "без отделки": "needs_repair",
+    "требует ремонта": "needs_repair",
+    "предчистовая": "needs_repair",
+}
+
+
+# ── DomClickDetailEnrichment ──────────────────────────────────────────────────
+@dataclass
+class DomClickDetailEnrichment:
+    """Результат парсинга detail-карточки DomClick (Layer B).
+
+    save_detail_enrichment() пишет подмножество в listings (+ offer_price_history).
+    Все поля None-safe: отсутствующее → None (не крэш).
+    """
+
+    item_id: str | None = None
+    source_url: str | None = None
+
+    repair_state: str | None = None  # enum: needs_repair/standard/good/excellent
+    repair_type: str | None = None  # сырое RU-значение renovation
+    living_area_m2: float | None = None
+    kitchen_area_m2: float | None = None
+    balconies_count: int | None = None
+    has_balcony: bool | None = None
+    sale_type: str | None = None
+    owners_count: int | None = None
+    encumbrances_clean: bool | None = None  # True=чисто, False=есть обременение
+    year_built: int | None = None
+    views_total: int | None = None
+
+    price_changes: list[dict[str, Any]] = field(default_factory=list)
+    raw_extra: dict[str, Any] = field(default_factory=dict)
+
+
+# ── SSR state extraction ──────────────────────────────────────────────────────
+
+
+def _scan_balanced_object(s: str, start: int) -> str:
+    """Сканирует от '{' по индексу ``start`` до парной '}'.
+
+    Отслеживает глубину по {/}, ИГНОРИРУЯ скобки внутри JS-строк (одинарные и
+    двойные кавычки с ``\\``-эскейпами). Возвращает литерал ``{...}`` включительно.
+
+    Raises:
+        DomClickParseError: если парная скобка не найдена (truncated HTML).
+    """
+    depth = 0
+    in_string = False
+    quote_char = ""
+    escaped = False
+    for i in range(start, len(s)):
+        ch = s[i]
+        if in_string:
+            if escaped:
+                escaped = False
+            elif ch == "\\":
+                escaped = True
+            elif ch == quote_char:
+                in_string = False
+            continue
+        if ch in ('"', "'"):
+            in_string = True
+            quote_char = ch
+        elif ch == "{":
+            depth += 1
+        elif ch == "}":
+            depth -= 1
+            if depth == 0:
+                return s[start : i + 1]
+    raise DomClickParseError("__SSR_STATE__ object not balanced (no matching brace)")
+
+
+def _extract_ssr_state(html: str) -> dict[str, Any]:
+    """Извлекает ``window.__SSR_STATE__`` object-литерал из HTML карточки.
+
+    Шаги:
+      1. Находим присваивание ``window.__SSR_STATE__ =``. Если нет — различаем
+         challenge-страницу (→ DomClickBlockedError) и просто отсутствие
+         стейта (→ DomClickParseError).
+      2. Balanced-brace scan от первого '{' → литерал.
+      3. Санитайз bare ``undefined`` → ``null`` (только value-позиция).
+      4. json.loads.
+
+    Raises:
+        DomClickBlockedError: страница без стейта + маркеры anti-bot.
+        DomClickParseError: стейт не найден / не сбалансирован / json.loads упал.
+    """
+    match = _SSR_ASSIGN_RE.search(html)
+    if match is None:
+        html_lower = html.lower()
+        if any(marker in html_lower for marker in _BLOCK_MARKERS):
+            raise DomClickBlockedError(
+                "DomClick detail: challenge page detected (no __SSR_STATE__, "
+                f"markers checked: {_BLOCK_MARKERS})"
+            )
+        raise DomClickParseError("__SSR_STATE__ not found")
+
+    brace_start = html.find("{", match.end())
+    if brace_start == -1:
+        raise DomClickParseError("__SSR_STATE__ assignment has no opening '{'")
+
+    literal = _scan_balanced_object(html, brace_start)
+    sanitized = _UNDEFINED_RE.sub("null", literal)
+    try:
+        data = json.loads(sanitized)
+    except json.JSONDecodeError as exc:
+        raise DomClickParseError(
+            f"__SSR_STATE__ json.loads failed: {exc} (head={sanitized[:200]!r})"
+        ) from exc
+    if not isinstance(data, dict):
+        raise DomClickParseError(f"__SSR_STATE__ is not an object: {type(data)}")
+    return data
+
+
+# ── Value coercion helpers ────────────────────────────────────────────────────
+
+
+def _to_float(value: Any) -> float | None:
+    """None-safe float. bool/неконвертируемое → None."""
+    if value is None or isinstance(value, bool):
+        return None
+    try:
+        return float(value)
+    except (ValueError, TypeError):
+        return None
+
+
+def _to_int(value: Any) -> int | None:
+    """None-safe int. bool/неконвертируемое → None."""
+    if value is None or isinstance(value, bool):
+        return None
+    try:
+        return int(value)
+    except (ValueError, TypeError):
+        return None
+
+
+def _parse_change_time(value: Any) -> datetime | None:
+    """Нормализует дату изменения цены → timezone-aware datetime.
+
+    Адаптация Layer A ``_parse_publish_date`` (которая возвращала date): здесь
+    поддерживаем epoch (сек/мс), числовую строку и ISO8601. Real priceHistory.date
+    приходит как ``2026-04-15T09:02:26.548728+03:00`` — fromisoformat СОХРАНЯЕТ
+    исходный offset (+03:00), НЕ конвертирует в UTC; naive-значения получают UTC.
+    epoch трактуем как UTC. Невалидное → None (caller пропускает запись).
+    """
+    if value is None or isinstance(value, bool):
+        return None
+    # epoch (число): >1e12 трактуем как миллисекунды
+    if isinstance(value, int | float):
+        ts = float(value)
+        if ts > 1e12:
+            ts /= 1000.0
+        try:
+            return datetime.fromtimestamp(ts, tz=UTC)
+        except (ValueError, OSError, OverflowError):
+            return None
+    s = str(value).strip()
+    if not s:
+        return None
+    if s.isdigit():
+        ts = float(s)
+        if ts > 1e12:
+            ts /= 1000.0
+        try:
+            return datetime.fromtimestamp(ts, tz=UTC)
+        except (ValueError, OSError, OverflowError):
+            return None
+    # ISO8601 (поддержим trailing 'Z')
+    try:
+        dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
+    except (ValueError, TypeError):
+        return None
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=UTC)
+    return dt
+
+
+def _map_repair_state(renovation: str | None) -> str | None:
+    """RU renovation → каноничный enum. Нет совпадения в карте → инференс из текста."""
+    if not renovation:
+        return None
+    mapped = _RENOVATION_MAP.get(renovation.strip().lower())
+    if mapped is not None:
+        return normalize_repair_state(mapped)
+    return infer_repair_state_from_text(renovation)
+
+
+def _compact(d: dict[str, Any]) -> dict[str, Any]:
+    """Выкидывает None-значения, чтобы raw_extra оставался компактным."""
+    return {k: v for k, v in d.items() if v is not None}
+
+
+def _pick(d: dict[str, Any], *keys: str) -> Any:
+    """Первое не-None значение среди ключей (defensive snake/camel spellings)."""
+    for k in keys:
+        v = d.get(k)
+        if v is not None:
+            return v
+    return None
+
+
+# ── item_id ───────────────────────────────────────────────────────────────────
+
+
+def _extract_item_id(url: str) -> str | None:
+    """sale__flat__<digits> из URL карточки."""
+    m = _ITEM_ID_RE.search(url or "")
+    return m.group(1) if m else None
+
+
+# ── parse_detail_html ──────────────────────────────────────────────────────────
+
+
+def parse_detail_html(html: str, source_url: str) -> DomClickDetailEnrichment:
+    """Pure parser detail-карточки DomClick (без сети/БД).
+
+    _extract_ssr_state может бросить DomClickBlockedError/DomClickParseError —
+    они propagate (оркестратор различает блок и parse-failure). Дальнейшая
+    навигация по полям обёрнута: per-field сюрприз формы → минимальный
+    enrichment (item_id+url), а не крэш.
+    """
+    state = _extract_ssr_state(html)
+    item_id = _extract_item_id(source_url)
+    try:
+        pc = state.get("productCard") or {}
+        oi = pc.get("objectInfo") or {}
+        hi = (state.get("houseInfo") or {}).get("info") or {}
+        egrn = pc.get("egrnData") or {}
+
+        # repair: enum + сырое значение
+        renovation = oi.get("renovation")
+        repair_type = renovation if isinstance(renovation, str) else None
+        repair_state = _map_repair_state(repair_type)
+
+        # площади
+        living_area_m2 = _to_float(oi.get("livingArea"))
+        kitchen_area_m2 = _to_float(oi.get("kitchenArea"))
+
+        # балконы: int → count+флаг; иное (строка/др.) → raw_extra, counts None
+        balconies = oi.get("balconies")
+        balconies_count: int | None = None
+        has_balcony: bool | None = None
+        balconies_raw: Any = None
+        if isinstance(balconies, bool):
+            balconies_raw = balconies
+        elif isinstance(balconies, int):
+            balconies_count = balconies
+            has_balcony = balconies > 0
+        elif balconies is not None:
+            balconies_raw = balconies
+
+        sale_type = (pc.get("legalOptions") or {}).get("saleType")
+
+        # owners_count — защищаемся от двух написаний ключа
+        owners_count = _to_int(egrn.get("owners_count"))
+        if owners_count is None:
+            owners_count = _to_int(egrn.get("ownersCount"))
+
+        # encumbrances_clean (ИНВЕРСИЯ): collateral/collateral_sber обозначают
+        # НАЛИЧИЕ обременения. Любой truthy → clean=False; оба отсутствуют/false
+        # при наличии egrnData → True; egrnData нет → None.
+        collateral = egrn.get("collateral")
+        collateral_sber = egrn.get("collateral_sber")
+        if egrn:
+            encumbrances_clean: bool | None = not bool(collateral or collateral_sber)
+        else:
+            encumbrances_clean = None
+
+        year_built = _to_int(hi.get("buildYear"))
+        views_total = _to_int(pc.get("viewsCount"))
+
+        # priceHistory → нормализованные записи (skip при битой дате / без цены)
+        raw_history = (pc.get("priceInfo") or {}).get("priceHistory") or []
+        price_changes: list[dict[str, Any]] = []
+        if isinstance(raw_history, list):
+            for entry in raw_history:
+                if not isinstance(entry, dict):
+                    continue
+                change_time = _parse_change_time(entry.get("date"))
+                price_rub = _to_int(entry.get("price"))
+                # Skip при нераспарсенной дате ИЛИ отсутствующей/невалидной цене.
+                if change_time is None or price_rub is None:
+                    continue
+                diff = entry.get("diff")
+                diff_percent = (
+                    diff if isinstance(diff, int | float) and not isinstance(diff, bool) else None
+                )
+                price_changes.append(
+                    {
+                        "change_time": change_time,
+                        "price_rub": price_rub,
+                        "diff_percent": diff_percent,
+                    }
+                )
+
+        # AVM (Layer C, тот же fetch) — DomClick price prediction живёт ТОП-УРОВНЕМ
+        # в SSR-стейте. Может быть плоским или обёрнут в .result/.data — defensively
+        # разворачиваем. В raw_extra пишем только присутствующие поля.
+        avm_root = state.get("pricePrediction") or {}
+        avm_src = avm_root
+        if isinstance(avm_root, dict):
+            avm_src = avm_root.get("result") or avm_root.get("data") or avm_root
+        if not isinstance(avm_src, dict):
+            avm_src = {}
+        avm = _compact(
+            {
+                "market_price": _pick(avm_src, "market_price", "marketPrice"),
+                "min": _pick(avm_src, "min_market_price", "minMarketPrice"),
+                "max": _pick(avm_src, "max_market_price", "maxMarketPrice"),
+                "repair_quality": _pick(avm_src, "repair_quality", "repairQuality"),
+            }
+        )
+
+        # raw_extra → merge в listings.raw_payload. wall/floor type ОСТАЮТСЯ тут,
+        # НЕ пишем в listings.house_type (кросс-источниковый словарь грязный).
+        # egrn area key — точное написание не подтверждено; пробуем несколько.
+        egrn_area = egrn.get("area") or egrn.get("rosreestrArea") or egrn.get("object_area")
+        demand = _compact(
+            {
+                "calls": pc.get("callsCount"),
+                "favorites": pc.get("favoriteOfferUsersCount"),
+                "duplicates": pc.get("duplicatesOfferCount"),
+            }
+        )
+        raw_extra = _compact(
+            {
+                "wall_type": hi.get("wallType") or (pc.get("house") or {}).get("wallType"),
+                "floor_type": hi.get("floorType"),
+                "entrance_count": hi.get("entranceCount"),
+                "quarters_count": hi.get("quartersCount"),
+                "energy_efficiency": hi.get("energyEfficiency"),
+                "building_series": hi.get("buildingSeries"),
+                "domclick_building_guid": (pc.get("address") or {}).get("guid"),
+                "egrn_area": egrn_area,
+                "demand": demand or None,
+                "collateral": collateral,
+                "collateral_sber": collateral_sber,
+                "balconies_raw": balconies_raw,
+                "avm": avm or None,
+            }
+        )
+
+        return DomClickDetailEnrichment(
+            item_id=item_id,
+            source_url=source_url,
+            repair_state=repair_state,
+            repair_type=repair_type,
+            living_area_m2=living_area_m2,
+            kitchen_area_m2=kitchen_area_m2,
+            balconies_count=balconies_count,
+            has_balcony=has_balcony,
+            sale_type=sale_type if isinstance(sale_type, str) else None,
+            owners_count=owners_count,
+            encumbrances_clean=encumbrances_clean,
+            year_built=year_built,
+            views_total=views_total,
+            price_changes=price_changes,
+            raw_extra=raw_extra,
+        )
+    except Exception:
+        logger.warning(
+            "domclick_detail: field navigation failed for %s — minimal enrichment",
+            source_url,
+            exc_info=True,
+        )
+        return DomClickDetailEnrichment(item_id=item_id, source_url=source_url)
+
+
+# ── fetch_detail ────────────────────────────────────────────────────────────────
+
+
+async def fetch_detail(
+    card_url: str,
+    *,
+    browser_fetcher: BrowserFetcher,
+) -> DomClickDetailEnrichment:
+    """Fetch абсолютного URL карточки через BrowserFetcher → parse_detail_html.
+
+    card_url уже абсолютный (listings.source_url из Layer A), URL не строим.
+    Сессией владеет оркестратор (передаёт уже открытый browser_fetcher).
+
+    Raises:
+        DomClickBlockedError: anti-bot challenge ИЛИ ошибка браузерного fetch
+            (нет curl-fallback — DataDome даёт 401 на curl). Оркестратор считает
+            это блоком и помечает listing failed.
+        DomClickParseError: HTTP 200, но SSR-стейт не парсится (дрейф схемы).
+    """
+    try:
+        html = await browser_fetcher.fetch(card_url)
+    except DomClickBlockedError:
+        raise
+    except Exception as exc:
+        raise DomClickBlockedError(
+            f"DomClick detail browser fetch failed for {card_url}: {exc}"
+        ) from exc
+    return parse_detail_html(html, card_url)
+
+
+# ── save_detail_enrichment ──────────────────────────────────────────────────────
+
+
+def save_detail_enrichment(
+    db: Session,
+    listing_id: int,
+    e: DomClickDetailEnrichment,
+) -> bool:
+    """Persist enrichment в одной транзакции (commit в конце; rollback — за caller).
+
+    1. UPDATE listings SET <cols> (COALESCE — None не затирает существующее),
+       raw_payload || raw_extra, detail_enriched_at=NOW() WHERE id=listing_id.
+    2. Для каждой записи priceHistory — INSERT INTO offer_price_history
+       (ON CONFLICT DO NOTHING → идемпотентно). source='domklik'.
+
+    Snapshot (upsert_listing_snapshot) НЕ пишем: enrichment не несёт текущую цену,
+    а detail-backfill цену не меняет — snapshot остаётся за SERP-путём (note).
+
+    Returns:
+        True если UPDATE задел строку (listing найден), иначе False.
+    """
+    raw_extra_json = json.dumps(e.raw_extra or {}, ensure_ascii=False, default=str)
+
+    result = db.execute(
+        text("""
+            UPDATE listings SET
+                repair_state = COALESCE(:repair_state, repair_state),
+                repair_type = COALESCE(:repair_type, repair_type),
+                living_area_m2 = COALESCE(:living_area_m2, living_area_m2),
+                kitchen_area_m2 = COALESCE(:kitchen_area_m2, kitchen_area_m2),
+                balconies_count = COALESCE(:balconies_count, balconies_count),
+                has_balcony = COALESCE(:has_balcony, has_balcony),
+                sale_type = COALESCE(:sale_type, sale_type),
+                owners_count = COALESCE(:owners_count, owners_count),
+                encumbrances_clean = COALESCE(:encumbrances_clean, encumbrances_clean),
+                year_built = COALESCE(:year_built, year_built),
+                views_total = COALESCE(:views_total, views_total),
+                raw_payload = COALESCE(raw_payload, '{}'::jsonb) || CAST(:raw_extra AS jsonb),
+                detail_enriched_at = NOW()
+            WHERE id = CAST(:lid AS bigint)
+        """),
+        {
+            "lid": listing_id,
+            "repair_state": e.repair_state,
+            "repair_type": e.repair_type,
+            "living_area_m2": e.living_area_m2,
+            "kitchen_area_m2": e.kitchen_area_m2,
+            "balconies_count": e.balconies_count,
+            "has_balcony": e.has_balcony,
+            "sale_type": e.sale_type,
+            "owners_count": e.owners_count,
+            "encumbrances_clean": e.encumbrances_clean,
+            "year_built": e.year_built,
+            "views_total": e.views_total,
+            "raw_extra": raw_extra_json,
+        },
+    )
+
+    # priceHistory — INSERT, де-дуп по (listing_id, change_time). SAVEPOINT per-row
+    # (backend.md): битый cast одной записи не валит весь UPDATE+commit.
+    for change in e.price_changes:
+        ct = change.get("change_time")
+        price = change.get("price_rub")
+        if not ct or not price:
+            continue
+        try:
+            with db.begin_nested():
+                db.execute(
+                    text("""
+                        INSERT INTO offer_price_history
+                            (listing_id, change_time, price_rub, diff_percent, source)
+                        VALUES (
+                            CAST(:lid AS bigint),
+                            CAST(:ct AS timestamptz),
+                            CAST(:price AS numeric),
+                            CAST(:diff AS numeric),
+                            'domklik'
+                        )
+                        ON CONFLICT ON CONSTRAINT offer_price_history_listing_change_uq
+                        DO NOTHING
+                    """),
+                    {
+                        "lid": listing_id,
+                        "ct": ct,
+                        "price": price,
+                        "diff": change.get("diff_percent"),
+                    },
+                )
+        except Exception as exc:
+            logger.warning(
+                "domclick_detail: price-history insert failed listing_id=%s ct=%s: %s",
+                listing_id,
+                ct,
+                exc,
+            )
+
+    db.commit()
+    found = result.rowcount > 0
+    if not found:
+        logger.warning(
+            "domclick_detail save: listing not found id=%s (item_id=%s)",
+            listing_id,
+            e.item_id,
+        )
+    else:
+        logger.info(
+            "domclick_detail save: listing id=%s enriched (price_changes=%d)",
+            listing_id,
+            len(e.price_changes),
+        )
+    return found

tradein-mvp/backend/app/services/scrapers/domclick_exceptions.py

@@ -11,3 +11,17 @@ class DomClickBlockedError(Exception):
     Обходится через shared mobile proxy (BrowserFetcher(source="domclick") →
     generic provider → мобильный egress).
     """
+
+
+class DomClickParseError(Exception):
+    """DomClick detail-карточка получена (HTTP 200, не block), но SSR-стейт
+    не извлекается / не парсится.
+
+    Возникает в Layer B (domclick_detail.py) когда:
+    - `window.__SSR_STATE__` не найден в HTML (но это и не challenge-страница);
+    - object-литерал не сбалансирован по скобкам;
+    - json.loads падает после санитайза bare `undefined` → `null`.
+
+    В отличие от DomClickBlockedError (anti-bot), это сигнал дрейфа схемы карточки
+    или повреждённого ответа — оркестратор считает его parse-failure, а не блоком.
+    """

tradein-mvp/backend/app/tasks/domclick_detail_backfill.py

@@ -0,0 +1,272 @@
+"""Scheduled backfill: DomClick Layer B detail-enrichment (#1846 follow-up).
+
+Nightly window 15:00-18:00 UTC (migration 139, source=domclick_detail_backfill).
+Offset after avito (09-12) / yandex (12-15) detail backfills + the city sweeps to
+avoid parallel egress on the shared mobile proxy.
+
+Problem: Layer A (domclick.py BFF sweep) собирает только базовые SERP-поля. Поля
+карточки (renovation/repair_state, living/kitchen area, balconies, sale_type,
+owners_count, encumbrances, year_built, views, priceHistory) живут только на
+странице карточки (.../card/sale__flat__<id>), за DataDome.
+
+Solution: single snapshot SELECT at start (guarantees termination). Транспорт —
+BrowserFetcher(source="cian"): card-хосты DomClick заблокированы QRATOR через
+domclick/avito-прокси, но чисты через cian-прокси (verified live 2026-06-27).
+Одна браузер-сессия открывается на весь прогон и передаётся в fetch_detail.
+Block-handling: DomClickBlockedError транзиентен (anti-bot / fetch-fail)
+→ abort после max_consecutive_blocks с mark_done (НЕ mark_failed; retry следующей ночью
+через NULL detail_enriched_at). DomClickParseError (HTTP 200, но SSR не парсится) →
+листинг остаётся необогащённым (rollback), переотбирается на следующем прогоне.
+
+Schedule ships DORMANT (seed 139, enabled=false) — enable manually after prod-smoke.
+Depends on Layer A (#1979 / domclick_city_sweep) having populated source='domklik'
+listings first.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass, field
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services import scrape_runs as runs_mod
+from app.services.scrapers.browser_fetcher import BrowserFetcher
+from app.services.scrapers.domclick_detail import fetch_detail, save_detail_enrichment
+from app.services.scrapers.domclick_exceptions import DomClickBlockedError, DomClickParseError
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "DomClickDetailBackfillResult",
+    "run_domclick_detail_backfill",
+]
+
+
+@dataclass
+class DomClickDetailBackfillResult:
+    """Counters for one domclick detail backfill run."""
+
+    attempted: int = 0
+    enriched: int = 0
+    blocked: int = 0
+    failed: int = 0
+    duration_sec: float = field(default=0.0)
+
+    def to_dict(self) -> dict[str, int]:
+        return {
+            "attempted": self.attempted,
+            "enriched": self.enriched,
+            "blocked": self.blocked,
+            "failed": self.failed,
+            "duration_sec": int(self.duration_sec),
+        }
+
+
+async def run_domclick_detail_backfill(
+    db: Session,
+    *,
+    run_id: int,
+    params: dict,
+) -> DomClickDetailBackfillResult:
+    """Backfill detail_enriched_at for legacy domclick listings via BrowserFetcher.
+
+    Params (from default_params jsonb in scrape_schedules):
+        batch_size: int -- snapshot size (SELECT LIMIT), default 300.
+        budget_sec: float -- wall-clock budget per run, default 3600s.
+        request_delay_sec: float -- delay between listings, default 8.0s.
+        max_consecutive_blocks: int -- consecutive blocks before abort, default 5.
+
+    Transport: ONE BrowserFetcher(source="cian") for the whole run (DomClick card
+    hosts are QRATOR-clean only via the cian proxy). fetch_detail receives the
+    already-open fetcher.
+
+    Lifecycle: update_heartbeat -> cancel-check -> snapshot -> loop with budget guard ->
+    mark_done (incl. partial / block-abort) / mark_failed (outer exception only).
+    """
+    batch_size = int(params.get("batch_size", 300))
+    budget_sec = float(params.get("budget_sec", 3600))
+    request_delay_sec = float(params.get("request_delay_sec", 8.0))
+    max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5))
+
+    counters = DomClickDetailBackfillResult()
+    current_counters: dict[str, int] = counters.to_dict()
+
+    start = time.monotonic()
+
+    try:
+        runs_mod.update_heartbeat(db, run_id, current_counters)
+
+        # Cooperative cancel before any browser launch / SELECT. The run is already in
+        # terminal 'cancelled' state (that is how is_cancelled detects it) — do NOT
+        # mark_done (would overwrite 'cancelled'), just return early.
+        if runs_mod.is_cancelled(db, run_id):
+            logger.info(
+                "domclick_detail_backfill: run_id=%d cancelled before loop -- returning early",
+                run_id,
+            )
+            return counters
+
+        # SNAPSHOT: single SELECT at start -- NOT re-selected in loop (termination guarantee).
+        snapshot = (
+            db.execute(
+                text(
+                    """
+                    SELECT id, source_id, source_url
+                    FROM listings
+                    WHERE source = 'domklik'
+                      AND detail_enriched_at IS NULL
+                      AND source_url IS NOT NULL
+                      AND is_active = TRUE
+                    ORDER BY scraped_at DESC NULLS LAST
+                    LIMIT CAST(:batch_size AS int)
+                    """
+                ),
+                {"batch_size": batch_size},
+            )
+            .mappings()
+            .all()
+        )
+
+        if not snapshot:
+            logger.info(
+                "domclick_detail_backfill: run_id=%d -- no pending listings "
+                "(detail_enriched_at IS NULL = 0), done",
+                run_id,
+            )
+            runs_mod.mark_done(db, run_id, current_counters)
+            return counters
+
+        logger.info(
+            "domclick_detail_backfill: run_id=%d snapshot=%d (budget=%.0fs "
+            "delay=%.1fs max_blocks=%d)",
+            run_id,
+            len(snapshot),
+            budget_sec,
+            request_delay_sec,
+            max_consecutive_blocks,
+        )
+
+        consecutive_blocks = 0
+        do_sleep = False
+
+        # ONE browser session for the whole run (mirror Layer A domclick.py fetch_city).
+        # DomClick card/offer-card hosts are QRATOR-blocked via the domclick/avito
+        # proxies but clean via the cian proxy (verified live 2026-06-27); route detail
+        # fetches through it.
+        async with BrowserFetcher(source="cian") as bf:
+            for idx, row in enumerate(snapshot):
+                # Budget guard
+                elapsed = time.monotonic() - start
+                if elapsed > budget_sec:
+                    logger.info(
+                        "domclick_detail_backfill: run_id=%d -- budget %.0fs exhausted "
+                        "(elapsed=%.1fs), stopping at #%d/%d",
+                        run_id,
+                        budget_sec,
+                        elapsed,
+                        idx,
+                        len(snapshot),
+                    )
+                    break
+
+                # Delay before each request except the first
+                if do_sleep:
+                    await asyncio.sleep(request_delay_sec)
+                do_sleep = True
+
+                listing_id: int = row["id"]
+                source_url: str = row["source_url"]
+                counters.attempted += 1
+
+                try:
+                    enrichment = await fetch_detail(source_url, browser_fetcher=bf)
+                    if save_detail_enrichment(db, listing_id, enrichment):
+                        counters.enriched += 1
+                    consecutive_blocks = 0
+
+                except DomClickBlockedError as exc:
+                    # Anti-bot challenge / browser fetch failure. Transient — keep the
+                    # inter-request delay (no IP rotation here, unlike avito) and abort
+                    # the whole run after too many in a row (retried next night).
+                    consecutive_blocks += 1
+                    counters.blocked += 1
+                    logger.warning(
+                        "domclick_detail_backfill: run_id=%d BLOCKED #%d/%d (consecutive=%d): %s",
+                        run_id,
+                        idx + 1,
+                        len(snapshot),
+                        consecutive_blocks,
+                        exc,
+                    )
+                    if consecutive_blocks >= max_consecutive_blocks:
+                        logger.error(
+                            "domclick_detail_backfill: run_id=%d ABORT -- %d consecutive blocks "
+                            "(anti-bot). enriched=%d attempted=%d",
+                            run_id,
+                            consecutive_blocks,
+                            counters.enriched,
+                            counters.attempted,
+                        )
+                        break
+
+                except DomClickParseError as exc:
+                    # HTTP 200 but SSR unparseable (schema drift). Do NOT mark the listing
+                    # enriched -> rollback so it is retried next run.
+                    counters.failed += 1
+                    logger.warning(
+                        "domclick_detail_backfill: run_id=%d listing_id=%d parse failure: %s",
+                        run_id,
+                        listing_id,
+                        exc,
+                    )
+                    try:
+                        db.rollback()
+                    except Exception:
+                        pass
+
+                except Exception as exc:
+                    counters.failed += 1
+                    logger.warning(
+                        "domclick_detail_backfill: run_id=%d listing_id=%d failed: %s",
+                        run_id,
+                        listing_id,
+                        exc,
+                    )
+                    try:
+                        db.rollback()
+                    except Exception:
+                        pass
+
+                if counters.attempted % 25 == 0:
+                    current_counters = counters.to_dict()
+                    runs_mod.update_heartbeat(db, run_id, current_counters)
+
+        counters.duration_sec = time.monotonic() - start
+        current_counters = counters.to_dict()
+        runs_mod.update_heartbeat(db, run_id, current_counters)
+        runs_mod.mark_done(db, run_id, current_counters)
+        logger.info(
+            "domclick_detail_backfill: run_id=%d DONE -- attempted=%d enriched=%d "
+            "blocked=%d failed=%d duration=%.1fs",
+            run_id,
+            counters.attempted,
+            counters.enriched,
+            counters.blocked,
+            counters.failed,
+            counters.duration_sec,
+        )
+        return counters
+
+    except Exception as exc:
+        counters.duration_sec = time.monotonic() - start
+        logger.exception(
+            "domclick_detail_backfill: run_id=%d FAILED after %.1fs",
+            run_id,
+            counters.duration_sec,
+        )
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters.to_dict())
+        raise

tradein-mvp/backend/data/sql/139_seed_domclick_detail_backfill.sql

@@ -0,0 +1,64 @@
+-- 139_seed_domclick_detail_backfill.sql
+-- Seed row для DomClick Layer B detail-backfill — периодическое обогащение карточек.
+--
+-- !!! DORMANT BY DESIGN !!! enabled = false.
+--   Layer A (domclick_city_sweep, #1979) собирает только базовые SERP-поля. Layer B
+--   дочитывает страницу карточки (.../card/sale__flat__<id>) и обогащает listings:
+--   renovation/repair_state, living/kitchen area, balconies, sale_type, owners_count,
+--   encumbrances, year_built, views, priceHistory.
+--
+--   Транспорт — ТОЛЬКО BrowserFetcher(source="domclick"): карточки за DataDome,
+--   curl_cffi отдаёт 401 (нет curl-fallback). Оркестратор открывает ОДНУ браузер-сессию
+--   на весь прогон. Block (DataDome / fetch-fail) транзиентен → abort после
+--   max_consecutive_blocks с mark_done (НЕ mark_failed); retry следующей ночью через
+--   detail_enriched_at IS NULL. Parse-failure (HTTP 200, но SSR не парсится) → листинг
+--   остаётся необогащённым (rollback) и переотбирается на следующем прогоне.
+--
+--   Capability полностью wired (run_domclick_detail_backfill + trigger_domclick_detail_
+--   backfill_run + dispatch в scheduler.py), но schedule засеян ВЫКЛЮЧЕННЫМ — live-форма
+--   SSR-стейта и поведение DataDome НЕ verified до ручного prod-smoke. ВКЛЮЧАТЬ ВРУЧНУЮ:
+--     UPDATE scrape_schedules SET enabled = true WHERE source = 'domclick_detail_backfill';
+--   Зеркалит yandex_detail_backfill (113) / avito_detail_backfill (112) и dormant-сиды
+--   domclick_city_sweep (122/138).
+--
+-- ЗАВИСИМОСТИ:
+--   - 052_scrape_schedules.sql (таблица + UNIQUE(source)).
+--   - Layer A (#1979 / domclick_city_sweep) должен СНАЧАЛА наполнить listings
+--     source='domklik' (иначе snapshot пуст → no-op mark_done на каждом прогоне).
+-- Idempotent: ON CONFLICT (source) DO NOTHING — безопасно запускать повторно.
+--
+-- Окно 15:00-18:00 UTC (18:00-21:00 МСК) — после avito (09-12) / yandex (12-15) detail
+--   backfills и city sweeps, чтобы не конкурировать за egress на shared mobile proxy.
+--   default_params (консервативно, под DataDome — browser-only транспорт медленный):
+--     batch_size              — размер snapshot (SELECT LIMIT) на один прогон.
+--     budget_sec              — wall-clock бюджет прогона (сек).
+--     request_delay_sec       — пауза между карточками (anti-DataDome).
+--     max_consecutive_blocks  — порог последовательных блоков до abort.
+
+BEGIN;
+
+-- next_run_at = следующее наступление окна (tomorrow + window_start_hour=15h).
+-- При enabled=false get_due_schedules() строку не подхватит, но заполняем по образцу
+-- 122/138: когда оператор включит schedule, next_run_at уже валиден и run не выстрелит
+-- мгновенно (вне окна). AT TIME ZONE 'UTC' фиксирует wall-clock час в UTC независимо
+-- от session TimeZone GUC.
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'domclick_detail_backfill',
+    false,                                       -- DORMANT — enable manually after prod-smoke
+    15,
+    18,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 15)) AT TIME ZONE 'UTC',
+    '{"batch_size": 300, "budget_sec": 3600, "request_delay_sec": 8, "max_consecutive_blocks": 5}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMIT;

tradein-mvp/backend/scripts/README.md

@@ -271,3 +271,48 @@ Caveats (also printed): CURRENT listings vs PAST deals (not point-in-time —
 needs `listing_source_snapshots` #570); asking-median + IQR core only; ДКП =
 registered price. Pure metric/ratio helpers are unit-tested in
 `tests/test_backtest_estimator.py` (no DB).
+
+---
+
+## `domclick_local_runner.py` — DomClick EKB вторичка с домашнего IP (offline)
+
+**SELF-CONTAINED** (без `app.*` импортов, stdlib + `playwright.async_api`).
+Запускается ОПЕРАТОРОМ на его машине с домашнего/резидентного IP — DomClick
+фронтит QRATOR, который банит datacenter/mobile-proxy, но пропускает домашний.
+Пишет JSONL (НЕ в БД); заливка в trade-in БД — отдельным ingest-шагом.
+
+Собирает Layer A (BFF search: `/api/offers/v1` + `/api/offers/count/v1`,
+room×price-бакетинг под cap offset<2000) + Layer B (detail-карточка,
+`window.__SSR_STATE__`: renovation, living/kitchen, priceHistory, egrnData
+owners/collateral, sale_type, views/calls, wall/floor type, AVM). Логика
+скопирована инлайн из `app/services/scrapers/domclick.py` + `domclick_detail.py`.
+
+### Режим окна (verified live 2026-06-27 на домашнем IP)
+
+QRATOR ловит СТАРЫЙ headless (`launch(headless=True)`) → `403 | Домклик` даже
+с домашнего IP. NEW-headless (`--headless=new`) И headed → `200` + полный
+`__SSR_STATE__`. Поэтому дефолт = new-headless (окно скрыто, детект проходит);
+`--headed` = видимое окно (debug). BFF-enumerate проходит в любом режиме.
+
+### Темп (явное требование — НЕ спалить домашний IP)
+
+- пауза между карточками `--min-delay` 12с / `--max-delay` 30с (`random.uniform`);
+- render-wait после goto `--render-min` 6с / `--render-max` 10с;
+- «человеческий перерыв» каждые 25-40 карточек → 60-180с;
+- на 403 — один ретрай через 45-90с, потом `status="blocked"` (не долбит).
+
+Дефолтный темп ≈ 100-150 карточек/час → полный свод ЕКБ (~6400) ≈ 45-65 ч
+(несколько ночей). Скрипт РЕЗЮМИРУЕМ: при старте читает JSONL и пропускает
+id с терминальным статусом (ok/blocked/parse_fail).
+
+```bash
+cd tradein-mvp/backend
+# боевой прогон (дефолтный медленный темп):
+python scripts/domclick_local_runner.py --out domclick_ekb.jsonl
+# быстрый smoke (маленькие задержки ТОЛЬКО для теста):
+python scripts/domclick_local_runner.py --limit 2 --min-delay 3 --max-delay 5 --out dc_test.jsonl
+# только перечислить (без detail):
+python scripts/domclick_local_runner.py --enumerate-only --out enum.jsonl
+```
+
+Запускать python'ом, где установлен playwright (`python -c "import playwright"`).

tradein-mvp/backend/scripts/ingest_domclick_jsonl.py

@@ -0,0 +1,291 @@
+"""Ингест JSONL от локального DomClick-раннера (`scripts/domclick_local_runner.py`) в БД tradein.
+
+Запускается ВНУТРИ tradein-контейнера (импортит `app`). Идемпотентно.
+
+Каждая ok-запись (status=="ok" и валидный price_rub) →
+  1. upsert в `listings` через save_listings() (house-match хук — fault-tolerant, в SAVEPOINT);
+  2. deep-detail enrichment через save_detail_enrichment()
+     (owners / encumbrances / kitchen / views / wall / floor / price-history).
+
+Listing.id не возвращается из save_listings — достаём отдельным SELECT по (source, source_id).
+
+Usage:
+    python -m scripts.ingest_domclick_jsonl --jsonl /tmp/domclick.jsonl [--limit N] [--dry-run]
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+from typing import Any
+
+from sqlalchemy import text
+
+from app.core.db import SessionLocal
+from app.services.scrapers.base import ScrapedLot, save_listings
+from app.services.scrapers.domclick_detail import (
+    DomClickDetailEnrichment,
+    save_detail_enrichment,
+)
+
+logger = logging.getLogger(__name__)
+
+SOURCE = "domklik"
+
+
+def _to_int(v: Any) -> int | None:
+    """int() None-safe: возвращает None на None / нечисловом мусоре (не крэшит прогон)."""
+    if v is None:
+        return None
+    try:
+        return int(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def _to_float(v: Any) -> float | None:
+    """float() None-safe: возвращает None на None / нечисловом мусоре."""
+    if v is None:
+        return None
+    try:
+        return float(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def _build_lot(rec: dict[str, Any]) -> ScrapedLot:
+    """Построить ScrapedLot из JSONL-записи. price_rub уже провалидирован caller'ом."""
+    raw_extra = rec.get("raw_extra") or {}
+    return ScrapedLot(
+        source=SOURCE,
+        source_url=rec["source_url"],
+        source_id=str(rec["id"]),
+        address=rec.get("address"),
+        lat=_to_float(rec.get("lat")),
+        lon=_to_float(rec.get("lon")),
+        rooms=_to_int(rec.get("rooms")),
+        area_m2=_to_float(rec.get("area_m2")),
+        floor=_to_int(rec.get("floor")),
+        total_floors=_to_int(rec.get("total_floors")),
+        year_built=_to_int(rec.get("year_built")),
+        price_rub=int(rec["price_rub"]),
+        listing_segment="vtorichka",
+        repair_state=rec.get("repair_state"),
+        raw_payload={
+            "is_sber_collateral": rec.get("is_sber_collateral"),
+            "square_price": raw_extra.get("square_price"),
+            "avm": rec.get("avm") or {},
+        },
+    )
+
+
+def _build_enrichment(rec: dict[str, Any]) -> DomClickDetailEnrichment:
+    """Построить deep-detail enrichment из JSONL-записи."""
+    raw_extra_in = rec.get("raw_extra") or {}
+    raw_extra: dict[str, Any] = {
+        "wall_type": rec.get("wall_type"),
+        "floor_type": rec.get("floor_type"),
+        "calls": rec.get("calls"),
+        "is_sber_collateral": rec.get("is_sber_collateral"),
+        "avm": rec.get("avm") or {},
+        **raw_extra_in,
+    }
+    return DomClickDetailEnrichment(
+        item_id=str(rec["id"]),
+        source_url=rec["source_url"],
+        repair_state=rec.get("repair_state"),
+        repair_type=rec.get("repair_raw"),
+        living_area_m2=_to_float(rec.get("living_area_m2")),
+        kitchen_area_m2=_to_float(rec.get("kitchen_area_m2")),
+        sale_type=rec.get("sale_type"),
+        owners_count=_to_int(rec.get("owners_count")),
+        encumbrances_clean=rec.get("encumbrances_clean"),
+        year_built=_to_int(rec.get("year_built")),
+        views_total=_to_int(rec.get("views")),
+        price_changes=rec.get("price_changes") or [],
+        raw_extra=raw_extra,
+    )
+
+
+def run(jsonl_path: str, limit: int | None, dry_run: bool) -> dict[str, int]:
+    """Прочитать JSONL, отфильтровать ok+price, upsert listings + enrichment.
+
+    Returns: dict со счётчиками для итогового лога / тестов.
+    """
+    lots: list[ScrapedLot] = []
+    valid_records: list[dict[str, Any]] = []  # (parallel to lots) для enrichment-шага
+
+    skipped_status = 0
+    skipped_no_price = 0
+    parse_failed = 0
+
+    with open(jsonl_path, encoding="utf-8") as fh:
+        for line in fh:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                rec = json.loads(line)
+            except json.JSONDecodeError as exc:
+                logger.warning("битая JSON-строка пропущена: %s", exc)
+                parse_failed += 1
+                continue
+
+            if rec.get("status") != "ok":
+                skipped_status += 1
+                continue
+            price = _to_int(rec.get("price_rub"))
+            if price is None or price <= 0:
+                skipped_no_price += 1
+                continue
+
+            try:
+                lot = _build_lot(rec)
+            except Exception as exc:
+                logger.warning("не удалось построить lot для id=%s: %s", rec.get("id"), exc)
+                parse_failed += 1
+                continue
+
+            lots.append(lot)
+            valid_records.append(rec)
+
+            if limit is not None and len(lots) >= limit:
+                break
+
+    total_valid = len(lots)
+    logger.info(
+        "прочитано: valid=%d skipped_status=%d skipped_no_price=%d parse_failed=%d",
+        total_valid,
+        skipped_status,
+        skipped_no_price,
+        parse_failed,
+    )
+
+    inserted = 0
+    updated = 0
+    house_matched = 0
+    match_failed = 0
+    detail_applied = 0
+    price_history_attempted = 0
+    enrich_failed = 0
+
+    if dry_run:
+        logger.info("[dry-run] %d записей распарсено, запись в БД пропущена", total_valid)
+        return {
+            "inserted": inserted,
+            "updated": updated,
+            "house_matched": house_matched,
+            "match_failed": match_failed,
+            "detail_applied": detail_applied,
+            "price_history_attempted": price_history_attempted,
+            "enrich_failed": enrich_failed,
+            "skipped_status": skipped_status,
+            "skipped_no_price": skipped_no_price,
+            "parse_failed": parse_failed,
+        }
+
+    db = SessionLocal()
+    try:
+        if lots:
+            inserted, updated = save_listings(db, lots)
+            logger.info("save_listings: inserted=%d updated=%d", inserted, updated)
+
+        # Достаём listing.id + house_id_fk одним SELECT по source_id.
+        source_ids = [lot.source_id for lot in lots if lot.source_id is not None]
+        id_map: dict[str, tuple[int, Any]] = {}
+        if source_ids:
+            rows = db.execute(
+                text(
+                    "SELECT source_id, id, house_id_fk FROM listings "
+                    "WHERE source = :source AND source_id = ANY(:ids)"
+                ),
+                {"source": SOURCE, "ids": source_ids},
+            ).fetchall()
+            for sid, lid, house_id_fk in rows:
+                id_map[str(sid)] = (int(lid), house_id_fk)
+
+        total_saved = len(id_map)
+        house_matched = sum(1 for _, hid in id_map.values() if hid is not None)
+        match_failed = total_saved - house_matched
+
+        # Deep-detail enrichment по каждой ok-записи.
+        for rec in valid_records:
+            sid = str(rec["id"])
+            entry = id_map.get(sid)
+            if entry is None:
+                logger.warning("listing id не найден для source_id=%s — enrichment пропущен", sid)
+                continue
+            listing_id, _ = entry
+            # Per-record изоляция: save_detail_enrichment коммитит сама, поэтому
+            # падение одной записи иначе оборвало бы остаток valid_records трейсбэком.
+            # Уже закоммиченные строки персистят → повторный прогон добьёт хвост.
+            try:
+                enrichment = _build_enrichment(rec)
+                price_history_attempted += len(enrichment.price_changes)
+                if save_detail_enrichment(db, listing_id, enrichment):
+                    detail_applied += 1
+            except Exception as exc:
+                db.rollback()
+                enrich_failed += 1
+                logger.warning("enrichment упал для source_id=%s: %s", sid, exc)
+                continue
+    finally:
+        db.close()
+
+    logger.info(
+        "ИТОГО: saved(inserted=%d/updated=%d) house_matched=%d match_failed=%d "
+        "detail_applied=%d price_history_attempted=%d enrich_failed=%d "
+        "skipped_status=%d skipped_no_price=%d parse_failed=%d",
+        inserted,
+        updated,
+        house_matched,
+        match_failed,
+        detail_applied,
+        price_history_attempted,
+        enrich_failed,
+        skipped_status,
+        skipped_no_price,
+        parse_failed,
+    )
+
+    return {
+        "inserted": inserted,
+        "updated": updated,
+        "house_matched": house_matched,
+        "match_failed": match_failed,
+        "detail_applied": detail_applied,
+        "price_history_attempted": price_history_attempted,
+        "enrich_failed": enrich_failed,
+        "skipped_status": skipped_status,
+        "skipped_no_price": skipped_no_price,
+        "parse_failed": parse_failed,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Ингест DomClick JSONL (от domclick_local_runner) в БД tradein.",
+    )
+    parser.add_argument("--jsonl", required=True, help="путь к JSONL-файлу раннера")
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=None,
+        help="обработать первые N валидных (ok+price) записей",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="парсить + логировать, но НЕ писать в БД",
+    )
+    args = parser.parse_args()
+    run(args.jsonl, args.limit, args.dry_run)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    main()

tradein-mvp/backend/tests/scrapers/test_domclick_detail.py

@@ -0,0 +1,404 @@
+"""Unit-тесты для DomClick Layer B detail-scraper (domclick_detail.py).
+
+Offline: parse/SSR-тесты без сети/БД; save-тест через MagicMock (зеркало
+test_cian_detail.py — в проекте нет live-DB фикстуры, идемпотентность
+гарантируется ON CONFLICT-клаузой, которая проверяется на уровне SQL-строки).
+
+Покрытие:
+- _extract_ssr_state: bare `undefined` → null; in-string "undefined" выживает;
+  скобки/эскейпы внутри строк не ломают balanced-scan; missing marker →
+  DomClickParseError; challenge → DomClickBlockedError; unbalanced/невалидный
+  JSON → DomClickParseError.
+- parse_detail_html (REAL __SSR_STATE__ paths, live card 2075729321):
+  productCard.objectInfo.{renovation→repair_state+repair_type,livingArea,kitchenArea};
+  balconies absent→None; productCard.legalOptions.saleType; egrnData snake_case
+  owners_count + encumbrances инверсия (collateral); houseInfo.info.{wallType,floorType}
+  → raw_extra (НЕ house_type); productCard.viewsCount; priceInfo.priceHistory
+  (ISO8601+offset kept tz-aware / bad-date + no-price skipped); top-level
+  pricePrediction → raw_extra.avm (плоский + nested .result); raw_extra без None;
+  item_id из URL.
+- fetch_detail: browser-branch + проброс ошибок.
+- save_detail_enrichment: UPDATE-параметры, detail_enriched_at, price-history INSERT.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime, timedelta
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from app.services.scrapers.domclick_detail import (
+    DomClickDetailEnrichment,
+    _extract_ssr_state,
+    fetch_detail,
+    parse_detail_html,
+    save_detail_enrichment,
+)
+from app.services.scrapers.domclick_exceptions import DomClickBlockedError, DomClickParseError
+
+_CARD_URL = "https://ekaterinburg.domclick.ru/card/sale__flat__2075729321"
+
+# Литерал с bare `undefined` (value-позиция), in-string "undefined", скобкой
+# внутри строки и эскейп-кавычками — написан вручную, json.dumps использовать
+# нельзя (нужен именно НЕвалидный-для-json undefined).
+# Real __SSR_STATE__ shape (ground truth, live card 2075729321, 2026-06-27).
+# bare `undefined` (value-pos), in-string "undefined", '{' внутри строки и
+# эскейп-кавычки оставлены для проверки санитайзера/balanced-scan; json.dumps
+# использовать нельзя (нужен именно НЕвалидный-для-json undefined).
+# priceHistory[0].date намеренно flush-left (длинная ISO+offset строка) — отступ
+# в литерале не влияет на парсер, но держит физическую строку ≤100 для ruff.
+_SSR_LITERAL = """{
+  "productCard": {
+    "objectInfo": {
+      "area": 38.0,
+      "rooms": 1,
+      "floor": 5,
+      "isApartment": false,
+      "renovation": "евроремонт",
+      "livingArea": 22.37,
+      "kitchenArea": 10.13,
+      "description": "Состояние пока undefined.",
+      "note": "угол {da}",
+      "quote": "сказал \\"привет\\" сосед"
+    },
+    "legalOptions": {"saleType": "Свободная продажа"},
+    "egrnData": {
+      "area": 38.2,
+      "floor": 5,
+      "owners_count": 1,
+      "collateral": true,
+      "collateral_sber": false
+    },
+    "priceInfo": {
+      "priceHistory": [
+{"date": "2026-04-15T09:02:26.548728+03:00", "price": 13400000, "diff": -400000, "state": "less"},
+        {"date": "not-a-date", "price": 4800000, "diff": -1.0, "state": "less"},
+        {"date": "2026-05-01T10:00:00Z", "price": null, "diff": 0, "state": "more"}
+      ]
+    },
+    "viewsCount": 338,
+    "callsCount": 5,
+    "favoriteOfferUsersCount": 12,
+    "duplicatesOfferCount": 0,
+    "address": {"guid": "abc-guid-123"},
+    "unknownField": undefined
+  },
+  "houseInfo": {
+    "info": {
+      "buildYear": 2015,
+      "wallType": "Кирпично-монолитный",
+      "floorType": "Железобетонный",
+      "entranceCount": 3,
+      "quartersCount": 120,
+      "energyEfficiency": "B",
+      "buildingSeries": "индивидуальный"
+    }
+  },
+  "pricePrediction": {
+    "market_price": 13000000,
+    "min_market_price": 12000000,
+    "max_market_price": 14000000,
+    "rent_short": 50000,
+    "rent_long": 35000,
+    "repair_quality": "good"
+  }
+}"""
+
+_HTML = (
+    "<html><head></head><body><script>"
+    f"window.__SSR_STATE__ = {_SSR_LITERAL};"
+    "</script></body></html>"
+)
+
+
+def _wrap(state: dict) -> str:
+    """Заворачивает валидный dict в SSR-HTML (для variant-фикстур без undefined)."""
+    literal = json.dumps(state, ensure_ascii=False)
+    return f"<html><body><script>window.__SSR_STATE__ = {literal};</script></body></html>"
+
+
+# ── _extract_ssr_state ────────────────────────────────────────────────────────
+
+
+def test_extract_ssr_state_parses_despite_undefined() -> None:
+    state = _extract_ssr_state(_HTML)
+    # bare undefined в value-позиции → null → None
+    assert state["productCard"]["unknownField"] is None
+    # in-string "undefined" НЕ испорчен санитайзером
+    assert state["productCard"]["objectInfo"]["description"] == "Состояние пока undefined."
+
+
+def test_extract_ssr_state_brace_and_escape_in_string() -> None:
+    state = _extract_ssr_state(_HTML)
+    oi = state["productCard"]["objectInfo"]
+    # '{' внутри строки не ломает balanced-brace scan
+    assert oi["note"] == "угол {da}"
+    # эскейп-кавычки внутри строки обработаны
+    assert oi["quote"] == 'сказал "привет" сосед'
+
+
+def test_extract_ssr_state_missing_marker_raises_parse() -> None:
+    with pytest.raises(DomClickParseError):
+        _extract_ssr_state("<html><body>no ssr state here</body></html>")
+
+
+def test_extract_ssr_state_challenge_raises_blocked() -> None:
+    html = "<html><body>Access denied by DataDome captcha</body></html>"
+    with pytest.raises(DomClickBlockedError):
+        _extract_ssr_state(html)
+
+
+def test_extract_ssr_state_unbalanced_raises_parse() -> None:
+    with pytest.raises(DomClickParseError):
+        _extract_ssr_state('window.__SSR_STATE__ = {"a": 1')
+
+
+def test_extract_ssr_state_invalid_json_raises_parse() -> None:
+    with pytest.raises(DomClickParseError):
+        _extract_ssr_state("window.__SSR_STATE__ = {this: is, not: json}")
+
+
+# ── parse_detail_html — полная карточка ───────────────────────────────────────
+
+
+def test_parse_detail_html_full() -> None:
+    e = parse_detail_html(_HTML, _CARD_URL)
+
+    assert e.item_id == "2075729321"
+    assert e.source_url == _CARD_URL
+
+    assert e.repair_state == "good"  # евроремонт → good (LOCAL map)
+    assert e.repair_type == "евроремонт"  # сырое значение сохранено
+
+    assert e.living_area_m2 == 22.37
+    assert e.kitchen_area_m2 == 10.13
+
+    # На реальной карточке нет ключа balconies в objectInfo → optional/None.
+    assert e.balconies_count is None
+    assert e.has_balcony is None
+
+    assert e.sale_type == "Свободная продажа"
+    assert e.owners_count == 1
+    assert e.encumbrances_clean is False  # collateral=true → есть обременение
+    assert e.year_built == 2015
+    assert e.views_total == 338
+
+
+def test_parse_detail_html_price_changes() -> None:
+    e = parse_detail_html(_HTML, _CARD_URL)
+    # entry1 ок; entry2 bad-date skip; entry3 price=null skip → 1 запись
+    assert len(e.price_changes) == 1
+    change = e.price_changes[0]
+    assert change["price_rub"] == 13400000
+    assert change["diff_percent"] == -400000
+    assert isinstance(change["change_time"], datetime)
+    # ISO8601-with-offset: tz-aware, offset СОХРАНЁН (+03:00), НЕ сконвертирован в UTC.
+    assert change["change_time"].tzinfo is not None
+    assert change["change_time"].utcoffset() == timedelta(hours=3)
+
+
+def test_parse_detail_html_raw_extra() -> None:
+    e = parse_detail_html(_HTML, _CARD_URL)
+    raw = e.raw_extra
+    assert raw["wall_type"] == "Кирпично-монолитный"
+    assert raw["floor_type"] == "Железобетонный"
+    # wall/floor type живут ТОЛЬКО в raw_extra, НЕ в listings.house_type.
+    assert "house_type" not in raw
+    assert raw["domclick_building_guid"] == "abc-guid-123"
+    assert raw["egrn_area"] == 38.2
+    assert raw["demand"]["calls"] == 5
+    assert raw["demand"]["favorites"] == 12
+    # AVM (Layer C, top-level pricePrediction) → raw_extra.avm
+    assert raw["avm"]["market_price"] == 13000000
+    assert raw["avm"]["min"] == 12000000
+    assert raw["avm"]["max"] == 14000000
+    assert raw["avm"]["repair_quality"] == "good"
+    # нет ключа balconies → raw_extra НЕ содержит balconies_raw
+    assert "balconies_raw" not in raw
+    # никаких None-значений на верхнем уровне (компактность)
+    assert all(v is not None for v in raw.values())
+
+
+# ── parse_detail_html — частные ветки ─────────────────────────────────────────
+
+
+def test_parse_balconies_string_goes_to_raw_extra() -> None:
+    html = _wrap({"productCard": {"objectInfo": {"balconies": "есть"}}})
+    e = parse_detail_html(html, _CARD_URL)
+    assert e.balconies_count is None
+    assert e.has_balcony is None
+    assert e.raw_extra["balconies_raw"] == "есть"
+
+
+def test_parse_encumbrances_clean_true_when_no_collateral() -> None:
+    html = _wrap({"productCard": {"egrnData": {"collateral": False, "collateral_sber": False}}})
+    e = parse_detail_html(html, _CARD_URL)
+    assert e.encumbrances_clean is True
+
+
+def test_parse_encumbrances_none_when_egrn_missing() -> None:
+    html = _wrap({"productCard": {"objectInfo": {"livingArea": 30}}})
+    e = parse_detail_html(html, _CARD_URL)
+    assert e.encumbrances_clean is None
+
+
+def test_parse_repair_map_hit() -> None:
+    html = _wrap({"productCard": {"objectInfo": {"renovation": "Косметический"}}})
+    e = parse_detail_html(html, _CARD_URL)
+    assert e.repair_state == "standard"
+    assert e.repair_type == "Косметический"
+
+
+def test_parse_repair_fallback_to_inference() -> None:
+    # Не точное совпадение в карте → инференс из текста (евроремонт → good)
+    html = _wrap({"productCard": {"objectInfo": {"renovation": "сделан свежий евроремонт"}}})
+    e = parse_detail_html(html, _CARD_URL)
+    assert e.repair_state == "good"
+    assert e.repair_type == "сделан свежий евроремонт"
+
+
+def test_parse_owners_count_camelcase_key() -> None:
+    html = _wrap({"productCard": {"egrnData": {"ownersCount": 3}}})
+    e = parse_detail_html(html, _CARD_URL)
+    assert e.owners_count == 3
+
+
+def test_parse_avm_nested_under_result() -> None:
+    # pricePrediction может приходить обёрнутым в .result — unwrap должен сработать.
+    html = _wrap(
+        {
+            "productCard": {},
+            "pricePrediction": {"result": {"marketPrice": 9000000, "repairQuality": "fine"}},
+        }
+    )
+    e = parse_detail_html(html, _CARD_URL)
+    assert e.raw_extra["avm"]["market_price"] == 9000000
+    assert e.raw_extra["avm"]["repair_quality"] == "fine"
+
+
+def test_parse_avm_absent_no_key() -> None:
+    # Нет pricePrediction → raw_extra без ключа avm (компактность).
+    e = parse_detail_html(_wrap({"productCard": {}}), _CARD_URL)
+    assert "avm" not in e.raw_extra
+
+
+def test_parse_item_id_none_for_bad_url() -> None:
+    e = parse_detail_html(_wrap({"productCard": {}}), "https://domclick.ru/card/rent__room__1")
+    assert e.item_id is None
+
+
+def test_dataclass_defaults() -> None:
+    e = DomClickDetailEnrichment()
+    assert e.item_id is None
+    assert e.price_changes == []
+    assert e.raw_extra == {}
+
+
+# ── fetch_detail (browser-branch) ─────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_parses_via_browser() -> None:
+    bf = MagicMock()
+    bf.fetch = AsyncMock(return_value=_HTML)
+    e = await fetch_detail(_CARD_URL, browser_fetcher=bf)
+    assert e.item_id == "2075729321"
+    assert e.repair_state == "good"
+    bf.fetch.assert_called_once_with(_CARD_URL)
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_browser_failure_raises_blocked() -> None:
+    bf = MagicMock()
+    bf.fetch = AsyncMock(side_effect=RuntimeError("502 bad gateway"))
+    with pytest.raises(DomClickBlockedError):
+        await fetch_detail(_CARD_URL, browser_fetcher=bf)
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_propagates_parse_error() -> None:
+    bf = MagicMock()
+    bf.fetch = AsyncMock(return_value="<html><body>no state</body></html>")
+    with pytest.raises(DomClickParseError):
+        await fetch_detail(_CARD_URL, browser_fetcher=bf)
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_propagates_blocked_from_html() -> None:
+    bf = MagicMock()
+    bf.fetch = AsyncMock(return_value="<html><body>Access denied datadome</body></html>")
+    with pytest.raises(DomClickBlockedError):
+        await fetch_detail(_CARD_URL, browser_fetcher=bf)
+
+
+# ── save_detail_enrichment (MagicMock — зеркало test_cian_detail) ─────────────
+
+
+def test_save_detail_enrichment_writes_columns() -> None:
+    db = MagicMock()
+    db.execute.return_value.rowcount = 1
+    ct = datetime(2026, 4, 1, tzinfo=UTC)
+    e = DomClickDetailEnrichment(
+        item_id="2075729321",
+        source_url=_CARD_URL,
+        repair_state="good",
+        repair_type="Евроремонт",
+        living_area_m2=45.5,
+        kitchen_area_m2=12.3,
+        balconies_count=2,
+        has_balcony=True,
+        sale_type="свободная",
+        owners_count=1,
+        encumbrances_clean=False,
+        year_built=2015,
+        views_total=1234,
+        price_changes=[{"change_time": ct, "price_rub": 5000000, "diff_percent": -2.5}],
+        raw_extra={"wall_type": "монолит"},
+    )
+
+    ok = save_detail_enrichment(db, 999, e)
+    assert ok is True
+
+    # UPDATE — первый execute
+    update_sql = str(db.execute.call_args_list[0][0][0])
+    update_params = db.execute.call_args_list[0][0][1]
+    assert "detail_enriched_at = NOW()" in update_sql
+    assert update_params["lid"] == 999
+    assert update_params["repair_state"] == "good"
+    assert update_params["living_area_m2"] == 45.5
+    assert update_params["balconies_count"] == 2
+    assert update_params["has_balcony"] is True
+    assert update_params["encumbrances_clean"] is False
+    assert "монолит" in update_params["raw_extra"]
+
+    # price-history INSERT с ON CONFLICT (идемпотентность)
+    insert_calls = [c for c in db.execute.call_args_list if "offer_price_history" in str(c[0][0])]
+    assert len(insert_calls) == 1
+    assert "ON CONFLICT ON CONSTRAINT offer_price_history_listing_change_uq" in str(
+        insert_calls[0][0][0]
+    )
+    ins_params = insert_calls[0][0][1]
+    assert ins_params["price"] == 5000000
+    assert ins_params["ct"] == ct
+    db.commit.assert_called_once()
+
+
+def test_save_detail_enrichment_listing_not_found() -> None:
+    db = MagicMock()
+    db.execute.return_value.rowcount = 0
+    e = DomClickDetailEnrichment(item_id="x", source_url=_CARD_URL, repair_state="good")
+    assert save_detail_enrichment(db, 123, e) is False
+
+
+def test_save_detail_enrichment_none_passthrough() -> None:
+    db = MagicMock()
+    db.execute.return_value.rowcount = 1
+    e = DomClickDetailEnrichment(item_id="x", source_url=_CARD_URL)  # все поля None
+    save_detail_enrichment(db, 1, e)
+    params = db.execute.call_args_list[0][0][1]
+    assert params["repair_state"] is None
+    assert params["living_area_m2"] is None
+    # пустой raw_extra → "{}" (merge no-op, не затирает)
+    assert params["raw_extra"] == "{}"

tradein-mvp/backend/tests/test_domclick_detail_backfill.py

@@ -0,0 +1,195 @@
+"""Offline unit tests for run_domclick_detail_backfill (Layer B orchestrator).
+
+Pure & fast: NO live network, NO DB, NO browser. Mirrors test_domclick_sweep mocking
+style. Mocks fetch_detail (AsyncMock) + save_detail_enrichment + scrape_runs lifecycle
++ a fake snapshot of listing rows via db.execute(...).mappings().all().
+
+Coverage:
+  - happy path: both rows enriched -> mark_done(attempted=2, enriched=2)
+  - block-abort: DomClickBlockedError x max_consecutive_blocks -> abort, mark_done (not failed)
+  - parse-failure: DomClickParseError for one row -> failed++, db.rollback, other still enriched
+  - cancel before loop -> returns early (no snapshot SELECT, no mark_done)
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from app.services.scrapers.domclick_exceptions import DomClickBlockedError, DomClickParseError
+from app.tasks import domclick_detail_backfill as mod
+from app.tasks.domclick_detail_backfill import run_domclick_detail_backfill
+
+
+class _FakeRuns:
+    """Stub для scrape_runs: отслеживает вызовы, поддерживает cooperative cancel."""
+
+    def __init__(self, *, cancel: bool = False) -> None:
+        self.cancel = cancel
+        self.heartbeats: list[dict[str, int]] = []
+        self.done: dict[str, int] | None = None
+        self.failed: tuple[str, dict[str, int]] | None = None
+
+    def is_cancelled(self, _db: Any, _run_id: int) -> bool:
+        return self.cancel
+
+    def update_heartbeat(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None:
+        self.heartbeats.append(dict(counters))
+
+    def mark_done(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None:
+        self.done = dict(counters)
+
+    def mark_failed(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None:
+        self.failed = (error, dict(counters))
+
+
+class _FakeBrowserFetcher:
+    """No-op async context manager standing in for BrowserFetcher (no browser launch)."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        pass
+
+    async def __aenter__(self) -> _FakeBrowserFetcher:
+        return self
+
+    async def __aexit__(self, *_args: Any) -> None:
+        return None
+
+
+def _fake_db(rows: list[dict[str, Any]]) -> MagicMock:
+    """MagicMock Session whose execute(...).mappings().all() returns rows."""
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.return_value = rows
+    return db
+
+
+def _row(listing_id: int, offer_id: str) -> dict[str, Any]:
+    return {
+        "id": listing_id,
+        "source_id": offer_id,
+        "source_url": f"https://ekaterinburg.domclick.ru/card/sale__flat__{offer_id}",
+    }
+
+
+@pytest.fixture(autouse=True)
+def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def _instant(_secs: float) -> None:
+        return None
+
+    monkeypatch.setattr(mod.asyncio, "sleep", _instant)
+
+
+@pytest.fixture(autouse=True)
+def _stub_browser(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(mod, "BrowserFetcher", _FakeBrowserFetcher)
+
+
+def _install_runs(monkeypatch: pytest.MonkeyPatch, fake: _FakeRuns) -> None:
+    monkeypatch.setattr(mod.runs_mod, "is_cancelled", fake.is_cancelled)
+    monkeypatch.setattr(mod.runs_mod, "update_heartbeat", fake.update_heartbeat)
+    monkeypatch.setattr(mod.runs_mod, "mark_done", fake.mark_done)
+    monkeypatch.setattr(mod.runs_mod, "mark_failed", fake.mark_failed)
+
+
+async def test_happy_path_enriches_all(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Both rows enriched → mark_done(attempted=2, enriched=2), save called per row."""
+    db = _fake_db([_row(1, "100"), _row(2, "101")])
+
+    fetch_mock = AsyncMock(return_value=MagicMock())
+    save_mock = MagicMock(return_value=True)
+    monkeypatch.setattr(mod, "fetch_detail", fetch_mock)
+    monkeypatch.setattr(mod, "save_detail_enrichment", save_mock)
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await run_domclick_detail_backfill(db, run_id=1, params={"request_delay_sec": 0})
+
+    assert counters.attempted == 2
+    assert counters.enriched == 2
+    assert counters.blocked == 0
+    assert counters.failed == 0
+    assert fetch_mock.await_count == 2
+    assert save_mock.call_count == 2
+    assert fake.done is not None
+    assert fake.done["attempted"] == 2
+    assert fake.done["enriched"] == 2
+    assert fake.failed is None
+
+
+async def test_block_abort_marks_done_not_failed(monkeypatch: pytest.MonkeyPatch) -> None:
+    """DomClickBlockedError max_consecutive_blocks in a row → abort, mark_done (not failed)."""
+    db = _fake_db([_row(i, str(i)) for i in range(1, 11)])  # 10 rows available
+
+    fetch_mock = AsyncMock(side_effect=DomClickBlockedError("datadome challenge"))
+    save_mock = MagicMock(return_value=True)
+    monkeypatch.setattr(mod, "fetch_detail", fetch_mock)
+    monkeypatch.setattr(mod, "save_detail_enrichment", save_mock)
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await run_domclick_detail_backfill(
+        db, run_id=2, params={"request_delay_sec": 0, "max_consecutive_blocks": 5}
+    )
+
+    # Abort after 5 consecutive blocks → attempted=5, all blocked, none enriched.
+    assert counters.blocked == 5
+    assert counters.attempted == 5
+    assert counters.enriched == 0
+    assert save_mock.call_count == 0
+    # Block is transient → mark_done, NOT mark_failed.
+    assert fake.done is not None
+    assert fake.done["blocked"] == 5
+    assert fake.failed is None
+
+
+async def test_parse_failure_rolls_back_and_continues(monkeypatch: pytest.MonkeyPatch) -> None:
+    """DomClickParseError on row 1 → failed++, db.rollback; row 2 still enriched; mark_done."""
+    db = _fake_db([_row(1, "100"), _row(2, "101")])
+
+    # First row -> parse error; second -> ok.
+    fetch_mock = AsyncMock(side_effect=[DomClickParseError("ssr schema drift"), MagicMock()])
+    save_mock = MagicMock(return_value=True)
+    monkeypatch.setattr(mod, "fetch_detail", fetch_mock)
+    monkeypatch.setattr(mod, "save_detail_enrichment", save_mock)
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await run_domclick_detail_backfill(db, run_id=3, params={"request_delay_sec": 0})
+
+    assert counters.attempted == 2
+    assert counters.failed == 1
+    assert counters.enriched == 1
+    db.rollback.assert_called()  # parse-failure rolled back the un-committed row
+    assert save_mock.call_count == 1  # only the second (parsed) row persisted
+    assert fake.done is not None
+    assert fake.failed is None
+
+
+async def test_cancel_before_loop_returns_early(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Cooperative cancel before loop → no snapshot SELECT, no fetch, no mark_done."""
+    db = _fake_db([_row(1, "100")])
+
+    fetch_mock = AsyncMock(return_value=MagicMock())
+    save_mock = MagicMock(return_value=True)
+    monkeypatch.setattr(mod, "fetch_detail", fetch_mock)
+    monkeypatch.setattr(mod, "save_detail_enrichment", save_mock)
+
+    fake = _FakeRuns(cancel=True)
+    _install_runs(monkeypatch, fake)
+
+    counters = await run_domclick_detail_backfill(db, run_id=4, params={"request_delay_sec": 0})
+
+    assert counters.attempted == 0
+    assert fetch_mock.await_count == 0
+    db.execute.assert_not_called()  # cancel short-circuits before the snapshot SELECT
+    assert fake.done is None
+    assert fake.failed is None