"""Cian.ru detail page scraper — single offer enrichment.""" from __future__ import annotations import json import logging from dataclasses import dataclass, field from typing import Any from curl_cffi.requests import AsyncSession from app.services.scrapers.cian_state_parser import extract_all_states, extract_state logger = logging.getLogger(__name__) @dataclass class DetailEnrichment: cian_id: int | None = None windows_view_type: str | None = None separate_wcs_count: int | None = None combined_wcs_count: int | None = None ceiling_height: float | None = None repair_type: str | None = None views_total: int | None = None views_today: int | None = None bti_data: dict[str, Any] | None = None price_changes: list[dict[str, Any]] = field(default_factory=list) agent_profile: dict[str, Any] | None = None newbuilding_data: dict[str, Any] | None = None raw_offer: dict[str, Any] | None = None raw_sister_states: dict[str, Any] = field(default_factory=dict) async def fetch_detail( offer_url: str, *, session: AsyncSession | None = None ) -> DetailEnrichment | None: """Fetch detail page → extract via state_parser → DetailEnrichment, or None on failure.""" close_session = False if session is None: session = AsyncSession( impersonate="chrome120", timeout=25.0, 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 %s → HTTP %d", offer_url, resp.status_code) return None html = resp.text 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=offer.get("windowsViewType"), separate_wcs_count=offer.get("separateWcsCount"), combined_wcs_count=offer.get("combinedWcsCount"), ceiling_height=_parse_float(offer.get("ceilingHeight")), repair_type=offer.get("repairType"), raw_offer=offer, ) stats = offer_state.get("stats", {}) result.views_total = _parse_views(stats.get("totalViewsFormattedString")) result.views_today = _parse_views(stats.get("todayViewsFormattedString")) result.price_changes = _extract_price_changes(offer, offer_state) result.agent_profile = _extract_agent_profile(offer) all_states = extract_all_states(html) bti_state = all_states.get("frontend-offer-card", {}).get("bti") if bti_state: result.bti_data = bti_state.get("houseData") result.raw_sister_states["bti"] = bti_state newbuilding = offer.get("newbuilding") if newbuilding and newbuilding.get("id"): result.newbuilding_data = newbuilding for k, v in all_states.get("frontend-offer-card", {}).items(): if k != "defaultState": result.raw_sister_states[k] = v return result finally: if close_session: await session.close() 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 | None) -> int | None: if not formatted: return None try: return int(str(formatted).replace(" ", "").replace(",", "")) except (ValueError, TypeError): return None def _extract_price_changes(offer: dict[str, Any], state: dict[str, Any]) -> list[dict[str, Any]]: raw = offer.get("priceChanges") or state.get("priceChanges") or [] if not isinstance(raw, list): return [] result = [] for entry in raw: if not isinstance(entry, dict): continue result.append( { "change_time": entry.get("changeTime") or entry.get("date"), "price_rub": entry.get("price") or entry.get("priceRub"), "diff_percent": entry.get("diffPercent") or entry.get("diff_percent"), } ) return result def _extract_agent_profile(offer: dict[str, Any]) -> dict[str, Any] | None: 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"), } async def save_detail_enrichment(db: Any, listing_id: int, enrichment: DetailEnrichment) -> None: """UPDATE listings + INSERT offer_price_history + UPSERT agents. psycopg v3 syntax — CAST(:x AS type), no :x::type. """ from sqlalchemy import text 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), views_total = COALESCE(:vt, views_total), views_today = COALESCE(:vd, views_today) 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, "vt": enrichment.views_total, "vd": enrichment.views_today, }, ) 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 DO NOTHING """), { "lid": listing_id, "ct": change["change_time"], "price": change["price_rub"], "diff": change.get("diff_percent"), }, ) if enrichment.agent_profile and enrichment.agent_profile.get("ext_agent_id"): ap = enrichment.agent_profile 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) """), { "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"), }, ) db.commit() logger.info( "Cian detail saved listing_id=%s (price_changes=%d)", listing_id, len(enrichment.price_changes), )