"""Cian.ru newbuilding (ЖК) catalog page scraper. URL: https://zhk---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 href inside its page

: # #

Купить квартиру в # ЖК «...»

# # (verified on the live ~2.8 MB prod SERP for nb 48853 — the header sits inside a # `NewbuildingHeaderSeo` block; there is NO / 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

… #

. `[^>]*` 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']*data-name="Title"[^>]*>.*?(https://zhk-[a-z0-9-]+\.cian\.ru)', re.DOTALL, ) # cat.php newbuilding-SERP template: one row per `newobject[0]=`. 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': }) 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//`` → canonical slug — NO LONGER EXISTS. Cian now serves HTTP **404** for ``/zhk//`` (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]=`` renders that ЖК's canonical ``https://zhk-.cian.ru`` URL as the ```` href inside its page heading ``

`` (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]=`` and extracts the canonical ``https://zhk-.cian.ru`` URL from its markup. This replaces the legacy :func:`resolve_cian_zhk_url`, whose ``/zhk//`` 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()