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

feat(tradein): cian_valuation.py — auth-required Calculator scraper + 24h cache #465

Merged
lekss361 merged 1 commit from feat/tradein-cian-valuation into main 2026-05-23 13:43:34 +00:00
Conversation 1 Commits 1 Files changed 2 +866
2 changed files with 866 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

465
tradein-mvp/backend/app/services/scrapers/cian_valuation.py Normal file
View file

@ -0,0 +1,465 @@
"""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 Any
from urllib.parse import urlencode
from curl_cffi.requests import AsyncSession
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.services.cian_session import load_session, mark_session_invalid
from app.services.scrapers.cian_state_parser import extract_state
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,
*,
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,
) -> 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)
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)
try:
async with AsyncSession(
impersonate="chrome120",
cookies=cookies,
timeout=25.0,
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)
# 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,
)
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 _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
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"],
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,
) -> 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,
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(: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
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,
"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

401
tradein-mvp/backend/tests/test_cian_valuation.py Normal file
View file

@ -0,0 +1,401 @@
"""Tests for cian_valuation.py — Stage 7 Cian Valuation Calculator scraper."""
import os
# Settings требует DATABASE_URL при инициализации (fail-fast).
# Для offline unit-тестов задаём dummy DSN до любого импорта из app.
os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
os.environ.setdefault("COOKIE_ENCRYPTION_KEY", "test-key-for-tests")
from unittest.mock import MagicMock
import pytest
from app.services.scrapers.cian_valuation import (
CianValuationResult,
_parse_change_pct,
_parse_num,
_parse_valuation_state,
compute_cache_key,
estimate_via_cian_valuation,
)
# ---------------------------------------------------------------------------
# compute_cache_key
# ---------------------------------------------------------------------------
def test_cache_key_stable():
"""Нормализация address: lower + strip дают одинаковый ключ."""
k1 = compute_cache_key("ЕКБ ул. Ленина 1", 50.0, 2, 5, "cosmetic", "sale")
k2 = compute_cache_key("екб ул. ленина 1 ", 50.0, 2, 5, "cosmetic", "sale")
assert k1 == k2
def test_cache_key_distinct_on_deal_type():
k1 = compute_cache_key("addr", 50.0, 2, 5, "cosmetic", "sale")
k2 = compute_cache_key("addr", 50.0, 2, 5, "cosmetic", "rent")
assert k1 != k2
def test_cache_key_distinct_on_area():
k1 = compute_cache_key("addr", 50.0, 2, 5, "cosmetic", "sale")
k2 = compute_cache_key("addr", 51.0, 2, 5, "cosmetic", "sale")
assert k1 != k2
def test_cache_key_is_hex_string():
k = compute_cache_key("addr", 50.0, 2, 5, "cosmetic", "sale")
assert len(k) == 64
assert all(c in "0123456789abcdef" for c in k)
# ---------------------------------------------------------------------------
# _parse_num
# ---------------------------------------------------------------------------
def test_parse_num_float():
assert _parse_num(5.5) == 5.5
def test_parse_num_str():
assert _parse_num("100") == 100.0
def test_parse_num_none():
assert _parse_num(None) is None
def test_parse_num_bad_str():
assert _parse_num("bad") is None
def test_parse_num_int():
assert _parse_num(0) == 0.0
# ---------------------------------------------------------------------------
# _parse_change_pct
# ---------------------------------------------------------------------------
def test_parse_change_pct_positive():
assert _parse_change_pct("8,3%") == pytest.approx(8.3)
def test_parse_change_pct_negative():
assert _parse_change_pct("-2,5%") == pytest.approx(-2.5)
def test_parse_change_pct_empty():
assert _parse_change_pct("") is None
def test_parse_change_pct_dot_format():
assert _parse_change_pct("1.5%") == pytest.approx(1.5)
# ---------------------------------------------------------------------------
# _parse_valuation_state — основные сценарии
# ---------------------------------------------------------------------------
def _make_state(
is_authenticated: bool = True,
user_id: int = 12345,
sale_price: int = 7500000,
sale_accuracy: int = 94,
rent_price: int = 35000,
rent_accuracy: int = 93,
chart_change: str = "increase",
chart_change_value: str = "8,3%",
house_id: int = 54016,
) -> dict:
return {
"user": {"isAuthenticated": is_authenticated, "userId": user_id},
"estimation": {
"sale": {
"data": {
"price": sale_price,
"accuracy": sale_accuracy,
"priceFrom": 6900000,
"priceTo": 8100000,
"priceSqm": 150000,
"dealType": "sale",
"filtersHash": "abc123",
},
"isError": False,
"isFetching": False,
},
"rent": {
"data": {
"price": rent_price,
"accuracy": rent_accuracy,
"priceFrom": 32000,
"priceTo": 38000,
"taxPrice": 1400,
"dealType": "rent",
},
},
},
"estimationChart": {
"data": {
"title": {
"title": "Стоимость этой квартиры увеличилась на {{change}} за полгода",
"changeValue": chart_change_value,
"change": chart_change,
},
"chartData": {
"maxXValue": 1777593600000,
"minXValue": 1761955200000,
"data": [
{"date": 1761955200000, "price": 6033438, "priceFormatted": "6,0 млн ₽"},
{"date": 1764547200000, "price": 6013573, "priceFormatted": "6,0 млн ₽"},
{"date": 1767225600000, "price": 6116587, "priceFormatted": "6,1 млн ₽"},
{"date": 1769904000000, "price": 6236983, "priceFormatted": "6,2 млн ₽"},
{"date": 1772323200000, "price": 6232172, "priceFormatted": "6,2 млн ₽"},
{"date": 1775001600000, "price": 6331112, "priceFormatted": "6,3 млн ₽"},
{"date": 1777593600000, "price": 6532872, "priceFormatted": "6,5 млн ₽"},
],
},
},
},
"houseInfo": {
"data": {
"items": [
{"title": "Год постройки", "value": 2005},
{"title": "Тип дома", "value": "Кирпичный"},
],
"isRenovation": False,
"isEmergency": False,
},
},
"filters": {"houseId": house_id, "roomsCount": 2},
"managementCompany": {
"data": {
"name": "ТСЖ «Учителей, 18»",
"address": "ул. Учителей, 18",
"phones": ["+7 (343) 000-00-00"],
"email": "tsj@example.ru",
"chiefName": "Иванов И.И.",
"openingHours": [{"Пн–Пт": ["9:00", "18:00"]}],
},
},
"page": {"pageType": "valuationPage"},
}
def test_parse_valuation_state_extracts_sale():
state = _make_state()
result = _parse_valuation_state(state)
assert result.is_authenticated is True
assert result.user_id == 12345
assert result.sale_price_rub == 7500000.0
assert result.sale_accuracy == 94.0
assert result.sale_price_from == 6900000.0
assert result.sale_price_to == 8100000.0
assert result.sale_price_sqm == 150000.0
assert result.filters_hash == "abc123"
def test_parse_valuation_state_extracts_rent():
state = _make_state()
result = _parse_valuation_state(state)
assert result.rent_price_rub == 35000.0
assert result.rent_accuracy == 93.0
assert result.rent_price_from == 32000.0
assert result.rent_price_to == 38000.0
assert result.rent_tax_price == 1400.0
def test_parse_valuation_state_chart_7_points():
state = _make_state()
result = _parse_valuation_state(state)
assert len(result.chart) == 7
# First point: unix ms 1761955200000 → 2025-11-01 (UTC)
assert result.chart[0]["month_date"] == "2025-11-01"
assert result.chart[0]["price"] == 6033438.0
def test_parse_valuation_state_chart_change_increase():
state = _make_state(chart_change="increase", chart_change_value="8,3%")
result = _parse_valuation_state(state)
assert result.chart_change_pct == pytest.approx(8.3)
assert result.chart_change_direction == "increase"
def test_parse_valuation_state_chart_change_decrease():
state = _make_state(chart_change="decrease", chart_change_value="-2,5%")
result = _parse_valuation_state(state)
assert result.chart_change_pct == pytest.approx(-2.5)
assert result.chart_change_direction == "decrease"
def test_parse_valuation_state_house_info():
state = _make_state()
result = _parse_valuation_state(state)
assert len(result.house_info) == 2
assert result.house_info[0]["title"] == "Год постройки"
assert result.house_info[0]["value"] == 2005
def test_parse_valuation_state_external_house_id():
state = _make_state(house_id=54016)
result = _parse_valuation_state(state)
assert result.external_house_id == 54016
def test_parse_valuation_state_management_company():
state = _make_state()
result = _parse_valuation_state(state)
assert result.management_company is not None
assert result.management_company["name"] == "ТСЖ «Учителей, 18»"
assert result.management_company["chiefName"] == "Иванов И.И."
def test_parse_valuation_state_no_auth():
state = _make_state(is_authenticated=False)
result = _parse_valuation_state(state)
assert result.is_authenticated is False
# Values still parsed — auth check is done by caller, not parser
assert result.sale_price_rub == 7500000.0
def test_parse_valuation_state_empty_estimation():
state = {"user": {"isAuthenticated": True, "userId": 1}, "estimation": {}}
result = _parse_valuation_state(state)
assert result.sale_price_rub is None
assert result.rent_price_rub is None
assert result.chart == []
def test_parse_valuation_state_missing_management_company():
state = _make_state()
del state["managementCompany"]
result = _parse_valuation_state(state)
assert result.management_company is None
# ---------------------------------------------------------------------------
# CianValuationResult dataclass defaults
# ---------------------------------------------------------------------------
def test_dataclass_defaults():
r = CianValuationResult()
assert r.chart == []
assert r.house_info == []
assert r.is_authenticated is False
assert r.sale_price_rub is None
assert r.management_company is None
# ---------------------------------------------------------------------------
# estimate_via_cian_valuation — integration (mocked)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_estimate_returns_none_when_no_cookies(monkeypatch):
"""Если load_session returns None → graceful return None без crash."""
monkeypatch.setattr(
"app.services.scrapers.cian_valuation.load_session",
lambda db: None,
)
db = MagicMock()
# cache miss
db.execute.return_value.mappings.return_value.first.return_value = None
result = await estimate_via_cian_valuation(
db,
address="ЕКБ",
total_area=50.0,
rooms_count=2,
floor=5,
use_cache=True,
)
assert result is None
@pytest.mark.asyncio
async def test_cache_hit_returns_cached(monkeypatch):
"""Если cache hit → return без fetching (load_session не вызывается)."""
db = MagicMock()
cached_row = {
"raw_payload": {"cached": True},
"sale_price_rub": 7000000,
"sale_accuracy": 80,
"rent_price_rub": 30000,
"rent_accuracy": 65,
"chart": [{"month_date": "2026-05-01", "price": 145000}],
"chart_change_pct": 0.5,
"chart_change_direction": "increase",
"external_house_id": 999,
"filters_hash": "xyz789",
}
db.execute.return_value.mappings.return_value.first.return_value = cached_row
# load_session должен НЕ вызываться при cache hit
monkeypatch.setattr(
"app.services.scrapers.cian_valuation.load_session",
lambda db: pytest.fail("load_session should not be called on cache hit"),
)
result = await estimate_via_cian_valuation(
db,
address="ЕКБ",
total_area=50.0,
rooms_count=2,
floor=5,
use_cache=True,
)
assert result is not None
assert result.sale_price_rub == 7000000.0
assert result.rent_price_rub == 30000.0
assert result.external_house_id == 999
assert result.filters_hash == "xyz789"
assert result.is_authenticated is True # cache implies authenticated
assert len(result.chart) == 1
@pytest.mark.asyncio
async def test_estimate_use_cache_false_skips_cache(monkeypatch):
"""use_cache=False → cache lookup не вызывается, идём сразу в load_session."""
load_session_called = []
def fake_load_session(db: object) -> None:
load_session_called.append(True)
return None # simulate no cookies → None result
monkeypatch.setattr(
"app.services.scrapers.cian_valuation.load_session",
fake_load_session,
)
db = MagicMock()
result = await estimate_via_cian_valuation(
db,
address="ЕКБ",
total_area=50.0,
rooms_count=2,
floor=5,
use_cache=False,
)
assert result is None
assert load_session_called # load_session был вызван (cache bypass worked)
# db.execute НЕ должен быть вызван для cache lookup
db.execute.assert_not_called()