From 19c0be5d5a6a0fd00902eab4dd229f6827f7fc2b Mon Sep 17 00:00:00 2001 From: lekss361 Date: Thu, 2 Jul 2026 16:11:31 +0000 Subject: [PATCH] feat(scraper-kit): copy cian providers with protocol injection, strangler (#2133 cian) --- .../test_scraper_kit_cian_golden_parity.py | 492 +++++++ .../scraper-kit/src/scraper_kit/contracts.py | 2 + .../src/scraper_kit/house_type_normalizer.py | 90 ++ .../scraper_kit/providers/cian/__init__.py | 17 + .../src/scraper_kit/providers/cian/detail.py | 450 +++++++ .../scraper_kit/providers/cian/newbuilding.py | 792 +++++++++++ .../src/scraper_kit/providers/cian/serp.py | 1196 +++++++++++++++++ .../src/scraper_kit/providers/cian/session.py | 265 ++++ .../scraper_kit/providers/cian/valuation.py | 556 ++++++++ 9 files changed, 3860 insertions(+) create mode 100644 tradein-mvp/backend/tests/test_scraper_kit_cian_golden_parity.py create mode 100644 tradein-mvp/packages/scraper-kit/src/scraper_kit/house_type_normalizer.py create mode 100644 tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/__init__.py create mode 100644 tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/detail.py create mode 100644 tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/newbuilding.py create mode 100644 tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/serp.py create mode 100644 tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/session.py create mode 100644 tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/valuation.py diff --git a/tradein-mvp/backend/tests/test_scraper_kit_cian_golden_parity.py b/tradein-mvp/backend/tests/test_scraper_kit_cian_golden_parity.py new file mode 100644 index 00000000..5ca9ace7 --- /dev/null +++ b/tradein-mvp/backend/tests/test_scraper_kit_cian_golden_parity.py @@ -0,0 +1,492 @@ +"""Golden-parity: `scraper_kit.providers.cian.*` парсинг ≡ `app.services.scrapers.cian*`. + +Strangler-инвариант (#2133 cian): новая scraper_kit-копия cian-провайдера должна давать +БАЙТ-ИДЕНТИЧНЫЙ результат парсинга старому боевому коду на одинаковом входе. Развязка +(settings→ScraperConfig, get_scraper_delay→delay_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 ( + "" + ) + + +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 ( + "" + ) + + +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 = ( + '

Купить квартиру в ' + 'ЖК «Парковый квартал»

' + ) + 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}" diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py index eb04dc5e..de6f5d54 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py @@ -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 diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/house_type_normalizer.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/house_type_normalizer.py new file mode 100644 index 00000000..f9ad280d --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/house_type_normalizer.py @@ -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 diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/__init__.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/__init__.py new file mode 100644 index 00000000..7c2c82b2 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/__init__.py @@ -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` +""" diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/detail.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/detail.py new file mode 100644 index 00000000..0152cdd0 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/detail.py @@ -0,0 +1,450 @@ +"""Cian.ru detail page scraper — single offer enrichment. + +Different from SERP (cian.py): +- URL: https://ekb.cian.ru/sale/flat// or https://www.cian.ru/sale/flat// +- 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, + ) diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/newbuilding.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/newbuilding.py new file mode 100644 index 00000000..eb74c544 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/newbuilding.py @@ -0,0 +1,792 @@ +"""Cian.ru newbuilding (ЖК) catalog page scraper. + +URL: https://zhk---i.cian.ru/ +MFE: 'newbuilding-card-desktop-frontend', key: 'initialState' + +Sister state containers extracted from same MFE initialState top-level keys: +- realtyValuation: 7-month price chart (data.priceDynamics.chart.data.{labels,values}) + → houses_price_dynamics +- reliabilityState.reliability: наш.дом.рф checkStatus + details[] → house_reliability_checks +- offers: grouped by roomType (fromDeveloperRooms[].layouts[]) +- reviews: house reviews list +- housesByTurn: корпуса по очередям (stored as jsonb on houses) + +Stage 6 of CianScraper v1. +""" + +from __future__ import annotations + +import json +import logging +import re +from dataclasses import dataclass, field +from typing import 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 href inside its page

: +# +#

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

+# +# (verified on the live ~2.8 MB prod SERP for nb 48853 — the header sits inside a +# `NewbuildingHeaderSeo` block; there is NO / og:url on this page +# type, so the H1 anchor is the only reliable canonical carrier). Anchoring on it avoids +# the WRONG-ЖК hazard of a naive first-match: cat.php pages also carry banner / "похожие +# ЖК" / recommendation blocks that can embed OTHER zhk-*.cian.ru urls, and those blocks +# may render BEFORE the title — a plain first-match could then resolve a foreign ЖК and +# silently enrich the house with another building's data. +# +# Matches the FIRST zhk-slug url that occurs inside the page's

… +#

. `[^>]*` keeps the attr scan within the opening tag; `.*?` (DOTALL) spans the +# inner markup up to the slug href. Falls back to naive first-match only when this finds +# nothing (markup drift), preserving the previous behaviour as a safety net. +_ZHK_TITLE_ANCHOR_RE = re.compile( + r']*data-name="Title"[^>]*>.*?(https://zhk-[a-z0-9-]+\.cian\.ru)', + re.DOTALL, +) + +# cat.php newbuilding-SERP template: one row per `newobject[0]=`. ekb. host is the +# Свердловская-обл. mirror tradein targets; the slug URL it returns is host-agnostic +# (the canonical zhk-*.cian.ru subdomain), so this resolves ЕКБ newbuildings correctly. +_CATPH_NEWBUILDING_SERP = ( + "https://ekb.cian.ru/cat.php?deal_type=sale&engine_version=2" + "&offer_type=flat&object_type[0]=2&newobject[0]={nb_id}" +) + + +@dataclass +class NewbuildingEnrichment: + """Result of fetch_newbuilding() — houses row + sister containers.""" + + # Core house metadata (from state.newbuilding) + cian_internal_house_id: int | None = None + # ZHK page URL — canonical slug form stored after fetch / redirect resolution + cian_zhk_url: str | None = None + name: str | None = None + address: str | None = None + year_built: int | None = None + deadline_year: int | None = None + floors_count_min: int | None = None + floors_count_max: int | None = None + building_class: str | None = None # 'comfort' / 'business' / 'premium' + + # Management + management_company: dict[str, Any] | None = None # name, phones[], email, ... + + # Aggregated metrics + transport_accessibility_rate: int | None = None + advantages: list[dict[str, Any]] = field(default_factory=list) + + # Time-series price chart (Stage 6 main deliverable) + # Each entry: {'month_date': str, 'room_count': str, 'prices_type': str, 'period': str, + # 'price_per_sqm': float | None} + realty_valuation_chart: list[dict[str, Any]] = field(default_factory=list) + + # Reliability (наш.дом.рф) — overall status + checks detail array + # {check_status: str, check_name: str, details: list[dict]} + reliability_checks: list[dict[str, Any]] = field(default_factory=list) + + # Builders / banks (related orgs) + builders: list[dict[str, Any]] = field(default_factory=list) + banks: list[dict[str, Any]] = field(default_factory=list) + + # Corpuses (houses_by_turn — корпуса по очередям) + houses_by_turn: list[dict[str, Any]] = field(default_factory=list) + + # Reviews + reviews: list[dict[str, Any]] = field(default_factory=list) + + # Mini-SERP offers (grouped by roomType, layouts) + nested_offers: list[dict[str, Any]] = field(default_factory=list) + + # Raw payloads для debugging + raw_state: dict[str, Any] | None = None + raw_sister_states: dict[str, Any] = field(default_factory=dict) + + +async def fetch_newbuilding( + zhk_url: str, + *, + session: AsyncSession | None = None, # kept for backward-compat; unused for page fetch +) -> NewbuildingEnrichment | None: + """Fetch ЖК catalog page и extract все containers. + + HTML страница ЖК теперь тянется через BrowserFetcher (camoufox), потому что + curl_cffi получает страницу без initialState под анти-ботом Cian (#972). + tradein-browser верифицирован живым: 1.17 MB, initialState присутствует. + + Args: + zhk_url: e.g. 'https://zhk-ekaterininskiy-park-ekb-i.cian.ru/' + session: зарезервирован для обратной совместимости с вызывающими; + больше не используется для fetch страницы ЖК. Resolve-функции + (resolve_cian_zhk_url_via_search) по-прежнему используют curl_cffi. + + Returns: NewbuildingEnrichment, or None если fetch / parse failed. + """ + async with BrowserFetcher(source="cian") as browser: + html = await browser.fetch(zhk_url) + + # Cian ЖК-карточка: MFE 'newbuilding-card-desktop-frontend', key 'initialState' + # (verified live 2026-06-15). + mfe = "newbuilding-card-desktop-frontend" + + nb_state = extract_state(html, mfe=mfe, key="initialState") + if nb_state is None: + logger.warning("Cian newbuilding %s: initialState extraction failed", zhk_url) + return None + + # Primary newbuilding object: state.newbuilding (60 fields per schema sec 15.4) + nb = nb_state.get("newbuilding") or {} + if not isinstance(nb, dict): + nb = {} + + result = NewbuildingEnrichment( + cian_internal_house_id=nb.get("id"), + cian_zhk_url=zhk_url, + name=nb.get("name") or nb.get("displayName"), + address=_extract_address(nb), + year_built=nb.get("yearBuilt") or nb.get("buildYear"), + deadline_year=_extract_deadline_year(nb), + floors_count_min=nb.get("floorsCountMin"), + floors_count_max=nb.get("floorsCountMax"), + building_class=nb.get("newbuildingClassName") or nb.get("buildingClass"), + management_company=nb.get("managementCompany"), + transport_accessibility_rate=_extract_transport_rate(nb), + advantages=_extract_advantages(nb), + builders=_safe_list(nb.get("builders")), + banks=_safe_list(nb.get("banks")), + houses_by_turn=_safe_list(nb.get("housesByTurn")), + raw_state=nb_state, + ) + + # Sister containers — all live under the same MFE initialState top-level keys + # (not separate push() entries) per schema sec 15.3 + all_states = extract_all_states(html) + mfe_states = all_states.get(mfe, {}) + # initialState top-level keys contain sister data: realtyValuation, reliability, offers + top = nb_state + + # realtyValuation 7-month chart (Stage 6 main deliverable) + # Located at initialState.realtyValuation (sec 15.5) + rv_state = top.get("realtyValuation") or mfe_states.get("realtyValuation") + if rv_state and isinstance(rv_state, dict): + result.realty_valuation_chart = _extract_chart(rv_state) + result.raw_sister_states["realtyValuation"] = rv_state + + # reliability (наш.дом.рф) — initialState.reliabilityState.reliability + # ({checkStatus, details[], banner, actions}, verified live 2026-06-15). + rel_state = (top.get("reliabilityState") or {}).get("reliability") + if rel_state and isinstance(rel_state, dict): + result.reliability_checks = _extract_reliability_checks(rel_state) + result.raw_sister_states["reliability"] = rel_state + + # reviews — located at initialState or separate state + reviews_state = top.get("reviews") or mfe_states.get("reviews") + if reviews_state and isinstance(reviews_state, dict): + items = reviews_state.get("items") or reviews_state.get("data") or [] + result.reviews = items if isinstance(items, list) else [] + result.raw_sister_states["reviews"] = reviews_state + + # offers (grouped by roomType) — located at initialState.offers (sec 15.7) + offers_state = top.get("offers") or mfe_states.get("offers") or mfe_states.get("miniSerp") + if offers_state and isinstance(offers_state, dict): + result.nested_offers = _extract_nested_offers(offers_state) + result.raw_sister_states["offers"] = offers_state + + logger.info( + "Cian newbuilding %s parsed: id=%s chart=%d reliability=%d offers=%d", + zhk_url, + result.cian_internal_house_id, + len(result.realty_valuation_chart), + len(result.reliability_checks), + len(result.nested_offers), + ) + return result + + +# ---- private extractors ---- + + +def _safe_list(v: Any) -> list[dict[str, Any]]: + """Return v as list[dict] or empty list.""" + if not isinstance(v, list): + return [] + return [item for item in v if isinstance(item, dict)] + + +def _extract_address(nb: dict[str, Any]) -> str | None: + """Extract display address string from newbuilding object.""" + addr = nb.get("fullAddress") or nb.get("address") + if isinstance(addr, str): + return addr + if isinstance(addr, dict): + return addr.get("full") or addr.get("display") or addr.get("name") + return None + + +def _extract_deadline_year(nb: dict[str, Any]) -> int | None: + """Extract planned completion year from newbuilding.deadline or similar field.""" + dl = nb.get("deadline") or nb.get("buildingDeadline") + if isinstance(dl, dict): + return dl.get("year") + if isinstance(dl, int): + return dl + return None + + +def _extract_transport_rate(nb: dict[str, Any]) -> int | None: + """Extract transport accessibility rate as int|None. + + Cian payload drift (#972): `transportAccessibilityRate` is now a nested dict + ({'transportAccessibility': }) rather than a bare int. Older pages + returned a plain int. Handle both — returning a dict here would violate the + NewbuildingEnrichment.transport_accessibility_rate: int|None contract and break + the CAST(:tar AS ...) bind in save_newbuilding_enrichment (psycopg cannot adapt + a dict). Anything non-int collapses to None. + """ + raw = nb.get("transportAccessibilityRate") + if isinstance(raw, dict): + raw = raw.get("transportAccessibility") or raw.get("rate") or raw.get("value") + if isinstance(raw, bool): # bool is an int subclass — exclude explicitly + return None + if isinstance(raw, int): + return raw + if isinstance(raw, float): + return int(raw) + return None + + +def _extract_advantages(nb: dict[str, Any]) -> list[dict[str, Any]]: + """Extract typed advantages list from newbuilding.advantages.""" + raw = nb.get("advantages") or nb.get("advantagesTyped") or [] + if not isinstance(raw, list): + return [] + return [a for a in raw if isinstance(a, dict)] + + +def _extract_chart(rv_state: dict[str, Any]) -> list[dict[str, Any]]: + """Extract realtyValuation time-series chart points. + + Actual Cian shape (sec 15.5): + rv_state.data.priceDynamics.chart.data = {labels: [date...], values: [int...]} + + Falls back to alternative shapes if structure varies. + Each output point: + {month_date: str, room_count: str, prices_type: str, period: str, price_per_sqm: float|None} + + Note: The initial page state provides aggregate (all room types) chart only. + Per-room-type data requires XHR requests (not done here). + """ + points: list[dict[str, Any]] = [] + + # Primary shape: data.priceDynamics.chart.data.{labels, values} + data_block = rv_state.get("data") + if isinstance(data_block, dict): + price_dynamics = data_block.get("priceDynamics") or {} + if isinstance(price_dynamics, dict): + chart = price_dynamics.get("chart") or {} + if isinstance(chart, dict): + chart_data = chart.get("data") or {} + if isinstance(chart_data, dict): + labels = chart_data.get("labels") or [] + values = chart_data.get("values") or [] + if isinstance(labels, list) and isinstance(values, list): + for label, value in zip(labels, values, strict=False): + if label is None or value is None: + continue + points.append( + { + "month_date": str(label)[:10], # YYYY-MM-DD + "room_count": "all", # aggregate (initial state) + "prices_type": "price", # total price (not priceSqm) + "period": "halfYear", # default period shown + "price_per_sqm": None, # chart shows total price, not sqm + } + ) + # Store raw value as price_per_sqm placeholder for DB + # (actual sqm price would require area context) + # Use the value as-is; caller / downstream can enrich + points[-1]["price_per_sqm"] = ( + float(value) if isinstance(value, int | float) else None + ) + if points: + return points + + # Fallback shape A: rv_state.chart (flat list with {month/date, price, roomType}) + chart_flat = rv_state.get("chart") or rv_state.get("prices") or rv_state.get("dynamics") + if isinstance(chart_flat, list): + for entry in chart_flat: + if not isinstance(entry, dict): + continue + month = entry.get("month") or entry.get("date") or entry.get("monthDate") + price = entry.get("price") or entry.get("value") or entry.get("pricePerSqm") + if month is None or price is None: + continue + price_val: float | None = None + try: + price_val = float(price) + except (TypeError, ValueError): + pass + points.append( + { + "month_date": str(month)[:10], + "room_count": str(entry.get("roomType") or entry.get("roomCount") or "all"), + "prices_type": str( + entry.get("pricesType") or entry.get("priceType") or "price" + ), + "period": str(entry.get("period") or "halfYear"), + "price_per_sqm": price_val, + } + ) + return points + + return [] + + +def _extract_reliability_checks(rel_state: dict[str, Any]) -> list[dict[str, Any]]: + """Extract reliability check summary from наш.дом.рф state. + + Actual Cian shape (sec 15.6): + rel_state = { + checkStatus: {status: 'reliable', title: '...', date: '...'}, + details: [{title, iconType, type}, ...], + ... + } + + Returns a list with ONE entry representing the overall check (per DB schema): + [{check_status, check_name, details}] + + The 'details' array is stored as jsonb (full check list) in house_reliability_checks. + """ + check_status_block = rel_state.get("checkStatus") + if not isinstance(check_status_block, dict): + # Alternative: flat list shape + raw = rel_state.get("checks") or rel_state.get("items") or [] + if isinstance(raw, list) and raw: + return [ + { + "check_name": entry.get("name") or entry.get("title") or entry.get("checkName"), + "check_status": entry.get("status") or entry.get("checkStatus"), + "details": entry.get("details") or entry, + } + for entry in raw + if isinstance(entry, dict) + ] + return [] + + status = check_status_block.get("status") + name = check_status_block.get("title") or check_status_block.get("mobileTitle") + details = rel_state.get("details") or [] + + return [ + { + "check_name": name, + "check_status": status, + "details": details if isinstance(details, list) else [details], + } + ] + + +def _extract_nested_offers(offers_state: dict[str, Any]) -> list[dict[str, Any]]: + """Extract mini-SERP offers from offers container. + + Actual Cian shape (sec 15.7): + offers_state.data.total.fromDeveloperRooms[].{roomType, layouts[]} + + Returns flat list of layout offers with room_count injected. + """ + result: list[dict[str, Any]] = [] + + # Shape A: miniSerp / items list + if "items" in offers_state: + items = offers_state["items"] + return items if isinstance(items, list) else [] + + # Shape B: data.total.fromDeveloperRooms[].layouts[] + data = offers_state.get("data") + if isinstance(data, dict): + total = data.get("total") or {} + if isinstance(total, dict): + rooms = total.get("fromDeveloperRooms") or [] + else: + rooms = [] + else: + rooms = [] + + if not isinstance(rooms, list): + return result + + for room in rooms: + if not isinstance(room, dict): + continue + room_type = room.get("roomType") or room.get("roomTypeDisplay") + layouts = room.get("layouts") or [] + if not isinstance(layouts, list): + continue + for layout in layouts: + if isinstance(layout, dict): + result.append( + { + "room_count": room_type, + **layout, + } + ) + + return result + + +# ---- save helpers ---- + + +def save_newbuilding_enrichment( + db: Any, + house_id: int, + enrichment: NewbuildingEnrichment, +) -> None: + """Persist NewbuildingEnrichment to DB. + + Steps: + 1. UPSERT management_companies (if present) → get mc_id + 2. UPDATE houses with Cian metadata (incl. cian_zhk_url if present) + 3. INSERT INTO houses_price_dynamics (chart points, ON CONFLICT DO UPDATE) + 4. INSERT INTO house_reliability_checks (overall + details) + """ + from sqlalchemy import text + + # 1. UPSERT management_company + mc_id: int | None = None + if enrichment.management_company and isinstance(enrichment.management_company, dict): + mc = enrichment.management_company + mc_name = mc.get("name") or "Unknown" + mc_ext_id = mc.get("id") + mc_result = db.execute( + text(""" + INSERT INTO management_companies ( + name, phones, email, opening_hours, chief_name, ext_source, ext_id + ) VALUES ( + :name, + CAST(:phones AS text[]), + :email, + :oh, + :chief, + 'cian', + CAST(:ext_id AS bigint) + ) + ON CONFLICT (ext_source, name, ext_id) DO UPDATE SET + phones = EXCLUDED.phones, + email = COALESCE(EXCLUDED.email, management_companies.email) + RETURNING id + """), + { + "name": mc_name, + "phones": mc.get("phones") or [], + "email": mc.get("email"), + "oh": mc.get("openingHours"), + "chief": mc.get("chiefName"), + "ext_id": mc_ext_id, + }, + ) + row = mc_result.fetchone() + mc_id = row[0] if row else None + + # 2. UPDATE houses + db.execute( + text(""" + UPDATE houses SET + cian_internal_house_id = COALESCE( + CAST(:cihi AS bigint), cian_internal_house_id + ), + cian_zhk_url = COALESCE(:zhk_url, cian_zhk_url), + year_built = COALESCE(:yb, year_built), + management_company_id = COALESCE(CAST(:mcid AS bigint), management_company_id), + transport_accessibility_rate = COALESCE( + :tar, transport_accessibility_rate + ), + advantages = CAST(:adv AS jsonb), + banks = CAST(:bk AS jsonb), + builders = CAST(:bld AS jsonb), + houses_by_turn = CAST(:hbt AS jsonb) + WHERE id = CAST(:hid AS bigint) + """), + { + "hid": house_id, + "cihi": enrichment.cian_internal_house_id, + "zhk_url": enrichment.cian_zhk_url, + "yb": enrichment.year_built, + "mcid": mc_id, + "tar": enrichment.transport_accessibility_rate, + "adv": json.dumps(enrichment.advantages, ensure_ascii=False), + "bk": json.dumps(enrichment.banks, ensure_ascii=False), + "bld": json.dumps(enrichment.builders, ensure_ascii=False), + "hbt": json.dumps(enrichment.houses_by_turn, ensure_ascii=False), + }, + ) + + # 3. INSERT houses_price_dynamics + # UNIQUE constraint: houses_price_dynamics_dim_key + # (house_id, source, room_count, prices_type, period, month_date) — per migration 029 + chart_saved = 0 + for point in enrichment.realty_valuation_chart: + if point.get("price_per_sqm") is None: + continue + room_count = point.get("room_count") or "all" + prices_type = point.get("prices_type") or "price" + period = point.get("period") or "halfYear" + db.execute( + text(""" + INSERT INTO houses_price_dynamics ( + house_id, month_date, source, + room_count, prices_type, period, + price_per_sqm, recorded_at + ) VALUES ( + CAST(:hid AS bigint), + CAST(:md AS date), + 'cian_realty_valuation', + :rc, :pt, :pd, + CAST(:pps AS numeric), + NOW() + ) + ON CONFLICT ON CONSTRAINT houses_price_dynamics_dim_key DO UPDATE SET + price_per_sqm = EXCLUDED.price_per_sqm, + recorded_at = NOW() + """), + { + "hid": house_id, + "md": point["month_date"], + "rc": room_count, + "pt": prices_type, + "pd": period, + "pps": point["price_per_sqm"], + }, + ) + chart_saved += 1 + + # 4. INSERT house_reliability_checks (stores overall check + details array) + # Schema (025): (house_id, check_status, check_name, details jsonb, source, recorded_at) + # No UNIQUE constraint — caller should manage duplicates if needed + for check in enrichment.reliability_checks: + if not check.get("check_name") and not check.get("check_status"): + continue + db.execute( + text(""" + INSERT INTO house_reliability_checks ( + house_id, check_status, check_name, details, source, recorded_at + ) VALUES ( + CAST(:hid AS bigint), + :cs, + :cn, + CAST(:det AS jsonb), + 'cian_nashdom', + NOW() + ) + """), + { + "hid": house_id, + "cs": check.get("check_status"), + "cn": check.get("check_name"), + "det": json.dumps(check.get("details") or [], ensure_ascii=False), + }, + ) + + db.commit() + logger.info( + "Cian newbuilding saved house_id=%s (chart=%d points, reliability=%d checks, mc_id=%s)", + house_id, + chart_saved, + len(enrichment.reliability_checks), + mc_id, + ) + + +async def resolve_cian_zhk_url( + cian_internal_house_id: int, + *, + 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//`` → canonical + slug — NO LONGER EXISTS. Cian now serves HTTP **404** for ``/zhk//`` + (verified 8/8 on prod, #972), so this returns None for every real id. Use + :func:`resolve_cian_zhk_url_via_search` instead: it fetches the cat.php + newbuilding-SERP and extracts the canonical ``zhk-*.cian.ru`` slug from its + markup (the WORKING path). Kept only for backward-compat / reference; do not + wire new callers to it. + + Args: + cian_internal_house_id: Cian's numeric ЖК identifier. + session: optional shared curl_cffi AsyncSession (caller owns lifecycle). + + Returns: + Canonical ZHK URL string, or None if request failed / redirect not followed. + + Note: + This function makes a real HTTP request — do NOT call it without rate limiting. + Caller must enforce per-request sleep matching scraper_settings 'cian' delay. + """ + close_session = False + if session is None: + # Mobile proxy wiring (#806 follow-up): resolve ЖК URL через мобильный прокси. + _proxy_url = 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]=`` renders that ЖК's canonical + ``https://zhk-.cian.ru`` URL as the ```` href inside its page heading + ``

`` (the ``NewbuildingHeaderSeo`` header). We anchor on that + heading rather than taking the first ``zhk-*.cian.ru`` match in the document, because + cat.php pages also carry banner / "похожие ЖК" / recommendation blocks that can embed + OTHER ЖК slug URLs — and some of those blocks render BEFORE the title. A naive + first-match could pick a foreign ЖК and enrich the house with the wrong building's + price_dynamics / reliability (and the backfill's idempotency would then lock the + corruption in). Anchoring on the title guarantees we resolve the queried ЖК. + + Falls back to a naive first ``zhk-*.cian.ru`` match ONLY when the title anchor finds + nothing (markup drift / unexpected page shape), preserving the previous behaviour as + a last-resort safety net. + + Returns the slug URL string, or None when the SERP carries no such anchor at all + (empty / no-results page, or markup drift). + """ + anchored = _ZHK_TITLE_ANCHOR_RE.search(html) + if anchored: + return anchored.group(1) + # Fallback: no title-anchored slug found — degrade to naive first-match so a markup + # drift that moves/renames the header still resolves *something* rather than failing. + match = _ZHK_SLUG_URL_RE.search(html) + return match.group(0) if match else None + + +async def resolve_cian_zhk_url_via_search( + nb_id: int, + *, + 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]=`` and + extracts the canonical ``https://zhk-.cian.ru`` URL from its markup. This + replaces the legacy :func:`resolve_cian_zhk_url`, whose ``/zhk//`` redirect path + now 404s (verified on prod, #972). The returned slug URL is exactly what + :func:`fetch_newbuilding` parses, so the enrichment chain is + ``nb_id → cat.php SERP → zhk-slug-url → fetch_newbuilding → enrich``. + + Verified live (HTTP 200, slug extracted): + nb 48853 → https://zhk-parkovyy-kvartal-ekb-i.cian.ru (ЖК «Парковый квартал») + nb 102791 → https://zhk-izumrudnyy-bor-ekb-i.cian.ru + nb 24991 → https://zhk-baltym-park-ekb-i.cian.ru + + Args: + nb_id: Cian newbuilding id (``house_sources.ext_id`` for cian houses). + session: optional shared curl_cffi AsyncSession (caller owns lifecycle). When + None, an own session is created using the same mobile-proxy wiring as the + rest of the Cian scrapers (``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() diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/serp.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/serp.py new file mode 100644 index 00000000..59ed1c03 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/serp.py @@ -0,0 +1,1196 @@ +"""Cian.ru scraper — state-based SERP parser (137 fields per offer). + +Стратегия (Stage 3 рефактор): +- Cian React micro-frontends хранят state в window._cianConfig['frontend-serp'].push(). +- URL pattern: https://ekb.cian.ru/cat.php?deal_type=sale&offer_type=flat&engine_version=2 + ekb.cian.ru — city-specific subdomain для ЕКБ (per Schema_Cian_SERP_Inventory sec 13). + +ВАЖНО: Циан блокирует httpx/curl по TLS+fingerprint и сыплет Google reCAPTCHA на +датацентр-IP. Транспорт SERP переведён на BrowserFetcher (camoufox real-browser +fingerprint через tradein-browser сервис, #1806 / epic #883 Phase 2). Браузер +отдаёт полный SSR-HTML с Redux initialState — парсинг state не меняется. +Мобильный прокси (ha.mobileproxy.space) применяется server-side в браузер-контейнере +и ротирует IP сам, поэтому code-side _rotate_ip (changeip) больше не нужен. + +НЕ используем anchor jitter: offer.geo.coordinates.{lat,lng} — точные координаты +прямо из SERP state. Jitter запрещён per implementation plan. + +Exhaustive load (fetch_all_secondary): +- Cian SERP — регион-wide (не geo-bbox). Anchor-loop в существующем city-sweep + избыточен для Cian (все anchor'ы дают одну выдачу). Полный сбор = ОДИН проход + с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление диапазона). +""" + +from __future__ import annotations + +import asyncio +import hashlib +import inspect +import logging +import math +import re +from collections.abc import Callable +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any +from urllib.parse import urlencode + +from scraper_kit.base import BaseScraper, ScrapedLot +from scraper_kit.browser_fetcher import BrowserFetcher +from scraper_kit.cian_state_parser import extract_state +from scraper_kit.house_type_normalizer import normalize_house_type +from scraper_kit.price_brackets import get_price_seed_brackets +from scraper_kit.repair_state_normalizer import ( + infer_repair_state_from_text, + normalize_repair_state, +) + +if TYPE_CHECKING: + from scraper_kit.contracts import ScraperConfig + +logger = logging.getLogger(__name__) + +# Регион 4743 = Свердловская область (Cian internal region ID) +CIAN_EKB_REGION_ID = 4743 + +# SERP MFE name и state key (per Schema sec 1.2) +_MFE_SERP = "frontend-serp" +_STATE_KEY = "initialState" + +# ── Константы exhaustive-загрузки ────────────────────────────────────────────── + +# Верхняя граница цены при первом рекурсивном делении (нет явного hi). +# 200 млн ₽ заведомо выше ТОПа ЕКБ — дальние бакеты дадут 0 офферов. +_MAX_PRICE = 200_000_000 + +# Минимальный ценовой диапазон для рекурсии; при hi - lo < MIN_BRACKET +# прекращаем деление и пагинируем как есть (принимаем возможный «хвост»). +_MIN_BRACKET = 50_000 + +# Cian SERP показывает ~28 офферов на страницу. +_CIAN_OFFERS_PER_PAGE = 28 + +# Комнатности по умолчанию для exhaustive-загрузки. +# Коды Cian: room1=1к, room2=2к, room3=3к, room4=4к, room5=5к, room6=6+. +# TODO: студии — Cian использует отдельный тип flat_type=studio, не room-param. +# Если добавить room9 — запрос может давать 0 офферов или смешивать типы. +# До верификации кода студии — студии НЕ включаем в exhaustive-загрузку. +_DEFAULT_ROOMS_BUCKETS: list[tuple[int, ...]] = [ + (1,), + (2,), + (3,), + (4,), + (5,), + (6,), +] + + +class CianScraper(BaseScraper): + """Cian SERP scraper. Тянет SSR-HTML через BrowserFetcher (camoufox). + + Извлекает 137 полей на offer из Redux initialState через cian_state_parser. + Координаты точные — НЕТ anchor jitter (offer.geo.coordinates). + + Транспорт: одна BrowserFetcher(source="cian") сессия открывается в __aenter__ + и переиспользуется на всех page-fetch'ах (#1806). camoufox-fingerprint + мобильный + прокси server-side обходят Cian reCAPTCHA; IP ротирует сам прокси. + """ + + name = "cian" + # ekb.cian.ru — city-specific subdomain для ЕКБ (per Schema sec 13, closed Q4) + base_url = "https://ekb.cian.ru" + # Класс-дефолт; реальное значение загружается из scraper_settings при создании экземпляра. + request_delay_sec = 5.0 # консервативно: Cian менее агрессивен чем Avito, но 5s безопасно + + def __init__( + self, + config: ScraperConfig, + *, + delay_provider: Callable[[str], float] | None = None, + ) -> None: + super().__init__() + # Strangler-инжекция (#2133): конфиг и провайдер задержки приходят снаружи + # вместо прямого импорта app.core.config.settings / + # app.services.scraper_settings.get_scraper_delay. Продуктовая сторона + # передаёт RealScraperConfig + get_scraper_delay; kit не знает про app / БД. + self._config = config + if delay_provider is not None: + self.request_delay_sec = delay_provider(self.name) + self._browser: BrowserFetcher | None = None + + async def __aenter__(self) -> CianScraper: + await super().__aenter__() + # Одна BrowserFetcher(source="cian") сессия на весь scrape — camoufox real-browser + # fingerprint через tradein-browser сервис (#1806). Сервер роутит /fetch на + # отдельный браузер+мобильный прокси для cian. Прокси ротирует IP сам, поэтому + # code-side warm-up cookies и changeip-ротация больше не нужны. + self._browser = BrowserFetcher(source="cian") + await self._browser.__aenter__() + return self + + async def __aexit__(self, *args: Any) -> None: + if self._browser is not None: + await self._browser.__aexit__(*args) + self._browser = None + await super().__aexit__(*args) + + # ── Anti-block: IP rotation ─────────────────────────────────────────────── + + async def _rotate_ip(self) -> bool: + """No-op после миграции на BrowserFetcher (#1806). + + Мобильный прокси в браузер-контейнере ротирует IP сам, code-side changeip + больше не нужен. Метод сохранён (returns False) для обратной совместимости + с вызовом в _walk_price_range и тестами, которые его мокают. + """ + return False + + async def fetch_around( + self, + lat: float, + lon: float, + radius_m: int = 1000, + rooms: tuple[int, ...] | None = None, + page: int = 1, + ) -> list[ScrapedLot]: + """Найти объявления Циан по ЕКБ (lat/lon для reference; Cian не поддерживает bbox). + + Cian SERP отдаёт весь ЕКБ — фильтрация по радиусу происходит в postgres + через ST_DWithin после сохранения (точные coords из state). + + rooms — список из 1,2,3,4 (Cian's room codes). Если None — все. + page — страница выдачи (Cian ~28 объявлений на страницу). + """ + url = self._build_url(rooms, page) + try: + assert self._browser is not None + html = await self._browser.fetch(url) + except Exception: + logger.exception("cian BrowserFetcher fetch failed for url=%s", url) + return [] + + lots = self._parse_serp_html(html) + logger.info( + "cian: %d lots fetched rooms=%s page=%d url=%s", + len(lots), + rooms, + page, + url, + ) + await self.sleep_between_requests() + return lots + + async def fetch_around_multi_room( + self, + lat: float, + lon: float, + radius_m: int = 1000, + pages: int = 8, + ) -> list[ScrapedLot]: + """Скрейп Циан по 1к/2к/3к/4+ × N страниц — расширяет выборку. + + 4 комнаты × N страниц × ~28 = до ~330+ лотов до дедупликации. + """ + seen: dict[str, ScrapedLot] = {} + for rooms in ((1,), (2,), (3,), (4,)): + for page in range(1, pages + 1): + try: + lots = await self.fetch_around(lat, lon, radius_m, rooms=rooms, page=page) + except Exception: + logger.exception("cian multi-room fetch failed rooms=%s page=%d", rooms, page) + continue + if not lots: + break # пустая страница → дальше смысла нет + for lot in lots: + key = lot.source_id or lot.source_url + if key and key not in seen: + seen[key] = lot + logger.info( + "cian multi-room: %d unique lots (lat=%.4f lon=%.4f radius=%dm)", + len(seen), + lat, + lon, + radius_m, + ) + return list(seen.values()) + + def _build_url( + self, + rooms: tuple[int, ...] | None = None, + page: int = 1, + min_price: int | None = None, + max_price: int | None = None, + ) -> str: + """URL для Cian каталога вторички ЕКБ. + + Используем ekb.cian.ru (city-specific subdomain). + Регион задаётся через region= param как fallback для reliability. + min_price / max_price — опциональные ценовые границы (руб.) для price-бакетов. + """ + params: list[tuple[str, Any]] = [ + ("deal_type", "sale"), + ("engine_version", "2"), + ("offer_type", "flat"), + ("region", CIAN_EKB_REGION_ID), + ("sort", "creation_date_desc"), + ] + if rooms: + for r in rooms: + params.append((f"room{r}", "1")) + if min_price is not None: + params.append(("minprice", min_price)) + if max_price is not None: + params.append(("maxprice", max_price)) + if page > 1: + params.append(("p", page)) + return f"{self.base_url}/cat.php?{urlencode(params)}" + + def _extract_total_offers(self, html: str) -> int | None: + """Извлечь totalOffers из Redux state Cian SERP. + + Использует тот же extract_state, что и _parse_serp_html. + Возвращает None при captcha/ошибке парсинга. + """ + state = extract_state(html, mfe=_MFE_SERP, key=_STATE_KEY) + if state is None: + return None + total = state.get("results", {}).get("totalOffers") + if total is None: + return None + try: + return int(total) + except (TypeError, ValueError): + return None + + async def _fetch_page_html( + self, + rooms: tuple[int, ...] | None, + page: int, + min_price: int | None, + max_price: int | None, + ) -> str | None: + """GET одной SERP-страницы через BrowserFetcher, возвращает HTML или None при ошибке.""" + assert self._browser is not None + url = self._build_url(rooms, page, min_price, max_price) + try: + return await self._browser.fetch(url) + except Exception: + logger.exception("cian: BrowserFetcher fetch failed url=%s", url) + return None + + async def fetch_all_secondary( + self, + *, + rooms_buckets: list[tuple[int, ...]] | None = None, + price_cap_per_bucket: int = 1400, + max_pages_per_bucket: int = 54, + concurrency: int = 5, + secondary_only: bool = True, + on_bucket: Callable[..., Any] | None = None, + on_progress: Callable[[int], None] | None = None, + skip_buckets: set[str] | None = None, + ) -> list[ScrapedLot]: + """Exhaustive-загрузка Cian ЕКБ вторички через партиционирование КОМНАТНОСТЬ × ЦЕНА. + + Обходит Cian SERP-cap (~54 стр/запрос ≈ 1500 результатов на запрос). Вместо + одной корень-бисекции [0, _MAX_PRICE] на комнатность (холостые probe-спуски + через пустой верх) — стартуем с общей data-driven seed-сетки ценовых брекетов + (get_price_seed_brackets() — shared EKB grid, общая с avito/yandex). Каждый + (комнатность × seed-брекет) дальше дробится _walk_price_range если + totalOffers > price_cap_per_bucket. Последний seed-брекет ОТКРЫТ (hi=None → + maxprice не ставится) — ловит весь хвост люкса без потолка. + Страницы внутри leaf-бакета запрашиваются параллельно (asyncio.gather + Semaphore). + Дедуп по source_id (dict seen), ОБЩИЙ по всем брекетам и комнатностям. + + Параметры: + rooms_buckets: список room-кодов Cian для перебора (default: 1-6). + price_cap_per_bucket: максимум офферов в бакете перед делением (< 1500). + max_pages_per_bucket: Cian hard cap ~54; не превышать. + concurrency: максимум параллельных page-фетчей в leaf-бакете (default=5). + secondary_only: если True (default) — отбрасывает новостройки + (listing_segment=="novostroyki") после парсинга, до save/on_bucket. + object_type=1 SERP-параметр Cian ненадёжен (~5% выдачи), поэтому + фильтруем по authoritative listing_segment из offer.newbuilding.id. + on_bucket: опциональный callback(bucket_key, list[ScrapedLot]) после каждого + leaf-бакета. Может быть async или sync. Если кидает исключение — прерывает прогон. + on_progress: опциональный callback(unique_count) для heartbeat (per room-bucket). + skip_buckets: множество ключей «room_label:lo:hi» уже завершённых бакетов — + пагинация и on_bucket для них пропускаются. Probe-запросы (для split-решения) + всё равно выполняются (их мало по сравнению с пагинацией). + + Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url). + """ + _buckets = rooms_buckets if rooms_buckets is not None else _DEFAULT_ROOMS_BUCKETS + seen: dict[str, ScrapedLot] = {} + + for rooms in _buckets: + room_label = f"room{'_'.join(str(r) for r in rooms)}" + logger.info( + "cian exhaustive: starting %s (price_cap=%d concurrency=%d secondary_only=%s)", + room_label, + price_cap_per_bucket, + concurrency, + secondary_only, + ) + before = len(seen) + # Перебираем seed-брекеты: для ЗАКРЫТЫХ передаём hi = br_hi - 1 + # (maxprice эксклюзивный сверху → соседние брекеты не пересекаются), для + # ОТКРЫТОГО (br_hi is None) — hi=None (без потолка). Дедуп по source_id + # в общем `seen` страхует на стыках. + for br_lo, br_hi in get_price_seed_brackets(): + walk_hi = br_hi - 1 if br_hi is not None else None + await self._walk_price_range( + rooms=rooms, + lo=br_lo, + hi=walk_hi, + seen=seen, + price_cap_per_bucket=price_cap_per_bucket, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + ) + room_collected = len(seen) - before + logger.info( + "cian exhaustive: %s done — collected %d (total unique=%d)", + room_label, + room_collected, + len(seen), + ) + if on_progress is not None: + on_progress(len(seen)) + + logger.info("cian exhaustive: DONE — total unique=%d lots", len(seen)) + return list(seen.values()) + + async def _walk_price_range( + self, + *, + rooms: tuple[int, ...], + lo: int, + hi: int | None, + seen: dict[str, ScrapedLot], + price_cap_per_bucket: int, + max_pages_per_bucket: int, + concurrency: int = 5, + secondary_only: bool = True, + on_bucket: Callable[..., Any] | None = None, + skip_buckets: set[str] | None = None, + _depth: int = 0, + ) -> None: + """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi]. + + Алгоритм для ЗАКРЫТОГО брекета (hi is not None): + 1. Запросить page=1 с min_price=lo, max_price=hi → totalOffers. + 2. Если totalOffers <= cap → пагинировать leaf-бакет параллельно (concurrency). + 3. Если totalOffers > cap → разбить бакет пополам (рекурсия). + Guard: hi - lo < _MIN_BRACKET → пагинировать как есть (логируем WARNING). + + ОТКРЫТЫЙ брекет (hi is None) — верхний seed-брекет без потолка (maxprice не + ставится, Cian отдаёт всё ≥lo). Делить нельзя (mid=(lo+hi)//2 невозможен) — + probe + пагинируем leaf напрямую. Хвост 250М+ крошечный; за cap обычно не + выходит, если вышел — пагинируем до max_pages + WARNING (tail-loss accepted). + + После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots) если задан. + bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый) — + checkpoint-ключ для resume. + on_bucket может быть async или sync. Исключение в on_bucket прерывает прогон. + skip_buckets: если bucket_key в skip_buckets — пагинацию и on_bucket пропускаем. + secondary_only: если True — новостройки (listing_segment=="novostroyki") отбрасываются + после сбора bucket_lots, до дедупа в seen и вызова on_bucket. + """ + room_label = f"room{'_'.join(str(r) for r in rooms)}" + _lo_param = lo if lo > 0 else None + _hi_repr = "open" if hi is None else str(hi) + + # ── Шаг 1: probe page 1 ──────────────────────────────────────────────── + html = await self._fetch_page_html(rooms, 1, _lo_param, hi) + await self.sleep_between_requests() + + total: int | None = None + if html is not None: + total = self._extract_total_offers(html) + + # Ретрай на captcha/ошибку: 1 повторный fetch. Мобильный прокси в браузер- + # контейнере ротирует IP сам (#1806), поэтому отдельный changeip-вызов + # (_rotate_ip) больше не нужен — просто повторяем запрос на чистом IP. + if total is None: + logger.warning( + "cian: totalOffers=None for %s [%d, %s] depth=%d — retry (proxy auto-rotates IP)", + room_label, + lo, + _hi_repr, + _depth, + ) + await self._rotate_ip() # no-op (back-compat); прокси ротирует IP сам + html = await self._fetch_page_html(rooms, 1, _lo_param, hi) + await self.sleep_between_requests() + if html is not None: + total = self._extract_total_offers(html) + + if total is None: + logger.error( + "cian: skipping bucket %s [%d, %s] — totalOffers unavailable after retry", + room_label, + lo, + _hi_repr, + ) + return + + logger.info( + "cian: %s [%d, %s] totalOffers=%d depth=%d", + room_label, + lo, + _hi_repr, + total, + _depth, + ) + + if total == 0: + return + + # ── ОТКРЫТЫЙ брекет (hi is None): делить нельзя — пагинируем leaf напрямую ─ + # total известен из probe → max_pages = min(ceil(total/per_page), cap). + # Если total > cap (не должно для 250М+, но guard) — пагинируем как есть до + # max_pages_per_bucket с WARNING (хвост люкса крошечный, tail-loss accepted). + if hi is None: + if total > price_cap_per_bucket: + logger.warning( + "cian: OPEN bucket %s [%d, open] totalOffers=%d > cap=%d — paginating " + "without split (tail-loss accepted, lux tail tiny)", + room_label, + lo, + total, + price_cap_per_bucket, + ) + await self._paginate_open_bucket( + rooms=rooms, + room_label=room_label, + lo=lo, + html=html, + total=total, + seen=seen, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + ) + return + + # ── Шаг 2: деление или пагинация (только ЗАКРЫТЫЙ брекет, hi: int) ───── + bracket_size = hi - lo + need_split = total > price_cap_per_bucket + too_narrow = bracket_size < _MIN_BRACKET + + if need_split and too_narrow: + logger.warning( + "cian: %s [%d, %d] totalOffers=%d > cap=%d but bracket=%d < MIN_BRACKET=%d " + "— paginating as-is (tail loss ~%d)", + room_label, + lo, + hi, + total, + price_cap_per_bucket, + bracket_size, + _MIN_BRACKET, + max(0, total - price_cap_per_bucket), + ) + need_split = False # принудительно пагинируем + + if need_split: + mid = (lo + hi) // 2 + # [lo, mid] + await self._walk_price_range( + rooms=rooms, + lo=lo, + hi=mid, + seen=seen, + price_cap_per_bucket=price_cap_per_bucket, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + _depth=_depth + 1, + ) + # [mid+1, hi] + await self._walk_price_range( + rooms=rooms, + lo=mid + 1, + hi=hi, + seen=seen, + price_cap_per_bucket=price_cap_per_bucket, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + _depth=_depth + 1, + ) + return + + # Закрытый leaf-бакет — общая пагинация (см. _paginate_leaf_bucket). + await self._paginate_leaf_bucket( + rooms=rooms, + room_label=room_label, + lo=lo, + hi=hi, + html=html, + total=total, + seen=seen, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + ) + + async def _paginate_open_bucket( + self, + *, + rooms: tuple[int, ...], + room_label: str, + lo: int, + html: str | None, + total: int, + seen: dict[str, ScrapedLot], + max_pages_per_bucket: int, + concurrency: int, + secondary_only: bool, + on_bucket: Callable[..., Any] | None, + skip_buckets: set[str] | None, + ) -> None: + """Пагинация ОТКРЫТОГО leaf-бакета (hi=None, без maxprice).""" + await self._paginate_leaf_bucket( + rooms=rooms, + room_label=room_label, + lo=lo, + hi=None, + html=html, + total=total, + seen=seen, + max_pages_per_bucket=max_pages_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=on_bucket, + skip_buckets=skip_buckets, + ) + + async def _paginate_leaf_bucket( + self, + *, + rooms: tuple[int, ...], + room_label: str, + lo: int, + hi: int | None, + html: str | None, + total: int, + seen: dict[str, ScrapedLot], + max_pages_per_bucket: int, + concurrency: int, + secondary_only: bool, + on_bucket: Callable[..., Any] | None, + skip_buckets: set[str] | None, + ) -> None: + """Параллельная пагинация leaf-бакета [lo, hi] (закрытого или открытого). + + ОТКРЫТЫЙ (hi is None): maxprice не ставится (_build_url отдаёт всё ≥lo), + bucket_key = "room_label:lo:open". Страница 1 берётся из probe-`html`. + """ + _lo_param = lo if lo > 0 else None + _hi_repr = "open" if hi is None else str(hi) + bucket_key = f"{room_label}:{lo}:{_hi_repr}" + + # Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket. + if skip_buckets and bucket_key in skip_buckets: + logger.info( + "cian: skip bucket %s — already done (resume)", + bucket_key, + ) + return + + pages_needed = math.ceil(total / _CIAN_OFFERS_PER_PAGE) + max_pages = min(pages_needed, max_pages_per_bucket) + + if pages_needed > max_pages_per_bucket: + # Cian hard-cap срабатывает: часть офферов недостижима через пагинацию. + # Это происходит когда bracket < _MIN_BRACKET но totalOffers всё равно > cap. + tail_loss = total - max_pages_per_bucket * _CIAN_OFFERS_PER_PAGE + logger.warning( + "cian: %s [%d, %s] totalOffers=%d exceeds page cap " + "(max_pages=%d × %d=%d offers) — tail loss ~%d offers, " + "consider narrowing price_cap_per_bucket or _MIN_BRACKET", + room_label, + lo, + _hi_repr, + total, + max_pages_per_bucket, + _CIAN_OFFERS_PER_PAGE, + max_pages_per_bucket * _CIAN_OFFERS_PER_PAGE, + tail_loss, + ) + + # Страница 1 уже есть (html из probe выше); остальные — параллельно. + sem = asyncio.Semaphore(concurrency) + + async def _one_page(p: int) -> list[ScrapedLot]: + if p == 1: + # Используем уже полученный HTML от probe + return self._parse_serp_html(html) if html else [] + async with sem: + page_html = await self._fetch_page_html(rooms, p, _lo_param, hi) + await asyncio.sleep(self.request_delay_sec) + if page_html is None: + logger.warning( + "cian: page_html=None %s [%d, %s] page=%d — skipping page", + room_label, + lo, + _hi_repr, + p, + ) + return [] + return self._parse_serp_html(page_html) + + page_results = await asyncio.gather( + *[_one_page(p) for p in range(1, max_pages + 1)], + return_exceptions=True, + ) + + bucket_lots: list[ScrapedLot] = [] + for p_idx, res in enumerate(page_results, start=1): + if isinstance(res, BaseException): + logger.warning( + "cian: page exception %s [%d, %s] page=%d — %r", + room_label, + lo, + _hi_repr, + p_idx, + res, + ) + else: + bucket_lots.extend(res) + + collected_this_bucket = len(bucket_lots) + + # ── Фильтр новостроек (secondary_only) ──────────────────────────────── + dropped_nb = 0 + if secondary_only: + filtered = [lot for lot in bucket_lots if lot.listing_segment != "novostroyki"] + dropped_nb = collected_this_bucket - len(filtered) + bucket_lots = filtered + + # Дедуп в общий seen + for lot in bucket_lots: + key = lot.source_id or lot.source_url + if key: + seen[key] = lot + + logger.info( + "cian: %s [%d, %s] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d", + room_label, + lo, + _hi_repr, + max_pages, + collected_this_bucket, + dropped_nb, + len(seen), + ) + + # ── on_bucket callback: инкрементальный save ────────────────────────── + if on_bucket is not None and bucket_lots: + res_cb = on_bucket(bucket_key, bucket_lots) + if inspect.isawaitable(res_cb): + await res_cb + + def _parse_serp_html(self, html: str) -> list[ScrapedLot]: + """Извлечь offers из Cian Redux state. + + Использует cian_state_parser.extract_state() — общая утилита Stage 2. + Возвращает пустой список если state не найден. + """ + state = extract_state(html, mfe=_MFE_SERP, key=_STATE_KEY) + if state is None: + logger.warning( + "cian SERP state extraction failed (mfe=%s key=%s) — " + "возможно Cian изменил структуру или вернул captcha", + _MFE_SERP, + _STATE_KEY, + ) + return [] + + offers_data: list[dict[str, Any]] = state.get("results", {}).get("offers", []) + if not offers_data: + logger.warning( + "cian state found but results.offers пуст (totalOffers=%s)", + state.get("results", {}).get("totalOffers", "?"), + ) + return [] + + logger.info( + "cian SERP state ok: %d offers (totalOffers=%s)", + len(offers_data), + state.get("results", {}).get("totalOffers", "?"), + ) + + lots: list[ScrapedLot] = [] + for offer in offers_data: + lot = self._offer_to_lot(offer) + if lot is not None: + lots.append(lot) + + raw_count = len(offers_data) + saved_count = len(lots) + # Охрана от silent-failure: если получили offers, но ни один не прошёл парсинг + # — likely schema regression (как descriptionMinhash str→list[int] 2026-05-23). + if raw_count > 0 and saved_count == 0: + logger.error( + "cian SERP: 0/%d offers прошли _offer_to_lot — возможна schema regression", + raw_count, + ) + try: + if self._config.glitchtip_dsn: + # Ленивый импорт: sentry_sdk — app-side error-reporting, не dep + # scraper_kit. Только когда glitchtip настроен И случилась + # schema-regression. ImportError глотается общим except ниже. + import sentry_sdk + + sentry_sdk.capture_message( + f"cian SERP: {raw_count}/{raw_count} offers failed _offer_to_lot" + " — possible schema regression", + level="error", + ) + except Exception: + pass # sentry_sdk not installed/initialised in dev + + return lots + + def _offer_to_lot(self, offer: dict[str, Any]) -> ScrapedLot | None: + """Парсинг одного Cian offer (137 fields) → ScrapedLot. + + Маппинг полей per Schema_Cian_SERP_Inventory sec 3, sec 8, sec 11. + Без anchor jitter: координаты из offer.geo.coordinates (точные). + """ + try: + # ── Identity ───────────────────────────────────────────────────── + offer_id = offer.get("cianId") or offer.get("id") + if not offer_id: + logger.debug("cian offer пропущен: нет cianId/id") + return None + + source_id = str(offer_id) + url = offer.get("fullUrl") + if not url: + url = f"{self.base_url}/sale/flat/{offer_id}/" + + # ── Цена ───────────────────────────────────────────────────────── + bargain: dict[str, Any] = offer.get("bargainTerms") or {} + price = bargain.get("priceRur") or bargain.get("price") + if not price: + logger.debug("cian offer %s пропущен: нет цены", offer_id) + return None + try: + price_rub = int(price) + except (TypeError, ValueError): + logger.debug("cian offer %s: не удалось распарсить цену %r", offer_id, price) + return None + if price_rub <= 0: + return None + + # ── Параметры квартиры ─────────────────────────────────────────── + rooms: int | None = offer.get("roomsCount") + # Студия: flatType == 'studio' → 0 комнат + if offer.get("flatType") == "studio" and rooms is None: + rooms = 0 + + area_raw = offer.get("totalArea") + area_m2: float | None = None + if area_raw is not None: + try: + area_m2 = float(area_raw) + except (TypeError, ValueError): + pass + + living_area_raw = offer.get("livingArea") + living_area_m2: float | None = None + if living_area_raw is not None: + try: + living_area_m2 = float(living_area_raw) + except (TypeError, ValueError): + pass + + # kitchen_area: сырое (str) в raw_payload + промоутим float в ScrapedLot + # колонку kitchen_area_m2 (#2008, Class B — дашборд/matching, не valuation). + kitchen_area_raw = offer.get("kitchenArea") + kitchen_area_m2: float | None = None + if kitchen_area_raw is not None: + try: + kitchen_area_m2 = float(kitchen_area_raw) + except (TypeError, ValueError): + pass + + floor: int | None = offer.get("floorNumber") + + # ── Здание ─────────────────────────────────────────────────────── + building: dict[str, Any] = offer.get("building") or {} + total_floors: int | None = building.get("floorsCount") + year_built: int | None = building.get("buildYear") + # house_type: cian materialType (camelCase: monolithBrick / gasSilicateBlock + # / stalin / ...) НЕ равен каноничному enum → estimator soft-penalty ложно + # штрафовал аналог. Нормализуем на ингесте (#2008, Class A); сырьё — в + # raw_payload['raw_material_type'] для аудита. + raw_material_type: str | None = building.get("materialType") + house_type: str | None = normalize_house_type(raw_material_type) + + # Newbuilding deadline year — если нет buildYear + if year_built is None: + deadline = building.get("deadline") or {} + year_built = deadline.get("year") + + # ── Кадастр ───────────────────────────────────────────────────── + # cadastralNumber — кадастр КВАРТИРЫ + cadastral_number: str | None = offer.get("cadastralNumber") + # buildingCadastralNumber — кадастр ДОМА + building_cadastral_number: str | None = offer.get("buildingCadastralNumber") + + # ── Геолокация (точные координаты — NO jitter) ─────────────────── + geo: dict[str, Any] = offer.get("geo") or {} + coords: dict[str, Any] = geo.get("coordinates") or {} + lat: float | None = coords.get("lat") + lon: float | None = coords.get("lng") + # Если нет coords → сохраняем без lat/lon (geocode-missing обработает позже) + if lat is None or lon is None: + logger.debug( + "cian offer %s: нет точных координат в state (geo=%s)", + offer_id, + coords, + ) + lat = lon = None + + # ── Адрес из geo.address[] ─────────────────────────────────────── + # newbuilding нужен ДО _format_address: для ЖК geo.address[] часто без + # части {type:"house"} → дом-номер восстанавливаем из building/кадастра/ + # названия ЖК (#1773), иначе все корпуса схлопываются в house-less + # catch-all и same-building anchor рвётся. + address_parts = geo.get("address") or [] + newbuilding: dict[str, Any] = offer.get("newbuilding") or {} + nb_id = newbuilding.get("id") + is_newbuilding = bool(nb_id) and int(nb_id) > 0 + address = _format_address( + address_parts, + building=building, + building_cadastral_number=building_cadastral_number, + newbuilding=newbuilding, + is_newbuilding=is_newbuilding, + ) + + # ── Metro stations ─────────────────────────────────────────────── + undergrounds: list[dict[str, Any]] = geo.get("undergrounds") or [] + metro_stations = [ + { + "name": u.get("name"), + "time": u.get("time"), + "mode": u.get("transportType"), + "line_color": u.get("lineColor"), + "line_id": u.get("lineId"), + "is_default": u.get("isDefault", False), + } + for u in undergrounds + if u.get("name") + ] + + # ── Newbuilding link ───────────────────────────────────────────── + # newbuilding / nb_id / is_newbuilding уже вычислены выше (до адреса). + # id == 0 — sentinel для вторички (per Schema sec 20.5) + house_source: str | None = None + house_ext_id: str | None = None + listing_segment: str | None = None + + if is_newbuilding: + # Предпочитаем кадастр ДОМА как same-building anchor (#1773): nb_id + # группирует по ЖК целиком, а один ЖК — это несколько корпусов + # (6679/11110/7019). buildingCadastralNumber стабильно идентифицирует + # ФИЗИЧЕСКИЙ дом → Tier 0 cadastr_exact в match_or_create_house + # схлопывает раздробленные корпуса. nb_id оставляем fallback'ом. + if building_cadastral_number: + house_source = "cian_building_cad" + house_ext_id = building_cadastral_number + else: + house_source = "cian_newbuilding" + house_ext_id = str(nb_id) + listing_segment = "novostroyki" + else: + house_source = "cian" + listing_segment = "vtorichka" + + # ── Дополнительные характеристики ──────────────────────────────── + balconies_count: int | None = offer.get("balconiesCount") + loggias_count: int | None = offer.get("loggiasCount") + bedrooms_count: int | None = offer.get("bedroomsCount") + + has_balcony: bool | None = None + if balconies_count is not None: + has_balcony = balconies_count > 0 + elif loggias_count is not None: + has_balcony = loggias_count > 0 + + # Мебель и repair_state + # hasFurniture (мебель вообще) и isSoldFurnished (продаётся с мебелью) — + # разная семантика, нельзя OR-склеивать: явный False у hasFurniture + # затирался бы коалесингом. Берём hasFurniture; только если он None, + # падаем на isSoldFurnished как fallback (#1388). + _has_furniture = offer.get("hasFurniture") + has_furniture: bool | None = ( + _has_furniture if _has_furniture is not None else offer.get("isSoldFurnished") + ) + # decoration — для новостроек (отделка); repair_state — для вторички + # Cian SERP: offer.decoration содержит raw значения (without/cosmetic/euro/design/...) + # Нормализуем до enum needs_repair/standard/good/excellent при инgesте. + repair_state: str | None = normalize_repair_state(offer.get("decoration")) + + # ── Продавец ───────────────────────────────────────────────────── + phones: list[dict[str, Any]] = offer.get("phones") or [] + is_homeowner: bool | None = offer.get("isByHomeowner") + is_pro_seller: bool | None = offer.get("isPro") + + # ── Сделка ─────────────────────────────────────────────────────── + sale_type: str | None = bargain.get("saleType") + bargain_allowed: bool | None = bargain.get("bargainAllowed") + + # ── Class B promote (#2008): ипотека / апартаменты / проверка ЕГРН ── + # Раньше лежали только в raw_payload; промоутим в колонки listings для + # дашборда покрытия и matching (НЕ для valuation — estimator их не читает). + mortgage_available: bool | None = bargain.get("mortgageAllowed") + is_apartments: bool | None = offer.get("isApartments") + is_rosreestr_checked: bool | None = offer.get("isRosreestrChecked") + + # ── Description + minhash ───────────────────────────────────────── + description: str | None = offer.get("description") + # Cian предоставляет готовый minhash для dedup и cross-source matching. + # С ~2026-05-23 Cian изменил тип descriptionMinhash: str → list[int]. + # Нормализуем оба варианта; НЕ sorted() — порядок band-ов значим для LSH. + _raw_minhash = offer.get("descriptionMinhash") + if isinstance(_raw_minhash, list): + description_minhash: str | None = ",".join(str(x) for x in _raw_minhash) or None + elif isinstance(_raw_minhash, str): + description_minhash = _raw_minhash or None + else: + description_minhash = None + # Если Cian не дал minhash — вычисляем простой SHA1 из description + if not description_minhash and description: + description_minhash = hashlib.sha1( + description.lower().encode("utf-8", errors="replace") + ).hexdigest()[:32] + + # Fallback: repair_state из описания, если decoration отсутствует (#622) + if repair_state is None and description: + repair_state = infer_repair_state_from_text(description) + + # ── Фото ───────────────────────────────────────────────────────── + photo_urls: list[str] = [ + p["fullUrl"] + for p in (offer.get("photos") or []) + if isinstance(p, dict) and p.get("fullUrl") + ] + + # ── Дата публикации ─────────────────────────────────────────────── + added_ts = offer.get("addedTimestamp") + listing_date = None + if added_ts: + try: + listing_date = datetime.fromtimestamp(int(added_ts), tz=UTC).date() + except (TypeError, ValueError, OSError): + pass + + # ── Raw payload (полный offer для enrichment в будущем) ─────────── + raw_payload: dict[str, Any] = { + "cian_id": offer_id, + "flat_type": offer.get("flatType"), + "is_apartments": offer.get("isApartments"), + "offer_type": offer.get("offerType"), + "category": offer.get("category"), + "has_furniture": has_furniture, + # сырой kitchenArea (str) — колонка kitchen_area_m2 (float) промоучена выше + "kitchen_area_m2": kitchen_area_raw, + # сырой materialType (camelCase) — house_type нормализован выше (#2008) + "raw_material_type": raw_material_type, + "mortgage_allowed": bargain.get("mortgageAllowed"), + "sale_type": sale_type, + "newbuilding_name": newbuilding.get("name"), + "is_from_developer": ( + offer.get("fromDeveloper") or newbuilding.get("isFromDeveloper") + ), + "builders_ids": offer.get("buildersIds"), + "is_rosreestr_checked": offer.get("isRosreestrChecked"), + "is_layout_approved": offer.get("isLayoutApproved"), + "user_id": offer.get("userId"), + "published_user_id": offer.get("publishedUserId"), + "is_by_commercial_owner": offer.get("isByCommercialOwner"), + "is_cian_partner": offer.get("isCianPartner"), + "description_words_highlighted": offer.get("descriptionWordsHighlighted"), + "building_parking": building.get("parking"), + "building_total_area": building.get("totalArea"), + } + + return ScrapedLot( + source="cian", + source_url=url, + source_id=source_id, + address=address, + lat=lat, + lon=lon, + rooms=rooms, + area_m2=area_m2, + floor=floor, + total_floors=total_floors, + year_built=year_built, + house_type=house_type, + repair_state=repair_state, + has_balcony=has_balcony, + kitchen_area_m2=kitchen_area_m2, + mortgage_available=mortgage_available, + is_apartments=is_apartments, + is_rosreestr_checked=is_rosreestr_checked, + kadastr_num=cadastral_number, + house_source=house_source, + house_ext_id=house_ext_id, + listing_segment=listing_segment, + price_rub=price_rub, + price_per_m2=None, # compute_price_per_m2() вычислит + listing_date=listing_date, + photo_urls=photo_urls, + raw_payload=raw_payload, + # Cian-specific (Stage 2 fields) + living_area_m2=living_area_m2, + bedrooms_count=bedrooms_count, + balconies_count=balconies_count, + loggias_count=loggias_count, + description_minhash=description_minhash, + cadastral_number=cadastral_number, + building_cadastral_number=building_cadastral_number, + phones=phones, + is_homeowner=is_homeowner, + is_pro_seller=is_pro_seller, + bargain_allowed=bargain_allowed, + sale_type=sale_type, + metro_stations=metro_stations, + ) + + except Exception: + _oid = offer.get("cianId") or offer.get("id") + logger.exception("cian _offer_to_lot failed for offer_id=%s", _oid) + return None + + +# ── Address formatter ──────────────────────────────────────────────────────── + +# Дом-токен в хвосте строки: «29», «29А», «12 к1», «5/2», «10 стр.3», «1С7». +# Якорим в конец строки (\Z) — берём именно завершающий номер, не цифры из +# названия ЖК («Суходольский квартал»). Не матчим «р-н», «мкр» и пр. слова. +_HOUSE_TOKEN_RE = re.compile( + r"(\d+[А-Яа-яA-Za-z]?" # базовый номер: 29 / 29А / 1С + r"(?:\s*(?:к|корп\.?|стр\.?|с)\s*\d+[А-Яа-яA-Za-z]?)?" # корпус/строение + r"(?:\s*/\s*\d+[А-Яа-яA-Za-z]?)?)" # дробь 5/2 + r"\s*\Z" +) + + +def _recover_newbuilding_house( + *, + building: dict[str, Any] | None, + building_cadastral_number: str | None, + newbuilding: dict[str, Any] | None, +) -> str | None: + """Восстановить дом-токен для новостройки, когда geo.address[] без {type:"house"}. + + Приоритет (#1773): + (a) building.houseNumber / building.address — если SERP-state их отдал; + (b) building_cadastral_number — стабильный same-building anchor: если ни + номер, ни название не дали дом, добавляем «кад. 66:41:...:11396» как + ключ-якорь, чтобы раздробленные корпуса схлопывались по кадастру; + (c) хвостовой дом-токен из newbuilding.name («… 29» / «… 29А к1»). + + НЕ выдумываем catch-all номер: если ничего из (a)/(b)/(c) не нашли — None. + Кадастр (b) предпочтительнее как house anchor (см. _offer_to_lot). + """ + building = building or {} + newbuilding = newbuilding or {} + + # (a) building.houseNumber / building.address — прямые поля SERP-state + house_number = building.get("houseNumber") + if house_number is not None: + token = str(house_number).strip() + if token: + return token + b_addr = (building.get("address") or "").strip() + if b_addr: + m = _HOUSE_TOKEN_RE.search(b_addr) + if m: + return m.group(1).strip() + + # (c) хвостовой дом-токен из названия ЖК (до кадастра — он читабельнее) + nb_name = (newbuilding.get("name") or "").strip() + if nb_name: + m = _HOUSE_TOKEN_RE.search(nb_name) + if m: + return m.group(1).strip() + + # (b) кадастр дома как anchor-ключ (последний приоритет в строке адреса, + # но самый надёжный для матчинга — Tier 0 cadastr_exact) + if building_cadastral_number: + cad = building_cadastral_number.strip() + if cad: + return f"кад. {cad}" + + return None + + +def _format_address( + address_parts: list[dict[str, Any]], + *, + building: dict[str, Any] | None = None, + building_cadastral_number: str | None = None, + newbuilding: dict[str, Any] | None = None, + is_newbuilding: bool = False, +) -> str: + """Сформировать читаемый адрес из geo.address[] Cian. + + Пропускаем location (страна/регион) и metro — берём раион/улицу/дом. + Пример: [location=Москва, raion=Пресненский, street=..., house=1С7, metro=Москва-Сити] + → 'Пресненский, улица ..., 1С7' + + #1773: для новостроек (is_newbuilding) geo.address[] часто без {type:"house"}. + Если части house нет — восстанавливаем номер дома через _recover_newbuilding_house + (building → название ЖК → кадастр) и дописываем его в конец строки, чтобы + same-building anchor (fingerprint/кадастр) не рвался по house-less улице. + """ + if not address_parts: + return "Екатеринбург (Cian)" + + skip_types = {"location", "metro"} + parts: list[str] = [] + has_house = False + for part in address_parts: + if not isinstance(part, dict): + continue + ptype = part.get("type", "") + if ptype == "house": + has_house = True + if ptype in skip_types: + continue + name = (part.get("fullName") or part.get("name") or "").strip() + if name: + parts.append(name) + + # Восстановление дома только для новостроек без части house — вторичка с + # house-less адресом не трогается (не выдумываем номер). + if is_newbuilding and not has_house: + recovered = _recover_newbuilding_house( + building=building, + building_cadastral_number=building_cadastral_number, + newbuilding=newbuilding, + ) + if recovered: + parts.append(recovered) + + return ", ".join(parts) if parts else "Екатеринбург (Cian)" diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/session.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/session.py new file mode 100644 index 00000000..ebdb1609 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/session.py @@ -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) diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/valuation.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/valuation.py new file mode 100644 index 00000000..11a18eea --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/valuation.py @@ -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.3–24.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.4–24.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