"""Shared utility для извлечения Cian Redux state из _cianConfig. Pattern: window._cianConfig[''].push({key:..., value:..., priority:..., filter:...}) Used by SERP / detail / newbuilding / valuation scrapers. """ import json import logging import re from typing import Any logger = logging.getLogger(__name__) # Match both patterns Cian uses: # 1. window._cianConfig['mfe'].push({key:..., value:..., priority:..., filter:...}) # 2. window._cianConfig['mfe'] = window._cianConfig['mfe'] || []; # window._cianConfig['mfe'].push({...}) _RE_CIAN_PUSH = re.compile( r"window\._cianConfig\['(?P[\w-]+)'\]" r"(?:\.push\(\s*|" r"\s*=\s*window\._cianConfig\['[\w-]+'\]\s*\|\|\s*\[\];\s*" r"window\._cianConfig\['[\w-]+'\]\.push\(\s*)" r"\{\s*key:\s*['\"](?P\w+)['\"]\s*,\s*" r"value:\s*(?P.+?)\s*,\s*" r"priority:", re.DOTALL, ) def extract_state(html: str, mfe: str, key: str = "initialState") -> dict[str, Any] | None: """Extract state.value where _cianConfig[] entry.key matches. Returns parsed JSON dict, or None if not found / parse failed. Logs warning at debug level on parse failure for diagnostics. """ for match in _RE_CIAN_PUSH.finditer(html): if match.group("mfe") != mfe or match.group("key") != key: continue value_str = match.group("value").strip() # Strategy 1: direct JSON parse (value уже JSON-like) try: return json.loads(value_str) except json.JSONDecodeError: pass # Strategy 2: JSON.parse("escaped string") # Use json.loads on the outer quoted string to unescape atomically — avoids # order-dependent manual .replace() chains that break when descriptions contain # backslashes (e.g. '\\' in listing text corrupts the rest of the parse). m_jp = re.match(r'JSON\.parse\(\s*("(?:[^"\\]|\\.)*")\s*\)\s*$', value_str, re.DOTALL) if m_jp: outer_json_str = m_jp.group(1) # the full "..." including surrounding quotes try: inner = json.loads(outer_json_str) # inner is now a Python str except json.JSONDecodeError as exc: logger.debug( "JSON.parse() outer unescape failed for mfe=%s key=%s: %s", mfe, key, exc ) continue try: return json.loads(inner) except json.JSONDecodeError as exc: logger.debug("JSON.parse() inner parse failed for mfe=%s key=%s: %s", mfe, key, exc) continue # Strategy 3: raw JS object literal — try demjson3 (optional dep) try: import demjson3 # type: ignore[import-untyped] return demjson3.decode(value_str) except ImportError: logger.debug( "demjson3 not installed; cannot parse raw JS object for mfe=%s key=%s", mfe, key ) except Exception as exc: logger.debug("demjson3 parse failed for mfe=%s key=%s: %s", mfe, key, exc) return None def extract_all_states(html: str) -> dict[str, dict[str, Any]]: """Returns {mfe_name: {key: parsed_value}} for all push() entries. Useful for discovery / debugging during scraper development. """ result: dict[str, dict[str, Any]] = {} for match in _RE_CIAN_PUSH.finditer(html): mfe = match.group("mfe") key = match.group("key") parsed = extract_state(html, mfe, key) if parsed is not None: result.setdefault(mfe, {})[key] = parsed return result