feat(scraper-kit): copy cian providers with protocol injection, strangler (#2133 cian)
All checks were successful
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / changes (pull_request) Successful in 8s
CI / openapi-codegen-check (pull_request) Has been skipped

This commit is contained in:
bot-backend 2026-07-02 19:10:18 +03:00
parent df5e9003df
commit ab8b9cb719
9 changed files with 3860 additions and 0 deletions

View file

@ -0,0 +1,492 @@
"""Golden-parity: `scraper_kit.providers.cian.*` парсинг ≡ `app.services.scrapers.cian*`.
Strangler-инвариант (#2133 cian): новая scraper_kit-копия cian-провайдера должна давать
БАЙТ-ИДЕНТИЧНЫЙ результат парсинга старому боевому коду на одинаковом входе. Развязка
(settingsScraperConfig, get_scraper_delaydelay_provider, cian_sessionлокальная копия)
НЕ меняет распарсенные данные.
Гоняем ОБА модуля на одних фикстурах и сравниваем результат:
- SERP `_parse_serp_html` (Redux state list[ScrapedLot]) на синтетическом _cianConfig
push-HTML (вторичка + новостройка + студия), сравнение model_dump();
- SERP `_offer_to_lot` пооферно + `_format_address` (pure, восстановление дома ЖК);
- `_extract_total_offers` (totalOffers из state);
- `extract_state` app-версия kit-версия на одном HTML (общий парсер state);
- detail `fetch_detail` (defaultState + bti sister DetailEnrichment, dataclasses.asdict);
- valuation `_parse_valuation_state` (estimation/chart/houseInfo CianValuationResult);
- newbuilding pure-экстракторы (_extract_chart / _extract_reliability_checks /
_extract_nested_offers / _extract_zhk_url_from_serp).
Идентичность результата = kit-парсер верно скопирован (развязка не задела логику).
"""
from __future__ import annotations
import dataclasses
import json
import os
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
# Старый app.services.scrapers.cian* импортирует app.core.config.settings=Settings(),
# которому нужен DATABASE_URL. Офлайн-парсинг БД не трогает — фиктивный DSN достаточен.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
# НОВЫЙ (scraper_kit) cian-провайдер.
from scraper_kit.cian_state_parser import extract_state as new_extract_state
from scraper_kit.providers.cian.detail import fetch_detail as new_fetch_detail
from scraper_kit.providers.cian.newbuilding import _extract_chart as new_extract_chart
from scraper_kit.providers.cian.newbuilding import (
_extract_nested_offers as new_extract_nested_offers,
)
from scraper_kit.providers.cian.newbuilding import (
_extract_reliability_checks as new_extract_reliability,
)
from scraper_kit.providers.cian.newbuilding import (
_extract_zhk_url_from_serp as new_extract_zhk_url,
)
from scraper_kit.providers.cian.serp import CianScraper as NewCianScraper
from scraper_kit.providers.cian.serp import _format_address as new_format_address
from scraper_kit.providers.cian.valuation import (
_parse_valuation_state as new_parse_valuation,
)
# СТАРЫЙ (app.services.scrapers) cian-провайдер.
from app.services.scrapers.cian import CianScraper as OldCianScraper
from app.services.scrapers.cian import _format_address as old_format_address
from app.services.scrapers.cian_detail import fetch_detail as old_fetch_detail
from app.services.scrapers.cian_newbuilding import _extract_chart as old_extract_chart
from app.services.scrapers.cian_newbuilding import (
_extract_nested_offers as old_extract_nested_offers,
)
from app.services.scrapers.cian_newbuilding import (
_extract_reliability_checks as old_extract_reliability,
)
from app.services.scrapers.cian_newbuilding import (
_extract_zhk_url_from_serp as old_extract_zhk_url,
)
from app.services.scrapers.cian_state_parser import extract_state as old_extract_state
from app.services.scrapers.cian_valuation import (
_parse_valuation_state as old_parse_valuation,
)
# ── Fixtures: SERP offers → _cianConfig push HTML ────────────────────────────
_CONFIG = SimpleNamespace(glitchtip_dsn=None)
def _serp_html(offers: list[dict[str, Any]]) -> str:
"""Обернуть offers в валидный _cianConfig['frontend-serp'].push() HTML."""
state = {"results": {"totalOffers": len(offers), "offers": offers}}
state_json = json.dumps(state, ensure_ascii=False)
return (
"<html><body><script>\n"
"window._cianConfig['frontend-serp'] = window._cianConfig['frontend-serp'] || [];\n"
"window._cianConfig['frontend-serp'].push({"
f"key: 'initialState', value: {state_json}, priority: 10, filter: null"
"});\n</script></body></html>"
)
def _vtorichka_offer() -> dict[str, Any]:
return {
"cianId": 328739308,
"id": 328739308,
"fullUrl": "https://ekb.cian.ru/sale/flat/328739308/",
"offerType": "flat",
"flatType": "rooms",
"category": "flatSale",
"isApartments": False,
"roomsCount": 2,
"totalArea": "52.0",
"livingArea": "35.0",
"kitchenArea": "11.5",
"balconiesCount": 1,
"loggiasCount": 0,
"bedroomsCount": None,
"hasFurniture": False,
"isSoldFurnished": False,
"decoration": "euro",
"floorNumber": 7,
"building": {
"floorsCount": 16,
"materialType": "monolithBrick",
"buildYear": 2010,
"parking": {"type": "underground"},
"totalArea": "3400",
},
"cadastralNumber": "66:41:0204016:1234",
"buildingCadastralNumber": "66:41:0204016:100",
"bargainTerms": {
"priceRur": 6500000,
"price": 6500000,
"saleType": "free",
"mortgageAllowed": True,
"bargainAllowed": True,
},
"geo": {
"coordinates": {"lat": 56.8385, "lng": 60.5940},
"address": [
{"type": "location", "fullName": "Свердловская область"},
{"type": "raion", "fullName": "Кировский"},
{"type": "street", "fullName": "улица Постовского"},
{"type": "house", "fullName": "17А"},
{"type": "metro", "fullName": "Геологическая"},
],
"undergrounds": [
{
"name": "Геологическая",
"time": 10,
"transportType": "walk",
"lineId": 10,
"lineColor": "FF0000",
"isDefault": True,
}
],
},
"newbuilding": {"id": 0, "name": ""},
"phones": [{"number": "9012345678", "countryCode": "7"}],
"userId": 13332612,
"publishedUserId": 13332612,
"isByHomeowner": True,
"isPro": False,
"isRosreestrChecked": True,
"description": "Продаётся отличная квартира с евроремонтом",
"descriptionMinhash": "abc123def456abc1",
"addedTimestamp": 1716451200,
"photos": [{"fullUrl": "https://images.cdn-cian.ru/images/328739308-1.jpg"}],
}
def _newbuilding_offer() -> dict[str, Any]:
"""Новостройка без части house в geo.address[] — жмёт _recover_newbuilding_house."""
return {
"cianId": 411112222,
"id": 411112222,
"fullUrl": "https://ekb.cian.ru/sale/flat/411112222/",
"offerType": "flat",
"flatType": "rooms",
"isApartments": False,
"roomsCount": 1,
"totalArea": "38.0",
"floorNumber": 12,
"building": {
"floorsCount": 25,
"materialType": "monolith",
"buildYear": None,
"deadline": {"year": 2026},
},
"buildingCadastralNumber": "66:41:0204016:11396",
"bargainTerms": {"priceRur": 5200000, "price": 5200000, "mortgageAllowed": True},
"geo": {
"coordinates": {"lat": 56.8000, "lng": 60.6000},
"address": [
{"type": "location", "fullName": "Свердловская область"},
{"type": "street", "fullName": "улица Новая"},
],
},
"newbuilding": {"id": 48853, "name": "ЖК Парковый квартал 29А"},
"isFromDeveloper": True,
"photos": [],
}
def _studio_offer() -> dict[str, Any]:
return {
"cianId": 500011111,
"id": 500011111,
"offerType": "flat",
"flatType": "studio",
"roomsCount": None,
"totalArea": "24.5",
"floorNumber": 3,
"building": {"floorsCount": 9, "materialType": "panel", "buildYear": 1980},
"bargainTerms": {"priceRur": 3100000, "price": 3100000},
"geo": {
"coordinates": {"lat": 56.81, "lng": 60.61},
"address": [
{"type": "street", "fullName": "улица Ленина"},
{"type": "house", "fullName": "5"},
],
},
"newbuilding": {"id": 0},
"photos": [],
}
_OFFERS = [_vtorichka_offer(), _newbuilding_offer(), _studio_offer()]
def _make_old() -> OldCianScraper:
with patch("app.services.scrapers.cian.get_scraper_delay", return_value=5.0):
return OldCianScraper()
def _make_new() -> NewCianScraper:
return NewCianScraper(config=_CONFIG)
def _dump(lots: list[Any]) -> list[dict[str, Any]]:
return [lot.model_dump() for lot in lots]
# ── SERP _parse_serp_html parity ─────────────────────────────────────────────
def test_serp_parse_serp_html_parity() -> None:
"""Redux offers → list[ScrapedLot] идентичны (вторичка + новостройка + студия)."""
html = _serp_html(_OFFERS)
old_lots = _make_old()._parse_serp_html(html)
new_lots = _make_new()._parse_serp_html(html)
assert len(old_lots) == len(new_lots) == 3
assert _dump(old_lots) == _dump(new_lots)
def test_serp_offer_to_lot_parity_per_offer() -> None:
"""Пооферный _offer_to_lot: model_dump идентичен для каждого offer."""
old, new = _make_old(), _make_new()
for offer in _OFFERS:
old_lot = old._offer_to_lot(offer)
new_lot = new._offer_to_lot(offer)
assert (old_lot is None) == (new_lot is None)
if old_lot is not None and new_lot is not None:
assert old_lot.model_dump() == new_lot.model_dump()
def test_serp_extract_total_offers_parity() -> None:
html = _serp_html(_OFFERS)
assert _make_old()._extract_total_offers(html) == _make_new()._extract_total_offers(html)
def test_format_address_parity() -> None:
"""_format_address (pure) — вкл. восстановление дома для ЖК без части house."""
building = {"houseNumber": None}
nb = {"name": "ЖК Парковый квартал 29А к1"}
cases = [
# (address_parts, is_newbuilding kwargs)
([], {}),
(
[
{"type": "location", "fullName": "Свердловская область"},
{"type": "street", "fullName": "улица Мира"},
{"type": "house", "fullName": "12"},
],
{},
),
(
[
{"type": "location", "fullName": "Свердловская область"},
{"type": "street", "fullName": "улица Новая"},
],
{
"building": building,
"building_cadastral_number": "66:41:0204016:11396",
"newbuilding": nb,
"is_newbuilding": True,
},
),
]
for parts, kwargs in cases:
assert old_format_address(parts, **kwargs) == new_format_address(parts, **kwargs)
def test_extract_state_parity() -> None:
"""app.extract_state ≡ scraper_kit.extract_state на SERP-фикстуре."""
html = _serp_html(_OFFERS)
old_state = old_extract_state(html, mfe="frontend-serp", key="initialState")
new_state = new_extract_state(html, mfe="frontend-serp", key="initialState")
assert old_state == new_state
assert new_state is not None
# ── detail fetch_detail parity ───────────────────────────────────────────────
def _detail_html() -> str:
"""defaultState + bti sister → _cianConfig['frontend-offer-card'] push HTML."""
default_state = {
"offerData": {
"offer": {
"cianId": 327830237,
"windowsViewType": "yardAndStreet",
"separateWcsCount": 1,
"combinedWcsCount": 0,
"building": {"ceilingHeight": 2.65},
"repairType": "cosmetic",
"kitchenArea": "11.5",
"description": "Отличная квартира",
"priceChanges": [
{"changeTime": "2026-04-01T00:00:00Z", "price": 5500000, "diffPercent": -1.8}
],
"agent": {
"id": 7654321,
"companyName": "Первичка Плюс",
"isPro": True,
"skills": [],
"accountType": "specialist",
},
"newbuilding": {"id": 0},
},
"stats": {
"totalViewsFormattedString": "1 042",
"todayViewsFormattedString": "7",
},
}
}
bti_state = {"houseData": {"series": "1-464", "wallMaterial": "brick"}}
ds_json = json.dumps(default_state, ensure_ascii=False)
bti_json = json.dumps(bti_state, ensure_ascii=False)
return (
"<html><body><script>\n"
"window._cianConfig['frontend-offer-card'] = "
"window._cianConfig['frontend-offer-card'] || [];\n"
"window._cianConfig['frontend-offer-card'].push({"
f"key: 'defaultState', value: {ds_json}, priority: 10, filter: null"
"});\n"
"window._cianConfig['frontend-offer-card'].push({"
f"key: 'bti', value: {bti_json}, priority: 5, filter: null"
"});\n</script></body></html>"
)
def _mock_session(html: str) -> MagicMock:
response = MagicMock()
response.status_code = 200
response.text = html
session = MagicMock()
session.get = AsyncMock(return_value=response)
session.close = AsyncMock()
return session
@pytest.mark.asyncio
async def test_detail_fetch_parity() -> None:
"""fetch_detail (defaultState + bti) → DetailEnrichment идентичен (все поля)."""
html = _detail_html()
url = "https://ekb.cian.ru/sale/flat/327830237/"
old_res = await old_fetch_detail(url, session=_mock_session(html))
new_res = await new_fetch_detail(url, session=_mock_session(html))
assert old_res is not None
assert new_res is not None
assert dataclasses.asdict(old_res) == dataclasses.asdict(new_res)
# ── valuation _parse_valuation_state parity ──────────────────────────────────
def _valuation_state() -> dict[str, Any]:
return {
"user": {"isAuthenticated": True, "userId": 42},
"estimation": {
"sale": {
"data": {
"price": 6800000,
"accuracy": 87,
"priceFrom": 6400000,
"priceTo": 7200000,
"priceSqm": 130000,
"filtersHash": "hash-abc",
}
},
"rent": {
"data": {
"price": 42000,
"accuracy": 80,
"priceFrom": 39000,
"priceTo": 45000,
"taxPrice": 5460,
}
},
},
"estimationChart": {
"data": {
"title": {"change": "increase", "changeValue": "8,3%"},
"chartData": {
"data": [
{"date": 1714521600000, "price": 6700000, "priceFormatted": "6,7 млн"},
{"date": 1717200000000, "price": 6800000, "priceFormatted": "6,8 млн"},
]
},
}
},
"houseInfo": {
"data": {
"items": [
{"title": "Год постройки", "value": "2010"},
{"title": "Тип дома", "value": "монолит"},
]
}
},
"filters": {"houseId": 998877},
"managementCompany": {"data": {"name": "УК Ромашка", "phones": ["+73430000000"]}},
}
def test_valuation_parse_state_parity() -> None:
"""_parse_valuation_state → CianValuationResult идентичен (все поля)."""
state = _valuation_state()
old_res = old_parse_valuation(state)
new_res = new_parse_valuation(state)
assert dataclasses.asdict(old_res) == dataclasses.asdict(new_res)
# ── newbuilding pure-экстракторы parity ──────────────────────────────────────
def test_newbuilding_extractors_parity() -> None:
rv_state = {
"data": {
"priceDynamics": {
"chart": {
"data": {"labels": ["2026-01-01", "2026-02-01"], "values": [120000, 125000]}
}
}
}
}
assert old_extract_chart(rv_state) == new_extract_chart(rv_state)
rel_state = {
"checkStatus": {"status": "reliable", "title": "Надёжный застройщик"},
"details": [{"title": "Проектная декларация", "type": "doc"}],
}
assert old_extract_reliability(rel_state) == new_extract_reliability(rel_state)
offers_state = {
"data": {
"total": {
"fromDeveloperRooms": [
{"roomType": "1", "layouts": [{"id": 1, "area": 38}]},
{"roomType": "2", "layouts": [{"id": 2, "area": 55}]},
]
}
}
}
assert old_extract_nested_offers(offers_state) == new_extract_nested_offers(offers_state)
serp_html = (
'<h1 data-name="Title">Купить квартиру в '
'<a href="https://zhk-parkovyy-kvartal-ekb-i.cian.ru/">ЖК «Парковый квартал»</a></h1>'
)
assert old_extract_zhk_url(serp_html) == new_extract_zhk_url(serp_html)
# ── strangler-guard: kit-провайдер не импортирует app.* ──────────────────────
def test_kit_cian_has_no_app_imports() -> None:
"""Ни один модуль scraper_kit.providers.cian.* не импортирует app.* (развязка)."""
import inspect
from scraper_kit.providers.cian import detail, newbuilding, serp, session, valuation
for mod in (serp, detail, newbuilding, valuation, session):
source = inspect.getsource(mod)
for line in source.splitlines():
stripped = line.strip()
assert not stripped.startswith("from app"), f"{mod.__name__}: {line!r}"
assert not stripped.startswith("import app"), f"{mod.__name__}: {line!r}"

View file

@ -108,6 +108,8 @@ class ScraperConfig(Protocol):
# Границы валидности cian-оценки, руб.
cian_valuation_min_rub: float
cian_valuation_max_rub: float
# Секрет pgp_sym_encrypt для cian_session_cookies (read в session.py).
cookie_encryption_key: str
# Sentry/GlitchTip DSN (опционально).
glitchtip_dsn: str | None

View file

@ -0,0 +1,90 @@
"""Нормализация house_type → каноничный enum для listings.house_type.
Целевой канон (сверено с estimator._IMV_HOUSE_TYPE_MAP, estimator.py:146):
panel / brick / monolith / monolith_brick / block / wood.
Зачем (#2007): estimator применяет soft-penalty по house_type — аналог с
house_type != target штрафуется. Yandex SERP отдаёт SCREAMING-значения
(MONOLIT / BRICK / PANEL / ...), которые НИКОГДА не равны каноничным
(monolith / brick / panel / ...) ~70% yandex-аналогов получали ложный штраф
и фактически выпадали из скоринга. Нормализация на ингесте чинит это.
Важно про None: неизвестное / 'other' / '' None, НЕ 'other'. В estimator
`house_type IS NULL` нейтрально (без штрафа), а любое не-канон значение всегда
!= target ложный штраф. Поэтому unknown лучше схлопнуть в NULL.
Источники raw-значений:
- yandex SERP: building.buildingType SCREAMING_SNAKE (MONOLIT, MONOLIT_BRICK, ...)
- cian SERP: building.materialType camelCase (monolith, monolithBrick,
gasSilicateBlock, ...) переиспользуется в #2008 (cian.py:815).
"""
from __future__ import annotations
import logging
logger = logging.getLogger(__name__)
# Канон listings.house_type — выровнен по estimator._IMV_HOUSE_TYPE_MAP.
_CANON: frozenset[str] = frozenset(
{"panel", "brick", "monolith", "monolith_brick", "block", "wood"}
)
# raw-токен → канон. Ключи — точные значения вокабуляров источников; сравнение
# регистронезависимое (см. _LOOKUP ниже), но строго по полному токену енума,
# без fuzzy-матчинга. Неизвестные токены сюда НЕ попадают → normalize вернёт None.
_RAW_TO_CANON: dict[str, str] = {
# ── yandex SERP (SCREAMING_SNAKE buildingType) ───────────────────────────
"MONOLIT": "monolith",
"BRICK": "brick",
"PANEL": "panel",
"MONOLIT_BRICK": "monolith_brick",
"BLOCK": "block",
"WOOD": "wood",
# ── cian SERP (camelCase materialType) — reuse в #2008 ───────────────────
"monolith": "monolith",
"brick": "brick",
"panel": "panel",
"block": "block",
"wood": "wood",
"monolithBrick": "monolith_brick",
"gasSilicateBlock": "block",
"aerocreteBlock": "block",
"foamConcreteBlock": "block",
"stalin": "brick", # «сталинка» — кирпич
}
# Регистронезависимый lookup. Лоуэркейс-ключи не коллизят между вокабулярами:
# 'MONOLIT'→'monolit' и 'monolith'→'monolith' — разные ключи.
_LOOKUP: dict[str, str] = {k.lower(): v for k, v in _RAW_TO_CANON.items()}
def normalize_house_type(raw: str | None) -> str | None:
"""Преобразовать raw house_type в каноничный enum.
Уже-каноничное значение возвращается as-is (идемпотентность нужна при
ре-обработке и для backfill-миграции). Неизвестное / 'other' / пустое /
None None (НЕ 'other': см. docstring модуля про estimator soft-penalty).
Args:
raw: сырое значение из парсера (e.g. «MONOLIT», «monolithBrick») или None.
Returns:
Одно из panel / brick / monolith / monolith_brick / block / wood, либо None.
"""
if raw is None:
return None
stripped = raw.strip()
if not stripped:
return None
# Pass-through: уже каноничное значение (idempotency — в т.ч. 'monolith_brick',
# которого нет среди raw-ключей карты).
if stripped in _CANON:
return stripped
# Регистронезависимый exact-token lookup по обоим вокабулярам.
canon = _LOOKUP.get(stripped) or _LOOKUP.get(stripped.lower())
if canon is None:
# Неизвестное / 'other' — нормально (NULL нейтрально для estimator).
# debug, не warning: 'other' встречается массово, warning засорил бы лог.
logger.debug("house_type_normalizer: unmapped raw value %r — stored as NULL", raw)
return canon

View file

@ -0,0 +1,17 @@
"""Cian-провайдер scraper_kit — strangler-копия `app.services.scrapers.cian*`.
Модули:
- serp SERP-sweep ядро (CianScraper, Redux state ScrapedLot, price-bisect)
- detail detail-страница объявления (fetch_detail DetailEnrichment)
- newbuilding каталог ЖК (fetch_newbuilding NewbuildingEnrichment, resolve zhk-url)
- valuation Cian Valuation Calculator (estimate_via_cian_valuation)
- session auth cookie management (load/save/verify/mark-invalid)
Развязка от `app.*`:
- `app.core.config.settings` инжектируемый `ScraperConfig` (scraper_kit.contracts)
- `app.services.scraper_settings.get_scraper_delay` инжектируемый `delay_provider`
- `app.services.scrapers.cian_state_parser` `scraper_kit.cian_state_parser`
- `app.services.scrapers.{base,browser_fetcher,price_brackets,repair_state_normalizer,
snapshot_writer,house_type_normalizer}` `scraper_kit.*`
- `app.services.cian_session` локальная копия `scraper_kit.providers.cian.session`
"""

View file

@ -0,0 +1,450 @@
"""Cian.ru detail page scraper — single offer enrichment.
Different from SERP (cian.py):
- URL: https://ekb.cian.ru/sale/flat/<offer_id>/ or https://www.cian.ru/sale/flat/<offer_id>/
- MFE: 'frontend-offer-card'
- State KEY: 'defaultState' (NOT 'initialState' SERP uses initialState)
- Sister containers in same _cianConfig: bti, priceChanges, stats, agent, newObject
Parses ~88 offer fields + sister data DetailEnrichment dataclass.
Stage 5 of CianScraper v1.
"""
from __future__ import annotations
import json
import logging
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
from curl_cffi.requests import AsyncSession
from sqlalchemy import text
from sqlalchemy.orm import Session
from scraper_kit.cian_state_parser import extract_all_states, extract_state
from scraper_kit.repair_state_normalizer import (
infer_repair_state_from_text,
normalize_repair_state,
)
from scraper_kit.snapshot_writer import upsert_listing_snapshot
if TYPE_CHECKING:
from scraper_kit.browser_fetcher import BrowserFetcher
from scraper_kit.contracts import ScraperConfig
logger = logging.getLogger(__name__)
@dataclass
class DetailEnrichment:
"""Result of cian_detail.fetch_detail() — fields для UPDATE listings + sister rows."""
# Core listing enrichment (extend existing listings row)
cian_id: int | None = None
windows_view_type: str | None = None # 'yard' / 'street' / 'yardAndStreet'
separate_wcs_count: int | None = None
combined_wcs_count: int | None = None
ceiling_height: float | None = None # meters — offer.building.ceilingHeight (#1791)
repair_type: str | None = None # raw Cian: offer.repairType OR offer.decoration (#1791)
repair_state: str | None = None # enum: needs_repair/standard/good/excellent
kitchen_area_m2: float | None = None # offer.kitchenArea (м²)
description: str | None = None # offer.description (текст объявления)
views_total: int | None = None # Cian stats.totalViewsFormattedString → int
views_today: int | None = None
# BTI (вторичка only — primary buildings don't have БТИ)
bti_data: dict[str, Any] | None = None # raw bti.houseData snapshot
# Price history (offer.priceChanges)
price_changes: list[dict[str, Any]] = field(default_factory=list)
# Agent profile (offer.agent.* expanded)
agent_profile: dict[str, Any] | None = None
# Newbuilding link (if offer is from a newbuilding)
newbuilding_data: dict[str, Any] | None = None
# Raw payload для debugging / future-extraction
raw_offer: dict[str, Any] | None = None
raw_sister_states: dict[str, Any] = field(default_factory=dict)
async def fetch_detail(
offer_url: str,
*,
config: ScraperConfig | None = None,
session: AsyncSession | None = None,
browser_fetcher: BrowserFetcher | None = None,
) -> DetailEnrichment | None:
"""Fetch detail page and extract all containers.
Args:
offer_url: e.g. 'https://ekb.cian.ru/sale/flat/327830237/'
config: инжектируемый ScraperConfig (#2133) — читается только на curl_cffi-пути
для прокси-egress (config.cian_proxy_url). None прямое подключение
(dev / browser_fetcher-путь, где прокси уже применён server-side).
session: optional shared curl_cffi session (для batched scrapes).
Ignored when browser_fetcher is provided.
browser_fetcher: optional BrowserFetcher instance (camoufox HTTP service).
When provided, fetches JS-rendered HTML via the browser service instead of
curl_cffi enables priceChanges extraction which requires JS rendering.
Caller is responsible for the context-manager lifecycle of the fetcher.
Returns: DetailEnrichment, or None если fetch / parse failed.
"""
if browser_fetcher is not None:
# Browser path: get fully JS-rendered HTML; same parse path follows.
try:
html = await browser_fetcher.fetch(offer_url)
except Exception as exc:
logger.warning("Cian detail browser fetch failed %s: %s", offer_url, exc)
return None
else:
# curl_cffi path (legacy, back-compat).
close_session = False
if session is None:
# proxies: mobile-proxy egress (#806) — без прокси datacenter-IP блокируется Cian.
# Пусто (config не передан / env не задан) → прямое подключение (dev/no-op).
_proxy_url = config.cian_proxy_url if config is not None else None
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
session = AsyncSession(
impersonate="chrome120",
timeout=25.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
try:
resp = await session.get(offer_url, allow_redirects=True)
if resp.status_code != 200:
logger.warning("Cian detail fetch %s → HTTP %d", offer_url, resp.status_code)
return None
html = resp.text
finally:
if close_session:
await session.close()
# ── Shared parse path (identical regardless of HTML source) ──────────────
# Primary state container — defaultState in frontend-offer-card MFE
# NOTE: detail pages use 'defaultState', SERP uses 'initialState'
offer_state = extract_state(html, mfe="frontend-offer-card", key="defaultState")
if offer_state is None:
logger.warning("Cian detail %s: defaultState extraction failed", offer_url)
return None
offer_data = offer_state.get("offerData", {})
offer = offer_data.get("offer", {})
result = DetailEnrichment(
cian_id=offer.get("cianId") or offer.get("id"),
windows_view_type=_extract_windows_view(offer),
separate_wcs_count=offer.get("separateWcsCount"),
combined_wcs_count=offer.get("combinedWcsCount"),
ceiling_height=_extract_ceiling_height(offer),
repair_type=_extract_repair_type(offer),
repair_state=_extract_repair_state(offer),
kitchen_area_m2=_parse_float(offer.get("kitchenArea")),
description=offer.get("description") or None,
raw_offer=offer,
)
# Stats (views) — from stats key in offerData or top-level state
stats = offer_data.get("stats") or offer_state.get("stats") or {}
result.views_total = _parse_views(stats.get("totalViewsFormattedString"))
result.views_today = _parse_views(stats.get("todayViewsFormattedString"))
# Price changes — browser HTML: offerData.priceChanges (priceData.price format)
# curl_cffi HTML: offer.priceChanges or state.priceChanges (flat price field)
result.price_changes = _extract_price_changes(offer, offer_data, offer_state)
# Agent profile expansion
result.agent_profile = _extract_agent_profile(offer)
# Sister state containers: bti, newObject, plus raw stash of everything else
all_states = extract_all_states(html)
offer_card_states = all_states.get("frontend-offer-card", {})
bti_state = offer_card_states.get("bti")
if bti_state:
result.bti_data = bti_state.get("houseData")
result.raw_sister_states["bti"] = bti_state
# Newbuilding link (if applicable)
newbuilding = offer.get("newbuilding")
if newbuilding and newbuilding.get("id"):
result.newbuilding_data = newbuilding
# Stash all non-defaultState sister containers for debugging / future-extraction
for k, v in offer_card_states.items():
if k != "defaultState":
result.raw_sister_states[k] = v
return result
# ── Field extractors ──────────────────────────────────────────────────────────
def _extract_windows_view(offer: dict[str, Any]) -> str | None:
return offer.get("windowsViewType")
def _extract_ceiling_height(offer: dict[str, Any]) -> float | None:
"""Высота потолков (м).
На detail-странице (defaultState) значение лежит в ``offer.building.ceilingHeight``,
а НЕ в ``offer.ceilingHeight`` (последнее всегда null) #1791. Top-level оставлен
как fallback на случай иной структуры карточки.
"""
building = offer.get("building") or {}
return _parse_float(building.get("ceilingHeight") or offer.get("ceilingHeight"))
def _extract_repair_type(offer: dict[str, Any]) -> str | None:
# Cian: 'cosmetic' / 'design' / 'no' / 'euro' / ...
# На detail-странице структурное поле чаще приходит как `decoration`
# ('preFine'/'fine'/...), а `repairType` пустое (#1791) — читаем оба.
return offer.get("repairType") or offer.get("decoration")
def _extract_repair_state(offer: dict[str, Any]) -> str | None:
"""Нормализованный enum из repairType/decoration → needs_repair/standard/good/excellent.
Если структурное поле пустое fallback на инференс из описания (#622).
Структурное значение всегда в приоритете.
"""
state = normalize_repair_state(offer.get("repairType") or offer.get("decoration"))
if state is None:
state = infer_repair_state_from_text(offer.get("description"))
return state
def _parse_float(value: Any) -> float | None:
if value is None:
return None
try:
return float(value)
except (ValueError, TypeError):
return None
def _parse_views(formatted_str: str | None) -> int | None:
"""Cian's stats.totalViewsFormattedString — '1 234' → 1234."""
if not formatted_str:
return None
try:
return int(str(formatted_str).replace(" ", "").replace(",", ""))
except (ValueError, TypeError):
return None
def _extract_price_changes(
offer: dict[str, Any],
offer_data: dict[str, Any],
state: dict[str, Any],
) -> list[dict[str, Any]]:
"""Normalize priceChanges from any of the known Cian locations/formats.
Lookup priority:
1. offer.priceChanges curl_cffi flat format: {changeTime, price, diffPercent}
2. offerData.priceChanges browser-rendered format: {changeTime, priceData:{price}}
3. state.priceChanges top-level fallback (rare)
Normalizes to: {change_time, price_rub, diff_percent}.
"""
raw = (
offer.get("priceChanges")
or offer_data.get("priceChanges")
or state.get("priceChanges")
or []
)
if not isinstance(raw, list):
return []
result = []
for entry in raw:
if not isinstance(entry, dict):
continue
# price_rub: flat field (curl_cffi) OR nested priceData.price (browser)
price_data = entry.get("priceData") or {}
price_rub = entry.get("price") or entry.get("priceRub") or price_data.get("price")
result.append(
{
"change_time": entry.get("changeTime") or entry.get("date"),
"price_rub": price_rub,
"diff_percent": entry.get("diffPercent") or entry.get("diff_percent"),
}
)
return result
def _extract_agent_profile(offer: dict[str, Any]) -> dict[str, Any] | None:
"""offer.agent or offer.agency or offer.user — extract canonical profile shape."""
agent = offer.get("agent") or {}
user = offer.get("user") or {}
if not agent and not user:
return None
return {
"ext_agent_id": agent.get("id") or user.get("cianUserId"),
"company_name": agent.get("companyName") or user.get("companyName"),
"is_pro": agent.get("isPro") or user.get("isPro"),
"skills": agent.get("skills") or [],
"account_type": agent.get("accountType") or user.get("accountType"),
}
# ── Save helpers ──────────────────────────────────────────────────────────────
def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnrichment) -> None:
"""Persist DetailEnrichment to DB:
- UPDATE listings SET windows_view_type, ceiling_height, ... WHERE id = listing_id
- INSERT INTO offer_price_history rows (if price_changes non-empty)
- INSERT INTO agents (if agent_profile) and link via agent_id_fk
- bti_data не сохраняется здесь это задача Stage 6 (houses), где есть house_id
Idempotency: re-running безопасно (UPSERT / ON CONFLICT DO NOTHING semantics).
"""
# Extend listings row with detail-enriched fields
db.execute(
text("""
UPDATE listings SET
windows_view_type = COALESCE(:wvt, windows_view_type),
separate_wcs_count = COALESCE(:swc, separate_wcs_count),
combined_wcs_count = COALESCE(:cwc, combined_wcs_count),
ceiling_height = COALESCE(:ch, ceiling_height),
repair_type = COALESCE(:rt, repair_type),
repair_state = COALESCE(:rs, repair_state),
kitchen_area_m2 = COALESCE(CAST(:ka AS double precision), kitchen_area_m2),
description = COALESCE(:descr, description),
views_total = COALESCE(:vt, views_total),
views_today = COALESCE(:vd, views_today),
detail_enriched_at = NOW()
WHERE id = CAST(:lid AS bigint)
"""),
{
"lid": listing_id,
"wvt": enrichment.windows_view_type,
"swc": enrichment.separate_wcs_count,
"cwc": enrichment.combined_wcs_count,
"ch": enrichment.ceiling_height,
"rt": enrichment.repair_type,
"rs": enrichment.repair_state,
"ka": enrichment.kitchen_area_m2,
"descr": enrichment.description,
"vt": enrichment.views_total,
"vd": enrichment.views_today,
},
)
# Snapshot: point-in-time observation в listings_snapshots.
# Берём price_rub / price_per_m2 из listings строки (только что обновлённой).
# Используем COALESCE — на случай если listings.price_rub NULL (не должно быть, но safe).
_snap_row = db.execute(
text("""
SELECT price_rub, price_per_m2
FROM listings
WHERE id = CAST(:lid AS bigint)
"""),
{"lid": listing_id},
).fetchone()
_snap_written = False
if _snap_row is not None and _snap_row.price_rub is not None:
try:
with db.begin_nested():
upsert_listing_snapshot(
db,
listing_id=listing_id,
price_rub=int(_snap_row.price_rub),
price_per_m2=int(_snap_row.price_per_m2) if _snap_row.price_per_m2 else None,
run_id=None,
status="active",
)
_snap_written = True
except Exception as exc:
logger.warning(
"save_detail_enrichment: snapshot write failed listing_id=%s: %s",
listing_id,
exc,
)
# Price changes — INSERT new rows, de-dup by listing_id + change_time
for change in enrichment.price_changes:
if not change.get("change_time") or not change.get("price_rub"):
continue
db.execute(
text("""
INSERT INTO offer_price_history
(listing_id, change_time, price_rub, diff_percent, source)
VALUES (
CAST(:lid AS bigint),
CAST(:ct AS timestamptz),
CAST(:price AS numeric),
CAST(:diff AS numeric),
'cian'
)
ON CONFLICT ON CONSTRAINT offer_price_history_listing_change_uq DO NOTHING
"""),
{
"lid": listing_id,
"ct": change["change_time"],
"price": change["price_rub"],
"diff": change.get("diff_percent"),
},
)
# Agent profile UPSERT — store agent row and link to listing
if enrichment.agent_profile and enrichment.agent_profile.get("ext_agent_id"):
ap = enrichment.agent_profile
row = db.execute(
text("""
INSERT INTO agents
(ext_source, ext_agent_id, company_name, is_pro, skills, account_type)
VALUES (
'cian',
:eai,
:cn,
:ip,
CAST(:sk AS jsonb),
:at
)
ON CONFLICT (ext_source, ext_agent_id) DO UPDATE SET
company_name = COALESCE(EXCLUDED.company_name, agents.company_name),
is_pro = COALESCE(EXCLUDED.is_pro, agents.is_pro),
skills = COALESCE(EXCLUDED.skills, agents.skills),
account_type = COALESCE(EXCLUDED.account_type, agents.account_type)
RETURNING id
"""),
{
"eai": str(ap["ext_agent_id"]),
"cn": ap.get("company_name"),
"ip": ap.get("is_pro"),
"sk": json.dumps(ap.get("skills") or []),
"at": ap.get("account_type"),
},
)
agent_db_id = row.scalar_one_or_none()
if agent_db_id is not None:
db.execute(
text("""
UPDATE listings
SET agent_id_fk = CAST(:aid AS bigint)
WHERE id = CAST(:lid AS bigint)
AND agent_id_fk IS NULL
"""),
{"aid": agent_db_id, "lid": listing_id},
)
db.commit()
logger.info(
"Cian detail enrichment saved for listing_id=%s (price_changes=%d, agent=%s, snapshot=%s)",
listing_id,
len(enrichment.price_changes),
bool(enrichment.agent_profile),
_snap_written,
)

View file

@ -0,0 +1,792 @@
"""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 TYPE_CHECKING, Any
from curl_cffi.requests import AsyncSession # type: ignore[import-untyped]
from scraper_kit.browser_fetcher import BrowserFetcher
from scraper_kit.cian_state_parser import extract_all_states, extract_state
if TYPE_CHECKING:
from scraper_kit.contracts import ScraperConfig
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,
*,
config: ScraperConfig | None = None,
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 = config.cian_proxy_url if config is not None else None
_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,
*,
config: ScraperConfig | None = None,
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 (``config.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 = config.cian_proxy_url if config is not None else None
_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()

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,265 @@
"""Cian session cookie management — load/save/verify encrypted cookies.
Used by cian_valuation.py (Stage 7) to authenticate Valuation Calculator API.
Cookies stored encrypted (pgp_sym_encrypt) in cian_session_cookies table.
"""
from __future__ import annotations
import json
import logging
from typing import TYPE_CHECKING, Any
from curl_cffi.requests import AsyncSession
from sqlalchemy import text
from sqlalchemy.orm import Session
from scraper_kit.cian_state_parser import extract_state
if TYPE_CHECKING:
from scraper_kit.contracts import ScraperConfig
logger = logging.getLogger(__name__)
# Cookies критичные для Cian auth — фильтр перед сохранением.
# Список обновлён по реальному DevTools-дампу из logged-in сессии cian.ru (2026-05-23).
# Старые записи оставлены как fallback (backward compat).
CIAN_REQUIRED_COOKIES: set[str] = {
# --- Cian auth & session (critical) ---
"DMIR_AUTH", # Cian auth token (httpOnly) — основной auth-сигнал
"_CIAN_GK", # Cian session GUID
# --- Yandex Metrika (обычно присутствуют, не critical) ---
"_ym_uid",
"_ym_d",
"_ym_isad",
"_ym_visorc",
"_yasc", # Yandex anti-spam
# --- Cian UX state ---
"uxfb_card_satisfaction",
# --- Cian session IDs (fallback / legacy deployments) ---
"_cian_visitor_session_id",
"_cian_app_session_id",
"_cian_uid",
# --- Misc (legacy / optional) ---
"_ga", # Google Analytics
"tlsr_id", # legacy fingerprint
"cf_clearance", # Cloudflare bot check
"session-id", # generic session cookie
"anti_bot",
"csrftoken",
}
_VALUATION_TEST_URL = (
"https://www.cian.ru/kalkulator-nedvizhimosti/"
"?address=%D0%A1%D0%B2%D0%B5%D1%80%D0%B4%D0%BB%D0%BE%D0%B2%D1%81%D0%BA%D0%B0%D1%8F"
"%20%D0%BE%D0%B1%D0%BB%2C%20%D0%95%D0%BA%D0%B0%D1%82%D0%B5%D1%80%D0%B8%D0%BD%D0%B1%D1%83%D1%80%D0%B3"
"&totalArea=40&roomsCount=1&valuationType=sale"
"&floor%5B0%5D=floorOther&repairType%5B0%5D=repairTypeCosmetic"
)
_MFE_VALUATION = "valuation-for-agent-frontend"
# #639 (2026-05): auth-сигнал (user.isAuthenticated/userId) уехал из valuation-MFE
# (там user теперь всегда anonymous) в site-header MFE `header-frontend`, который
# присутствует на любой cian-странице. Verify читает auth именно оттуда.
_MFE_AUTH = "header-frontend"
# Returned by verify_session when a TLS/bot-fingerprint ban (HTTP 403) is detected.
# Callers that check `if state is None` will NOT treat this as "expired cookies" —
# the sentinel is truthy (non-None), so cookie-refresh alerts are suppressed.
# Callers that need to distinguish ban from valid auth should check state.get("_ban").
VERIFY_BAN_SENTINEL: dict[str, Any] = {"_ban": True}
def _classify_verify_response(
status_code: int,
html: str | None,
) -> dict[str, Any] | None:
"""Pure classifier — maps (status_code, html) to verify_session outcome.
Returns:
VERIFY_BAN_SENTINEL 403/TLS ban (cookies may be fine, server is blocking)
None 401 or isAuthenticated=false (cookies genuinely expired)
state dict authenticated successfully
"""
if status_code == 403:
return VERIFY_BAN_SENTINEL
if status_code == 401:
return None
if html is None:
return None
state = extract_state(html, mfe=_MFE_AUTH, key="initialState")
if state is None:
return None
user = state.get("user", {}) or {}
if not user.get("isAuthenticated"):
return None
return state
async def verify_session(
cookies: dict[str, str],
*,
config: ScraperConfig | None = None,
) -> dict[str, Any] | None:
"""Hit Cian Valuation Calculator with cookies — return parsed state if authenticated.
Uses curl_cffi with impersonate='chrome120' (same as prod scrapers) to avoid
TLS-fingerprint bans that httpx would trigger.
Returns:
state dict authenticated (contains user.isAuthenticated + userId)
VERIFY_BAN_SENTINEL HTTP 403 TLS/bot ban; cookies may still be valid
callers should NOT trigger a cookie-refresh alert
None HTTP 401 or isAuthenticated=false; cookies expired
Никогда не логирует сырые значения cookies.
"""
try:
# proxies: mobile-proxy egress (#806) — Cian блокирует datacenter-IP даже
# при валидных DMIR_AUTH cookies. Без прокси verify всегда вернёт 403.
# Пусто (env не задан) → прямое подключение (dev/no-op).
_proxy_url = config.cian_proxy_url if config is not None else None
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
async with AsyncSession(
impersonate="chrome120",
cookies=cookies,
timeout=20.0,
proxies=_proxies,
) as session:
resp = await session.get(_VALUATION_TEST_URL, allow_redirects=True)
status = resp.status_code
html: str | None = resp.text if status == 200 else None
result = _classify_verify_response(status, html)
if result is VERIFY_BAN_SENTINEL:
logger.warning(
"Cian cookies verify: HTTP 403 TLS/bot ban — cookies NOT marked expired"
)
elif result is None:
logger.warning("Cian cookies verify: expired/unauthenticated (status=%d)", status)
else:
user = result.get("user", {}) or {}
logger.info("Cian cookies verified — userId=%s", user.get("userId"))
return result
except Exception as exc:
logger.warning("Cian cookies verify failed: %s", exc)
return None
def save_session(
db: Session,
account_user_id: int,
cookies: dict[str, str],
ttl_days: int = 30,
*,
config: ScraperConfig,
) -> None:
"""Encrypt cookies via pgp_sym_encrypt and UPSERT into cian_session_cookies.
Uses config.cookie_encryption_key as the encryption secret.
Никогда не логирует сырые значения cookies.
"""
cookies_json = json.dumps(cookies)
db.execute(
text("""
INSERT INTO cian_session_cookies (
account_user_id,
cookies_encrypted,
expires_at_estimate,
uploaded_at
) VALUES (
CAST(:uid AS bigint),
pgp_sym_encrypt(:cookies_json, :key),
NOW() + (CAST(:ttl_days AS int) || ' days')::interval,
NOW()
)
ON CONFLICT (account_user_id) DO UPDATE SET
cookies_encrypted = EXCLUDED.cookies_encrypted,
expires_at_estimate = EXCLUDED.expires_at_estimate,
uploaded_at = NOW(),
last_invalid_at = NULL
"""),
{
"uid": account_user_id,
"cookies_json": cookies_json,
"key": config.cookie_encryption_key,
"ttl_days": ttl_days,
},
)
db.commit()
logger.info(
"Cian cookies saved for userId=%s (count=%d, ttl=%d days)",
account_user_id,
len(cookies),
ttl_days,
)
def load_session(db: Session, *, config: ScraperConfig) -> dict[str, str] | None:
"""Load most-recently-uploaded valid Cian cookies (decrypt).
Returns dict[cookie_name, cookie_value] или None если нет валидной session.
Выбирает только записи где expires_at_estimate > NOW() и сессия не
была инвалидирована после последнего upload'а.
"""
row = (
db.execute(
text("""
SELECT
account_user_id,
pgp_sym_decrypt(cookies_encrypted, :key)::text AS cookies_json,
expires_at_estimate
FROM cian_session_cookies
WHERE expires_at_estimate > NOW()
AND (last_invalid_at IS NULL OR last_invalid_at < uploaded_at)
ORDER BY uploaded_at DESC
LIMIT 1
"""),
{"key": config.cookie_encryption_key},
)
.mappings()
.first()
)
if row is None:
logger.warning("No valid Cian session cookies in DB")
return None
cookies: dict[str, str] = json.loads(row["cookies_json"])
# Обновляем last_used_at — не критично, игнорируем ошибки.
try:
db.execute(
text(
"UPDATE cian_session_cookies SET last_used_at = NOW()"
" WHERE account_user_id = CAST(:uid AS bigint)"
),
{"uid": row["account_user_id"]},
)
db.commit()
except Exception as exc:
logger.warning(
"Failed to update last_used_at for userId=%s: %s", row["account_user_id"], exc
)
logger.info(
"Cian cookies loaded for userId=%s (count=%d)",
row["account_user_id"],
len(cookies),
)
return cookies
def mark_session_invalid(db: Session, account_user_id: int) -> None:
"""Flag session как expired/invalid (например после 401 во время scrape)."""
db.execute(
text(
"UPDATE cian_session_cookies SET last_invalid_at = NOW()"
" WHERE account_user_id = CAST(:uid AS bigint)"
),
{"uid": account_user_id},
)
db.commit()
logger.warning("Cian session marked invalid for userId=%s", account_user_id)

View file

@ -0,0 +1,556 @@
"""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.324.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
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.cian.session import load_session, mark_session_invalid
if TYPE_CHECKING:
from scraper_kit.contracts import 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,
) -> 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. Используем cian_proxy_url. proxy=None → прямое подключение (dev).
_proxy_url = config.cian_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.424.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,
},
)
db.commit()
except Exception as exc:
logger.error("Cian valuation cache save failed: %s", exc, exc_info=True)
raise