"""Cian.ru Valuation Calculator scraper — auth required.

URL: https://www.cian.ru/kalkulator-nedvizhimosti/?address=...&totalArea=...
MFE: 'valuation-for-agent-frontend'
Key: 'initialState'

Auth: loads Cian session cookies from cian_session_cookies table (Stage 4).
Cache: 24h TTL по sha256(address + params) in external_valuations table.

State shape (sec 24.3–24.6 Schema_Cian_SERP_Inventory):
  estimation.sale.data.{price, accuracy, priceFrom, priceTo, priceSqm}
  estimation.rent.data.{price, accuracy, priceFrom, priceTo, taxPrice}
  estimationChart.data.{title.{change, changeValue}, chartData.data[{date, price}]}
  houseInfo.data.{items[{title, value}], isRenovation, isEmergency, isCulturalHeritage}
  managementCompany.data.{name, address, phones, email, openingHours, chiefName}
  filters.houseId  — Cian's internal house_id

Used as 6th evaluation source в estimator.py (Stage 9).
"""

from __future__ import annotations

import hashlib
import json
import logging
import re
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
from urllib.parse import urlencode

from curl_cffi.requests import AsyncSession
from sqlalchemy import text
from sqlalchemy.orm import Session

from scraper_kit.cian_state_parser import extract_state
from scraper_kit.providers._proxy import curl_proxy_url
from scraper_kit.providers.cian.session import load_session, mark_session_invalid

if TYPE_CHECKING:
    from scraper_kit.contracts import ProxyProvider, ScraperConfig

logger = logging.getLogger(__name__)

VALUATION_BASE_URL = "https://www.cian.ru/kalkulator-nedvizhimosti/"
VALUATION_MFE = "valuation-for-agent-frontend"

# Cian floor enum: floor 1 → floorOne, floor 2 → floorTwo,
# last floor → floorLast, все остальные → floorOther.
_FLOOR_ENUM_SPECIAL = {1: "floorOne", 2: "floorTwo"}

# Cian repair type enum
_REPAIR_TYPE_MAP: dict[str, str] = {
    "without": "repairTypeWithout",
    "no": "repairTypeWithout",
    "cosmetic": "repairTypeCosmetic",
    "euro": "repairTypeEuro",
    "design": "repairTypeDesign",
}

# Cian chart change direction → normalized storage values (per DDL: increase/decrease/neutral)
_CHANGE_DIR_MAP: dict[str, str] = {
    "increase": "increase",
    "decrease": "decrease",
    "neutral": "neutral",
    "up": "increase",
    "down": "decrease",
}


def _cian_floor_enum(floor: int, total_floors: int | None) -> str:
    if floor in _FLOOR_ENUM_SPECIAL:
        return _FLOOR_ENUM_SPECIAL[floor]
    if total_floors is not None and floor == total_floors:
        return "floorLast"
    return "floorOther"


@dataclass
class CianValuationResult:
    """Cian Valuation Calculator response parsed."""

    # Sale estimate
    sale_price_rub: float | None = None
    sale_accuracy: float | None = None
    sale_price_from: float | None = None
    sale_price_to: float | None = None
    sale_price_sqm: float | None = None

    # Rent estimate (Cian-specific — both sale + rent per 1 request)
    rent_price_rub: float | None = None
    rent_accuracy: float | None = None
    rent_price_from: float | None = None
    rent_price_to: float | None = None
    rent_tax_price: float | None = None

    # Chart (7 monthly points for this specific apartment)
    chart: list[dict[str, Any]] = field(default_factory=list)
    chart_change_pct: float | None = None
    chart_change_direction: str | None = None  # increase/decrease/neutral

    # House info (15 items: year, type, floors, etc.)
    house_info: list[dict[str, Any]] = field(default_factory=list)
    external_house_id: int | None = None  # filters.houseId
    filters_hash: str | None = None  # estimation.sale.data.filtersHash

    # Management company (unique to Cian Valuation)
    management_company: dict[str, Any] | None = None

    # Authentication state
    is_authenticated: bool = False
    user_id: int | None = None

    # Raw payload (full state for raw_payload column)
    raw_state: dict[str, Any] | None = None


def compute_cache_key(
    address: str,
    total_area: float,
    rooms_count: int,
    floor: int,
    repair_type: str,
    deal_type: str,
) -> str:
    """sha256 of normalized params for 24h cache lookup."""
    norm = (
        f"{address.strip().lower()}|{total_area}|{rooms_count}|{floor}"
        f"|{repair_type}|{deal_type}"
    )
    return hashlib.sha256(norm.encode("utf-8")).hexdigest()


async def estimate_via_cian_valuation(
    db: Session,
    *,
    config: ScraperConfig,
    address: str,
    total_area: float,
    rooms_count: int,
    floor: int,
    total_floors: int | None = None,
    repair_type: str = "cosmetic",
    deal_type: str = "sale",
    use_cache: bool = True,
    house_id: int | None = None,
    listing_id: int | None = None,
    proxy_provider: ProxyProvider | None = None,
) -> CianValuationResult | None:
    """Estimate value via Cian's authenticated Valuation Calculator.

    Выполняет 1 GET на cian.ru/kalkulator-nedvizhimosti с cookies из DB,
    парсит state['estimation'] + estimationChart + houseInfo + managementCompany.

    Returns CianValuationResult, or None если cookies expired/invalid/fetch failed.
    Graceful: не кидает исключений — caller может продолжить без этого source.
    """
    cache_key = compute_cache_key(address, total_area, rooms_count, floor, repair_type, deal_type)

    # 1. Cache lookup (только если use_cache=True)
    if use_cache:
        cached = _load_from_cache(db, cache_key)
        if cached is not None:
            logger.info("Cian valuation cache HIT for cache_key=%s...", cache_key[:12])
            return cached

    # 2. Load auth cookies из cian_session_cookies
    cookies = load_session(db, config=config)
    if cookies is None:
        logger.warning("Cian valuation: no auth session in DB — skipping (graceful)")
        return None

    # 3. Build URL с корректными enum values
    params: dict[str, str] = {
        "address": address,
        "totalArea": str(total_area),
        "roomsCount": str(rooms_count),
        "valuationType": deal_type,
        "floor[0]": _cian_floor_enum(floor, total_floors),
        "repairType[0]": _REPAIR_TYPE_MAP.get(repair_type, "repairTypeCosmetic"),
    }
    url = f"{VALUATION_BASE_URL}?{urlencode(params, safe='[]')}"

    # 4. Fetch с TLS fingerprint (curl_cffi, impersonate chrome120)
    # Mobile proxy wiring (#806 follow-up): Cian валюация — такой же datacenter-бан риск
    # как SERP. Прокси: пул за флагом use_proxy_pool_curl (#2163), иначе env cian_proxy_url.
    # proxy=None → прямое подключение (dev). curl_proxy_url на выходе mark_health + release.
    with curl_proxy_url(
        config, proxy_provider, "cian", env_fallback_url=config.cian_proxy_url
    ) as _proxy_url:
        _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
        try:
            async with AsyncSession(
                impersonate="chrome120",
                cookies=cookies,
                timeout=25.0,
                proxies=_proxies,
                headers={
                    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
                },
            ) as session:
                resp = await session.get(url, allow_redirects=True)
                if resp.status_code != 200:
                    logger.warning("Cian valuation fetch → HTTP %d for %s", resp.status_code, url)
                    return None
                html = resp.text
        except Exception as exc:
            logger.warning("Cian valuation fetch failed: %s", exc)
            return None

    # 5. Parse state из _cianConfig['valuation-for-agent-frontend']
    state = extract_state(html, mfe=VALUATION_MFE, key="initialState")
    if state is None:
        logger.warning("Cian valuation: state extraction failed (mfe=%s)", VALUATION_MFE)
        return None

    # 6. Verify auth — если false → session expired
    user = state.get("user", {}) or {}
    if not user.get("isAuthenticated"):
        user_id = user.get("userId")
        logger.warning(
            "Cian valuation: session not authenticated (userId=%s) — marking invalid", user_id
        )
        if user_id:
            mark_session_invalid(db, user_id)
        return None

    # 7. Extract structured result
    result = _parse_valuation_state(state)

    # 7a. Sanity-check price bounds (Mera-audit fix-1).
    # Битый API-ответ (999_999 < min или 9_999_999_999 > max) → не кэшируем, возвращаем None.
    # Аналогично проверяем low/high bound: low>high или отрицательные → drop.
    if not _is_price_sane(result, config):
        return None

    # 8. Persist to cache (24h TTL)
    _save_to_cache(
        db,
        cache_key=cache_key,
        address=address,
        total_area=total_area,
        rooms_count=rooms_count,
        floor=floor,
        total_floors=total_floors,
        repair_type=repair_type,
        deal_type=deal_type,
        result=result,
        house_id=house_id,
        listing_id=listing_id,
    )

    logger.info(
        "Cian valuation extracted: sale=%s (acc=%s%%) rent=%s",
        result.sale_price_rub,
        result.sale_accuracy,
        result.rent_price_rub,
    )
    return result


def _parse_valuation_state(state: dict[str, Any]) -> CianValuationResult:
    """Extract CianValuationResult from Cian valuation initialState.

    Mapping (per Schema_Cian_SERP_Inventory sec 24.4–24.6):
      estimation.sale.data.{price, accuracy, priceFrom, priceTo, priceSqm, filtersHash}
      estimation.rent.data.{price, accuracy, priceFrom, priceTo, taxPrice}
      estimationChart.data.{title.{change, changeValue}, chartData.data[]}
      houseInfo.data.{items[], ...}
      filters.houseId
      managementCompany.data
    """
    result = CianValuationResult(raw_state=state)

    user = state.get("user") or {}
    result.is_authenticated = bool(user.get("isAuthenticated"))
    result.user_id = user.get("userId")

    # --- estimation: sale ---
    estimation = state.get("estimation") or {}
    sale_wrapper = estimation.get("sale") or {}
    sale_data = sale_wrapper.get("data") or {}
    result.sale_price_rub = _parse_num(sale_data.get("price"))
    result.sale_accuracy = _parse_num(sale_data.get("accuracy"))
    result.sale_price_from = _parse_num(sale_data.get("priceFrom"))
    result.sale_price_to = _parse_num(sale_data.get("priceTo"))
    result.sale_price_sqm = _parse_num(sale_data.get("priceSqm"))
    result.filters_hash = sale_data.get("filtersHash")

    # --- estimation: rent ---
    rent_wrapper = estimation.get("rent") or {}
    rent_data = rent_wrapper.get("data") or {}
    result.rent_price_rub = _parse_num(rent_data.get("price"))
    result.rent_accuracy = _parse_num(rent_data.get("accuracy"))
    result.rent_price_from = _parse_num(rent_data.get("priceFrom"))
    result.rent_price_to = _parse_num(rent_data.get("priceTo"))
    result.rent_tax_price = _parse_num(rent_data.get("taxPrice"))

    # --- estimationChart: 7 monthly time-series points ---
    est_chart = state.get("estimationChart") or {}
    chart_outer = est_chart.get("data") or {}
    chart_title = chart_outer.get("title") or {}

    # changeValue: "8,3%" → parse out float
    change_value_str = chart_title.get("changeValue") or ""
    result.chart_change_pct = _parse_change_pct(change_value_str)

    # change direction: increase / decrease / neutral
    change_dir_raw = chart_title.get("change")
    if change_dir_raw:
        result.chart_change_direction = _CHANGE_DIR_MAP.get(change_dir_raw, change_dir_raw)

    chart_data_wrapper = chart_outer.get("chartData") or {}
    chart_points = chart_data_wrapper.get("data") or []
    if isinstance(chart_points, list):
        for entry in chart_points:
            if not isinstance(entry, dict):
                continue
            # date: unix_ms → ISO string "YYYY-MM-DD"
            date_ms = entry.get("date")
            date_str: str | None = None
            if isinstance(date_ms, int | float) and date_ms > 0:
                import datetime

                date_str = datetime.datetime.fromtimestamp(
                    date_ms / 1000, tz=datetime.UTC
                ).strftime("%Y-%m-%d")
            result.chart.append(
                {
                    "month_date": date_str or "",
                    "price": _parse_num(entry.get("price")),
                    "price_formatted": entry.get("priceFormatted"),
                }
            )

    # --- houseInfo: 15 items о доме ---
    house_info_wrapper = state.get("houseInfo") or {}
    house_info_data = house_info_wrapper.get("data") or {}
    items = house_info_data.get("items")
    if isinstance(items, list):
        result.house_info = [i for i in items if isinstance(i, dict)]

    # --- filters.houseId: Cian internal house_id ---
    filters = state.get("filters") or {}
    result.external_house_id = filters.get("houseId")

    # --- managementCompany ---
    mc_wrapper = state.get("managementCompany") or {}
    mc_data = mc_wrapper.get("data")
    if isinstance(mc_data, dict) and mc_data:
        result.management_company = mc_data

    return result


def _parse_num(value: Any) -> float | None:
    if value is None:
        return None
    try:
        return float(value)
    except (ValueError, TypeError):
        return None


def _parse_change_pct(value_str: str) -> float | None:
    """Parse '8,3%' → 8.3, '-2,5%' → -2.5, '' → None."""
    if not value_str:
        return None
    cleaned = value_str.replace(",", ".").replace("%", "").strip()
    try:
        return float(cleaned)
    except ValueError:
        return None


def _is_price_sane(result: CianValuationResult, config: ScraperConfig) -> bool:
    """Mera-audit fix-1: проверка разумного диапазона цены Cian Valuation.

    Возвращает False (→ caller вернёт None, не кэшируем) если:
    - sale_price_rub задан и вне [config.cian_valuation_min_rub, cian_valuation_max_rub]
    - sale_price_from или sale_price_to отрицательные
    - sale_price_from > sale_price_to (инвертированный диапазон)

    None-поля не судим: Cian может не вернуть диапазон — это не ошибка.
    """
    price = result.sale_price_rub
    if price is not None:
        min_rub = config.cian_valuation_min_rub
        max_rub = config.cian_valuation_max_rub
        if price < min_rub or price > max_rub:
            logger.warning(
                "Cian valuation sanity: sale_price_rub=%.0f вне [%.0f, %.0f] → drop",
                price,
                min_rub,
                max_rub,
            )
            return False

    low = result.sale_price_from
    high = result.sale_price_to
    if low is not None and low < 0:
        logger.warning("Cian valuation sanity: sale_price_from=%.0f < 0 → drop", low)
        return False
    if high is not None and high < 0:
        logger.warning("Cian valuation sanity: sale_price_to=%.0f < 0 → drop", high)
        return False
    if low is not None and high is not None and low > high:
        logger.warning(
            "Cian valuation sanity: low=%.0f > high=%.0f (инвертированный диапазон) → drop",
            low,
            high,
        )
        return False

    return True


def _load_from_cache(db: Session, cache_key: str) -> CianValuationResult | None:
    """SELECT from external_valuations where cache_key + source='cian_valuation' + not expired."""
    row = (
        db.execute(
            text("""
            SELECT
                raw_payload,
                sale_price_rub,
                sale_accuracy,
                rent_price_rub,
                rent_accuracy,
                chart,
                chart_change_pct,
                chart_change_direction,
                external_house_id,
                filters_hash,
                low_price,
                high_price
            FROM external_valuations
            WHERE source = 'cian_valuation'
              AND cache_key = :ck
              AND expires_at > NOW()
            ORDER BY fetched_at DESC
            LIMIT 1
        """),
            {"ck": cache_key},
        )
        .mappings()
        .first()
    )

    if row is None:
        return None

    return CianValuationResult(
        sale_price_rub=_parse_num(row["sale_price_rub"]),
        sale_accuracy=_parse_num(row["sale_accuracy"]),
        rent_price_rub=_parse_num(row["rent_price_rub"]),
        rent_accuracy=_parse_num(row["rent_accuracy"]),
        chart=row["chart"] if isinstance(row["chart"], list) else [],
        chart_change_pct=_parse_num(row["chart_change_pct"]),
        chart_change_direction=row["chart_change_direction"],
        external_house_id=row["external_house_id"],
        filters_hash=row["filters_hash"],
        sale_price_from=_parse_num(row["low_price"]),
        sale_price_to=_parse_num(row["high_price"]),
        raw_state=row["raw_payload"] if isinstance(row["raw_payload"], dict) else None,
        is_authenticated=True,  # cache только authenticated results
    )


def _save_to_cache(
    db: Session,
    *,
    cache_key: str,
    address: str,
    total_area: float,
    rooms_count: int,
    floor: int,
    total_floors: int | None,
    repair_type: str,
    deal_type: str,
    result: CianValuationResult,
    house_id: int | None = None,
    listing_id: int | None = None,
) -> None:
    """INSERT result into external_valuations (source='cian_valuation') с 24h expiry.

    Использует ON CONFLICT (source, cache_key) DO UPDATE для идемпотентности.
    Все числовые поля через CAST(:x AS numeric) — psycopg v3 safe, без ::type.
    """
    try:
        db.execute(
            text("""
                INSERT INTO external_valuations (
                    source, cache_key, address,
                    total_area, rooms_count, floor, total_floors,
                    repair_type, deal_type,
                    price_rub, accuracy,
                    low_price, high_price,
                    house_id, listing_id,
                    sale_price_rub, sale_accuracy,
                    rent_price_rub, rent_accuracy,
                    chart, chart_change_pct, chart_change_direction,
                    external_house_id, filters_hash,
                    raw_payload, fetched_at, expires_at
                ) VALUES (
                    'cian_valuation', :ck, :addr,
                    CAST(:ta AS numeric), :rc, :fl, :tf,
                    :rt, :dt,
                    CAST(:sp AS numeric), CAST(:sa AS numeric),
                    CAST(:lp AS numeric), CAST(:hp AS numeric),
                    CAST(:hid AS bigint), CAST(:lid AS bigint),
                    CAST(:sp AS numeric), CAST(:sa AS numeric),
                    CAST(:rp AS numeric), CAST(:ra AS numeric),
                    CAST(:ch AS jsonb), CAST(:ccp AS numeric), :ccd,
                    :ehi, :fh,
                    CAST(:raw AS jsonb), NOW(), NOW() + INTERVAL '24 hours'
                )
                ON CONFLICT (source, cache_key) DO UPDATE SET
                    price_rub             = EXCLUDED.price_rub,
                    accuracy              = EXCLUDED.accuracy,
                    low_price             = EXCLUDED.low_price,
                    high_price            = EXCLUDED.high_price,
                    house_id   = COALESCE(EXCLUDED.house_id, external_valuations.house_id),
                    listing_id = COALESCE(EXCLUDED.listing_id, external_valuations.listing_id),
                    sale_price_rub        = EXCLUDED.sale_price_rub,
                    sale_accuracy         = EXCLUDED.sale_accuracy,
                    rent_price_rub        = EXCLUDED.rent_price_rub,
                    rent_accuracy         = EXCLUDED.rent_accuracy,
                    chart                 = EXCLUDED.chart,
                    chart_change_pct      = EXCLUDED.chart_change_pct,
                    chart_change_direction = EXCLUDED.chart_change_direction,
                    external_house_id     = EXCLUDED.external_house_id,
                    filters_hash          = EXCLUDED.filters_hash,
                    raw_payload           = EXCLUDED.raw_payload,
                    fetched_at            = NOW(),
                    expires_at            = NOW() + INTERVAL '24 hours'
            """),
            {
                "ck": cache_key,
                "addr": address,
                "ta": total_area,
                "rc": rooms_count,
                "fl": floor,
                "tf": total_floors,
                "rt": repair_type,
                "dt": deal_type,
                "sp": result.sale_price_rub,
                "sa": result.sale_accuracy,
                "lp": result.sale_price_from,
                "hp": result.sale_price_to,
                "hid": house_id,
                "lid": listing_id,
                "rp": result.rent_price_rub,
                "ra": result.rent_accuracy,
                "ch": json.dumps(result.chart),
                "ccp": result.chart_change_pct,
                "ccd": result.chart_change_direction,
                "ehi": result.external_house_id,
                "fh": result.filters_hash,
                "raw": json.dumps(result.raw_state) if result.raw_state else None,
            },
        )

        # Best-effort house enrichment (#2435): management_company/house_info/
        # cian_internal_house_id → канонический houses-ряд. house_id уже резолвлен
        # ВЫЗЫВАЮЩИМ (estimator.py::match_house_readonly — read-only, не создаёт дом из
        # ephemeral valuation-запроса) — здесь только COALESCE-UPDATE, никакого
        # match_or_create_house. SAVEPOINT изолирует сбой от уже выполненного INSERT
        # выше (тот же паттерн, что estimator.py::_backfill_house_fias).
        try:
            with db.begin_nested():
                _persist_cian_valuation_house(db, house_id=house_id, result=result)
        except Exception as exc:
            logger.warning(
                "Cian valuation house-persist failed for house_id=%s (graceful): %s",
                house_id,
                exc,
                exc_info=True,
            )

        db.commit()
    except Exception as exc:
        logger.error("Cian valuation cache save failed: %s", exc, exc_info=True)
        raise


# ── House enrichment (#2435) ──────────────────────────────────────────────────

# houseInfo.data.items title (RU, lowercased) → houses column. Только поля с чистым
# 1:1 маппингом на существующую колонку (Schema_Cian_SERP_Inventory sec 24.6).
# Пропущенные (мусоропровод/реновация/спортплощадка/фонд капремонта) — нет колонки
# в houses, см. gap-заметку в docstring _persist_cian_valuation_house.
_HOUSE_INFO_TITLE_MAP: dict[str, str] = {
    "год постройки": "year_built",
    "тип дома": "house_type",
    "этажность": "total_floors",
    "газоснабжение": "gas_supply_type",
    "отопление": "heat_supply_type",
    "тип перекрытий": "overlap_type",
    "подъездов": "entrances",
    "квартир": "flat_count",
    "аварийность": "is_emergency",
    "детская площадка": "has_playground",
}
_HOUSE_INFO_BOOL_FIELDS = {"is_emergency", "has_playground"}
_BOOL_YES_RU = {"да"}
_BOOL_NO_RU = {"нет"}

# "Количество лифтов": free-text "1 пассажирский, 1 грузовой" (не структурировано Cian'ом).
_ELEVATORS_PASSENGER_RE = re.compile(r"(\d+)\s*пассажир", re.IGNORECASE)
_ELEVATORS_CARGO_RE = re.compile(r"(\d+)\s*груз", re.IGNORECASE)


def _map_house_info_items(items: list[dict[str, Any]]) -> dict[str, Any]:
    """houseInfo.data.items (RU title/value пары) → known houses columns (#2435).

    Best-effort парсинг: неизвестные/непонятные title просто пропускаются (нет
    исключений). "Количество лифтов" — единственное поле с free-text значением
    ("N пассажирский, M грузовой") — парсится regex'ом, при несовпадении паттерна
    соответствующий счётчик остаётся не заполненным (не 0 — 0 означало бы "нет лифтов").
    """
    out: dict[str, Any] = {}
    for item in items:
        if not isinstance(item, dict):
            continue
        title = str(item.get("title") or "").strip().lower()
        value = item.get("value")

        if title == "количество лифтов" and isinstance(value, str):
            pm = _ELEVATORS_PASSENGER_RE.search(value)
            if pm:
                out["passenger_elevators"] = int(pm.group(1))
            cm = _ELEVATORS_CARGO_RE.search(value)
            if cm:
                out["cargo_elevators"] = int(cm.group(1))
            continue

        field_name = _HOUSE_INFO_TITLE_MAP.get(title)
        if field_name is None:
            continue

        if field_name in _HOUSE_INFO_BOOL_FIELDS:
            if isinstance(value, bool):
                out[field_name] = value
            elif isinstance(value, str):
                v_norm = value.strip().lower()
                if v_norm in _BOOL_YES_RU:
                    out[field_name] = True
                elif v_norm in _BOOL_NO_RU:
                    out[field_name] = False
        else:
            out[field_name] = value
    return out


def _upsert_cian_valuation_management_company(
    db: Session, mc: dict[str, Any] | None
) -> int | None:
    """UPSERT management_companies из Cian Valuation Calculator's managementCompany.data.

    Тот же UPSERT-паттерн, что providers/cian/newbuilding.py::save_newbuilding_enrichment
    (независимая копия — newbuilding.py вне scope #2435, не рефакторим/не переиспользуем).
    ext_source='cian_valuation' (не 'cian') — валюация не даёт числового id УК (в отличие
    от newbuilding.managementCompany.id), поэтому dedup-ключ (ext_source, name, ext_id=NULL)
    сходится по имени УК среди повторных valuation-запросов.

    opening_hours — Cian отдаёт список словарей ([{"Пн–Пт": ["10:00-20:00"]}, ...]), не
    строку; json.dumps в text-колонку (в отличие от newbuilding.py, которая пишет объект
    напрямую — тот путь не в scope этой задачи).
    """
    if not mc or not isinstance(mc, dict):
        return None
    name = mc.get("name")
    if not name:
        return None

    opening_hours = mc.get("openingHours")
    row = db.execute(
        text("""
            INSERT INTO management_companies (
                name, phones, email, opening_hours, chief_name, ext_source, ext_id
            ) VALUES (
                :name, CAST(:phones AS text[]), :email, :oh, :chief, 'cian_valuation', NULL
            )
            ON CONFLICT (ext_source, name, ext_id) DO UPDATE SET
                phones = COALESCE(EXCLUDED.phones, management_companies.phones),
                email  = COALESCE(EXCLUDED.email, management_companies.email)
            RETURNING id
        """),
        {
            "name": name,
            "phones": mc.get("phones") or [],
            "email": mc.get("email"),
            "oh": json.dumps(opening_hours, ensure_ascii=False) if opening_hours else None,
            "chief": mc.get("chiefName"),
        },
    ).fetchone()
    return int(row[0]) if row else None


def _persist_cian_valuation_house(
    db: Session,
    *,
    house_id: int | None,
    result: CianValuationResult,
) -> None:
    """Пишет managementCompany/houseInfo/houseId в канонический houses-ряд (#2435).

    `house_id` уже резолвлен ВЫЗЫВАЮЩИМ (estimator.py, read-only match_house_readonly) —
    valuation-запрос по эфемерному целевому адресу не должен САМ создавать дом (в отличие
    от cian_bti #2435 Part 1, где match_or_create_house уместен — там источник привязан
    к конкретному листингу/дому, а не к произвольному введённому адресу). house_id=None →
    no-op, best-effort.

    COALESCE-направление — по app.services.matching.conflict_resolution.HOUSE_FIELD_PRIORITY:
      - management_company_id: HOUSE_FIELD_PRIORITY["management_company_id"] == ["cian_valuation"]
        — единственный источник → COALESCE(new, existing) (свежее значение побеждает).
      - cian_internal_house_id (filters.houseId): не в реестре, но тот же принцип, что
        newbuilding.py применяет к этой же колонке → COALESCE(new, existing).
      - houseInfo-производные (year_built/house_type/total_floors/entrances/flat_count/
        is_emergency/heat_supply_type/gas_supply_type/overlap_type/has_playground/
        passenger_elevators/cargo_elevators): cian_valuation НЕ входит в приоритет для этих
        полей вовсе (или cian_bti/serp/avito ранжированы выше) → COALESCE(existing, new) —
        валюация только заполняет пробелы, не перетирает более авторитетные источники.
    """
    if house_id is None:
        return

    mc_id = _upsert_cian_valuation_management_company(db, result.management_company)
    info = _map_house_info_items(result.house_info)

    db.execute(
        text("""
            UPDATE houses SET
                management_company_id  = COALESCE(
                    CAST(:mcid AS bigint), houses.management_company_id
                ),
                cian_internal_house_id = COALESCE(
                    CAST(:cihi AS bigint), houses.cian_internal_house_id
                ),
                year_built              = COALESCE(houses.year_built, CAST(:yb AS int)),
                house_type              = COALESCE(houses.house_type, CAST(:ht AS text)),
                total_floors            = COALESCE(houses.total_floors, CAST(:tf AS int)),
                entrances               = COALESCE(houses.entrances, CAST(:ent AS int)),
                flat_count              = COALESCE(houses.flat_count, CAST(:fc AS int)),
                is_emergency            = COALESCE(houses.is_emergency, CAST(:ie AS boolean)),
                heat_supply_type        = COALESCE(
                    houses.heat_supply_type, CAST(:hst AS text)
                ),
                gas_supply_type         = COALESCE(houses.gas_supply_type, CAST(:gst AS text)),
                overlap_type            = COALESCE(houses.overlap_type, CAST(:ot AS text)),
                has_playground          = COALESCE(
                    houses.has_playground, CAST(:hp AS boolean)
                ),
                passenger_elevators     = COALESCE(
                    houses.passenger_elevators, CAST(:pe AS int)
                ),
                cargo_elevators         = COALESCE(houses.cargo_elevators, CAST(:ce AS int))
            WHERE id = CAST(:hid AS bigint)
        """),
        {
            "hid": house_id,
            "mcid": mc_id,
            "cihi": result.external_house_id,
            "yb": info.get("year_built"),
            "ht": info.get("house_type"),
            "tf": info.get("total_floors"),
            "ent": info.get("entrances"),
            "fc": info.get("flat_count"),
            "ie": info.get("is_emergency"),
            "hst": info.get("heat_supply_type"),
            "gst": info.get("gas_supply_type"),
            "ot": info.get("overlap_type"),
            "hp": info.get("has_playground"),
            "pe": info.get("passenger_elevators"),
            "ce": info.get("cargo_elevators"),
        },
    )
    logger.info(
        "_persist_cian_valuation_house: house_id=%s mc_id=%s cian_internal_house_id=%s",
        house_id,
        mc_id,
        result.external_house_id,
    )