feat(tradein): cian_state_parser shared utility + ScrapedLot Cian fields
- Add cian_state_parser.py: regex-based extractor for window._cianConfig push() entries, supports direct JSON, JSON.parse() escaped form, and optional demjson3 fallback - Extend ScrapedLot in base.py with 13 Cian-specific fields (living_area_m2, bedrooms_count, balconies_count, loggias_count, description_minhash, cadastral_number, building_cadastral_number, phones, is_homeowner, is_pro_seller, bargain_allowed, sale_type, metro_stations) - Add tests/test_cian_state_parser.py with 4 unit tests covering all parse strategies
This commit is contained in:
parent
23e6cbc773
commit
7f03a5b634
3 changed files with 154 additions and 0 deletions
|
|
@ -93,6 +93,21 @@ class ScrapedLot(BaseModel):
|
|||
photo_urls: list[str] = Field(default_factory=list)
|
||||
raw_payload: dict[str, Any] | None = None
|
||||
|
||||
# Cian-specific extensions (Stage 2 of CianScraper v1)
|
||||
living_area_m2: float | None = None
|
||||
bedrooms_count: int | None = None
|
||||
balconies_count: int | None = None
|
||||
loggias_count: int | None = None
|
||||
description_minhash: str | None = None
|
||||
cadastral_number: str | None = None
|
||||
building_cadastral_number: str | None = None
|
||||
phones: list[dict] = Field(default_factory=list)
|
||||
is_homeowner: bool | None = None
|
||||
is_pro_seller: bool | None = None
|
||||
bargain_allowed: bool | None = None
|
||||
sale_type: str | None = None
|
||||
metro_stations: list[dict] = Field(default_factory=list)
|
||||
|
||||
def compute_dedup_hash(self) -> str:
|
||||
"""SHA256(source + (source_id или source_url) + price_rub) — стабильный uniqueness key.
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,93 @@
|
|||
"""Shared utility для извлечения Cian Redux state из _cianConfig.
|
||||
|
||||
Pattern: window._cianConfig['<mfe>'].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<mfe>[\w-]+)'\]"
|
||||
r"(?:\.push\(\s*|"
|
||||
r"\s*=\s*window\._cianConfig\['[\w-]+'\]\s*\|\|\s*\[\];\s*"
|
||||
r"window\._cianConfig\['[\w-]+'\]\.push\(\s*)"
|
||||
r"\{\s*key:\s*['\"](?P<key>\w+)['\"]\s*,\s*"
|
||||
r"value:\s*(?P<value>.+?)\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[<mfe>] 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")
|
||||
m_jp = re.match(r'JSON\.parse\(\s*"(.+?)"\s*\)\s*$', value_str, re.DOTALL)
|
||||
if m_jp:
|
||||
escaped = m_jp.group(1)
|
||||
unescaped = (
|
||||
escaped
|
||||
.replace('\\"', '"')
|
||||
.replace('\\\\', '\\')
|
||||
.replace('\\n', '\n')
|
||||
.replace('\\/', '/')
|
||||
)
|
||||
try:
|
||||
return json.loads(unescaped)
|
||||
except json.JSONDecodeError as exc:
|
||||
logger.debug(
|
||||
"JSON.parse() unescape 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
|
||||
46
tradein-mvp/backend/tests/test_cian_state_parser.py
Normal file
46
tradein-mvp/backend/tests/test_cian_state_parser.py
Normal file
|
|
@ -0,0 +1,46 @@
|
|||
"""Tests for cian_state_parser."""
|
||||
import pytest
|
||||
from app.services.scrapers.cian_state_parser import extract_state
|
||||
|
||||
|
||||
def test_extract_state_simple_json():
|
||||
html = '''
|
||||
<html><body>
|
||||
<script>
|
||||
window._cianConfig['frontend-serp'] = window._cianConfig['frontend-serp'] || [];
|
||||
window._cianConfig['frontend-serp'].push({
|
||||
key: 'initialState',
|
||||
value: {"results": {"offers": [{"id": 1}]}},
|
||||
priority: 1
|
||||
});
|
||||
</script>
|
||||
</body></html>
|
||||
'''
|
||||
state = extract_state(html, mfe='frontend-serp', key='initialState')
|
||||
assert state is not None
|
||||
assert state['results']['offers'][0]['id'] == 1
|
||||
|
||||
|
||||
def test_extract_state_json_parse_form():
|
||||
html = r'''
|
||||
window._cianConfig['frontend-offer-card'].push({
|
||||
key: 'defaultState',
|
||||
value: JSON.parse("{\"offerData\":{\"offer\":{\"cianId\":123}}}"),
|
||||
priority: 1
|
||||
});
|
||||
'''
|
||||
state = extract_state(html, mfe='frontend-offer-card', key='defaultState')
|
||||
assert state is not None
|
||||
assert state['offerData']['offer']['cianId'] == 123
|
||||
|
||||
|
||||
def test_extract_state_missing_returns_none():
|
||||
html = "<html></html>"
|
||||
assert extract_state(html, mfe='frontend-serp') is None
|
||||
|
||||
|
||||
def test_extract_state_wrong_mfe_returns_none():
|
||||
html = '''
|
||||
window._cianConfig['other-mfe'].push({key: 'initialState', value: {}, priority: 1});
|
||||
'''
|
||||
assert extract_state(html, mfe='frontend-serp', key='initialState') is None
|
||||
Loading…
Add table
Reference in a new issue