fix(scraper-kit): бедный re-scrape больше не стирает признаки продавца

Апсерт listings присваивал шесть полей напрямую (= EXCLUDED.X): phones,
is_homeowner, is_pro_seller, bargain_allowed, sale_type, metro_stations.

Комментарий рядом («Cian-specific: обновляем при каждом re-scrape») был
верен, пока в listings писал один cian, который отдаёт эти поля на каждом
проходе. Сейчас в ту же таблицу пишут yandex/avito/domklik, чей SERP их не
отдаёт вовсе — и NULL молча затирал значение, добытое detail-обогащением.
Ни ошибки, ни лога, ни изменения статуса прогона: колонка просто
откатывалась к NULL на обычном переобходе.

Прод 23-24.08, yandex: is_homeowner NULL у 98.6% (16 232 из 16 460)
активных листингов, is_pro_seller — у 76.6%.

Соседние поля в том же операторе уже защищены COALESCE ровно от этого, с
явными объяснениями #2007 / #2594 / #2777 — address, city, kitchen_area_m2,
ceiling_height_m, mortgage_available, is_apartments, is_rosreestr_checked.
Эти шесть в защиту просто не попали.

Правка в ДВУХ местах, не в одном. Кроме SET изменена правая часть гейта
`IS DISTINCT FROM` (#2992): он решает, переписывать ли строку, сравнивая
текущие значения с ИТОГОВЫМИ (post-COALESCE). Оставить его на сыром
EXCLUDED значило бы, что гейт видит «NULL против значения» и считает
строку изменившейся там, где она не меняется — вернулись бы ровно те
лишние UPDATE и TOAST-чанки, ради которых #2992 делался.

Пустые phones/metro_stations сериализуются как None (`_to_json(...) if
lot.X else None`), то есть приходят SQL NULL — COALESCE их удерживает, а
не подменяет пустым массивом.

Тест сторожит и паритет тоже: сравнивает оба кортежа гейта поэлементно,
поэтому поймает будущее расхождение по ЛЮБОЙ колонке, не только по этим
шести.

Проверено:
  - фальсификация: все 7 тестов падают на коде до правки
  - 770 passed, 10 skipped (-k "upsert or listing or scrape or writer or
    snapshot or timestamps")
  - выравнивание кортежей гейта: 39 = 39 элементов, порядок колонок совпал

Из #3063 сделан только пункт 1. Пункт 2 (выводить is_pro_seller из
agency_name) НЕ вошёл — см. комментарий в issue: у него есть риск ложного
срабатывания, которого нет у этой правки.

Refs #3063

tradein-mvp/backend/tests/test_3063_seller_fields_not_eroded.py

@@ -0,0 +1,241 @@
+"""#3063: бедный re-scrape больше не стирает признаки продавца.
+
+Апсерт listings присваивал шесть полей напрямую (`= EXCLUDED.X`):
+phones, is_homeowner, is_pro_seller, bargain_allowed, sale_type, metro_stations.
+Комментарий рядом («Cian-specific: обновляем при каждом re-scrape») был верен,
+пока в listings писал один cian, который отдаёт их на каждом проходе. Сейчас в ту
+же таблицу пишут yandex/avito/domklik, чей SERP их не отдаёт вовсе — и NULL молча
+затирал значение, добытое detail-обогащением. Ни ошибки, ни лога, ни изменения
+статуса прогона: колонка просто откатывалась к NULL на обычном переобходе.
+
+Прод 23-24.08: у yandex is_homeowner NULL у 98.6 % (16 232 из 16 460) активных
+листингов, is_pro_seller — у 76.6 %.
+
+Соседние поля (address, city, kitchen_area_m2, ceiling_height_m,
+mortgage_available, is_apartments, is_rosreestr_checked) уже защищены COALESCE
+ровно от этого — с явными объяснениями #2007/#2594/#2777. Эти шесть в защиту
+просто не попали.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import inspect
+import uuid
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+import scraper_kit.base as base_module
+from scraper_kit.base import ScrapedLot, save_listings
+from sqlalchemy import text
+
+# Поля, которые detail добывает, а SERP чужого источника не отдаёт.
+ERODED_FIELDS = [
+    "phones",
+    "is_homeowner",
+    "is_pro_seller",
+    "bargain_allowed",
+    "sale_type",
+    "metro_stations",
+]
+
+
+def _upsert_sql() -> str:
+    return inspect.getsource(base_module)
+
+
+# ── 1. Статические гарды: работают всегда, даже без БД ───────────────────────
+
+
+@pytest.mark.parametrize("field", ERODED_FIELDS)
+def test_field_is_coalesce_protected_in_set(field: str) -> None:
+    """SET присваивает поле через COALESCE, а не сырым EXCLUDED.
+
+    Фальсификация: на коде до правки каждый кейс падает — там стоит
+    `is_homeowner = EXCLUDED.is_homeowner` без COALESCE.
+    """
+    src = _upsert_sql()
+    assert not re.search(rf"^\s*{field} = EXCLUDED\.{field},", src, re.M), (
+        f"{field} присваивается сырым EXCLUDED — бедный re-scrape затрёт значение"
+    )
+    assert re.search(
+        rf"{field} = COALESCE\(\s*EXCLUDED\.{field}, listings\.{field}\s*\)", src
+    ), f"{field} должен присваиваться через COALESCE(EXCLUDED.{field}, listings.{field})"
+
+
+def test_unchanged_gate_mirrors_the_set_clause() -> None:
+    """Правая часть `IS DISTINCT FROM` зеркалит то, что реально запишется.
+
+    Это НЕ стилистика. Гейт #2992 решает, переписывать ли строку, сравнивая
+    текущие значения с ИТОГОВЫМИ (post-COALESCE). Если SET станет COALESCE, а
+    гейт останется на сыром EXCLUDED, они разъедутся: гейт увидит «NULL против
+    значения» и посчитает строку изменившейся там, где она не меняется —
+    вернутся ровно те лишние UPDATE и TOAST-чанки, ради которых #2992 делался.
+
+    Гард структурный: сравниваем два кортежа поэлементно, поэтому он поймает и
+    будущее расхождение по ЛЮБОЙ колонке, не только по шести из #3063.
+    """
+    src = _upsert_sql()
+    m = re.search(
+        r"WHERE \(\n(.*?)\n\s*\) IS DISTINCT FROM \(\n(.*?)\n\s*\)\n\s*OR \(listings\.last_seen_at",
+        src,
+        re.S,
+    )
+    assert m is not None, "не найден гейт `IS DISTINCT FROM` — тест устарел вместе с кодом"
+
+    def items(block: str) -> list[str]:
+        block = re.sub(r"--[^\n]*", "", block)  # комментарии внутри кортежа
+        out: list[str] = []
+        depth, cur = 0, ""
+        for ch in block:
+            if ch == "(":
+                depth += 1
+            if ch == ")":
+                depth -= 1
+            if ch == "," and depth == 0:
+                out.append(cur.strip())
+                cur = ""
+            else:
+                cur += ch
+        if cur.strip():
+            out.append(cur.strip())
+        return [" ".join(x.split()) for x in out]
+
+    left, right = items(m.group(1)), items(m.group(2))
+    assert len(left) == len(right), (
+        f"кортежи гейта разной длины: слева {len(left)}, справа {len(right)} — "
+        "сравнение поехало бы по колонкам молча"
+    )
+    for i, (lhs, rhs) in enumerate(zip(left, right)):
+        column = lhs.split(".")[-1]
+        if column == "is_active":
+            continue  # SET пишет литерал true, справа он же — намеренно
+        assert column in rhs, f"позиция {i}: слева {lhs}, справа {rhs} — колонки разъехались"
+        if column in ERODED_FIELDS:
+            assert rhs.startswith("COALESCE("), (
+                f"{column} в гейте сравнивается сырым EXCLUDED, а SET пишет COALESCE"
+            )
+
+
+# ── 2. Поведенческий тест на живом Postgres (в CI есть, локально skip) ───────
+
+
+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)
+        conn = engine.connect()
+        conn.execute(text("SELECT 1"))
+        conn.close()
+        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 _lot(src_id: str, *, rich: bool) -> ScrapedLot:
+    """rich=True — как detail-обогащение; rich=False — как бедный SERP чужого
+    источника (поля не заполнены вовсе)."""
+    extra: dict[str, Any] = {}
+    if rich:
+        extra = {
+            "is_homeowner": True,
+            "is_pro_seller": False,
+            "phones": ["+79000000000"],
+            "sale_type": "free",
+        }
+    return ScrapedLot(
+        source="cian",
+        source_url=f"https://ekb.cian.ru/sale/flat/t3063-{src_id}/",
+        source_id=f"t3063-{src_id}",
+        price_rub=5_000_000,
+        **extra,
+    )
+
+
+def _cleanup(db: Any) -> None:
+    try:
+        db.rollback()
+        db.execute(
+            text(
+                "DELETE FROM listings_snapshots WHERE listing_id IN "
+                "(SELECT id FROM listings WHERE source='cian' AND source_id LIKE 't3063-%')"
+            )
+        )
+        db.execute(text("DELETE FROM listing_sources WHERE ext_id LIKE 't3063-%'"))
+        db.execute(text("DELETE FROM listings WHERE source='cian' AND source_id LIKE 't3063-%'"))
+        db.commit()
+    finally:
+        db.close()
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_poor_rescrape_does_not_erase_seller_fields() -> None:
+    """Головной: обогащённая строка переживает последующий бедный проход.
+
+    На коде до правки второй save_listings обнуляет is_homeowner/phones/sale_type.
+    """
+    db = _live_session()
+    sid = uuid.uuid4().hex[:8]
+    try:
+        save_listings(db, [_lot(sid, rich=True)], matcher=_matcher(), region_code=66)
+        save_listings(db, [_lot(sid, rich=False)], matcher=_matcher(), region_code=66)
+        row = db.execute(
+            text(
+                "SELECT is_homeowner, is_pro_seller, phones, sale_type FROM listings "
+                "WHERE source='cian' AND source_id = :sid"
+            ),
+            {"sid": f"t3063-{sid}"},
+        ).fetchone()
+        assert row is not None, "строка исчезла"
+        assert row.is_homeowner is True, "бедный re-scrape стёр is_homeowner"
+        assert row.is_pro_seller is False, "бедный re-scrape стёр is_pro_seller (False -> NULL)"
+        assert row.phones is not None, "бедный re-scrape стёр phones"
+        assert row.sale_type == "free", "бедный re-scrape стёр sale_type"
+    finally:
+        _cleanup(db)
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_real_change_still_overwrites() -> None:
+    """Обратная сторона: COALESCE не превращает поля в write-once.
+
+    Настоящая смена признака (частник -> агентство) приходит НЕ NULL'ом, значит
+    обязана перезаписать. Без этого теста фикс мог бы «защитить» данные ценой
+    того, что они перестали бы обновляться вообще.
+    """
+    db = _live_session()
+    sid = uuid.uuid4().hex[:8]
+    try:
+        save_listings(db, [_lot(sid, rich=True)], matcher=_matcher(), region_code=66)
+        flipped = _lot(sid, rich=True)
+        flipped.is_homeowner = False
+        flipped.is_pro_seller = True
+        save_listings(db, [flipped], matcher=_matcher(), region_code=66)
+        row = db.execute(
+            text(
+                "SELECT is_homeowner, is_pro_seller FROM listings "
+                "WHERE source='cian' AND source_id = :sid"
+            ),
+            {"sid": f"t3063-{sid}"},
+        ).fetchone()
+        assert row.is_homeowner is False, "настоящая смена признака не записалась"
+        assert row.is_pro_seller is True, "настоящая смена признака не записалась"
+    finally:
+        _cleanup(db)

tradein-mvp/packages/scraper-kit/src/scraper_kit/base.py

@@ -594,12 +594,29 @@ def save_listings(
                     description_minhash = EXCLUDED.description_minhash,
                     cadastral_number = EXCLUDED.cadastral_number,
                     building_cadastral_number = EXCLUDED.building_cadastral_number,
-                    phones = EXCLUDED.phones,
-                    is_homeowner = EXCLUDED.is_homeowner,
-                    is_pro_seller = EXCLUDED.is_pro_seller,
-                    bargain_allowed = EXCLUDED.bargain_allowed,
-                    sale_type = EXCLUDED.sale_type,
-                    metro_stations = EXCLUDED.metro_stations,
+                    -- #3063: COALESCE, а не прямая перезапись. Комментарий выше
+                    -- («Cian-specific: обновляем при каждом re-scrape») был верен, пока
+                    -- в listings писал один cian, который отдаёт эти поля на каждом
+                    -- проходе. Сейчас в ту же таблицу пишут yandex/avito/domklik, чей
+                    -- SERP их не отдаёт вовсе → NULL молча затирал значение, добытое
+                    -- detail-обогащением. Ни ошибки, ни лога: колонка просто
+                    -- откатывалась к NULL на обычном переобходе.
+                    -- Прод 23-24.08: у yandex is_homeowner NULL у 98.6 % (16 232 из
+                    -- 16 460) активных листингов, is_pro_seller — у 76.6 %.
+                    -- Пустые phones/metro_stations сериализуются как None (см.
+                    -- `_to_json(...) if lot.X else None` в сборке параметров), то есть
+                    -- приходят SQL NULL — COALESCE их тоже удерживает, а не подменяет
+                    -- пустым массивом.
+                    phones = COALESCE(EXCLUDED.phones, listings.phones),
+                    is_homeowner = COALESCE(EXCLUDED.is_homeowner, listings.is_homeowner),
+                    is_pro_seller = COALESCE(EXCLUDED.is_pro_seller, listings.is_pro_seller),
+                    bargain_allowed = COALESCE(
+                        EXCLUDED.bargain_allowed, listings.bargain_allowed
+                    ),
+                    sale_type = COALESCE(EXCLUDED.sale_type, listings.sale_type),
+                    metro_stations = COALESCE(
+                        EXCLUDED.metro_stations, listings.metro_stations
+                    ),
                     listing_date = COALESCE(EXCLUDED.listing_date, listings.listing_date),
                     area_m2 = COALESCE(EXCLUDED.area_m2, listings.area_m2),
                     -- #2777: ДОзаполнение адреса — порядок аргументов обратный остальным,
@@ -728,9 +745,16 @@ def save_listings(
                 EXCLUDED.price_rub, EXCLUDED.price_per_m2, EXCLUDED.living_area_m2,
                 EXCLUDED.bedrooms_count, EXCLUDED.balconies_count, EXCLUDED.loggias_count,
                 EXCLUDED.description_minhash, EXCLUDED.cadastral_number,
-                EXCLUDED.building_cadastral_number, EXCLUDED.phones, EXCLUDED.is_homeowner,
-                EXCLUDED.is_pro_seller, EXCLUDED.bargain_allowed, EXCLUDED.sale_type,
-                EXCLUDED.metro_stations,
+                EXCLUDED.building_cadastral_number,
+                -- #3063: те же COALESCE, что в SET выше. Правая часть гейта ОБЯЗАНА
+                -- зеркалить то, что реально запишется, иначе строка считалась бы
+                -- изменившейся там, где она не меняется.
+                COALESCE(EXCLUDED.phones, listings.phones),
+                COALESCE(EXCLUDED.is_homeowner, listings.is_homeowner),
+                COALESCE(EXCLUDED.is_pro_seller, listings.is_pro_seller),
+                COALESCE(EXCLUDED.bargain_allowed, listings.bargain_allowed),
+                COALESCE(EXCLUDED.sale_type, listings.sale_type),
+                COALESCE(EXCLUDED.metro_stations, listings.metro_stations),
                 COALESCE(EXCLUDED.listing_date, listings.listing_date),
                 COALESCE(EXCLUDED.area_m2, listings.area_m2),
                 COALESCE(listings.address, EXCLUDED.address),