tradein-mvp/backend/app/services/scrapers/cian_detail.py

@@ -0,0 +1,234 @@
+"""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),
+    )

tradein-mvp/backend/tests/test_cian_detail.py

@@ -0,0 +1,142 @@
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+
+from app.services.scrapers.cian_detail import (
+    DetailEnrichment,
+    _parse_float,
+    _parse_views,
+    _extract_price_changes,
+    _extract_agent_profile,
+    fetch_detail,
+)
+
+
+def test_parse_views():
+    assert _parse_views("1 234") == 1234
+    assert _parse_views("567") == 567
+    assert _parse_views(None) is None
+    assert _parse_views("nope") is None
+
+
+def test_parse_float():
+    assert _parse_float("2.7") == 2.7
+    assert _parse_float(3) == 3.0
+    assert _parse_float(None) is None
+    assert _parse_float("bad") is None
+
+
+def test_extract_price_changes_normalizes():
+    offer = {
+        "priceChanges": [
+            {"changeTime": "2026-04-01T00:00:00Z", "price": 5000000, "diffPercent": -2.5},
+        ]
+    }
+    result = _extract_price_changes(offer, {})
+    assert len(result) == 1
+    assert result[0]["change_time"] == "2026-04-01T00:00:00Z"
+    assert result[0]["price_rub"] == 5000000
+    assert result[0]["diff_percent"] == -2.5
+
+
+def test_extract_price_changes_empty():
+    assert _extract_price_changes({}, {}) == []
+    assert _extract_price_changes({"priceChanges": None}, {}) == []
+
+
+def test_extract_agent_profile_from_agent():
+    offer = {"agent": {"id": 999, "companyName": "ACME", "isPro": True, "skills": ["x"]}}
+    profile = _extract_agent_profile(offer)
+    assert profile is not None
+    assert profile["ext_agent_id"] == 999
+    assert profile["company_name"] == "ACME"
+    assert profile["is_pro"] is True
+
+
+def test_extract_agent_profile_from_user_fallback():
+    offer = {"user": {"cianUserId": 111, "companyName": "Sole"}}
+    profile = _extract_agent_profile(offer)
+    assert profile is not None
+    assert profile["ext_agent_id"] == 111
+    assert profile["company_name"] == "Sole"
+
+
+def test_extract_agent_profile_none_when_empty():
+    assert _extract_agent_profile({}) is None
+
+
+def test_dataclass_defaults():
+    e = DetailEnrichment()
+    assert e.cian_id is None
+    assert e.price_changes == []
+    assert e.raw_sister_states == {}
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_returns_none_on_no_state(monkeypatch):
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.extract_state",
+        lambda html, mfe, key: None,
+    )
+    session = MagicMock()
+    response = MagicMock()
+    response.status_code = 200
+    response.text = "<html></html>"
+    session.get = AsyncMock(return_value=response)
+    session.close = AsyncMock()
+    result = await fetch_detail("https://ekb.cian.ru/sale/flat/123/", session=session)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_returns_none_on_http_error(monkeypatch):
+    session = MagicMock()
+    response = MagicMock()
+    response.status_code = 403
+    response.text = ""
+    session.get = AsyncMock(return_value=response)
+    session.close = AsyncMock()
+    result = await fetch_detail("https://ekb.cian.ru/sale/flat/123/", session=session)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_fetch_detail_parses_core_fields(monkeypatch):
+    fake_state = {
+        "offerData": {
+            "offer": {
+                "cianId": 42,
+                "windowsViewType": "yard",
+                "separateWcsCount": 1,
+                "combinedWcsCount": 0,
+                "ceilingHeight": 2.7,
+                "repairType": "cosmetic",
+            }
+        },
+        "stats": {
+            "totalViewsFormattedString": "1 234",
+            "todayViewsFormattedString": "12",
+        },
+    }
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.extract_state",
+        lambda html, mfe, key: fake_state,
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.extract_all_states",
+        lambda html: {"frontend-offer-card": {"defaultState": fake_state}},
+    )
+    session = MagicMock()
+    response = MagicMock()
+    response.status_code = 200
+    response.text = "<html></html>"
+    session.get = AsyncMock(return_value=response)
+    session.close = AsyncMock()
+    result = await fetch_detail("https://ekb.cian.ru/sale/flat/42/", session=session)
+    assert result is not None
+    assert result.cian_id == 42
+    assert result.windows_view_type == "yard"
+    assert result.separate_wcs_count == 1
+    assert result.ceiling_height == 2.7
+    assert result.repair_type == "cosmetic"
+    assert result.views_total == 1234
+    assert result.views_today == 12