lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(cian): collect price history via BrowserFetcher (fixes silent no-op backfill) #1657

Merged
bot-reviewer merged 1 commit from feat/cian-browser-price-history into main 2026-06-16 19:47:16 +00:00
Conversation 0 Commits 1 Files changed 3 +336 -122
3 changed files with 336 additions and 122 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

188
tradein-mvp/backend/app/services/scrapers/cian_detail.py
View file

@ -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"),
}
)

97
tradein-mvp/backend/app/tasks/cian_history_backfill.py
View file

@ -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:

173
tradein-mvp/backend/tests/test_cian_detail.py
View file

@ -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="<html>browser-rendered</html>")
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="<html></html>")
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="<html>no state here</html>")
result = await fetch_detail(
"https://ekb.cian.ru/sale/flat/777777/",
browser_fetcher=mock_bf,
)
assert result is None