gendesign/tradein-mvp/backend/scripts/domclick_local_runner.py

#!/usr/bin/env python3
"""DomClick EKB вторичка — SELF-CONTAINED локальный сборщик (search + detail).

ЗАЧЕМ ОТДЕЛЬНЫЙ СКРИПТ
=====================
DomClick фронтит QRATOR anti-bot. Проверено вживую (2026-06-27):
  • запрос с ДОМАШНЕГО (резидентного) IP → отдаёт полную карточку с
    ``window.__SSR_STATE__`` (renovation / priceHistory / egrnData / AVM);
  • запрос с datacenter / mobile-proxy IP → 403 | Домклик (bot_mitigation).
Поэтому продовый pipeline (BrowserFetcher через shared proxy) карточки НЕ
вытягивает. Этот скрипт запускается ОПЕРАТОРОМ на его собственной машине с
домашнего IP — без прокси — и собирает данные в JSONL, который потом
заливается в trade-in БД ОТДЕЛЬНЫМ шагом (этот скрипт в БД НЕ пишет).

САМОДОСТАТОЧНОСТЬ
=================
Никаких ``app.*`` импортов: вся логика Layer A (BFF enumerate) и Layer B
(detail ``__SSR_STATE__`` parse) скопирована ИНЛАЙН из
``app/services/scrapers/domclick.py`` + ``domclick_detail.py`` +
``repair_state_normalizer.py``. Зависимости: stdlib + ``playwright.async_api``.

ТРАНСПОРТ
=========
Playwright Chromium, БЕЗ прокси (домашний IP):
  • enumerate (BFF JSON) — ``context.request.get(url)`` (тот же сетевой
    стек/IP/куки браузера) → парсим первый ``{`` … последний ``}``;
  • detail — ``page.goto(card_url)`` → jittered wait → ``page.evaluate`` с
    ``fetch(location.href)`` чтобы получить СЫРОЙ HTML (SPA удаляет
    ``window.__SSR_STATE__`` после гидрации, поэтому нужен повторный fetch).

ВАЖНО — РЕЖИМ ОКНА (verified 2026-06-27 на домашнем IP этой машины)
------------------------------------------------------------------
QRATOR ДЕТЕКТИТ СТАРЫЙ HEADLESS-РЕЖИМ: ``launch(headless=True)`` → даже с
домашнего IP карточка отдаёт ``401 / 403 | Домклик`` (нет ``__SSR_STATE__``).
А вот НОВЫЙ headless (``--headless=new``) И обычный headed → ``200`` + полный
``__SSR_STATE__`` (~330 КБ). Поэтому по умолчанию запускаемся в new-headless
(окно НЕ видно, но детект проходит). ``--headed`` → видимое окно (debug).
BFF-enumerate проходит в ЛЮБОМ режиме (это не браузерный рендер).

ВАЖНО — ЧЕЛОВЕКОПОДОБНЫЙ ТЕМП (явное требование оператора)
=========================================================
Домашний IP — расходник: спалим частотой → бан и потеря единственного
рабочего канала. Поэтому темп НАМЕРЕННО медленный и настраиваемый:
  • случайная пауза МЕЖДУ карточками: ``--min-delay`` (12с) / ``--max-delay``
    (30с), ``random.uniform``;
  • jittered 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 (status ok/blocked/parse_fail), так что прогон можно дробить на сессии.

ЗАПУСК
======
Нужен python с установленным playwright (``python -c "import playwright"``).
::

    cd .../tradein-mvp/backend
    # боевой прогон (дефолтный медленный темп), дописывает в JSONL:
    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 /tmp/dc_test.jsonl
    # только перечислить карточки, без detail:
    python scripts/domclick_local_runner.py --enumerate-only --out enum.jsonl

OUTPUT
======
JSONL, по одной записи на карточку (append). Поля см. ``build_record``.
Файл потребляется ОТДЕЛЬНЫМ ingest-шагом в trade-in БД (listings +
offer_price_history); этот скрипт сам в БД ничего не пишет.
"""

from __future__ import annotations

import argparse
import asyncio
import json
import logging
import random
import re
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from urllib.parse import urlencode

from playwright.async_api import async_playwright

logger = logging.getLogger("domclick_local")

# ── API / гео константы (ground truth, verified 2026-06-27) ──────────────────

_BFF_BASE = "https://bff-search-web.domclick.ru"
_EKB_ADDRESS_GUID = "0d475b79-88de-4054-818c-37d8f9d0d440"
_EKB_AREA_ID = "20561"
_CARD_HOST_FALLBACK = "https://ekaterinburg.domclick.ru"
_WARMUP_URL = "https://ekaterinburg.domclick.ru/"

# Порядок бакетов комнат влияет на логи. urlencode("5+") → "5%2B".
ROOM_BUCKETS: tuple[str, ...] = ("st", "1", "2", "3", "4", "5+")

OFFSET_CAP: int = 2000  # макс. offset, принимаемый BFF
PAGE_SIZE: int = 20  # items на страницу (жёстко задан API)
LTE_MAX: int = 1_000_000_000  # потолок цены для binary-split unbounded бакета
MIN_PRICE_SPAN: int = 100_000  # ниже этого span деление прекращаем
MAX_SPLIT_DEPTH: int = 30  # guard рекурсии binary-split (log2(1e9/1e5) ≈ 13)

# Гео-guard ЕКБ (bbox). aids=20561 уже даёт чистый ЕКБ; bbox — defensive.
# Deviation от app: ослаблено строгое равенство offerRegionName до bbox-only
# (устойчивее к дрейфу имени региона; aids уже ограничивает выборку ЕКБ).
_EKB_LAT_MIN, _EKB_LAT_MAX = 56.6, 57.0
_EKB_LON_MIN, _EKB_LON_MAX = 60.2, 60.9

# Маркеры QRATOR / anti-bot challenge.
_BLOCK_MARKERS: tuple[str, ...] = (
    "qrator",
    "bot_mitigation",
    "система защиты",
    "403 | домклик",
    "captcha",
    "datadome",
    "access denied",
)

# Реалистичный desktop-Chrome UA (headless-UA палится anti-bot'ами).
_USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
)

# SSR-стейт встроен как JS-литерал (НЕ чистый JSON: bare `undefined`).
_SSR_ASSIGN_RE = re.compile(r"window\.__SSR_STATE__\s*=\s*")
_UNDEFINED_RE = re.compile(r"(?<=[:\[,\s])undefined(?=[\s,\]\}])")
_ITEM_ID_RE = re.compile(r"sale__flat__(\d+)")

# JS для повторного fetch сырого HTML карточки из контекста страницы.
_FETCH_RAW_JS = (
    "async () => { const r = await fetch(window.location.href, "
    "{credentials: 'include'}); return await r.text(); }"
)


# ── Исключения ───────────────────────────────────────────────────────────────


class DomClickBlockedError(Exception):
    """Anti-bot challenge (QRATOR/DataDome/403) — НЕ parse-failure."""


class DomClickParseError(Exception):
    """HTTP 200, но __SSR_STATE__ не найден / не парсится (дрейф схемы)."""


class _EnumLimitReachedError(Exception):
    """Внутренний сигнал: набрали target новых карточек — стоп enumerate."""


# ── repair_state normalizer (инлайн из repair_state_normalizer.py) ───────────

_RAW_TO_ENUM: dict[str, str] = {
    "required": "needs_repair",
    "without": "needs_repair",
    "no": "needs_repair",
    "rough": "needs_repair",
    "prefine": "needs_repair",
    "cosmetic": "standard",
    "euro": "good",
    "fine": "good",
    "designer": "excellent",
    "design": "excellent",
}
_VALID_ENUM: frozenset[str] = frozenset({"needs_repair", "standard", "good", "excellent"})

# renovation (RU) → каноничный enum (локальная карта Layer B).
# Расширено относительно app-версии РЕАЛЬНЫМИ короткими токенами DomClick,
# наблюдёнными вживую 2026-06-27 ("Евро", "Косметический"): SSR-поле renovation
# отдаёт КОРОТКУЮ форму ("Евро"), а не "евроремонт" — иначе good терялся в None.
_RENOVATION_MAP: dict[str, str] = {
    "евроремонт": "good",
    "евро": "good",
    "косметический": "standard",
    "косметический ремонт": "standard",
    "дизайнерский": "excellent",
    "дизайнерский ремонт": "excellent",
    "черновая": "needs_repair",
    "черновая отделка": "needs_repair",
    "без отделки": "needs_repair",
    "без ремонта": "needs_repair",
    "требует ремонта": "needs_repair",
    "предчистовая": "needs_repair",
    "предчистовая отделка": "needs_repair",
}

_TEXT_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
    (
        re.compile(
            r"дизайнерск(?:ий|ого|ом|ая)\s+ремонт"
            r"|дизайнерск(?:ая|ой)\s+отделк"
            r"|премиальн\w*\s+ремонт"
            r"|эксклюзивн\w*\s+ремонт"
            r"|авторск\w*\s+ремонт",
            re.IGNORECASE,
        ),
        "excellent",
    ),
    (
        re.compile(
            r"евроремонт"
            r"|евро[\s-]?ремонт"
            r"|отличн\w*\s+ремонт"
            r"|свеж\w*\s+ремонт"
            r"|современн\w*\s+ремонт"
            r"|качественн\w*\s+ремонт"
            r"|ремонт\s+в\s+отличн",
            re.IGNORECASE,
        ),
        "good",
    ),
    (
        re.compile(
            r"без\s+отделк"
            r"|без\s+ремонт"
            r"|черновая\s+отделк"
            r"|черновой\s+ремонт"
            r"|требует(?:ся)?\s+ремонт"
            r"|требует(?:ся)?\s+космет"
            r"|нужен\s+ремонт"
            r"|под\s+ремонт"
            r"|под\s+чистов\w+\s+отделк"
            r"|предчистов",
            re.IGNORECASE,
        ),
        "needs_repair",
    ),
    (
        re.compile(
            r"косметическ\w*\s+ремонт"
            r"|косметическ\w*\s+отделк"
            r"|обычн\w*\s+ремонт"
            r"|жил(?:ое|ом)\s+состоян"
            r"|хорош\w*\s+состоян",
            re.IGNORECASE,
        ),
        "standard",
    ),
)


def normalize_repair_state(raw: str | None) -> str | None:
    """raw repair → каноничный enum (идемпотентно). Неизвестное → None."""
    if raw is None:
        return None
    if raw in _VALID_ENUM:
        return raw
    result = _RAW_TO_ENUM.get(raw) or _RAW_TO_ENUM.get(raw.lower())
    if result is None:
        logger.debug("repair_state: unknown raw %r — stored NULL", raw)
    return result


def infer_repair_state_from_text(description: str | None) -> str | None:
    """repair_state из текста описания (fallback). Первое совпадение по силе."""
    if not description:
        return None
    for pattern, enum_value in _TEXT_PATTERNS:
        if pattern.search(description):
            return enum_value
    return None


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)


# ── value coercion (инлайн из domclick_detail.py) ────────────────────────────


def _to_float(value: Any) -> float | 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:
    if value is None or isinstance(value, bool):
        return None
    try:
        return int(value)
    except (ValueError, TypeError):
        return None


def _pos_float(value: Any) -> float | None:
    """float > 0, иначе None. DomClick эмитит 0 как sentinel «неизвестно»
    для living/kitchen area (особенно у студий) — 0.0 ввело бы в заблуждение."""
    f = _to_float(value)
    if f is None or f <= 0:
        return None
    return f


def _parse_change_time(value: Any) -> datetime | None:
    """date изменения цены → tz-aware datetime. epoch(сек/мс)/ISO8601/num-str."""
    if value is None or isinstance(value, bool):
        return None
    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
    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 _compact(d: dict[str, Any]) -> dict[str, Any]:
    return {k: v for k, v in d.items() if v is not None}


def _pick(d: dict[str, Any], *keys: str) -> Any:
    for k in keys:
        v = d.get(k)
        if v is not None:
            return v
    return None


def _extract_item_id(url: str) -> str | None:
    m = _ITEM_ID_RE.search(url or "")
    return m.group(1) if m else None


# ── BFF JSON extraction (инлайн из domclick.py) ──────────────────────────────


def _is_block_text(text: str) -> bool:
    low = text.lower()
    return any(m in low for m in _BLOCK_MARKERS)


def _extract_json(body: str) -> dict[str, Any]:
    """Первый ``{`` … последний ``}`` из ответа BFF (raw JSON или <pre>-wrap).

    Raises:
        DomClickBlockedError: QRATOR/captcha маркеры в теле.
        ValueError: JSON не найден / не dict.
    """
    if _is_block_text(body):
        raise DomClickBlockedError("BFF: QRATOR block page detected")
    start = body.find("{")
    end = body.rfind("}")
    if start == -1 or end == -1 or end <= start:
        raise ValueError(f"No JSON object in BFF response (len={len(body)})")
    data = json.loads(body[start : end + 1])
    if not isinstance(data, dict):
        raise ValueError(f"BFF JSON is not a dict: {type(data)}")
    return data


def _build_offers_url(rooms: str, gte: int | None, lte: int | None, offset: int) -> str:
    params: list[tuple[str, str]] = [
        ("address", _EKB_ADDRESS_GUID),
        ("aids", _EKB_AREA_ID),
        ("deal_type", "sale"),
        ("category", "living"),
        ("offer_type", "flat"),
        ("rooms", rooms),
        ("sort", "qi"),
        ("sort_dir", "desc"),
        ("offset", str(offset)),
        ("limit", str(PAGE_SIZE)),
    ]
    if gte is not None:
        params.append(("sale_price__gte", str(gte)))
    if lte is not None:
        params.append(("sale_price__lte", str(lte)))
    return f"{_BFF_BASE}/api/offers/v1?{urlencode(params)}"


def _build_count_url(rooms: str, gte: int | None, lte: int | None) -> str:
    params: list[tuple[str, str]] = [
        ("address", _EKB_ADDRESS_GUID),
        ("aids", _EKB_AREA_ID),
        ("deal_type", "sale"),
        ("category", "living"),
        ("offer_type", "flat"),
        ("rooms", rooms),
        ("sort", "qi"),
        ("sort_dir", "desc"),
    ]
    if gte is not None:
        params.append(("sale_price__gte", str(gte)))
    if lte is not None:
        params.append(("sale_price__lte", str(lte)))
    return f"{_BFF_BASE}/api/offers/count/v1?{urlencode(params)}"


def _is_geo_ok(item: dict[str, Any]) -> bool:
    """bbox-guard ЕКБ. lat/lon обязательны и в пределах bbox."""
    loc = item.get("location") or {}
    lat = loc.get("lat")
    lon = loc.get("lon")
    if lat is None or lon is None:
        return False
    try:
        latf, lonf = float(lat), float(lon)
    except (ValueError, TypeError):
        return False
    return _EKB_LAT_MIN <= latf <= _EKB_LAT_MAX and _EKB_LON_MIN <= lonf <= _EKB_LON_MAX


# ── SERP item → плоский dict (инлайн из domclick.py _map_item) ────────────────


def map_serp_item(item: dict[str, Any], force_rooms: int | None) -> dict[str, Any] | None:
    """offer-item из BFF → SERP-dict. None при price<=0 / без id / ошибке."""
    try:
        price = item.get("price") or 0
        if price <= 0:
            return None
        item_id = item.get("id")
        if item_id is None:
            return None
        sid = str(item_id)

        path = item.get("path") or ""
        if path.startswith("/"):
            path = _CARD_HOST_FALLBACK + path

        loc = item.get("location") or {}
        oi = item.get("objectInfo") or {}
        house = item.get("house") or {}
        addr = item.get("address") or {}

        area_raw = oi.get("area")
        floor_raw = oi.get("floor")
        floors_raw = house.get("floors")
        year_raw = house.get("buildYear")
        sq_raw = item.get("squarePrice")

        if force_rooms is not None:
            rooms: int | None = force_rooms
        else:
            rr = oi.get("rooms")
            rooms = int(rr) if rr is not None else None

        return {
            "id": sid,
            "source_url": path,
            "price_rub": int(price),
            "square_price": int(sq_raw) if sq_raw else None,
            "area_m2": float(area_raw) if area_raw else None,
            "rooms": rooms,
            "floor": int(floor_raw) if floor_raw else None,
            "total_floors": int(floors_raw) if floors_raw else None,
            "year_built": int(year_raw) if year_raw else None,
            "lat": loc.get("lat"),
            "lon": loc.get("lon"),
            "address": addr.get("displayName"),
            "is_apartment": oi.get("isApartment"),
            "offer_region_name": item.get("offerRegionName"),
            "description": item.get("description"),
            "is_sber_collateral": item.get("isSberCollateral"),
            "published_date": item.get("publishedDate"),
        }
    except Exception:
        logger.warning("map_serp_item failed id=%r", item.get("id"), exc_info=True)
        return None


# ── SSR state extraction (инлайн из domclick_detail.py) ──────────────────────


def _scan_balanced_object(s: str, start: int) -> str:
    """От '{' по индексу start до парной '}', игнорируя скобки внутри строк."""
    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 карточки → dict.

    Raises:
        DomClickBlockedError: нет стейта + anti-bot маркеры.
        DomClickParseError: стейт не найден / не сбалансирован / json.loads упал.
    """
    match = _SSR_ASSIGN_RE.search(html)
    if match is None:
        if _is_block_text(html):
            raise DomClickBlockedError("detail: challenge page (no __SSR_STATE__)")
        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[:160]!r})"
        ) from exc
    if not isinstance(data, dict):
        raise DomClickParseError(f"__SSR_STATE__ is not an object: {type(data)}")
    return data


def _extract_owners_count(egrn: dict[str, Any]) -> int | None:
    """egrnData.owners_count. Ground truth: {status, value}; fallback на скаляр."""
    oc = egrn.get("owners_count")
    if oc is None:
        oc = egrn.get("ownersCount")
    if isinstance(oc, dict):
        return _to_int(oc.get("value"))
    return _to_int(oc)


def parse_detail_html(html: str, source_url: str) -> dict[str, Any]:
    """Парсер detail-карточки → dict обогащения. _extract_ssr_state может бросить
    Blocked/ParseError (propagate); навигация по полям обёрнута → минимум при сюрпризе.
    """
    state = _extract_ssr_state(html)
    try:
        pc = state.get("productCard") or {}
        oi = pc.get("objectInfo") or {}
        hi = (state.get("houseInfo") or {}).get("info") or {}
        egrn = pc.get("egrnData") or {}

        renovation = oi.get("renovation")
        repair_raw = renovation if isinstance(renovation, str) else None
        repair_state = _map_repair_state(repair_raw)

        living = _pos_float(oi.get("livingArea"))
        kitchen = _pos_float(oi.get("kitchenArea"))

        sale_type = (pc.get("legalOptions") or {}).get("saleType")
        owners_count = _extract_owners_count(egrn)

        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

        wall_type = hi.get("wallType") or (pc.get("house") or {}).get("wallType")
        floor_type = hi.get("floorType")
        year_built = _to_int(hi.get("buildYear"))
        views = _to_int(pc.get("viewsCount"))
        calls = _to_int(pc.get("callsCount"))

        # 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
                ct = _parse_change_time(entry.get("date"))
                price = _to_int(entry.get("price"))
                if ct is None or price is None:
                    continue
                diff = entry.get("diff")
                diff_pct = (
                    diff if isinstance(diff, int | float) and not isinstance(diff, bool) else None
                )
                price_changes.append(
                    {
                        "change_time": ct.isoformat(),
                        "price_rub": price,
                        "diff_percent": diff_pct,
                        "state": entry.get("state"),
                    }
                )

        # AVM — top-level pricePrediction (часто пусто); defensively разворачиваем.
        avm_root = state.get("pricePrediction") or {}
        avm_src: Any = 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"),
            }
        )

        demand = _compact(
            {
                "calls": calls,
                "favorites": pc.get("favoriteOfferUsersCount"),
                "duplicates": pc.get("duplicatesOfferCount"),
            }
        )
        raw_extra = _compact(
            {
                "wall_type": wall_type,
                "floor_type": floor_type,
                "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"),
                "demand": demand or None,
                "collateral": collateral,
                "collateral_sber": collateral_sber,
            }
        )

        return {
            "repair_state": repair_state,
            "repair_raw": repair_raw,
            "living_area_m2": living,
            "kitchen_area_m2": kitchen,
            "sale_type": sale_type if isinstance(sale_type, str) else None,
            "owners_count": owners_count,
            "encumbrances_clean": encumbrances_clean,
            "wall_type": wall_type,
            "floor_type": floor_type,
            "views": views,
            "calls": calls,
            "price_changes": price_changes,
            "collateral_sber": bool(collateral_sber) if collateral_sber is not None else None,
            "year_built": year_built,
            "avm": avm or None,
            "raw_extra": raw_extra,
        }
    except (DomClickBlockedError, DomClickParseError):
        raise
    except Exception:
        logger.warning("detail field navigation failed for %s — minimal", source_url, exc_info=True)
        return {}


# ── Output record ────────────────────────────────────────────────────────────


def build_record(
    serp: dict[str, Any], detail: dict[str, Any] | None, status: str
) -> dict[str, Any]:
    """SERP + detail → итоговая JSONL-запись."""
    d = detail or {}
    repair_state = d.get("repair_state")
    if not repair_state:
        # fallback-инференс из SERP-описания (как в проде для пустого структурного)
        repair_state = infer_repair_state_from_text(serp.get("description"))

    is_sber = serp.get("is_sber_collateral")
    if not is_sber and d.get("collateral_sber"):
        is_sber = True

    return {
        "id": serp.get("id"),
        "source_url": serp.get("source_url"),
        "status": status,
        "fetched_at": datetime.now(UTC).isoformat(),
        "price_rub": serp.get("price_rub"),
        "area_m2": serp.get("area_m2"),
        "rooms": serp.get("rooms"),
        "floor": serp.get("floor"),
        "total_floors": serp.get("total_floors"),
        "year_built": d.get("year_built") or serp.get("year_built"),
        "lat": serp.get("lat"),
        "lon": serp.get("lon"),
        "address": serp.get("address"),
        "repair_state": repair_state,
        "repair_raw": d.get("repair_raw"),
        "living_area_m2": d.get("living_area_m2"),
        "kitchen_area_m2": d.get("kitchen_area_m2"),
        "sale_type": d.get("sale_type"),
        "owners_count": d.get("owners_count"),
        "encumbrances_clean": d.get("encumbrances_clean"),
        "wall_type": d.get("wall_type"),
        "floor_type": d.get("floor_type"),
        "views": d.get("views"),
        "calls": d.get("calls"),
        "price_changes": d.get("price_changes") or [],
        "is_sber_collateral": is_sber,
        "avm": d.get("avm") or {},
        "raw_extra": {
            "square_price": serp.get("square_price"),
            "is_apartment": serp.get("is_apartment"),
            "offer_region_name": serp.get("offer_region_name"),
            "published_date": serp.get("published_date"),
            **(d.get("raw_extra") or {}),
        },
    }


# ── Resume / output IO ───────────────────────────────────────────────────────

_TERMINAL_STATUSES = frozenset({"ok", "blocked", "parse_fail"})


def load_done_ids(path: Path) -> set[str]:
    """id с терминальным detail-статусом из существующего JSONL (для resume)."""
    done: set[str] = set()
    if not path.exists():
        return done
    with path.open("r", encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                rec = json.loads(line)
            except json.JSONDecodeError:
                continue
            rid = rec.get("id")
            if rid and rec.get("status") in _TERMINAL_STATUSES:
                done.add(str(rid))
    return done


def append_record(path: Path, rec: dict[str, Any]) -> None:
    with path.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n")
        fh.flush()


# ── BFF transport (playwright APIRequestContext) ─────────────────────────────


async def fetch_bff_json(context: Any, url: str) -> dict[str, Any]:
    """GET BFF через context.request (тот же IP/куки браузера) → dict."""
    resp = await context.request.get(
        url,
        headers={
            "Accept": "application/json, text/plain, */*",
            "Referer": "https://ekaterinburg.domclick.ru/search?deal_type=sale",
            "Accept-Language": "ru-RU,ru;q=0.9",
        },
        timeout=60000,
    )
    body = await resp.text()
    if resp.status == 403:
        raise DomClickBlockedError(f"BFF 403 for {url}")
    return _extract_json(body)


async def get_count(context: Any, rooms: str, gte: int | None, lte: int | None) -> int:
    data = await fetch_bff_json(context, _build_count_url(rooms, gte, lte))
    res = data.get("result") or {}
    raw = res.get("snippetsCount")
    return int(raw) if raw else 0


# ── Enumerate (mirror domclick.py _sweep_bucket / _paginate) ─────────────────


async def _paginate(
    context: Any,
    rooms: str,
    gte: int | None,
    lte: int | None,
    seen: set[str],
    done: set[str],
    out: list[dict[str, Any]],
    limit_pages: int,
    target_new: int | None,
    enum_jitter: tuple[float, float],
) -> None:
    force_rooms = 0 if rooms == "st" else None
    page_idx = 0
    while page_idx < limit_pages:
        offset = page_idx * PAGE_SIZE
        if offset >= OFFSET_CAP:
            break
        data = await fetch_bff_json(context, _build_offers_url(rooms, gte, lte, offset))
        items = (data.get("result") or {}).get("items") or []
        if not items:
            break
        for item in items:
            serp = map_serp_item(item, force_rooms)
            if serp is None:
                continue
            if not _is_geo_ok(item):
                continue
            sid = serp["id"]
            if sid in seen or sid in done:
                continue
            seen.add(sid)
            out.append(serp)
            if target_new is not None and len(out) >= target_new:
                raise _EnumLimitReachedError
        page_idx += 1
        if len(items) < PAGE_SIZE:
            break
        await asyncio.sleep(random.uniform(*enum_jitter))


async def _sweep_bucket(
    context: Any,
    rooms: str,
    gte: int | None,
    lte: int | None,
    seen: set[str],
    done: set[str],
    out: list[dict[str, Any]],
    limit_pages: int,
    target_new: int | None,
    enum_jitter: tuple[float, float],
    depth: int,
) -> None:
    count = await get_count(context, rooms, gte, lte)
    if count == 0:
        return
    if count > OFFSET_CAP and depth < MAX_SPLIT_DEPTH:
        _gte = gte if gte is not None else 0
        _lte = lte if lte is not None else LTE_MAX
        span = _lte - _gte
        if span <= MIN_PRICE_SPAN:
            logger.warning(
                "rooms=%r price=[%s,%s] count=%d > cap but span<=%d — paginating as-is",
                rooms,
                gte,
                lte,
                count,
                MIN_PRICE_SPAN,
            )
            await _paginate(
                context, rooms, gte, lte, seen, done, out, limit_pages, target_new, enum_jitter
            )
            return
        mid = (_gte + _lte) // 2
        logger.info(
            "rooms=%r count=%d > cap — split [%d,%d]+[%d,%d]",
            rooms,
            count,
            _gte,
            mid,
            mid + 1,
            _lte,
        )
        await _sweep_bucket(
            context,
            rooms,
            _gte,
            mid,
            seen,
            done,
            out,
            limit_pages,
            target_new,
            enum_jitter,
            depth + 1,
        )
        await _sweep_bucket(
            context,
            rooms,
            mid + 1,
            _lte,
            seen,
            done,
            out,
            limit_pages,
            target_new,
            enum_jitter,
            depth + 1,
        )
    else:
        await _paginate(
            context, rooms, gte, lte, seen, done, out, limit_pages, target_new, enum_jitter
        )


async def enumerate_cards(
    context: Any,
    done: set[str],
    limit_pages: int,
    target_new: int | None,
    enum_jitter: tuple[float, float],
) -> list[dict[str, Any]]:
    """Свод ЕКБ-вторички через BFF: все room-бакеты, dedup, skip done."""
    out: list[dict[str, Any]] = []
    seen: set[str] = set()
    try:
        for bucket in ROOM_BUCKETS:
            logger.info("enumerate: bucket rooms=%r (collected=%d)", bucket, len(out))
            await _sweep_bucket(
                context,
                bucket,
                None,
                None,
                seen,
                done,
                out,
                limit_pages,
                target_new,
                enum_jitter,
                0,
            )
    except _EnumLimitReachedError:
        logger.info("enumerate: reached target=%d new cards — stop sweep", target_new)
    except DomClickBlockedError as exc:
        logger.error(
            "enumerate: QRATOR block (%s) — aborting sweep, keeping %d collected", exc, len(out)
        )
    logger.info("enumerate: done, %d new cards collected", len(out))
    return out


# ── Detail transport ─────────────────────────────────────────────────────────


async def fetch_card_html(
    page: Any, url: str, render_wait: tuple[float, float]
) -> tuple[str, str, int]:
    """goto → jittered render-wait → re-fetch сырого HTML. (html, title, status)."""
    resp = await page.goto(url, wait_until="domcontentloaded", timeout=70000)
    await asyncio.sleep(random.uniform(*render_wait))
    title = ""
    try:
        title = (await page.title()) or ""
    except Exception:
        title = ""
    html = await page.evaluate(_FETCH_RAW_JS)
    status = resp.status if resp else 0
    return html, title, status


async def process_card(
    page: Any,
    serp: dict[str, Any],
    render_wait: tuple[float, float],
) -> tuple[str, dict[str, Any] | None]:
    """Один detail-fetch с одним ретраем на блок. → (status, detail|None)."""
    url = serp["source_url"]
    for attempt in (1, 2):
        try:
            html, title, http_status = await fetch_card_html(page, url, render_wait)
            if http_status == 403 or _is_block_text(title) or _is_block_text(html[:4096]):
                raise DomClickBlockedError(f"403/challenge title={title!r}")
            detail = parse_detail_html(html, url)
            return "ok", detail
        except DomClickBlockedError as exc:
            if attempt == 1:
                wait = random.uniform(45, 90)
                logger.warning(
                    "card %s blocked (%s) — retry once after %.0fs", serp["id"], exc, wait
                )
                await asyncio.sleep(wait)
                continue
            logger.warning("card %s still blocked after retry — status=blocked", serp["id"])
            return "blocked", None
        except DomClickParseError as exc:
            logger.warning("card %s parse_fail: %s", serp["id"], exc)
            return "parse_fail", None
        except Exception as exc:
            if attempt == 1:
                logger.warning("card %s fetch error (%s) — retry once", serp["id"], exc)
                await asyncio.sleep(random.uniform(8, 15))
                continue
            logger.warning("card %s fetch error after retry — parse_fail", serp["id"])
            return "parse_fail", None
    return "parse_fail", None


# ── Orchestration ────────────────────────────────────────────────────────────


class Summary:
    """Счётчики прогона (для финального лога + KeyboardInterrupt flush)."""

    def __init__(self) -> None:
        self.enumerated = 0
        self.ok = 0
        self.blocked = 0
        self.parse_fail = 0
        self.skipped_done = 0
        self.started = datetime.now(UTC)

    def log_final(self) -> None:
        elapsed = (datetime.now(UTC) - self.started).total_seconds()
        logger.info(
            "SUMMARY enumerated=%d ok=%d blocked=%d parse_fail=%d skipped_done=%d elapsed=%.0fs",
            self.enumerated,
            self.ok,
            self.blocked,
            self.parse_fail,
            self.skipped_done,
            elapsed,
        )


async def run(args: argparse.Namespace) -> None:
    out_path = Path(args.out)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    done = load_done_ids(out_path)
    logger.info("resume: %d ids already done in %s", len(done), out_path)

    summary = Summary()
    render_wait = (args.render_min, args.render_max)
    target_new = args.limit if args.limit and args.limit > 0 else None
    enum_jitter = (1.5, 4.0)

    # ВАЖНО: launch(headless=True) = СТАРЫЙ headless → QRATOR 403 даже с дом. IP.
    # Запускаемся с headless=False, но добавляем --headless=new (новый headless,
    # окно не видно, детект проходит). --headed → без флага = видимое окно.
    launch_args = ["--disable-blink-features=AutomationControlled"]
    if not args.headed:
        launch_args.append("--headless=new")

    async with async_playwright() as pw:
        browser = await pw.chromium.launch(headless=False, args=launch_args)
        context = await browser.new_context(
            user_agent=_USER_AGENT,
            locale="ru-RU",
            timezone_id="Asia/Yekaterinburg",
            viewport={"width": 1366, "height": 768},
            extra_http_headers={"Accept-Language": "ru-RU,ru;q=0.9"},
        )
        page = await context.new_page()

        # Warm-up: устанавливаем QRATOR-куки .domclick.ru перед BFF-запросами.
        try:
            await page.goto(_WARMUP_URL, wait_until="domcontentloaded", timeout=70000)
            await asyncio.sleep(random.uniform(2.0, 4.0))
            logger.info("warmup: loaded %s", _WARMUP_URL)
        except Exception as exc:
            logger.warning("warmup navigation failed (%s) — continuing", exc)

        # 1. Enumerate.
        cards = await enumerate_cards(context, done, args.limit_pages, target_new, enum_jitter)
        summary.enumerated = len(cards)

        if args.enumerate_only:
            logger.info("enumerate-only: writing %d SERP records (status=enumerated)", len(cards))
            for serp in cards:
                append_record(out_path, build_record(serp, None, "enumerated"))
            summary.log_final()
            await context.close()
            await browser.close()
            return

        # 2. Detail.
        next_break_at = random.randint(25, 40)
        processed_since_break = 0
        try:
            for idx, serp in enumerate(cards, start=1):
                if args.limit and (summary.ok + summary.blocked + summary.parse_fail) >= args.limit:
                    break
                sid = serp["id"]
                if sid in done:
                    summary.skipped_done += 1
                    continue

                status, detail = await process_card(page, serp, render_wait)
                rec = build_record(serp, detail, status)
                append_record(out_path, rec)
                done.add(sid)

                if status == "ok":
                    summary.ok += 1
                elif status == "blocked":
                    summary.blocked += 1
                else:
                    summary.parse_fail += 1

                logger.info(
                    "[%d/%d] id=%s status=%s repair=%s living=%s owners=%s price_changes=%d "
                    "| totals ok=%d blocked=%d fail=%d",
                    idx,
                    len(cards),
                    sid,
                    status,
                    rec.get("repair_state"),
                    rec.get("living_area_m2"),
                    rec.get("owners_count"),
                    len(rec.get("price_changes") or []),
                    summary.ok,
                    summary.blocked,
                    summary.parse_fail,
                )

                processed_since_break += 1
                if processed_since_break >= next_break_at:
                    brk = random.uniform(60, 180)
                    logger.info(
                        "human break: sleeping %.0fs after %d cards", brk, processed_since_break
                    )
                    await asyncio.sleep(brk)
                    processed_since_break = 0
                    next_break_at = random.randint(25, 40)

                # Пауза между карточками (главный темп-контроль).
                await asyncio.sleep(random.uniform(args.min_delay, args.max_delay))
        except (KeyboardInterrupt, asyncio.CancelledError):
            logger.warning("interrupted — flushing summary and exiting gracefully")
        finally:
            summary.log_final()
            await context.close()
            await browser.close()


def build_arg_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="DomClick EKB вторичка local runner (home IP, human-paced).",
    )
    p.add_argument("--out", default="domclick_ekb.jsonl", help="JSONL output (append, resumable)")
    p.add_argument("--limit", type=int, default=None, help="cap total cards (для теста)")
    p.add_argument("--min-delay", type=float, default=12.0, help="мин. пауза между карточками, с")
    p.add_argument("--max-delay", type=float, default=30.0, help="макс. пауза между карточками, с")
    p.add_argument("--render-min", type=float, default=6.0, help="мин. render-wait после goto, с")
    p.add_argument("--render-max", type=float, default=10.0, help="макс. render-wait после goto, с")
    p.add_argument("--limit-pages", type=int, default=100, help="макс. страниц на room-бакет")
    p.add_argument("--enumerate-only", action="store_true", help="только перечислить, без detail")
    p.add_argument(
        "--headed",
        action="store_true",
        help="видимое окно браузера (debug); по умолчанию new-headless (окно скрыто)",
    )
    return p


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        datefmt="%H:%M:%S",
    )
    args = build_arg_parser().parse_args()
    if args.max_delay < args.min_delay:
        args.max_delay = args.min_delay
    if args.render_max < args.render_min:
        args.render_max = args.render_min

    # Graceful Ctrl+C: asyncio.run превратит KeyboardInterrupt в CancelledError
    # внутри run(); финальный flush — в finally.
    try:
        asyncio.run(run(args))
    except KeyboardInterrupt:
        logger.warning("KeyboardInterrupt at top level — exiting")


if __name__ == "__main__":
    main()