fix(tradein/scraper): listing_segment самочинится при upsert вместо вечного NULL

Раньше listing_segment писался ТОЛЬКО при INSERT в save_listings — строка,
единожды родившаяся с NULL-сегментом (классификатор не сработал на первом
скрейпе), никогда не получала сегмент на повторных сборах, даже когда он
становился определим. Симптом лечила отдельная джоба деактивации
(deactivate_stale_{cian,yandex}_null_segment, миграция 266, PR #2908) — чистит
мусор, но не саму запись.

Добавлен listing_segment = COALESCE(EXCLUDED.listing_segment,
listings.listing_segment) в ON CONFLICT DO UPDATE, и тот же идиом в
reconcile-UPDATE (dedup_hash-drift fallback path) — иначе self-heal был бы
неполным для строк, прошедших через этот путь. COALESCE, а не голый EXCLUDED,
чтобы повторный скрейп без определённого сегмента не затирал уже известное
значение пустым (тот же паттерн, что уже применён для city/kitchen/ceiling).

Тесты: tests/test_listing_segment_upsert_selfheal.py — INSERT-путь,
ON CONFLICT COALESCE (обе стороны), reconcile-UPDATE COALESCE.

tradein-mvp/backend/tests/test_listing_segment_upsert_selfheal.py

@@ -0,0 +1,215 @@
+"""listing_segment upsert self-heal: строка не должна вечно застревать с NULL-сегментом.
+
+Баг: `scraper_kit.base.save_listings` писал `listing_segment` ТОЛЬКО в INSERT-ветке
+upsert'а — колонки не было ни в `ON CONFLICT (dedup_hash) DO UPDATE SET`, ни в
+reconcile-UPDATE (dedup_hash-drift fallback, срабатывает при UniqueViolation по
+(source, source_id)). Итог: если первый скрейп объявления не смог определить сегмент
+(классификатор вернул None), строка рождалась с `listing_segment IS NULL` и
+НИКОГДА не самочинялась на последующих пересборах, даже когда сегмент становился
+определим. Симптом лечили отдельной джобой деактивации
+(`deactivate_stale_{cian,yandex}_null_segment`, миграция 266, PR #2908) — чистит
+мусор в `is_active`, но не лечит саму запись сегмента.
+
+Fix: `listing_segment = COALESCE(EXCLUDED.listing_segment, listings.listing_segment)`
+в ON CONFLICT DO UPDATE + `listing_segment = COALESCE(:listing_segment, listing_segment)`
+в reconcile UPDATE — тот же идиом, что уже применён для `city` (#2594,
+test_listings_city_from_sweep.py) и `kitchen_area_m2`/`ceiling_height_m` (#2007):
+новое значение обновляет строку, но НЕ затирает уже известное пустым.
+
+Тесты здесь, как и соседний test_listings_city_from_sweep.py, мокают db.execute и
+проверяют SQL-текст + bind-параметры (unit-уровень, без реальной Postgres) —
+COALESCE-семантику "новое непустое побеждает / пустое не затирает старое" исполняет
+сама база при выполнении запроса.
+"""
+
+from __future__ import annotations
+
+import os
+from contextlib import contextmanager
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from scraper_kit.base import ScrapedLot as KitLot
+from scraper_kit.base import save_listings as kit_save_listings
+
+
+@contextmanager
+def _nested_ctx() -> Any:
+    yield MagicMock()
+
+
+def _mock_db_insert_path(listing_id: int = 42) -> MagicMock:
+    """Session mock для fresh INSERT path (xmax = 0 → inserted)."""
+    insert_row = MagicMock()
+    insert_row.id = listing_id
+    insert_row.inserted = True
+
+    db = MagicMock()
+
+    def _execute(sql: Any, params: dict[str, Any] | None = None) -> MagicMock:
+        s = str(sql)
+        res = MagicMock()
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
+            res.fetchone.return_value = None
+        elif "FROM listings_snapshots" in s:
+            res.fetchone.return_value = None
+        elif "INSERT INTO listings (" in s:
+            res.fetchone.return_value = insert_row
+        else:
+            res.fetchone.return_value = None
+        return res
+
+    db.execute.side_effect = _execute
+    db.begin_nested.side_effect = _nested_ctx
+    return db
+
+
+def _find_call(db: MagicMock, needle: str) -> tuple[str, dict[str, Any]]:
+    for call in db.execute.call_args_list:
+        sql = str(call.args[0])
+        if needle in sql:
+            params = call.args[1] if len(call.args) > 1 else {}
+            return sql, params
+    raise AssertionError(f"SQL containing {needle!r} not found")
+
+
+def _kit_matcher() -> MagicMock:
+    matcher = MagicMock()
+    matcher.match_or_create_house.return_value = (101, 1.0, "new")
+    matcher.upsert_listing_source.return_value = None
+    return matcher
+
+
+def _lot(
+    source: str = "avito",
+    source_id: str = "1",
+    listing_segment: str | None = None,
+) -> KitLot:
+    return KitLot(
+        source=source,
+        source_url=f"https://www.{source}.ru/item/{source_id}",
+        source_id=source_id,
+        address="ул. Победы, 30",
+        listing_segment=listing_segment,
+        price_rub=3_000_000,
+    )
+
+
+# ── save_listings(...) — INSERT path ────────────────────────────────────────
+
+
+def test_save_listings_writes_listing_segment_into_insert_sql() -> None:
+    """listing_segment передаётся в SQL params И колонка есть в INSERT-списке."""
+    db = _mock_db_insert_path()
+    lot = _lot(listing_segment="vtorichka")
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66)
+
+    sql, params = _find_call(db, "INSERT INTO listings (")
+    assert "listing_segment" in sql, "listing_segment column must be in INSERT column list"
+    assert params["listing_segment"] == "vtorichka"
+
+
+def test_save_listings_listing_segment_none_backward_compat() -> None:
+    """Классификатор не определил сегмент (None) — INSERT всё равно проходит, NULL."""
+    db = _mock_db_insert_path()
+    lot = _lot(listing_segment=None)
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66)
+
+    _sql, params = _find_call(db, "INSERT INTO listings (")
+    assert params["listing_segment"] is None
+
+
+# ── ON CONFLICT DO UPDATE — COALESCE self-heal (главный фикс) ──────────────
+
+
+def test_save_listings_on_conflict_coalesces_listing_segment() -> None:
+    """ON CONFLICT DO UPDATE — listing_segment = COALESCE(EXCLUDED.listing_segment,
+    listings.listing_segment), не blind overwrite и не "никогда не обновляется"."""
+    db = _mock_db_insert_path()
+    lot = _lot(listing_segment="vtorichka")
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66)
+
+    sql, params = _find_call(db, "INSERT INTO listings (")
+    assert "listing_segment = COALESCE(" in sql
+    assert "EXCLUDED.listing_segment, listings.listing_segment" in sql
+    # Повторный скрейп с ОПРЕДЕЛЁННЫМ сегментом — новое значение уходит в EXCLUDED,
+    # COALESCE на стороне Postgres применит его к прежде-NULL строке (self-heal).
+    assert params["listing_segment"] == "vtorichka"
+
+
+def test_save_listings_on_conflict_listing_segment_none_does_not_blind_overwrite() -> None:
+    """Повторный скрейп БЕЗ сегмента (classifier снова None) — SQL всё равно
+    несёт COALESCE (не голый EXCLUDED), значит уже известный сегмент строки
+    в БД НЕ будет затёрт пустым при выполнении запроса."""
+    db = _mock_db_insert_path()
+    lot = _lot(listing_segment=None)
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66)
+
+    sql, params = _find_call(db, "INSERT INTO listings (")
+    assert "listing_segment = COALESCE(" in sql
+    assert "EXCLUDED.listing_segment, listings.listing_segment" in sql
+    assert params["listing_segment"] is None
+
+
+# ── Reconcile UPDATE (dedup_hash drift) — тот же self-heal ─────────────────
+
+
+def test_save_listings_reconcile_update_coalesces_listing_segment() -> None:
+    """dedup_hash-drift reconcile UPDATE path — тоже COALESCE(:listing_segment,
+    listing_segment), не blind overwrite. Без этого фикса строки, дошедшие до
+    reconcile (content дрейфит, старый dedup_hash не находится, INSERT ловит
+    UniqueViolation по (source, source_id)), остались бы незалеченными."""
+    import psycopg.errors
+    from sqlalchemy.exc import IntegrityError
+
+    uv_orig = psycopg.errors.UniqueViolation()
+    integrity_err = IntegrityError("INSERT INTO listings ...", {}, uv_orig)
+    rec_row = MagicMock()
+    rec_row.id = 88
+
+    db = MagicMock()
+
+    def _execute(sql: Any, params: dict[str, Any] | None = None) -> MagicMock:
+        s = str(sql)
+        res = MagicMock()
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
+            res.fetchone.return_value = None
+        elif "FROM listings_snapshots" in s:
+            res.fetchone.return_value = None
+        elif "INSERT INTO listings (" in s:
+            raise integrity_err
+        elif "UPDATE listings" in s and "SET dedup_hash" in s:
+            res.fetchone.return_value = rec_row
+        else:
+            res.fetchone.return_value = None
+        return res
+
+    db.execute.side_effect = _execute
+
+    @contextmanager
+    def _nested() -> Any:
+        try:
+            yield MagicMock()
+        except IntegrityError:
+            raise
+
+    db.begin_nested.side_effect = _nested
+
+    lot = _lot(source="avito", source_id="7960764619", listing_segment="novostroyki")
+
+    with patch("scraper_kit.base.upsert_listing_snapshot", return_value=None):
+        kit_save_listings(db, [lot], matcher=_kit_matcher(), region_code=66)
+
+    sql, params = _find_call(db, "SET dedup_hash")
+    assert "listing_segment = COALESCE(:listing_segment, listing_segment)" in sql
+    assert params["listing_segment"] == "novostroyki"

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

@@ -636,6 +636,19 @@ def save_listings(
                     is_rosreestr_checked = COALESCE(
                         EXCLUDED.is_rosreestr_checked, listings.is_rosreestr_checked
                     ),
+                    -- listing_segment раньше писался ТОЛЬКО при INSERT — строка, единожды
+                    -- родившаяся с NULL-сегментом (классификатор не сработал в первый
+                    -- скрейп), никогда не самочинилась на повторных сборах, даже когда
+                    -- сегмент становился определим. Симптом лечили отдельной джобой
+                    -- деактивации (deactivate_stale_{cian,yandex}_null_segment,
+                    -- миграция 266, PR #2908) — это чистит мусор, но не устраняет причину.
+                    -- COALESCE, а не голый EXCLUDED: если СЕЙЧАС скрейп снова не смог
+                    -- определить сегмент (EXCLUDED.listing_segment IS NULL), нельзя затирать
+                    -- уже известное значение пустым — та же защита, что и для
+                    -- city/kitchen/ceiling выше.
+                    listing_segment = COALESCE(
+                        EXCLUDED.listing_segment, listings.listing_segment
+                    ),
                     -- Yandex rich fields (+ shared description/agency/publish_date):
                     -- COALESCE so a source that does not provide them never wipes
                     -- a value previously written by another source / scrape.
@@ -738,6 +751,11 @@ def save_listings(
                                is_rosreestr_checked = COALESCE(
                                    :is_rosreestr_checked, is_rosreestr_checked
                                ),
+                               -- см. ON CONFLICT DO UPDATE выше — тот же self-heal для
+                               -- listing_segment, теперь и на reconcile-пути (dedup_hash
+                               -- drift). Без этого строки, прошедшие через reconcile,
+                               -- остались бы с тем же незалеченным NULL-сегментом.
+                               listing_segment = COALESCE(:listing_segment, listing_segment),
                                publish_date = COALESCE(:publish_date, publish_date),
                                days_on_market = COALESCE(:days_on_market, days_on_market),
                                description = COALESCE(:description, description),