From 66433dfe1b46ddff59dfe10d7b055ed8f462852c Mon Sep 17 00:00:00 2001 From: lekss361 Date: Tue, 16 Jun 2026 19:47:16 +0000 Subject: [PATCH] feat(cian): collect price history via BrowserFetcher (fixes silent no-op backfill) (#1657) Co-authored-by: lekss361 Co-committed-by: lekss361 --- .../app/services/scrapers/cian_detail.py | 188 +++++++++++------- .../app/tasks/cian_history_backfill.py | 97 +++++---- tradein-mvp/backend/tests/test_cian_detail.py | 173 +++++++++++++++- 3 files changed, 336 insertions(+), 122 deletions(-) diff --git a/tradein-mvp/backend/app/services/scrapers/cian_detail.py b/tradein-mvp/backend/app/services/scrapers/cian_detail.py index 4ccfdcaf..55b54092 100644 --- a/tradein-mvp/backend/app/services/scrapers/cian_detail.py +++ b/tradein-mvp/backend/app/services/scrapers/cian_detail.py @@ -15,7 +15,7 @@ from __future__ import annotations import json import logging from dataclasses import dataclass, field -from typing import Any +from typing import TYPE_CHECKING, Any from curl_cffi.requests import AsyncSession from sqlalchemy import text @@ -29,6 +29,9 @@ from app.services.scrapers.repair_state_normalizer import ( ) from app.services.scrapers.snapshot_writer import upsert_listing_snapshot +if TYPE_CHECKING: + from app.services.scrapers.browser_fetcher import BrowserFetcher + logger = logging.getLogger(__name__) @@ -69,96 +72,113 @@ async def fetch_detail( offer_url: str, *, 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/' - session: optional shared curl_cffi session (для batched scrapes) + 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. """ - close_session = False - if session is None: - # proxies: mobile-proxy egress (#806) — без прокси datacenter-IP блокируется Cian. - # Пусто (env не задан) → прямое подключение (dev/no-op). - _proxy_url = settings.cian_proxy_url - _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) + 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 - html = resp.text + else: + # curl_cffi path (legacy, back-compat). + close_session = False + if session is None: + # proxies: mobile-proxy egress (#806) — без прокси datacenter-IP блокируется Cian. + # Пусто (env не задан) → прямое подключение (dev/no-op). + _proxy_url = settings.cian_proxy_url + _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 - # 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 + 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() - offer_data = offer_state.get("offerData", {}) - offer = offer_data.get("offer", {}) + # ── Shared parse path (identical regardless of HTML source) ────────────── - 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=_parse_float(offer.get("ceilingHeight")), - repair_type=_extract_repair_type(offer), - repair_state=_extract_repair_state(offer), - kitchen_area_m2=_parse_float(offer.get("kitchenArea")), - raw_offer=offer, - ) + # 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 - # 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")) + offer_data = offer_state.get("offerData", {}) + offer = offer_data.get("offer", {}) - # Price changes — from offer.priceChanges or state-level priceChanges - result.price_changes = _extract_price_changes(offer, offer_state) + 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=_parse_float(offer.get("ceilingHeight")), + repair_type=_extract_repair_type(offer), + repair_state=_extract_repair_state(offer), + kitchen_area_m2=_parse_float(offer.get("kitchenArea")), + raw_offer=offer, + ) - # Agent profile expansion - result.agent_profile = _extract_agent_profile(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")) - # 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", {}) + # 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) - bti_state = offer_card_states.get("bti") - if bti_state: - result.bti_data = bti_state.get("houseData") - result.raw_sister_states["bti"] = bti_state + # Agent profile expansion + result.agent_profile = _extract_agent_profile(offer) - # Newbuilding link (if applicable) - newbuilding = offer.get("newbuilding") - if newbuilding and newbuilding.get("id"): - result.newbuilding_data = newbuilding + # 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", {}) - # 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 + bti_state = offer_card_states.get("bti") + if bti_state: + result.bti_data = bti_state.get("houseData") + result.raw_sister_states["bti"] = bti_state - return result + # Newbuilding link (if applicable) + newbuilding = offer.get("newbuilding") + if newbuilding and newbuilding.get("id"): + result.newbuilding_data = newbuilding - finally: - if close_session: - await session.close() + # 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 ────────────────────────────────────────────────────────── @@ -204,19 +224,39 @@ def _parse_views(formatted_str: str | None) -> int | None: return None -def _extract_price_changes(offer: dict[str, Any], state: dict[str, Any]) -> list[dict[str, Any]]: - """offer.priceChanges or state.priceChanges — normalize to common form.""" - raw = offer.get("priceChanges") or state.get("priceChanges") or [] +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": entry.get("price") or entry.get("priceRub"), + "price_rub": price_rub, "diff_percent": entry.get("diffPercent") or entry.get("diff_percent"), } ) diff --git a/tradein-mvp/backend/app/tasks/cian_history_backfill.py b/tradein-mvp/backend/app/tasks/cian_history_backfill.py index abbc03c0..41d7a508 100644 --- a/tradein-mvp/backend/app/tasks/cian_history_backfill.py +++ b/tradein-mvp/backend/app/tasks/cian_history_backfill.py @@ -8,7 +8,9 @@ 1. SELECT listings WHERE source='cian' AND source_url IS NOT NULL AND NOT EXISTS (SELECT 1 FROM offer_price_history WHERE listing_id=...) LIMIT batch_size - → fetch_detail + save_detail_enrichment per listing + → fetch_detail(browser_fetcher=bf) + save_detail_enrichment per listing + Использует BrowserFetcher (camoufox) для получения JS-rendered HTML, потому что + curl_cffi не рендерит priceChanges (они hidden в raw HTML). 2. Houses block: SELECT houses WHERE cian_zhk_url IS NOT NULL AND NOT EXISTS (SELECT 1 FROM houses_price_dynamics WHERE house_id=...) LIMIT batch_size @@ -17,6 +19,7 @@ Rate limit: scraper_settings.get_scraper_delay('cian') between requests. """ + from __future__ import annotations import asyncio @@ -28,6 +31,7 @@ from sqlalchemy import text from sqlalchemy.orm import Session from app.services.scraper_settings import get_scraper_delay +from app.services.scrapers.browser_fetcher import BrowserFetcher from app.services.scrapers.cian_detail import fetch_detail, save_detail_enrichment from app.services.scrapers.cian_valuation import estimate_via_cian_valuation @@ -107,46 +111,49 @@ async def backfill_cian_history( if dry_run: logger.info("dry_run: would process %d cian listings", result.listings_total) else: - for row in rows: - listing_id: int = row["id"] - source_url: str = row["source_url"] - result.listings_processed += 1 + # One BrowserFetcher instance shared across all listings in this batch. + # priceChanges requires JS rendering — curl_cffi returns empty list (#1574). + async with BrowserFetcher() as bf: + for row in rows: + listing_id: int = row["id"] + source_url: str = row["source_url"] + result.listings_processed += 1 + + enrichment = None + try: + enrichment = await fetch_detail(source_url, browser_fetcher=bf) + except Exception as exc: + logger.warning( + "cian_detail fetch failed for listing_id=%s url=%s: %s", + listing_id, + source_url, + exc, + ) + result.listings_failed_fetch += 1 + await asyncio.sleep(delay) + continue + + if enrichment is None: + logger.warning( + "cian_detail fetch returned None for listing_id=%s url=%s", + listing_id, + source_url, + ) + result.listings_failed_fetch += 1 + await asyncio.sleep(delay) + continue + + try: + save_detail_enrichment(db, listing_id, enrichment) + result.listings_succeeded += 1 + result.price_changes_attempted += len(enrichment.price_changes or []) + except Exception as exc: + logger.warning( + "cian_detail save failed for listing_id=%s: %s", listing_id, exc + ) + result.listings_failed_save += 1 - enrichment = None - try: - enrichment = await fetch_detail(source_url) - except Exception as exc: - logger.warning( - "cian_detail fetch failed for listing_id=%s url=%s: %s", - listing_id, - source_url, - exc, - ) - result.listings_failed_fetch += 1 await asyncio.sleep(delay) - continue - - if enrichment is None: - logger.warning( - "cian_detail fetch returned None for listing_id=%s url=%s", - listing_id, - source_url, - ) - result.listings_failed_fetch += 1 - await asyncio.sleep(delay) - continue - - try: - save_detail_enrichment(db, listing_id, enrichment) - result.listings_succeeded += 1 - result.price_changes_attempted += len(enrichment.price_changes or []) - except Exception as exc: - logger.warning( - "cian_detail save failed for listing_id=%s: %s", listing_id, exc - ) - result.listings_failed_save += 1 - - await asyncio.sleep(delay) # ── 2. Houses: missing houses_price_dynamics ────────────────────────────── if do_houses: @@ -230,8 +237,9 @@ async def backfill_cian_history( # ── 3. Cian listings без external_valuations (price prediction backfill) ── if do_valuations: - rows = db.execute( - text(""" + rows = ( + db.execute( + text(""" SELECT l.id, l.address, l.area_m2, l.rooms, l.floor, l.total_floors FROM listings l WHERE l.source = 'cian' @@ -248,8 +256,11 @@ async def backfill_cian_history( ) LIMIT :lim """), - {"lim": batch_size}, - ).mappings().all() + {"lim": batch_size}, + ) + .mappings() + .all() + ) result.valuations_total = len(rows) if dry_run: diff --git a/tradein-mvp/backend/tests/test_cian_detail.py b/tradein-mvp/backend/tests/test_cian_detail.py index da8d5859..b303373f 100644 --- a/tradein-mvp/backend/tests/test_cian_detail.py +++ b/tradein-mvp/backend/tests/test_cian_detail.py @@ -70,13 +70,14 @@ def test_repair_state_none_when_both_missing(): def test_extract_price_changes_normalizes(): + """curl_cffi flat format: {changeTime, price, diffPercent}.""" offer = { "priceChanges": [ {"changeTime": "2026-04-01T00:00:00Z", "price": 5000000, "diffPercent": -2.5}, {"changeTime": "2026-04-10T00:00:00Z", "price": 4900000, "diffPercent": -2.0}, ] } - result = _extract_price_changes(offer, {}) + result = _extract_price_changes(offer, {}, {}) assert len(result) == 2 assert result[0]["change_time"] == "2026-04-01T00:00:00Z" assert result[0]["price_rub"] == 5000000 @@ -84,11 +85,11 @@ def test_extract_price_changes_normalizes(): def test_extract_price_changes_handles_empty_offer(): - assert _extract_price_changes({}, {}) == [] + assert _extract_price_changes({}, {}, {}) == [] def test_extract_price_changes_handles_none_value(): - assert _extract_price_changes({"priceChanges": None}, {}) == [] + assert _extract_price_changes({"priceChanges": None}, {}, {}) == [] def test_extract_price_changes_falls_back_to_state(): @@ -97,17 +98,55 @@ def test_extract_price_changes_falls_back_to_state(): {"changeTime": "2026-03-01T00:00:00Z", "price": 6000000, "diffPercent": 0}, ] } - result = _extract_price_changes({}, state) + result = _extract_price_changes({}, {}, state) assert len(result) == 1 assert result[0]["price_rub"] == 6000000 def test_extract_price_changes_skips_non_dict_entries(): offer = {"priceChanges": [{"changeTime": "2026-04-01T00:00:00Z", "price": 5000000}, "bad"]} - result = _extract_price_changes(offer, {}) + result = _extract_price_changes(offer, {}, {}) assert len(result) == 1 +def test_extract_price_changes_offer_data_browser_format(): + """Browser-rendered HTML: offerData.priceChanges with nested priceData.price.""" + offer_data = { + "priceChanges": [ + { + "changeTime": "2026-06-05T15:32:41Z", + "priceData": {"currency": "rur", "price": 13300000}, + }, + { + "changeTime": "2026-06-03T08:05:04Z", + "priceData": {"currency": "rur", "price": 13600000}, + }, + ] + } + result = _extract_price_changes({}, offer_data, {}) + assert len(result) == 2 + assert result[0]["change_time"] == "2026-06-05T15:32:41Z" + assert result[0]["price_rub"] == 13300000 + assert result[1]["price_rub"] == 13600000 + + +def test_extract_price_changes_offer_wins_over_offer_data(): + """offer.priceChanges takes priority over offerData.priceChanges (flat format wins).""" + offer = { + "priceChanges": [ + {"changeTime": "2026-04-01T00:00:00Z", "price": 5000000}, + ] + } + offer_data = { + "priceChanges": [ + {"changeTime": "2026-06-01T00:00:00Z", "priceData": {"price": 9000000}}, + ] + } + result = _extract_price_changes(offer, offer_data, {}) + assert len(result) == 1 + assert result[0]["price_rub"] == 5000000 + + # ── _extract_agent_profile ──────────────────────────────────────────────────── @@ -463,3 +502,127 @@ def test_save_detail_enrichment_kitchen_area_none_passthrough(): update_call_params = db.execute.call_args_list[0][0][1] assert update_call_params["ka"] is None + + +# ── fetch_detail browser_fetcher param ─────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_fetch_detail_browser_fetcher_uses_browser_html(monkeypatch): + """browser_fetcher provided → HTML comes from BrowserFetcher.fetch, not curl_cffi. + + Simulates the browser-rendered format where priceChanges is at offerData level + with nested priceData.price (verified on prod 2026-06-16). + """ + fake_state = { + "offerData": { + "offer": { + "cianId": 328647045, + }, + # Browser-rendered: priceChanges at offerData level, priceData.price format + "priceChanges": [ + { + "changeTime": "2026-06-05T15:32:41Z", + "priceData": {"currency": "rur", "price": 13300000}, + }, + { + "changeTime": "2026-06-03T08:05:04Z", + "priceData": {"currency": "rur", "price": 13600000}, + }, + ], + } + } + + monkeypatch.setattr( + "app.services.scrapers.cian_detail.extract_state", + lambda html, mfe, key: fake_state if key == "defaultState" else None, + ) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.extract_all_states", + lambda html: {}, + ) + + # Mock BrowserFetcher: fetch() returns HTML tagged so we can verify it was called. + mock_bf = MagicMock() + mock_bf.fetch = AsyncMock(return_value="browser-rendered") + + result = await fetch_detail( + "https://ekb.cian.ru/sale/flat/328647045/", + browser_fetcher=mock_bf, + ) + + assert result is not None + assert result.cian_id == 328647045 + # price_changes populated from browser-rendered HTML via offerData.priceChanges + assert len(result.price_changes) == 2 + assert result.price_changes[0]["price_rub"] == 13300000 + assert result.price_changes[0]["change_time"] == "2026-06-05T15:32:41Z" + assert result.price_changes[1]["price_rub"] == 13600000 + # BrowserFetcher.fetch was called with the offer URL + mock_bf.fetch.assert_called_once_with("https://ekb.cian.ru/sale/flat/328647045/") + + +@pytest.mark.asyncio +async def test_fetch_detail_browser_fetcher_no_curl_session_used(monkeypatch): + """When browser_fetcher is given, curl_cffi AsyncSession is NOT instantiated.""" + fake_state = {"offerData": {"offer": {"cianId": 555555}}} + + monkeypatch.setattr( + "app.services.scrapers.cian_detail.extract_state", + lambda html, mfe, key: fake_state if key == "defaultState" else None, + ) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.extract_all_states", + lambda html: {}, + ) + + # Patch AsyncSession so we can assert it was NOT called. + mock_session_cls = MagicMock() + monkeypatch.setattr( + "app.services.scrapers.cian_detail.AsyncSession", + mock_session_cls, + ) + + mock_bf = MagicMock() + mock_bf.fetch = AsyncMock(return_value="") + + result = await fetch_detail( + "https://ekb.cian.ru/sale/flat/555555/", + browser_fetcher=mock_bf, + ) + + assert result is not None + mock_session_cls.assert_not_called() + + +@pytest.mark.asyncio +async def test_fetch_detail_browser_fetcher_returns_none_on_exception(monkeypatch): + """If BrowserFetcher.fetch raises, fetch_detail returns None (no crash).""" + mock_bf = MagicMock() + mock_bf.fetch = AsyncMock(side_effect=RuntimeError("browser service unavailable")) + + result = await fetch_detail( + "https://ekb.cian.ru/sale/flat/999999/", + browser_fetcher=mock_bf, + ) + + assert result is None + + +@pytest.mark.asyncio +async def test_fetch_detail_browser_fetcher_state_missing_returns_none(monkeypatch): + """browser HTML parsed but no defaultState → fetch_detail returns None.""" + monkeypatch.setattr( + "app.services.scrapers.cian_detail.extract_state", + lambda html, mfe, key: None, + ) + + mock_bf = MagicMock() + mock_bf.fetch = AsyncMock(return_value="no state here") + + result = await fetch_detail( + "https://ekb.cian.ru/sale/flat/777777/", + browser_fetcher=mock_bf, + ) + + assert result is None