gendesign/tradein-mvp/backend/tests/test_3252_domclick_card_fields.py

"""#3252: три тихие дыры в полях карточки Домклика.

1. owners_count. egrnData отдаёт поля ЕГРН обёрткой ``{"status", "value"}``, а парсер
   делал ``int()`` от словаря и молча получал None: на проде 0 из ~2970 карточек,
   обогащённых с 29.08 (при 5075 из 6296 у ручного прогона 18.07, который обёртку
   разворачивал).
2. Нулевые площади. ``livingArea: 0`` / ``kitchenArea: 0`` у Домклика значит «не
   указано», а в колонку ложился 0.00 (303 кухни и 84 жилые площади).
3. Жилая площадь и балконы стирались переобходом выдачи: апсерт в scraper_kit.base
   писал их сырым ``EXCLUDED``, а SERP Домклика и Авито этих полей не отдаёт. У
   domklik, переобойдённых после обогащения, living_area_m2 = 0 из 5217.

Состояние карточек ниже — урезанный ``__SSR_STATE__`` живых карточек 2074362051 и
2078257603, снятый 17.09 (значения как есть, лишние ветки выкинуты).
"""

from __future__ import annotations

import json
import os
import uuid
from decimal import Decimal
from typing import Any
from unittest.mock import MagicMock

import pytest
from scraper_kit.base import ScrapedLot, save_listings
from scraper_kit.providers.domclick.detail import parse_detail_html
from sqlalchemy import text

_LIVE_2074362051 = {
    "productCard": {
        "objectInfo": {
            "area": 59.8,
            "balconies": 1,
            "kitchenArea": 14.9,
            "livingArea": 25.8,
            "renovation": "Косметический",
        },
        "egrnData": {
            "area": {"status": "success", "value": 59.8},
            "collateral": False,
            "collateral_sber": False,
            "floor": {"status": "success", "value": 14},
            "owners_count": {"status": "error", "value": 6},
        },
    },
}

_LIVE_2078257603 = {
    "productCard": {
        "objectInfo": {
            "area": 67.4,
            "balconies": 0,
            "kitchenArea": 0,
            "livingArea": 0,
            "renovation": "",
        },
        "egrnData": {
            "area": {"status": "error", "value": 65.3},
            "collateral": False,
            "collateral_sber": False,
            "floor": {"status": "success", "value": 21},
            "owners_count": {"status": "success", "value": 1},
        },
    },
}


def _parse(state: dict, card_id: str):
    html = f"<script>window.__SSR_STATE__ = {json.dumps(state, ensure_ascii=False)};</script>"
    return parse_detail_html(html, f"https://ekaterinburg.domclick.ru/card/sale__flat__{card_id}")


# ── 1-2. Разбор живой формы ──────────────────────────────────────────────────


def test_owners_count_is_read_from_egrn_wrapper_whatever_the_status() -> None:
    # status=error у owners_count = «много собственников», value — сам факт ЕГРН.
    assert _parse(_LIVE_2074362051, "2074362051").owners_count == 6
    assert _parse(_LIVE_2078257603, "2078257603").owners_count == 1


def test_egrn_area_keeps_wrapper_in_raw_payload() -> None:
    e = _parse(_LIVE_2078257603, "2078257603")
    assert e.raw_extra["egrn_area"] == {"status": "error", "value": 65.3}


def test_zero_areas_mean_unknown_not_zero() -> None:
    zero = _parse(_LIVE_2078257603, "2078257603")
    assert (zero.living_area_m2, zero.kitchen_area_m2) == (None, None)
    real = _parse(_LIVE_2074362051, "2074362051")
    assert (real.living_area_m2, real.kitchen_area_m2) == (25.8, 14.9)


# ── 3. Переобход выдачи не стирает то, что добыла карточка (живой Postgres) ──


def _live_session() -> Any | None:
    try:
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
        if not dsn or "localhost:5432/test" in dsn:
            return None
        engine = create_engine(dsn, future=True)
        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
        return sessionmaker(bind=engine, future=True)()
    except Exception:
        return None


def _matcher() -> MagicMock:
    m = MagicMock()
    m.match_or_create_house.return_value = (None, 0.0, "no_address")
    m.upsert_listing_source.return_value = None
    return m


def _serp_lot(sid: str, **detail: Any) -> ScrapedLot:
    return ScrapedLot(
        source="domklik",
        source_url=f"https://ekaterinburg.domclick.ru/card/sale__flat__t3252{sid}",
        source_id=f"t3252-{sid}",
        price_rub=7_000_000,
        **detail,
    )


def _row(db: Any, sid: str) -> Any:
    return db.execute(
        text(
            "SELECT living_area_m2, balconies_count FROM listings "
            "WHERE source='domklik' AND source_id = :sid"
        ),
        {"sid": f"t3252-{sid}"},
    ).fetchone()


def _cleanup(db: Any) -> None:
    try:
        db.rollback()
        ids = "(SELECT id FROM listings WHERE source='domklik' AND source_id LIKE 't3252-%')"
        db.execute(text(f"DELETE FROM listings_snapshots WHERE listing_id IN {ids}"))
        db.execute(text("DELETE FROM listing_sources WHERE ext_id LIKE 't3252-%'"))
        db.execute(text("DELETE FROM listings WHERE source='domklik' AND source_id LIKE 't3252-%'"))
        db.commit()
    finally:
        db.close()


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_serp_rescrape_keeps_detail_living_area_and_balconies() -> None:
    db = _live_session()
    sid = uuid.uuid4().hex[:8]
    try:
        save_listings(
            db,
            [_serp_lot(sid, living_area_m2=25.8, balconies_count=1)],
            matcher=_matcher(),
            region_code=66,
        )
        save_listings(db, [_serp_lot(sid)], matcher=_matcher(), region_code=66)  # выдача: полей нет
        row = _row(db, sid)
        assert (row.living_area_m2, row.balconies_count) == (Decimal("25.80"), 1)
    finally:
        _cleanup(db)


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_real_new_value_still_overwrites() -> None:
    db = _live_session()
    sid = uuid.uuid4().hex[:8]
    try:
        save_listings(
            db,
            [_serp_lot(sid, living_area_m2=25.8, balconies_count=1)],
            matcher=_matcher(),
            region_code=66,
        )
        save_listings(
            db,
            [_serp_lot(sid, living_area_m2=30.5, balconies_count=2)],
            matcher=_matcher(),
            region_code=66,
        )
        row = _row(db, sid)
        assert (row.living_area_m2, row.balconies_count) == (Decimal("30.50"), 2)
    finally:
        _cleanup(db)