gendesign/tradein-mvp/backend/app/services/scrapers/cian_newbuilding.py

"""Cian.ru newbuilding (ЖК) catalog page scraper.

URL: https://zhk-<slug>-<city>-i.cian.ru/
MFE: 'newbuilding-card-desktop-frontend', key: 'initialState'

Sister state containers extracted from same MFE initialState top-level keys:
- realtyValuation: 7-month price chart (data.priceDynamics.chart.data.{labels,values})
  → houses_price_dynamics
- reliabilityState.reliability: наш.дом.рф checkStatus + details[] → house_reliability_checks
- offers: grouped by roomType (fromDeveloperRooms[].layouts[])
- reviews: house reviews list
- housesByTurn: корпуса по очередям (stored as jsonb on houses)

Stage 6 of CianScraper v1.
"""

from __future__ import annotations

import json
import logging
import re
from dataclasses import dataclass, field
from typing import Any

from curl_cffi.requests import AsyncSession  # type: ignore[import-untyped]

from app.core.config import settings
from app.services.scrapers.browser_fetcher import BrowserFetcher
from app.services.scrapers.cian_state_parser import extract_all_states, extract_state

logger = logging.getLogger(__name__)

# Canonical ЖК-slug subdomain URL embedded in any Cian SERP/page that references a
# newbuilding (e.g. https://zhk-parkovyy-kvartal-ekb-i.cian.ru). The cat.php SERP for a
# single newobject[] id puts this slug in its title block — extracting it is the WORKING
# nb_id → ЖК-url resolution (#972; see resolve_cian_zhk_url_via_search).
_ZHK_SLUG_URL_RE = re.compile(r"https://zhk-[a-z0-9-]+\.cian\.ru")

# Title-anchored extraction (#972 hardening). The cat.php newbuilding-SERP renders the
# canonical ЖК-slug of the QUERIED newobject as the <a> href inside its page <h1>:
#
#   <h1 data-name="Title" ...>Купить квартиру в
#       <a href="https://zhk-<slug>.cian.ru/" ...>ЖК «...»</a></h1>
#
# (verified on the live ~2.8 MB prod SERP for nb 48853 — the header sits inside a
# `NewbuildingHeaderSeo` block; there is NO <link rel=canonical> / og:url on this page
# type, so the H1 anchor is the only reliable canonical carrier). Anchoring on it avoids
# the WRONG-ЖК hazard of a naive first-match: cat.php pages also carry banner / "похожие
# ЖК" / recommendation blocks that can embed OTHER zhk-*.cian.ru urls, and those blocks
# may render BEFORE the title — a plain first-match could then resolve a foreign ЖК and
# silently enrich the house with another building's data.
#
# Matches the FIRST zhk-slug url that occurs inside the page's <h1 data-name="Title"> …
# </h1>. `[^>]*` keeps the attr scan within the opening tag; `.*?` (DOTALL) spans the
# inner markup up to the slug href. Falls back to naive first-match only when this finds
# nothing (markup drift), preserving the previous behaviour as a safety net.
_ZHK_TITLE_ANCHOR_RE = re.compile(
    r'<h1[^>]*data-name="Title"[^>]*>.*?(https://zhk-[a-z0-9-]+\.cian\.ru)',
    re.DOTALL,
)

# cat.php newbuilding-SERP template: one row per `newobject[0]=<nb_id>`. ekb. host is the
# Свердловская-обл. mirror tradein targets; the slug URL it returns is host-agnostic
# (the canonical zhk-*.cian.ru subdomain), so this resolves ЕКБ newbuildings correctly.
_CATPH_NEWBUILDING_SERP = (
    "https://ekb.cian.ru/cat.php?deal_type=sale&engine_version=2"
    "&offer_type=flat&object_type[0]=2&newobject[0]={nb_id}"
)


@dataclass
class NewbuildingEnrichment:
    """Result of fetch_newbuilding() — houses row + sister containers."""

    # Core house metadata (from state.newbuilding)
    cian_internal_house_id: int | None = None
    # ZHK page URL — canonical slug form stored after fetch / redirect resolution
    cian_zhk_url: str | None = None
    name: str | None = None
    address: str | None = None
    year_built: int | None = None
    deadline_year: int | None = None
    floors_count_min: int | None = None
    floors_count_max: int | None = None
    building_class: str | None = None  # 'comfort' / 'business' / 'premium'

    # Management
    management_company: dict[str, Any] | None = None  # name, phones[], email, ...

    # Aggregated metrics
    transport_accessibility_rate: int | None = None
    advantages: list[dict[str, Any]] = field(default_factory=list)

    # Time-series price chart (Stage 6 main deliverable)
    # Each entry: {'month_date': str, 'room_count': str, 'prices_type': str, 'period': str,
    #              'price_per_sqm': float | None}
    realty_valuation_chart: list[dict[str, Any]] = field(default_factory=list)

    # Reliability (наш.дом.рф) — overall status + checks detail array
    # {check_status: str, check_name: str, details: list[dict]}
    reliability_checks: list[dict[str, Any]] = field(default_factory=list)

    # Builders / banks (related orgs)
    builders: list[dict[str, Any]] = field(default_factory=list)
    banks: list[dict[str, Any]] = field(default_factory=list)

    # Corpuses (houses_by_turn — корпуса по очередям)
    houses_by_turn: list[dict[str, Any]] = field(default_factory=list)

    # Reviews
    reviews: list[dict[str, Any]] = field(default_factory=list)

    # Mini-SERP offers (grouped by roomType, layouts)
    nested_offers: list[dict[str, Any]] = field(default_factory=list)

    # Raw payloads для debugging
    raw_state: dict[str, Any] | None = None
    raw_sister_states: dict[str, Any] = field(default_factory=dict)


async def fetch_newbuilding(
    zhk_url: str,
    *,
    session: AsyncSession | None = None,  # kept for backward-compat; unused for page fetch
) -> NewbuildingEnrichment | None:
    """Fetch ЖК catalog page и extract все containers.

    HTML страница ЖК теперь тянется через BrowserFetcher (camoufox), потому что
    curl_cffi получает страницу без initialState под анти-ботом Cian (#972).
    tradein-browser верифицирован живым: 1.17 MB, initialState присутствует.

    Args:
        zhk_url: e.g. 'https://zhk-ekaterininskiy-park-ekb-i.cian.ru/'
        session: зарезервирован для обратной совместимости с вызывающими;
            больше не используется для fetch страницы ЖК. Resolve-функции
            (resolve_cian_zhk_url_via_search) по-прежнему используют curl_cffi.

    Returns: NewbuildingEnrichment, or None если fetch / parse failed.
    """
    async with BrowserFetcher(source="cian") as browser:
        html = await browser.fetch(zhk_url)

    # Cian ЖК-карточка: MFE 'newbuilding-card-desktop-frontend', key 'initialState'
    # (verified live 2026-06-15).
    mfe = "newbuilding-card-desktop-frontend"

    nb_state = extract_state(html, mfe=mfe, key="initialState")
    if nb_state is None:
        logger.warning("Cian newbuilding %s: initialState extraction failed", zhk_url)
        return None

    # Primary newbuilding object: state.newbuilding (60 fields per schema sec 15.4)
    nb = nb_state.get("newbuilding") or {}
    if not isinstance(nb, dict):
        nb = {}

    result = NewbuildingEnrichment(
        cian_internal_house_id=nb.get("id"),
        cian_zhk_url=zhk_url,
        name=nb.get("name") or nb.get("displayName"),
        address=_extract_address(nb),
        year_built=nb.get("yearBuilt") or nb.get("buildYear"),
        deadline_year=_extract_deadline_year(nb),
        floors_count_min=nb.get("floorsCountMin"),
        floors_count_max=nb.get("floorsCountMax"),
        building_class=nb.get("newbuildingClassName") or nb.get("buildingClass"),
        management_company=nb.get("managementCompany"),
        transport_accessibility_rate=_extract_transport_rate(nb),
        advantages=_extract_advantages(nb),
        builders=_safe_list(nb.get("builders")),
        banks=_safe_list(nb.get("banks")),
        houses_by_turn=_safe_list(nb.get("housesByTurn")),
        raw_state=nb_state,
    )

    # Sister containers — all live under the same MFE initialState top-level keys
    # (not separate push() entries) per schema sec 15.3
    all_states = extract_all_states(html)
    mfe_states = all_states.get(mfe, {})
    # initialState top-level keys contain sister data: realtyValuation, reliability, offers
    top = nb_state

    # realtyValuation 7-month chart (Stage 6 main deliverable)
    # Located at initialState.realtyValuation (sec 15.5)
    rv_state = top.get("realtyValuation") or mfe_states.get("realtyValuation")
    if rv_state and isinstance(rv_state, dict):
        result.realty_valuation_chart = _extract_chart(rv_state)
        result.raw_sister_states["realtyValuation"] = rv_state

    # reliability (наш.дом.рф) — initialState.reliabilityState.reliability
    # ({checkStatus, details[], banner, actions}, verified live 2026-06-15).
    rel_state = (top.get("reliabilityState") or {}).get("reliability")
    if rel_state and isinstance(rel_state, dict):
        result.reliability_checks = _extract_reliability_checks(rel_state)
        result.raw_sister_states["reliability"] = rel_state

    # reviews — located at initialState or separate state
    reviews_state = top.get("reviews") or mfe_states.get("reviews")
    if reviews_state and isinstance(reviews_state, dict):
        items = reviews_state.get("items") or reviews_state.get("data") or []
        result.reviews = items if isinstance(items, list) else []
        result.raw_sister_states["reviews"] = reviews_state

    # offers (grouped by roomType) — located at initialState.offers (sec 15.7)
    offers_state = top.get("offers") or mfe_states.get("offers") or mfe_states.get("miniSerp")
    if offers_state and isinstance(offers_state, dict):
        result.nested_offers = _extract_nested_offers(offers_state)
        result.raw_sister_states["offers"] = offers_state

    logger.info(
        "Cian newbuilding %s parsed: id=%s chart=%d reliability=%d offers=%d",
        zhk_url,
        result.cian_internal_house_id,
        len(result.realty_valuation_chart),
        len(result.reliability_checks),
        len(result.nested_offers),
    )
    return result


# ---- private extractors ----


def _safe_list(v: Any) -> list[dict[str, Any]]:
    """Return v as list[dict] or empty list."""
    if not isinstance(v, list):
        return []
    return [item for item in v if isinstance(item, dict)]


def _extract_address(nb: dict[str, Any]) -> str | None:
    """Extract display address string from newbuilding object."""
    addr = nb.get("fullAddress") or nb.get("address")
    if isinstance(addr, str):
        return addr
    if isinstance(addr, dict):
        return addr.get("full") or addr.get("display") or addr.get("name")
    return None


def _extract_deadline_year(nb: dict[str, Any]) -> int | None:
    """Extract planned completion year from newbuilding.deadline or similar field."""
    dl = nb.get("deadline") or nb.get("buildingDeadline")
    if isinstance(dl, dict):
        return dl.get("year")
    if isinstance(dl, int):
        return dl
    return None


def _extract_transport_rate(nb: dict[str, Any]) -> int | None:
    """Extract transport accessibility rate as int|None.

    Cian payload drift (#972): `transportAccessibilityRate` is now a nested dict
    ({'transportAccessibility': <int|null>}) rather than a bare int. Older pages
    returned a plain int. Handle both — returning a dict here would violate the
    NewbuildingEnrichment.transport_accessibility_rate: int|None contract and break
    the CAST(:tar AS ...) bind in save_newbuilding_enrichment (psycopg cannot adapt
    a dict). Anything non-int collapses to None.
    """
    raw = nb.get("transportAccessibilityRate")
    if isinstance(raw, dict):
        raw = raw.get("transportAccessibility") or raw.get("rate") or raw.get("value")
    if isinstance(raw, bool):  # bool is an int subclass — exclude explicitly
        return None
    if isinstance(raw, int):
        return raw
    if isinstance(raw, float):
        return int(raw)
    return None


def _extract_advantages(nb: dict[str, Any]) -> list[dict[str, Any]]:
    """Extract typed advantages list from newbuilding.advantages."""
    raw = nb.get("advantages") or nb.get("advantagesTyped") or []
    if not isinstance(raw, list):
        return []
    return [a for a in raw if isinstance(a, dict)]


def _extract_chart(rv_state: dict[str, Any]) -> list[dict[str, Any]]:
    """Extract realtyValuation time-series chart points.

    Actual Cian shape (sec 15.5):
      rv_state.data.priceDynamics.chart.data = {labels: [date...], values: [int...]}

    Falls back to alternative shapes if structure varies.
    Each output point:
      {month_date: str, room_count: str, prices_type: str, period: str, price_per_sqm: float|None}

    Note: The initial page state provides aggregate (all room types) chart only.
    Per-room-type data requires XHR requests (not done here).
    """
    points: list[dict[str, Any]] = []

    # Primary shape: data.priceDynamics.chart.data.{labels, values}
    data_block = rv_state.get("data")
    if isinstance(data_block, dict):
        price_dynamics = data_block.get("priceDynamics") or {}
        if isinstance(price_dynamics, dict):
            chart = price_dynamics.get("chart") or {}
            if isinstance(chart, dict):
                chart_data = chart.get("data") or {}
                if isinstance(chart_data, dict):
                    labels = chart_data.get("labels") or []
                    values = chart_data.get("values") or []
                    if isinstance(labels, list) and isinstance(values, list):
                        for label, value in zip(labels, values, strict=False):
                            if label is None or value is None:
                                continue
                            points.append(
                                {
                                    "month_date": str(label)[:10],  # YYYY-MM-DD
                                    "room_count": "all",  # aggregate (initial state)
                                    "prices_type": "price",  # total price (not priceSqm)
                                    "period": "halfYear",  # default period shown
                                    "price_per_sqm": None,  # chart shows total price, not sqm
                                }
                            )
                            # Store raw value as price_per_sqm placeholder for DB
                            # (actual sqm price would require area context)
                            # Use the value as-is; caller / downstream can enrich
                            points[-1]["price_per_sqm"] = (
                                float(value) if isinstance(value, int | float) else None
                            )
                        if points:
                            return points

    # Fallback shape A: rv_state.chart (flat list with {month/date, price, roomType})
    chart_flat = rv_state.get("chart") or rv_state.get("prices") or rv_state.get("dynamics")
    if isinstance(chart_flat, list):
        for entry in chart_flat:
            if not isinstance(entry, dict):
                continue
            month = entry.get("month") or entry.get("date") or entry.get("monthDate")
            price = entry.get("price") or entry.get("value") or entry.get("pricePerSqm")
            if month is None or price is None:
                continue
            price_val: float | None = None
            try:
                price_val = float(price)
            except (TypeError, ValueError):
                pass
            points.append(
                {
                    "month_date": str(month)[:10],
                    "room_count": str(entry.get("roomType") or entry.get("roomCount") or "all"),
                    "prices_type": str(
                        entry.get("pricesType") or entry.get("priceType") or "price"
                    ),
                    "period": str(entry.get("period") or "halfYear"),
                    "price_per_sqm": price_val,
                }
            )
        return points

    return []


def _extract_reliability_checks(rel_state: dict[str, Any]) -> list[dict[str, Any]]:
    """Extract reliability check summary from наш.дом.рф state.

    Actual Cian shape (sec 15.6):
      rel_state = {
        checkStatus: {status: 'reliable', title: '...', date: '...'},
        details: [{title, iconType, type}, ...],
        ...
      }

    Returns a list with ONE entry representing the overall check (per DB schema):
      [{check_status, check_name, details}]

    The 'details' array is stored as jsonb (full check list) in house_reliability_checks.
    """
    check_status_block = rel_state.get("checkStatus")
    if not isinstance(check_status_block, dict):
        # Alternative: flat list shape
        raw = rel_state.get("checks") or rel_state.get("items") or []
        if isinstance(raw, list) and raw:
            return [
                {
                    "check_name": entry.get("name") or entry.get("title") or entry.get("checkName"),
                    "check_status": entry.get("status") or entry.get("checkStatus"),
                    "details": entry.get("details") or entry,
                }
                for entry in raw
                if isinstance(entry, dict)
            ]
        return []

    status = check_status_block.get("status")
    name = check_status_block.get("title") or check_status_block.get("mobileTitle")
    details = rel_state.get("details") or []

    return [
        {
            "check_name": name,
            "check_status": status,
            "details": details if isinstance(details, list) else [details],
        }
    ]


def _extract_nested_offers(offers_state: dict[str, Any]) -> list[dict[str, Any]]:
    """Extract mini-SERP offers from offers container.

    Actual Cian shape (sec 15.7):
      offers_state.data.total.fromDeveloperRooms[].{roomType, layouts[]}

    Returns flat list of layout offers with room_count injected.
    """
    result: list[dict[str, Any]] = []

    # Shape A: miniSerp / items list
    if "items" in offers_state:
        items = offers_state["items"]
        return items if isinstance(items, list) else []

    # Shape B: data.total.fromDeveloperRooms[].layouts[]
    data = offers_state.get("data")
    if isinstance(data, dict):
        total = data.get("total") or {}
        if isinstance(total, dict):
            rooms = total.get("fromDeveloperRooms") or []
        else:
            rooms = []
    else:
        rooms = []

    if not isinstance(rooms, list):
        return result

    for room in rooms:
        if not isinstance(room, dict):
            continue
        room_type = room.get("roomType") or room.get("roomTypeDisplay")
        layouts = room.get("layouts") or []
        if not isinstance(layouts, list):
            continue
        for layout in layouts:
            if isinstance(layout, dict):
                result.append(
                    {
                        "room_count": room_type,
                        **layout,
                    }
                )

    return result


# ---- save helpers ----


def save_newbuilding_enrichment(
    db: Any,
    house_id: int,
    enrichment: NewbuildingEnrichment,
) -> None:
    """Persist NewbuildingEnrichment to DB.

    Steps:
    1. UPSERT management_companies (if present) → get mc_id
    2. UPDATE houses with Cian metadata (incl. cian_zhk_url if present)
    3. INSERT INTO houses_price_dynamics (chart points, ON CONFLICT DO UPDATE)
    4. INSERT INTO house_reliability_checks (overall + details)
    """
    from sqlalchemy import text

    # 1. UPSERT management_company
    mc_id: int | None = None
    if enrichment.management_company and isinstance(enrichment.management_company, dict):
        mc = enrichment.management_company
        mc_name = mc.get("name") or "Unknown"
        mc_ext_id = mc.get("id")
        mc_result = 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',
                    CAST(:ext_id AS bigint)
                )
                ON CONFLICT (ext_source, name, ext_id) DO UPDATE SET
                    phones = EXCLUDED.phones,
                    email  = COALESCE(EXCLUDED.email, management_companies.email)
                RETURNING id
            """),
            {
                "name": mc_name,
                "phones": mc.get("phones") or [],
                "email": mc.get("email"),
                "oh": mc.get("openingHours"),
                "chief": mc.get("chiefName"),
                "ext_id": mc_ext_id,
            },
        )
        row = mc_result.fetchone()
        mc_id = row[0] if row else None

    # 2. UPDATE houses
    db.execute(
        text("""
            UPDATE houses SET
                cian_internal_house_id  = COALESCE(
                    CAST(:cihi AS bigint), cian_internal_house_id
                ),
                cian_zhk_url            = COALESCE(:zhk_url, cian_zhk_url),
                year_built              = COALESCE(:yb, year_built),
                management_company_id   = COALESCE(CAST(:mcid AS bigint), management_company_id),
                transport_accessibility_rate = COALESCE(
                    :tar, transport_accessibility_rate
                ),
                advantages              = CAST(:adv AS jsonb),
                banks                   = CAST(:bk  AS jsonb),
                builders                = CAST(:bld AS jsonb),
                houses_by_turn          = CAST(:hbt AS jsonb)
            WHERE id = CAST(:hid AS bigint)
        """),
        {
            "hid": house_id,
            "cihi": enrichment.cian_internal_house_id,
            "zhk_url": enrichment.cian_zhk_url,
            "yb": enrichment.year_built,
            "mcid": mc_id,
            "tar": enrichment.transport_accessibility_rate,
            "adv": json.dumps(enrichment.advantages, ensure_ascii=False),
            "bk": json.dumps(enrichment.banks, ensure_ascii=False),
            "bld": json.dumps(enrichment.builders, ensure_ascii=False),
            "hbt": json.dumps(enrichment.houses_by_turn, ensure_ascii=False),
        },
    )

    # 3. INSERT houses_price_dynamics
    # UNIQUE constraint: houses_price_dynamics_dim_key
    #   (house_id, source, room_count, prices_type, period, month_date) — per migration 029
    chart_saved = 0
    for point in enrichment.realty_valuation_chart:
        if point.get("price_per_sqm") is None:
            continue
        room_count = point.get("room_count") or "all"
        prices_type = point.get("prices_type") or "price"
        period = point.get("period") or "halfYear"
        db.execute(
            text("""
                INSERT INTO houses_price_dynamics (
                    house_id, month_date, source,
                    room_count, prices_type, period,
                    price_per_sqm, recorded_at
                ) VALUES (
                    CAST(:hid AS bigint),
                    CAST(:md  AS date),
                    'cian_realty_valuation',
                    :rc, :pt, :pd,
                    CAST(:pps AS numeric),
                    NOW()
                )
                ON CONFLICT ON CONSTRAINT houses_price_dynamics_dim_key DO UPDATE SET
                    price_per_sqm = EXCLUDED.price_per_sqm,
                    recorded_at   = NOW()
            """),
            {
                "hid": house_id,
                "md": point["month_date"],
                "rc": room_count,
                "pt": prices_type,
                "pd": period,
                "pps": point["price_per_sqm"],
            },
        )
        chart_saved += 1

    # 4. INSERT house_reliability_checks (stores overall check + details array)
    # Schema (025): (house_id, check_status, check_name, details jsonb, source, recorded_at)
    # No UNIQUE constraint — caller should manage duplicates if needed
    for check in enrichment.reliability_checks:
        if not check.get("check_name") and not check.get("check_status"):
            continue
        db.execute(
            text("""
                INSERT INTO house_reliability_checks (
                    house_id, check_status, check_name, details, source, recorded_at
                ) VALUES (
                    CAST(:hid AS bigint),
                    :cs,
                    :cn,
                    CAST(:det AS jsonb),
                    'cian_nashdom',
                    NOW()
                )
            """),
            {
                "hid": house_id,
                "cs": check.get("check_status"),
                "cn": check.get("check_name"),
                "det": json.dumps(check.get("details") or [], ensure_ascii=False),
            },
        )

    db.commit()
    logger.info(
        "Cian newbuilding saved house_id=%s (chart=%d points, reliability=%d checks, mc_id=%s)",
        house_id,
        chart_saved,
        len(enrichment.reliability_checks),
        mc_id,
    )


async def resolve_cian_zhk_url(
    cian_internal_house_id: int,
    *,
    session: AsyncSession | None = None,
) -> str | None:
    """Resolve canonical ZHK slug URL from a Cian internal house ID (LEGACY — BROKEN).

    .. deprecated::
        The redirect path this relies on — ``https://cian.ru/zhk/<id>/`` → canonical
        slug — NO LONGER EXISTS. Cian now serves HTTP **404** for ``/zhk/<id>/``
        (verified 8/8 on prod, #972), so this returns None for every real id. Use
        :func:`resolve_cian_zhk_url_via_search` instead: it fetches the cat.php
        newbuilding-SERP and extracts the canonical ``zhk-*.cian.ru`` slug from its
        markup (the WORKING path). Kept only for backward-compat / reference; do not
        wire new callers to it.

    Args:
        cian_internal_house_id: Cian's numeric ЖК identifier.
        session: optional shared curl_cffi AsyncSession (caller owns lifecycle).

    Returns:
        Canonical ZHK URL string, or None if request failed / redirect not followed.

    Note:
        This function makes a real HTTP request — do NOT call it without rate limiting.
        Caller must enforce per-request sleep matching scraper_settings 'cian' delay.
    """
    close_session = False
    if session is None:
        # Mobile proxy wiring (#806 follow-up): resolve ЖК URL через мобильный прокси.
        _proxy_url = settings.cian_proxy_url
        _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
        session = AsyncSession(
            impersonate="chrome120",
            timeout=15.0,
            proxies=_proxies,
            headers={
                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
            },
        )
        close_session = True

    fallback_url = f"https://cian.ru/zhk/{cian_internal_house_id}/"
    try:
        resp = await session.get(fallback_url, allow_redirects=True)
        final_url = str(resp.url)
        if final_url and final_url != fallback_url:
            logger.debug("resolve_cian_zhk_url id=%s → %s", cian_internal_house_id, final_url)
            return final_url
        # Redirect not followed or same URL — return fallback as canonical
        if resp.status_code == 200:
            return fallback_url
        logger.warning(
            "resolve_cian_zhk_url id=%s: HTTP %d, no usable URL",
            cian_internal_house_id,
            resp.status_code,
        )
        return None
    except Exception as exc:
        logger.warning("resolve_cian_zhk_url id=%s failed: %s", cian_internal_house_id, exc)
        return None
    finally:
        if close_session:
            await session.close()


def _extract_zhk_url_from_serp(html: str) -> str | None:
    """Extract the canonical ЖК-slug URL of the QUERIED newobject from a cat.php SERP.

    Pure helper (no I/O) so the extraction can be unit-tested against a saved fixture.
    The cat.php SERP for a single ``newobject[0]=<nb_id>`` renders that ЖК's canonical
    ``https://zhk-<slug>.cian.ru`` URL as the ``<a>`` href inside its page heading
    ``<h1 data-name="Title">`` (the ``NewbuildingHeaderSeo`` header). We anchor on that
    heading rather than taking the first ``zhk-*.cian.ru`` match in the document, because
    cat.php pages also carry banner / "похожие ЖК" / recommendation blocks that can embed
    OTHER ЖК slug URLs — and some of those blocks render BEFORE the title. A naive
    first-match could pick a foreign ЖК and enrich the house with the wrong building's
    price_dynamics / reliability (and the backfill's idempotency would then lock the
    corruption in). Anchoring on the title guarantees we resolve the queried ЖК.

    Falls back to a naive first ``zhk-*.cian.ru`` match ONLY when the title anchor finds
    nothing (markup drift / unexpected page shape), preserving the previous behaviour as
    a last-resort safety net.

    Returns the slug URL string, or None when the SERP carries no such anchor at all
    (empty / no-results page, or markup drift).
    """
    anchored = _ZHK_TITLE_ANCHOR_RE.search(html)
    if anchored:
        return anchored.group(1)
    # Fallback: no title-anchored slug found — degrade to naive first-match so a markup
    # drift that moves/renames the header still resolves *something* rather than failing.
    match = _ZHK_SLUG_URL_RE.search(html)
    return match.group(0) if match else None


async def resolve_cian_zhk_url_via_search(
    nb_id: int,
    *,
    session: AsyncSession | None = None,
) -> str | None:
    """Resolve the canonical ЖК-slug URL for a Cian newbuilding id (the WORKING path).

    Fetches the cat.php newbuilding-SERP for a single ``newobject[0]=<nb_id>`` and
    extracts the canonical ``https://zhk-<slug>.cian.ru`` URL from its markup. This
    replaces the legacy :func:`resolve_cian_zhk_url`, whose ``/zhk/<id>/`` redirect path
    now 404s (verified on prod, #972). The returned slug URL is exactly what
    :func:`fetch_newbuilding` parses, so the enrichment chain is
    ``nb_id → cat.php SERP → zhk-slug-url → fetch_newbuilding → enrich``.

    Verified live (HTTP 200, slug extracted):
        nb 48853  → https://zhk-parkovyy-kvartal-ekb-i.cian.ru   (ЖК «Парковый квартал»)
        nb 102791 → https://zhk-izumrudnyy-bor-ekb-i.cian.ru
        nb 24991  → https://zhk-baltym-park-ekb-i.cian.ru

    Args:
        nb_id: Cian newbuilding id (``house_sources.ext_id`` for cian houses).
        session: optional shared curl_cffi AsyncSession (caller owns lifecycle). When
            None, an own session is created using the same mobile-proxy wiring as the
            rest of the Cian scrapers (``settings.cian_proxy_url`` when set, else direct).

    Returns:
        The canonical ЖК-slug URL string, or None on non-200 / empty SERP / no match /
        request failure (each logs a warning).

    Note:
        Makes ONE real HTTP request and does NOT sleep — the CALLER enforces the
        anti-bot delay (matching scraper_settings 'cian'). At scale this needs the
        mobile proxy; low-volume direct fetches work for the bounded proof.
    """
    close_session = False
    if session is None:
        # Mobile proxy wiring (#806): Cian блокирует datacenter-IP. proxy=None → прямое
        # подключение (dev / proxy-down fallback — single fetches survive direct).
        _proxy_url = settings.cian_proxy_url
        _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
        session = AsyncSession(
            impersonate="chrome120",
            timeout=30.0,
            proxies=_proxies,
            headers={
                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
            },
        )
        close_session = True

    serp_url = _CATPH_NEWBUILDING_SERP.format(nb_id=nb_id)
    try:
        resp = await session.get(serp_url, allow_redirects=True)
        if resp.status_code != 200:
            logger.warning(
                "resolve_cian_zhk_url_via_search nb_id=%s: cat.php → HTTP %d",
                nb_id,
                resp.status_code,
            )
            return None
        zhk_url = _extract_zhk_url_from_serp(resp.text)
        if zhk_url is None:
            logger.warning(
                "resolve_cian_zhk_url_via_search nb_id=%s: no zhk-slug URL in SERP "
                "(empty results / markup drift?)",
                nb_id,
            )
            return None
        logger.info("resolve_cian_zhk_url_via_search nb_id=%s → %s", nb_id, zhk_url)
        return zhk_url
    except Exception as exc:
        logger.warning("resolve_cian_zhk_url_via_search nb_id=%s failed: %s", nb_id, exc)
        return None
    finally:
        if close_session:
            await session.close()