"""DomClick.ru Layer B — detail-card enrichment (#1846 follow-up).

Layer A (domclick.py) собирает базовые поля через BFF JSON API. Layer B дочитывает
страницу карточки (``https://ekaterinburg.domclick.ru/card/sale__flat__<id>``) и
обогащает listings: renovation/repair_state, living/kitchen area, balconies,
sale_type, owners_count, encumbrances, year_built, views и priceHistory.

Транспорт — BrowserFetcher через QRATOR-clean прокси (source="cian"): card-хосты
DomClick заблокированы QRATOR через domclick/avito-прокси, но чисты через cian
(verified live 2026-06-27). curl_cffi на card-хосте отдаёт 401/стаб. fetch_detail
принимает уже открытый browser_fetcher; жизненным циклом сессии и выбором source
владеет оркестратор.

SSR-quirk
---------
HTML карточки встраивает ``window.__SSR_STATE__ = {...}`` — это JS object literal,
а НЕ чистый JSON: в value-позиции встречается bare ``undefined`` (json.loads падает).
_extract_ssr_state делает balanced-brace scan (с пропуском скобок внутри строк) и
заменяет bare ``undefined`` → ``null`` строго в value-позиции, не трогая строки.

Форма SSR-стейта подтверждена на живой карточке 2075729321 (2026-06-27):
productCard.objectInfo.{renovation,livingArea,kitchenArea}, productCard.priceInfo.
priceHistory ({date ISO8601+tz, price, diff, state}), productCard.egrnData
(snake_case owners_count/collateral/collateral_sber), productCard.legalOptions.
saleType, productCard.viewsCount/callsCount, houseInfo.info.{wallType,floorType},
и ТОП-УРОВНЕМ pricePrediction (DomClick AVM → raw_extra.avm).
Schedule для Layer B ships DORMANT — включается вручную после prod-smoke.
"""

from __future__ import annotations

import json
import logging
import re
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.services.scrapers.domclick_exceptions import DomClickBlockedError, DomClickParseError
from app.services.scrapers.repair_state_normalizer import (
    infer_repair_state_from_text,
    normalize_repair_state,
)

if TYPE_CHECKING:
    from app.services.scrapers.browser_fetcher import BrowserFetcher

logger = logging.getLogger(__name__)

# ── Маркеры anti-bot challenge (DataDome / QRATOR) ───────────────────────────
# Зеркало идеи Layer A: если __SSR_STATE__ отсутствует И страница похожа на
# challenge — это блок, а не parse-failure.
_BLOCK_MARKERS: tuple[str, ...] = ("qrator", "captcha", "datadome", "access denied")

# ── item_id из URL карточки: .../card/sale__flat__2075729321 ─────────────────
_ITEM_ID_RE = re.compile(r"sale__flat__(\d+)")

# ── SSR assignment marker ────────────────────────────────────────────────────
_SSR_ASSIGN_RE = re.compile(r"window\.__SSR_STATE__\s*=\s*")

# ── bare `undefined` (value-позиция) → null, не трогая строки ─────────────────
_UNDEFINED_RE = re.compile(r"(?<=[:\[,\s])undefined(?=[\s,\]\}])")

# ── renovation (RU) → каноничный repair_state enum ───────────────────────────
# ЛОКАЛЬНАЯ карта (не трогаем shared _RAW_TO_ENUM в repair_state_normalizer).
# Значения уже каноничны; прогоняются через normalize_repair_state для валидации.
_RENOVATION_MAP: dict[str, str] = {
    "евроремонт": "good",
    "косметический": "standard",
    "дизайнерский": "excellent",
    "черновая": "needs_repair",
    "без отделки": "needs_repair",
    "требует ремонта": "needs_repair",
    "предчистовая": "needs_repair",
}


# ── DomClickDetailEnrichment ──────────────────────────────────────────────────
@dataclass
class DomClickDetailEnrichment:
    """Результат парсинга detail-карточки DomClick (Layer B).

    save_detail_enrichment() пишет подмножество в listings (+ offer_price_history).
    Все поля None-safe: отсутствующее → None (не крэш).
    """

    item_id: str | None = None
    source_url: str | None = None

    repair_state: str | None = None  # enum: needs_repair/standard/good/excellent
    repair_type: str | None = None  # сырое RU-значение renovation
    living_area_m2: float | None = None
    kitchen_area_m2: float | None = None
    balconies_count: int | None = None
    has_balcony: bool | None = None
    sale_type: str | None = None
    owners_count: int | None = None
    encumbrances_clean: bool | None = None  # True=чисто, False=есть обременение
    year_built: int | None = None
    views_total: int | None = None

    price_changes: list[dict[str, Any]] = field(default_factory=list)
    raw_extra: dict[str, Any] = field(default_factory=dict)


# ── SSR state extraction ──────────────────────────────────────────────────────


def _scan_balanced_object(s: str, start: int) -> str:
    """Сканирует от '{' по индексу ``start`` до парной '}'.

    Отслеживает глубину по {/}, ИГНОРИРУЯ скобки внутри JS-строк (одинарные и
    двойные кавычки с ``\\``-эскейпами). Возвращает литерал ``{...}`` включительно.

    Raises:
        DomClickParseError: если парная скобка не найдена (truncated HTML).
    """
    depth = 0
    in_string = False
    quote_char = ""
    escaped = False
    for i in range(start, len(s)):
        ch = s[i]
        if in_string:
            if escaped:
                escaped = False
            elif ch == "\\":
                escaped = True
            elif ch == quote_char:
                in_string = False
            continue
        if ch in ('"', "'"):
            in_string = True
            quote_char = ch
        elif ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                return s[start : i + 1]
    raise DomClickParseError("__SSR_STATE__ object not balanced (no matching brace)")


def _extract_ssr_state(html: str) -> dict[str, Any]:
    """Извлекает ``window.__SSR_STATE__`` object-литерал из HTML карточки.

    Шаги:
      1. Находим присваивание ``window.__SSR_STATE__ =``. Если нет — различаем
         challenge-страницу (→ DomClickBlockedError) и просто отсутствие
         стейта (→ DomClickParseError).
      2. Balanced-brace scan от первого '{' → литерал.
      3. Санитайз bare ``undefined`` → ``null`` (только value-позиция).
      4. json.loads.

    Raises:
        DomClickBlockedError: страница без стейта + маркеры anti-bot.
        DomClickParseError: стейт не найден / не сбалансирован / json.loads упал.
    """
    match = _SSR_ASSIGN_RE.search(html)
    if match is None:
        html_lower = html.lower()
        if any(marker in html_lower for marker in _BLOCK_MARKERS):
            raise DomClickBlockedError(
                "DomClick detail: challenge page detected (no __SSR_STATE__, "
                f"markers checked: {_BLOCK_MARKERS})"
            )
        raise DomClickParseError("__SSR_STATE__ not found")

    brace_start = html.find("{", match.end())
    if brace_start == -1:
        raise DomClickParseError("__SSR_STATE__ assignment has no opening '{'")

    literal = _scan_balanced_object(html, brace_start)
    sanitized = _UNDEFINED_RE.sub("null", literal)
    try:
        data = json.loads(sanitized)
    except json.JSONDecodeError as exc:
        raise DomClickParseError(
            f"__SSR_STATE__ json.loads failed: {exc} (head={sanitized[:200]!r})"
        ) from exc
    if not isinstance(data, dict):
        raise DomClickParseError(f"__SSR_STATE__ is not an object: {type(data)}")
    return data


# ── Value coercion helpers ────────────────────────────────────────────────────


def _to_float(value: Any) -> float | None:
    """None-safe float. bool/неконвертируемое → None."""
    if value is None or isinstance(value, bool):
        return None
    try:
        return float(value)
    except (ValueError, TypeError):
        return None


def _to_int(value: Any) -> int | None:
    """None-safe int. bool/неконвертируемое → None."""
    if value is None or isinstance(value, bool):
        return None
    try:
        return int(value)
    except (ValueError, TypeError):
        return None


def _parse_change_time(value: Any) -> datetime | None:
    """Нормализует дату изменения цены → timezone-aware datetime.

    Адаптация Layer A ``_parse_publish_date`` (которая возвращала date): здесь
    поддерживаем epoch (сек/мс), числовую строку и ISO8601. Real priceHistory.date
    приходит как ``2026-04-15T09:02:26.548728+03:00`` — fromisoformat СОХРАНЯЕТ
    исходный offset (+03:00), НЕ конвертирует в UTC; naive-значения получают UTC.
    epoch трактуем как UTC. Невалидное → None (caller пропускает запись).
    """
    if value is None or isinstance(value, bool):
        return None
    # epoch (число): >1e12 трактуем как миллисекунды
    if isinstance(value, int | float):
        ts = float(value)
        if ts > 1e12:
            ts /= 1000.0
        try:
            return datetime.fromtimestamp(ts, tz=UTC)
        except (ValueError, OSError, OverflowError):
            return None
    s = str(value).strip()
    if not s:
        return None
    if s.isdigit():
        ts = float(s)
        if ts > 1e12:
            ts /= 1000.0
        try:
            return datetime.fromtimestamp(ts, tz=UTC)
        except (ValueError, OSError, OverflowError):
            return None
    # ISO8601 (поддержим trailing 'Z')
    try:
        dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
    except (ValueError, TypeError):
        return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=UTC)
    return dt


def _map_repair_state(renovation: str | None) -> str | None:
    """RU renovation → каноничный enum. Нет совпадения в карте → инференс из текста."""
    if not renovation:
        return None
    mapped = _RENOVATION_MAP.get(renovation.strip().lower())
    if mapped is not None:
        return normalize_repair_state(mapped)
    return infer_repair_state_from_text(renovation)


def _compact(d: dict[str, Any]) -> dict[str, Any]:
    """Выкидывает None-значения, чтобы raw_extra оставался компактным."""
    return {k: v for k, v in d.items() if v is not None}


def _pick(d: dict[str, Any], *keys: str) -> Any:
    """Первое не-None значение среди ключей (defensive snake/camel spellings)."""
    for k in keys:
        v = d.get(k)
        if v is not None:
            return v
    return None


# ── item_id ───────────────────────────────────────────────────────────────────


def _extract_item_id(url: str) -> str | None:
    """sale__flat__<digits> из URL карточки."""
    m = _ITEM_ID_RE.search(url or "")
    return m.group(1) if m else None


# ── parse_detail_html ──────────────────────────────────────────────────────────


def parse_detail_html(html: str, source_url: str) -> DomClickDetailEnrichment:
    """Pure parser detail-карточки DomClick (без сети/БД).

    _extract_ssr_state может бросить DomClickBlockedError/DomClickParseError —
    они propagate (оркестратор различает блок и parse-failure). Дальнейшая
    навигация по полям обёрнута: per-field сюрприз формы → минимальный
    enrichment (item_id+url), а не крэш.
    """
    state = _extract_ssr_state(html)
    item_id = _extract_item_id(source_url)
    try:
        pc = state.get("productCard") or {}
        oi = pc.get("objectInfo") or {}
        hi = (state.get("houseInfo") or {}).get("info") or {}
        egrn = pc.get("egrnData") or {}

        # repair: enum + сырое значение
        renovation = oi.get("renovation")
        repair_type = renovation if isinstance(renovation, str) else None
        repair_state = _map_repair_state(repair_type)

        # площади
        living_area_m2 = _to_float(oi.get("livingArea"))
        kitchen_area_m2 = _to_float(oi.get("kitchenArea"))

        # балконы: int → count+флаг; иное (строка/др.) → raw_extra, counts None
        balconies = oi.get("balconies")
        balconies_count: int | None = None
        has_balcony: bool | None = None
        balconies_raw: Any = None
        if isinstance(balconies, bool):
            balconies_raw = balconies
        elif isinstance(balconies, int):
            balconies_count = balconies
            has_balcony = balconies > 0
        elif balconies is not None:
            balconies_raw = balconies

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

        # owners_count — защищаемся от двух написаний ключа
        owners_count = _to_int(egrn.get("owners_count"))
        if owners_count is None:
            owners_count = _to_int(egrn.get("ownersCount"))

        # encumbrances_clean (ИНВЕРСИЯ): collateral/collateral_sber обозначают
        # НАЛИЧИЕ обременения. Любой truthy → clean=False; оба отсутствуют/false
        # при наличии egrnData → True; egrnData нет → None.
        collateral = egrn.get("collateral")
        collateral_sber = egrn.get("collateral_sber")
        if egrn:
            encumbrances_clean: bool | None = not bool(collateral or collateral_sber)
        else:
            encumbrances_clean = None

        year_built = _to_int(hi.get("buildYear"))
        views_total = _to_int(pc.get("viewsCount"))

        # priceHistory → нормализованные записи (skip при битой дате / без цены)
        raw_history = (pc.get("priceInfo") or {}).get("priceHistory") or []
        price_changes: list[dict[str, Any]] = []
        if isinstance(raw_history, list):
            for entry in raw_history:
                if not isinstance(entry, dict):
                    continue
                change_time = _parse_change_time(entry.get("date"))
                price_rub = _to_int(entry.get("price"))
                # Skip при нераспарсенной дате ИЛИ отсутствующей/невалидной цене.
                if change_time is None or price_rub is None:
                    continue
                diff = entry.get("diff")
                diff_percent = (
                    diff if isinstance(diff, int | float) and not isinstance(diff, bool) else None
                )
                price_changes.append(
                    {
                        "change_time": change_time,
                        "price_rub": price_rub,
                        "diff_percent": diff_percent,
                    }
                )

        # AVM (Layer C, тот же fetch) — DomClick price prediction живёт ТОП-УРОВНЕМ
        # в SSR-стейте. Может быть плоским или обёрнут в .result/.data — defensively
        # разворачиваем. В raw_extra пишем только присутствующие поля.
        avm_root = state.get("pricePrediction") or {}
        avm_src = avm_root
        if isinstance(avm_root, dict):
            avm_src = avm_root.get("result") or avm_root.get("data") or avm_root
        if not isinstance(avm_src, dict):
            avm_src = {}
        avm = _compact(
            {
                "market_price": _pick(avm_src, "market_price", "marketPrice"),
                "min": _pick(avm_src, "min_market_price", "minMarketPrice"),
                "max": _pick(avm_src, "max_market_price", "maxMarketPrice"),
                "repair_quality": _pick(avm_src, "repair_quality", "repairQuality"),
            }
        )

        # raw_extra → merge в listings.raw_payload. wall/floor type ОСТАЮТСЯ тут,
        # НЕ пишем в listings.house_type (кросс-источниковый словарь грязный).
        # egrn area key — точное написание не подтверждено; пробуем несколько.
        egrn_area = egrn.get("area") or egrn.get("rosreestrArea") or egrn.get("object_area")
        demand = _compact(
            {
                "calls": pc.get("callsCount"),
                "favorites": pc.get("favoriteOfferUsersCount"),
                "duplicates": pc.get("duplicatesOfferCount"),
            }
        )
        raw_extra = _compact(
            {
                "wall_type": hi.get("wallType") or (pc.get("house") or {}).get("wallType"),
                "floor_type": hi.get("floorType"),
                "entrance_count": hi.get("entranceCount"),
                "quarters_count": hi.get("quartersCount"),
                "energy_efficiency": hi.get("energyEfficiency"),
                "building_series": hi.get("buildingSeries"),
                "domclick_building_guid": (pc.get("address") or {}).get("guid"),
                "egrn_area": egrn_area,
                "demand": demand or None,
                "collateral": collateral,
                "collateral_sber": collateral_sber,
                "balconies_raw": balconies_raw,
                "avm": avm or None,
            }
        )

        return DomClickDetailEnrichment(
            item_id=item_id,
            source_url=source_url,
            repair_state=repair_state,
            repair_type=repair_type,
            living_area_m2=living_area_m2,
            kitchen_area_m2=kitchen_area_m2,
            balconies_count=balconies_count,
            has_balcony=has_balcony,
            sale_type=sale_type if isinstance(sale_type, str) else None,
            owners_count=owners_count,
            encumbrances_clean=encumbrances_clean,
            year_built=year_built,
            views_total=views_total,
            price_changes=price_changes,
            raw_extra=raw_extra,
        )
    except Exception:
        logger.warning(
            "domclick_detail: field navigation failed for %s — minimal enrichment",
            source_url,
            exc_info=True,
        )
        return DomClickDetailEnrichment(item_id=item_id, source_url=source_url)


# ── fetch_detail ────────────────────────────────────────────────────────────────


async def fetch_detail(
    card_url: str,
    *,
    browser_fetcher: BrowserFetcher,
) -> DomClickDetailEnrichment:
    """Fetch абсолютного URL карточки через BrowserFetcher → parse_detail_html.

    card_url уже абсолютный (listings.source_url из Layer A), URL не строим.
    Сессией владеет оркестратор (передаёт уже открытый browser_fetcher).

    Raises:
        DomClickBlockedError: anti-bot challenge ИЛИ ошибка браузерного fetch
            (нет curl-fallback — DataDome даёт 401 на curl). Оркестратор считает
            это блоком и помечает listing failed.
        DomClickParseError: HTTP 200, но SSR-стейт не парсится (дрейф схемы).
    """
    try:
        html = await browser_fetcher.fetch(card_url)
    except DomClickBlockedError:
        raise
    except Exception as exc:
        raise DomClickBlockedError(
            f"DomClick detail browser fetch failed for {card_url}: {exc}"
        ) from exc
    return parse_detail_html(html, card_url)


# ── save_detail_enrichment ──────────────────────────────────────────────────────


def save_detail_enrichment(
    db: Session,
    listing_id: int,
    e: DomClickDetailEnrichment,
) -> bool:
    """Persist enrichment в одной транзакции (commit в конце; rollback — за caller).

    1. UPDATE listings SET <cols> (COALESCE — None не затирает существующее),
       raw_payload || raw_extra, detail_enriched_at=NOW() WHERE id=listing_id.
    2. Для каждой записи priceHistory — INSERT INTO offer_price_history
       (ON CONFLICT DO NOTHING → идемпотентно). source='domklik'.

    Snapshot (upsert_listing_snapshot) НЕ пишем: enrichment не несёт текущую цену,
    а detail-backfill цену не меняет — snapshot остаётся за SERP-путём (note).

    Returns:
        True если UPDATE задел строку (listing найден), иначе False.
    """
    raw_extra_json = json.dumps(e.raw_extra or {}, ensure_ascii=False, default=str)

    result = db.execute(
        text("""
            UPDATE listings SET
                repair_state = COALESCE(:repair_state, repair_state),
                repair_type = COALESCE(:repair_type, repair_type),
                living_area_m2 = COALESCE(:living_area_m2, living_area_m2),
                kitchen_area_m2 = COALESCE(:kitchen_area_m2, kitchen_area_m2),
                balconies_count = COALESCE(:balconies_count, balconies_count),
                has_balcony = COALESCE(:has_balcony, has_balcony),
                sale_type = COALESCE(:sale_type, sale_type),
                owners_count = COALESCE(:owners_count, owners_count),
                encumbrances_clean = COALESCE(:encumbrances_clean, encumbrances_clean),
                year_built = COALESCE(:year_built, year_built),
                views_total = COALESCE(:views_total, views_total),
                raw_payload = COALESCE(raw_payload, '{}'::jsonb) || CAST(:raw_extra AS jsonb),
                detail_enriched_at = NOW()
            WHERE id = CAST(:lid AS bigint)
        """),
        {
            "lid": listing_id,
            "repair_state": e.repair_state,
            "repair_type": e.repair_type,
            "living_area_m2": e.living_area_m2,
            "kitchen_area_m2": e.kitchen_area_m2,
            "balconies_count": e.balconies_count,
            "has_balcony": e.has_balcony,
            "sale_type": e.sale_type,
            "owners_count": e.owners_count,
            "encumbrances_clean": e.encumbrances_clean,
            "year_built": e.year_built,
            "views_total": e.views_total,
            "raw_extra": raw_extra_json,
        },
    )

    # priceHistory — INSERT, де-дуп по (listing_id, change_time). SAVEPOINT per-row
    # (backend.md): битый cast одной записи не валит весь UPDATE+commit.
    for change in e.price_changes:
        ct = change.get("change_time")
        price = change.get("price_rub")
        if not ct or not price:
            continue
        try:
            with db.begin_nested():
                db.execute(
                    text("""
                        INSERT INTO offer_price_history
                            (listing_id, change_time, price_rub, diff_percent, source)
                        VALUES (
                            CAST(:lid AS bigint),
                            CAST(:ct AS timestamptz),
                            CAST(:price AS numeric),
                            CAST(:diff AS numeric),
                            'domklik'
                        )
                        ON CONFLICT ON CONSTRAINT offer_price_history_listing_change_uq
                        DO NOTHING
                    """),
                    {
                        "lid": listing_id,
                        "ct": ct,
                        "price": price,
                        "diff": change.get("diff_percent"),
                    },
                )
        except Exception as exc:
            logger.warning(
                "domclick_detail: price-history insert failed listing_id=%s ct=%s: %s",
                listing_id,
                ct,
                exc,
            )

    db.commit()
    found = result.rowcount > 0
    if not found:
        logger.warning(
            "domclick_detail save: listing not found id=%s (item_id=%s)",
            listing_id,
            e.item_id,
        )
    else:
        logger.info(
            "domclick_detail save: listing id=%s enriched (price_changes=%d)",
            listing_id,
            len(e.price_changes),
        )
    return found