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 (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, и detail идут
ОДНИМ браузерным транспортом ``page.goto`` (verified 2026-06-27):
  • enumerate — ``page.goto(SERP_url)`` на фронтовый
    ``ekaterinburg.domclick.ru/search?...`` (SSR). Прямой BFF
    ``bff-search-web.domclick.ru/api/offers/v1`` режет QRATOR-403 уже с
    домашнего IP, а фронтовый SERP грузится чисто (тот же транспорт, что
    card-detail). id офферов берём из DOM (``a[href*="/card/"]``), total —
    из ``<title>`` ("… N объявлений"). Прямого offers/v1 XHR фронт НЕ шлёт.
    Пагинация — параметром ``offset`` (page-size 20); при total>2000 —
    price-bisection бакета (границы тоже из title-total).
  • detail — ``page.goto(card_url)`` → jittered wait → ``page.evaluate`` с
    ``fetch(location.href)`` чтобы получить СЫРОЙ HTML (SPA удаляет
    ``window.__SSR_STATE__`` после гидрации, поэтому нужен повторный fetch).
    Layer-A (price/area/rooms/floor/total_floors/lat/lon/address) теперь
    берётся ИЗ card-SSR (enumerate отдаёт только id+url).

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

ВАЖНО — ЧЕЛОВЕКОПОДОБНЫЙ ТЕМП (явное требование оператора)
=========================================================
Домашний 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) ──────────────────

_EKB_AREA_ID = "20561"  # aids= фильтр ЕКБ-вторички на фронтовом SERP
_SERP_HOST = "https://ekaterinburg.domclick.ru"
_WARMUP_URL = "https://ekaterinburg.domclick.ru/"

# Порядок бакетов комнат влияет на логи. КРИТИЧНО: "5+" обязан кодироваться как
# "5%2B" — литеральный '+' в query фронт читает как пробел → фильтр игнорится и
# отдаётся ПОЛНЫЙ свод (verified 2026-06-27). urlencode("5+") → "5%2B" — корректно.
# "st" фронт принимает как студию. "5" в одиночку фронт НЕ понимает (только "5+").
ROOM_BUCKETS: tuple[str, ...] = ("st", "1", "2", "3", "4", "5+")

OFFSET_CAP: int = 2000  # практический потолок offset фронтового SERP
_SERP_PAGE_SIZE: int = 20  # офферов на страницу SERP (verified)
# total из <title>: "... 2 269 объявлений" — число с разрядами (пробел/nbsp/thin).
_SERP_TOTAL_RE = re.compile(r"(\d[\d\s]*)\s*объявлен")
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)

# Гео-фильтр ЕКБ обеспечивает aids=20561 в SERP-URL; bbox-guard убран вместе с
# BFF-enumerate (фронт отдаёт только id+url, координаты — из card-SSR).

# Маркеры 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(); }"
)

# JS: id+url всех офферов из DOM фронтового SERP (anchor'ы /card/...<8+ digits>).
# Дедуп через Set по сериализованному объекту. Фронт — SSR, offers/v1 XHR НЕ шлёт.
_SERP_IDS_JS = (
    "() => [...new Set([...document.querySelectorAll('a[href*=\"/card/\"]')]"
    ".map(a => { const h=a.getAttribute('href')||'';"
    " const m=h.match(/\\/card\\/[a-z_]*?(\\d{8,})/);"
    " return m ? {id:m[1], url: h.startsWith('http')? h :"
    " 'https://ekaterinburg.domclick.ru'+h} : null; })"
    ".filter(Boolean).map(o => JSON.stringify(o)))].map(s => JSON.parse(s))"
)


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


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


# ── Block detection / SERP URL & parsing ─────────────────────────────────────


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


def _build_serp_url(rooms: str, gte: int | None, lte: int | None, offset: int) -> str:
    """URL фронтового SERP. urlencode кодирует "5+" → "5%2B" (иначе фронт читает
    '+' как пробел и игнорит room-фильтр)."""
    params: list[tuple[str, str]] = [
        ("deal_type", "sale"),
        ("category", "living"),
        ("offer_type", "flat"),
        ("aids", _EKB_AREA_ID),
        ("from", "topline2020"),
        ("offset", str(offset)),
        ("rooms", rooms),
    ]
    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"{_SERP_HOST}/search?{urlencode(params)}"


def _parse_serp_total(title: str) -> int | None:
    """total из <title> ("… 2 269 объявлений"). None если числа нет."""
    if not title:
        return None
    m = _SERP_TOTAL_RE.search(title)
    if not m:
        return None
    digits = re.sub(r"\D", "", m.group(1))
    return int(digits) if digits else None


def _canon_card_url(sid: str, url: str | None) -> str:
    """Канонизируем source_url к /card/sale__flat__<id>, если href неполный."""
    if url and url.startswith("http") and "/card/" in url:
        return url
    return f"{_SERP_HOST}/card/sale__flat__{sid}"


def _serp_offer_to_dict(raw: Any) -> dict[str, Any] | None:
    """offer из _SERP_IDS_JS ({id,url}) → минимальный SERP-dict {id, source_url}.
    Layer-A (price/area/...) теперь добирается из card-SSR в parse_detail_html."""
    if not isinstance(raw, dict):
        return None
    sid = raw.get("id")
    if sid is None:
        return None
    sid = str(sid)
    if not sid.isdigit():
        return None
    return {"id": sid, "source_url": _canon_card_url(sid, raw.get("url"))}


# ── 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"))

        # Layer-A из card-SSR (раньше шёл из BFF serp; теперь enumerate даёт
        # только id+url). Пути verified 2026-06-27: цена/squarePrice — в
        # priceInfo; area/rooms/floor — в objectInfo; total_floors — houseInfo;
        # lat/lon/address — productCard.address(.position).
        price_info = pc.get("priceInfo") or {}
        address = pc.get("address") or {}
        position = address.get("position") or {}
        price_rub = _to_int(price_info.get("price"))
        square_price = _to_int(price_info.get("squarePrice"))
        area_m2 = _pos_float(oi.get("area"))
        rooms = _to_int(oi.get("rooms"))
        floor = _to_int(oi.get("floor"))
        total_floors = _to_int(hi.get("floors")) or _to_int(
            (pc.get("house") or {}).get("info", {}).get("floors")
        )
        lat = _to_float(position.get("lat"))
        lon = _to_float(position.get("lon"))
        address_name = address.get("displayName")
        is_apartment = oi.get("isApartment")
        description = oi.get("description")

        # 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 {
            # Layer-A (из card-SSR; раньше из BFF serp).
            "price_rub": price_rub,
            "square_price": square_price,
            "area_m2": area_m2,
            "rooms": rooms,
            "floor": floor,
            "total_floors": total_floors,
            "lat": lat,
            "lon": lon,
            "address": address_name,
            "is_apartment": is_apartment,
            "description": description,
            # Layer-B.
            "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-инференс из описания (теперь из card-SSR; serp-описания больше нет)
        repair_state = infer_repair_state_from_text(d.get("description") or serp.get("description"))

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

    # Layer-A: приоритет card-SSR (d), фолбэк на serp (на случай если serp когда-то
    # снова начнёт отдавать поля). enumerate теперь даёт только id+source_url.
    return {
        "id": serp.get("id"),
        "source_url": serp.get("source_url"),
        "status": status,
        "fetched_at": datetime.now(UTC).isoformat(),
        "price_rub": d.get("price_rub") or serp.get("price_rub"),
        "area_m2": d.get("area_m2") or serp.get("area_m2"),
        "rooms": d.get("rooms") if d.get("rooms") is not None else serp.get("rooms"),
        "floor": d.get("floor") or serp.get("floor"),
        "total_floors": d.get("total_floors") or serp.get("total_floors"),
        "year_built": d.get("year_built") or serp.get("year_built"),
        "lat": d.get("lat") or serp.get("lat"),
        "lon": d.get("lon") or serp.get("lon"),
        "address": d.get("address") or 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": d.get("square_price") or serp.get("square_price"),
            "is_apartment": (
                d.get("is_apartment")
                if d.get("is_apartment") is not None
                else serp.get("is_apartment")
            ),
            "offer_region_name": serp.get("offer_region_name"),
            "published_date": serp.get("published_date"),
            **(d.get("raw_extra") or {}),
            **({"sold_similar": d["sold_similar"]} if d.get("sold_similar") else {}),
        },
    }


# ── 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()


# ── Enumerate cache (полный список id, чтобы НЕ гонять SERP заново) ────────────


def write_enum_cache(path: str, offers: list[dict[str, Any]]) -> None:
    """Перезаписать кэш-файл полным списком offer-id + sentinel завершённости.

    Каждый offer пишется как {"id":..., "source_url":...} (только 2 поля),
    финальной строкой — sentinel {"_enum_complete": true, "count": N}. Sentinel
    означает «enumerate завершился полностью» — без него load_enum_cache считает
    кэш неполным и заставляет re-enumerate.
    """
    cache_path = Path(path)
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    with cache_path.open("w", encoding="utf-8") as fh:
        for serp in offers:
            row = {"id": serp.get("id"), "source_url": serp.get("source_url")}
            fh.write(json.dumps(row, ensure_ascii=False) + "\n")
        fh.write(json.dumps({"_enum_complete": True, "count": len(offers)}) + "\n")
        fh.flush()
    logger.info("enum-cache: wrote %d ids → %s", len(offers), path)


def load_enum_cache(path: str) -> list[dict[str, Any]] | None:
    """Загрузить полный список offer-id из кэша. None если файла нет/кэш неполный.

    Кэш валиден ТОЛЬКО если последняя валидная строка — sentinel с
    ``_enum_complete == True`` (enumerate завершился). Битые строки пропускаются;
    нет sentinel / пустой / повреждённый → None (заставляет re-enumerate).
    """
    cache_path = Path(path)
    if not cache_path.exists():
        return None
    records: list[dict[str, Any]] = []
    with cache_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
            if isinstance(rec, dict):
                records.append(rec)
    if not records:
        return None
    last = records[-1]
    if not (isinstance(last, dict) and last.get("_enum_complete") is True):
        return None  # кэш неполный (enumerate не завершился) → re-enumerate
    offers: list[dict[str, Any]] = []
    for rec in records[:-1]:
        sid = rec.get("id")
        if sid is None:
            continue
        offers.append({"id": str(sid), "source_url": rec.get("source_url")})
    logger.info("enum-cache: loaded %d ids (complete)", len(offers))
    return offers


# ── SERP transport (playwright page navigation, чистый браузерный транспорт) ──


async def fetch_serp_page(
    page: Any,
    url: str,
    render_wait: tuple[float, float],
) -> tuple[list[dict[str, Any]], int | None, int]:
    """Навигация на фронтовый SERP → (offers, total, http_status).

    offers — список {id, source_url} из DOM; total — из <title>.
    Raises:
        DomClickBlockedError: 403 или QRATOR-маркеры в title.
    """
    resp = await page.goto(url, wait_until="domcontentloaded", timeout=70000)
    # SERP рендерит карточки клиентской гидрацией через ~6-12с после domcontentloaded.
    # Фиксированный sleep хрупок: при медленном ответе DOM ещё пуст → 0 ids (silent
    # 0-coverage). Ждём появления card-ссылок ЯВНО; genuinely-пустой ценовой band →
    # timeout → 0 offers (корректно). render_wait остаётся коротким settle поверх.
    try:
        await page.wait_for_selector("a[href*='/card/']", timeout=12000)
    except Exception:
        pass  # пустой band или медленный рендер — читаем как есть (0 offers)
    await asyncio.sleep(random.uniform(*render_wait))  # settle: дорисовать все 20 + title
    status = resp.status if resp else 0
    title = ""
    try:
        title = (await page.title()) or ""
    except Exception:
        title = ""
    if status == 403 or _is_block_text(title):
        raise DomClickBlockedError(f"SERP block status={status} title={title!r}")
    raw = await page.evaluate(_SERP_IDS_JS)
    offers: list[dict[str, Any]] = []
    if isinstance(raw, list):
        for item in raw:
            serp = _serp_offer_to_dict(item)
            if serp is not None:
                offers.append(serp)
    total = _parse_serp_total(title)
    return offers, total, status


# ── Enumerate (mirror domclick.py _sweep_bucket / _paginate, на title-total) ──


def _add_offers(
    offers: list[dict[str, Any]],
    seen: set[str],
    done: set[str],
    out: list[dict[str, Any]],
    target_new: int | None,
) -> int:
    """Добавить новые офферы в out (dedup по seen/done). → число добавленных.
    Бросает _EnumLimitReachedError при достижении target_new."""
    added = 0
    for serp in offers:
        sid = serp["id"]
        if sid in seen or sid in done:
            continue
        seen.add(sid)
        out.append(serp)
        added += 1
        if target_new is not None and len(out) >= target_new:
            raise _EnumLimitReachedError
    return added


async def _paginate(
    page: Any,
    rooms: str,
    gte: int | None,
    lte: int | None,
    total: int | None,
    seen: set[str],
    done: set[str],
    out: list[dict[str, Any]],
    limit_pages: int,
    target_new: int | None,
    render_wait: tuple[float, float],
    enum_jitter: tuple[float, float],
    start_page: int,
) -> None:
    """Пагинация SERP с offset=start_page*20 далее. Стоп: достигли total/cap,
    кончились limit_pages, или страница не дала новых id."""
    bound = OFFSET_CAP if total is None else min(total, OFFSET_CAP)
    page_idx = start_page
    while page_idx < limit_pages:
        offset = page_idx * _SERP_PAGE_SIZE
        if offset >= bound:
            break
        await asyncio.sleep(random.uniform(*enum_jitter))
        offers, _, _ = await fetch_serp_page(
            page, _build_serp_url(rooms, gte, lte, offset), render_wait
        )
        if not offers:
            break
        # NB: НЕ break при added==0. При resume первые страницы бакета — это топ-qi
        # офферы, уже лежащие в `done` → added=0, но новые id есть на след. страницах.
        # Терминаторы — только пустая страница (выше) и offset>=bound (total/cap).
        _add_offers(offers, seen, done, out, target_new)
        page_idx += 1


async def _sweep_bucket(
    page: 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,
    render_wait: tuple[float, float],
    enum_jitter: tuple[float, float],
    depth: int,
) -> None:
    """offset=0 → total + page-0 офферы. total>cap → price-bisection, иначе
    пагинируем остаток. total читаем из <title> (чистый фронтовый транспорт)."""
    offers0, total, _ = await fetch_serp_page(
        page, _build_serp_url(rooms, gte, lte, 0), render_wait
    )
    _add_offers(offers0, seen, done, out, target_new)
    if not total:
        # нет числа / 0 объявлений: офферов на page-0 нет → выходим; есть → допагинируем.
        if offers0:
            await _paginate(
                page,
                rooms,
                gte,
                lte,
                None,
                seen,
                done,
                out,
                limit_pages,
                target_new,
                render_wait,
                enum_jitter,
                1,
            )
        return
    if total > 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:
            mid = (_gte + _lte) // 2
            logger.info(
                "rooms=%r total=%d > cap — split [%d,%d]+[%d,%d]",
                rooms,
                total,
                _gte,
                mid,
                mid + 1,
                _lte,
            )
            await _sweep_bucket(
                page,
                rooms,
                _gte,
                mid,
                seen,
                done,
                out,
                limit_pages,
                target_new,
                render_wait,
                enum_jitter,
                depth + 1,
            )
            await _sweep_bucket(
                page,
                rooms,
                mid + 1,
                _lte,
                seen,
                done,
                out,
                limit_pages,
                target_new,
                render_wait,
                enum_jitter,
                depth + 1,
            )
            return
        logger.warning(
            "rooms=%r price=[%s,%s] total=%d > cap но span<=%d — пагинируем как есть",
            rooms,
            gte,
            lte,
            total,
            MIN_PRICE_SPAN,
        )
    await _paginate(
        page,
        rooms,
        gte,
        lte,
        total,
        seen,
        done,
        out,
        limit_pages,
        target_new,
        render_wait,
        enum_jitter,
        1,
    )


async def enumerate_cards(
    page: Any,
    done: set[str],
    limit_pages: int,
    target_new: int | None,
    render_wait: tuple[float, float],
    enum_jitter: tuple[float, float],
) -> list[dict[str, Any]]:
    """Свод ЕКБ-вторички через фронтовый SERP: все 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(
                page,
                bucket,
                None,
                None,
                seen,
                done,
                out,
                limit_pages,
                target_new,
                render_wait,
                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


# JS: дозабор offer-card v3 (AVM price_prediction + опц. sold_similar) из
# контекста УЖЕ загруженной страницы — та же сессия/куки, что прошли QRATOR.
_OFFER_CARD_EXTRAS_JS = """async (args) => {
  const [id, withSold, soldLimit] = args;
  const out = {};
  const get = async (u) => {
    try {
      const r = await fetch(u, {credentials: 'include'});
      return r.ok ? await r.json() : {__status: r.status};
    } catch (e) {
      return {__err: String(e)};
    }
  };
  out.price_prediction = await get(
    'https://offer-card.domclick.ru/api/v3/offers/price_prediction?offer_id=' + id
  );
  if (withSold) {
    out.sold_similar = await get(
      'https://offer-card.domclick.ru/api/v3/offers/sold_similar/flat/' + id +
      '?in_house=false&limit=' + soldLimit + '&offset=0'
    );
  }
  return out;
}"""


def _parse_offer_card_avm(price_prediction: Any) -> dict[str, Any] | None:
    """price_prediction.result → AVM-форма (как d["avm"] из SSR). None если пусто."""
    if not isinstance(price_prediction, dict):
        return None
    result = price_prediction.get("result")
    if not isinstance(result, dict):
        return None
    market_price = result.get("market_price")
    if market_price is None:
        return None
    avm = {
        "market_price": market_price,
        "min": result.get("min_market_price"),
        "max": result.get("max_market_price"),
        "repair_quality": result.get("repair_quality"),
        "rent_long": result.get("rent_long"),
        "rent_short": result.get("rent_short"),
    }
    return {k: v for k, v in avm.items() if v is not None}


def _parse_offer_card_sold(sold_similar: Any) -> dict[str, Any] | None:
    """sold_similar → {total, items}. None если нет result.items."""
    if not isinstance(sold_similar, dict):
        return None
    result = sold_similar.get("result")
    if not isinstance(result, dict):
        return None
    items = result.get("items")
    if not isinstance(items, list):
        return None
    pagination = sold_similar.get("pagination") or {}
    total = pagination.get("total") if isinstance(pagination, dict) else None
    return {"total": total, "items": items}


async def fetch_offer_card_extras(
    page: Any,
    item_id: str,
    with_sold_similar: bool,
    sold_limit: int,
) -> dict[str, Any]:
    """Дозабор AVM (всегда) + sold_similar (опц.) с offer-card v3 из контекста
    загруженной страницы. None-safe: любая ошибка/__status/__err → поле None."""
    raw = await page.evaluate(_OFFER_CARD_EXTRAS_JS, [str(item_id), with_sold_similar, sold_limit])
    if not isinstance(raw, dict):
        return {"avm": None, "sold_similar": None}
    avm = _parse_offer_card_avm(raw.get("price_prediction"))
    sold = _parse_offer_card_sold(raw.get("sold_similar")) if with_sold_similar else None
    return {"avm": avm, "sold_similar": sold}


async def process_card(
    page: Any,
    serp: dict[str, Any],
    render_wait: tuple[float, float],
    with_sold_similar: bool,
    sold_limit: int,
) -> 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)
            try:
                extras = await fetch_offer_card_extras(
                    page, serp["id"], with_sold_similar, sold_limit
                )
                if extras.get("avm"):
                    detail["avm"] = extras["avm"]
                if extras.get("sold_similar"):
                    detail["sold_similar"] = extras["sold_similar"]
            except Exception as exc:
                logger.warning("card %s offer-card extras failed: %s", serp["id"], exc)
            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
    # SERP-навигация требует render-wait на SSR; enum_jitter — доп. пейсинг между
    # страницами (SERP-навигаций много при свипе, бережём home-IP).
    enum_render = (max(2.0, args.render_min - 2.0), max(3.0, args.render_max - 2.0))
    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 перед SERP-навигацией.
        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 через фронтовый SERP (тот же чистый браузерный транспорт).
        # enum-cache: полный список id живёт в файле → следующий запуск ПРОПУСКАЕТ
        # enumerate (~1.5-2ч) и сразу идёт в detail. detail-resume по `done`
        # композится поверх: кэш даёт ВСЕ id, `done` (из JSONL) скипает готовые.
        cards: list[dict[str, Any]] | None = None
        if args.enum_cache:
            cards = load_enum_cache(args.enum_cache)
        if cards is None:
            cards = await enumerate_cards(
                page, done, args.limit_pages, target_new, enum_render, enum_jitter
            )
            # Sentinel пишем ТОЛЬКО после реально завершённого enumerate (offers
            # полный) — прерванный/упавший свод кэш не перезаписывает.
            if args.enum_cache:
                write_enum_cache(args.enum_cache, cards)
        else:
            logger.info("enum-cache HIT: %d ids — enumerate ПРОПУЩЕН", len(cards))
        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, args.with_sold_similar, args.sold_limit
                )
                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(args.break_min, args.break_max)
                    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("--break-min", type=float, default=40.0, help="мин. длительность human-break, с")
    p.add_argument(
        "--break-max", type=float, default=60.0, help="макс. длительность human-break, с"
    )
    p.add_argument("--limit-pages", type=int, default=100, help="макс. страниц на room-бакет")
    p.add_argument("--enumerate-only", action="store_true", help="только перечислить, без detail")
    p.add_argument(
        "--enum-cache",
        type=str,
        default=None,
        help="файл-кэш полного списка id (enumerate один раз → resume detail поверх)",
    )
    p.add_argument(
        "--with-sold-similar",
        action="store_true",
        help="дозабор sold_similar с offer-card v3 (+1 запрос/карточку)",
    )
    p.add_argument(
        "--sold-limit", type=int, default=5, help="limit для sold_similar (если включён)"
    )
    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()