fix(scrapers): save_listings — UPDATE by (source,source_id) при dedup_hash-дрейфе (avito_full_load краш)

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

@@ -25,8 +25,10 @@ from typing import Any
 from zoneinfo import ZoneInfo
 
 import httpx
+import psycopg.errors
 from pydantic import BaseModel, Field
 from sqlalchemy import text
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 from tenacity import retry, stop_after_attempt, wait_exponential
 
@@ -282,6 +284,7 @@ def save_listings(
     inserted = 0
     updated = 0
     skipped = 0
+    reconciled = 0  # UPDATE by (source,source_id) при dedup_hash-дрейфе
     matched = 0
     match_failures = 0
 
@@ -312,176 +315,277 @@ def save_listings(
                 skipped += 1
                 continue
 
-        result = db.execute(
-            text(
-                """
-                INSERT INTO listings (
-                    source, source_url, source_id, dedup_hash,
-                    address, lat, lon, region_code,
-                    rooms, area_m2, floor, total_floors, year_built,
-                    house_type, repair_state, has_balcony,
-                    house_source, house_ext_id, house_url, listing_segment,
-                    newbuilding_id, newbuilding_url,
-                    price_rub, price_per_m2,
-                    listing_date, publish_date, days_on_market, description,
-                    photo_urls, raw_payload,
-                    living_area_m2, bedrooms_count, balconies_count, loggias_count,
-                    description_minhash, cadastral_number, building_cadastral_number,
-                    phones, is_homeowner, is_pro_seller,
-                    bargain_allowed, sale_type, metro_stations, agency_name,
-                    yandex_offer_id, predicted_price_rub,
-                    predicted_price_min, predicted_price_max,
-                    price_trend, price_previous_rub,
-                    geo_precision, card_hash,
-                    scraped_at, last_seen_at
-                ) VALUES (
-                    :source, :source_url, :source_id, :dedup,
-                    :address, :lat, :lon, 66,
-                    :rooms, :area_m2, :floor, :total_floors, :year_built,
-                    :house_type, :repair_state, :has_balcony,
-                    :house_source, :house_ext_id, :house_url, :listing_segment,
-                    :newbuilding_id, :newbuilding_url,
-                    :price_rub, :ppm2,
-                    :listing_date, :publish_date, :days_on_market, :description,
-                    CAST(:photos AS jsonb),
-                    CAST(:raw AS jsonb),
-                    :living_area_m2, :bedrooms_count, :balconies_count, :loggias_count,
-                    :description_minhash, :cadastral_number, :building_cadastral_number,
-                    CAST(:phones AS jsonb), :is_homeowner, :is_pro_seller,
-                    :bargain_allowed, :sale_type, CAST(:metro_stations AS jsonb), :agency_name,
-                    :yandex_offer_id, :predicted_price_rub,
-                    :predicted_price_min, :predicted_price_max,
-                    :price_trend, :price_previous_rub,
-                    :geo_precision, :card_hash,
-                    NOW(), NOW()
-                )
-                -- Конфликт-арбитр — dedup_hash (sha256(source|source_id)).
-                -- Для одного (source, source_id) формула даёт ОДИН dedup_hash,
-                -- поэтому этот апсерт уже не вставит второй раз → согласован с
-                -- частичным UNIQUE (source, source_id) WHERE source_id IS NOT NULL
-                -- (133_listings_uq_source_source_id.sql). Менять арбитр на
-                -- (source, source_id) НЕЛЬЗЯ: yandex/url-only дают source_id=NULL,
-                -- а NULL не годится как conflict-target → дубли вернулись бы (#1773).
-                ON CONFLICT (dedup_hash) DO UPDATE
-                    SET last_seen_at = NOW(),
-                        is_active = true,
-                        -- если цена изменилась — обновляем
-                        price_rub = EXCLUDED.price_rub,
-                        price_per_m2 = EXCLUDED.price_per_m2,
-                        -- Cian-specific: обновляем при каждом re-scrape
-                        living_area_m2 = EXCLUDED.living_area_m2,
-                        bedrooms_count = EXCLUDED.bedrooms_count,
-                        balconies_count = EXCLUDED.balconies_count,
-                        loggias_count = EXCLUDED.loggias_count,
-                        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,
-                        listing_date = COALESCE(EXCLUDED.listing_date, listings.listing_date),
-                        area_m2 = COALESCE(EXCLUDED.area_m2, listings.area_m2),
-                        -- 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.
-                        publish_date = COALESCE(EXCLUDED.publish_date, listings.publish_date),
-                        days_on_market = COALESCE(
-                            EXCLUDED.days_on_market, listings.days_on_market
-                        ),
-                        description = COALESCE(EXCLUDED.description, listings.description),
-                        agency_name = COALESCE(EXCLUDED.agency_name, listings.agency_name),
-                        yandex_offer_id = COALESCE(
-                            EXCLUDED.yandex_offer_id, listings.yandex_offer_id
-                        ),
-                        predicted_price_rub = COALESCE(
-                            EXCLUDED.predicted_price_rub, listings.predicted_price_rub
-                        ),
-                        predicted_price_min = COALESCE(
-                            EXCLUDED.predicted_price_min, listings.predicted_price_min
-                        ),
-                        predicted_price_max = COALESCE(
-                            EXCLUDED.predicted_price_max, listings.predicted_price_max
-                        ),
-                        price_trend = COALESCE(EXCLUDED.price_trend, listings.price_trend),
-                        price_previous_rub = COALESCE(
-                            EXCLUDED.price_previous_rub, listings.price_previous_rub
-                        ),
-                        newbuilding_id = COALESCE(
-                            EXCLUDED.newbuilding_id, listings.newbuilding_id
-                        ),
-                        newbuilding_url = COALESCE(
-                            EXCLUDED.newbuilding_url, listings.newbuilding_url
-                        ),
-                        card_hash = EXCLUDED.card_hash
-                RETURNING id, (xmax = 0) AS inserted
-                """
-            ),
-            {
-                "source": lot.source,
-                "source_url": lot.source_url,
-                "source_id": lot.source_id,
-                "dedup": dedup,
-                "address": lot.address,
-                "lat": lot.lat,
-                "lon": lot.lon,
-                "rooms": lot.rooms,
-                "area_m2": lot.area_m2,
-                "floor": lot.floor,
-                "total_floors": lot.total_floors,
-                "year_built": lot.year_built,
-                "house_type": lot.house_type,
-                "repair_state": lot.repair_state,
-                "has_balcony": lot.has_balcony,
-                "house_source": lot.house_source,
-                "house_ext_id": lot.house_ext_id,
-                "house_url": lot.house_url,
-                "listing_segment": lot.listing_segment,
-                "newbuilding_id": lot.newbuilding_id,
-                "newbuilding_url": lot.newbuilding_url,
-                "price_rub": lot.price_rub,
-                "ppm2": ppm2,
-                "listing_date": lot.listing_date,
-                "publish_date": lot.publish_date,
-                "days_on_market": lot.days_on_market,
-                "description": lot.description,
-                "photos": _to_json(lot.photo_urls),
-                "raw": _to_json(lot.raw_payload) if lot.raw_payload else None,
-                # Cian-specific columns — None for non-Cian scrapers (→ SQL NULL)
-                "living_area_m2": lot.living_area_m2,
-                "bedrooms_count": lot.bedrooms_count,
-                "balconies_count": lot.balconies_count,
-                "loggias_count": lot.loggias_count,
-                "description_minhash": lot.description_minhash,
-                "cadastral_number": lot.cadastral_number,
-                "building_cadastral_number": lot.building_cadastral_number,
-                "phones": _to_json(lot.phones) if lot.phones else None,
-                "is_homeowner": lot.is_homeowner,
-                "is_pro_seller": lot.is_pro_seller,
-                "bargain_allowed": lot.bargain_allowed,
-                "sale_type": lot.sale_type,
-                "metro_stations": _to_json(lot.metro_stations) if lot.metro_stations else None,
-                "agency_name": lot.agency_name,
-                "yandex_offer_id": lot.yandex_offer_id,
-                "predicted_price_rub": lot.predicted_price_rub,
-                "predicted_price_min": lot.predicted_price_min,
-                "predicted_price_max": lot.predicted_price_max,
-                "price_trend": lot.price_trend,
-                "price_previous_rub": lot.price_previous_rub,
-                "geo_precision": lot.geo_precision,
-                "card_hash": card_hash,
-            },
-        ).fetchone()
+        # ── Per-lot SAVEPOINT: изолируем listings-upsert ────────────────────
+        # Когда контент существующего ad дрейфит → новый dedup_hash → INSERT
+        # находит prior_row=None (SELECT по старому dedup_hash пуст) → INSERT
+        # летит в UniqueViolation по (source, source_id) при source_id IS NOT NULL
+        # (constraint 133_listings_uq_source_source_id). Без SAVEPOINT это роняет
+        # весь прогон. С SAVEPOINT ловим IntegrityError, rollback SAVEPOINT'а,
+        # и делаем прямой UPDATE по (source, source_id) — reconcile drifted row.
+        # source_id IS NULL (yandex) → constraint не фаерит, обычный путь.
+        _upsert_params = {
+            "source": lot.source,
+            "source_url": lot.source_url,
+            "source_id": lot.source_id,
+            "dedup": dedup,
+            "address": lot.address,
+            "lat": lot.lat,
+            "lon": lot.lon,
+            "rooms": lot.rooms,
+            "area_m2": lot.area_m2,
+            "floor": lot.floor,
+            "total_floors": lot.total_floors,
+            "year_built": lot.year_built,
+            "house_type": lot.house_type,
+            "repair_state": lot.repair_state,
+            "has_balcony": lot.has_balcony,
+            "house_source": lot.house_source,
+            "house_ext_id": lot.house_ext_id,
+            "house_url": lot.house_url,
+            "listing_segment": lot.listing_segment,
+            "newbuilding_id": lot.newbuilding_id,
+            "newbuilding_url": lot.newbuilding_url,
+            "price_rub": lot.price_rub,
+            "ppm2": ppm2,
+            "listing_date": lot.listing_date,
+            "publish_date": lot.publish_date,
+            "days_on_market": lot.days_on_market,
+            "description": lot.description,
+            "photos": _to_json(lot.photo_urls),
+            "raw": _to_json(lot.raw_payload) if lot.raw_payload else None,
+            # Cian-specific columns — None for non-Cian scrapers (→ SQL NULL)
+            "living_area_m2": lot.living_area_m2,
+            "bedrooms_count": lot.bedrooms_count,
+            "balconies_count": lot.balconies_count,
+            "loggias_count": lot.loggias_count,
+            "description_minhash": lot.description_minhash,
+            "cadastral_number": lot.cadastral_number,
+            "building_cadastral_number": lot.building_cadastral_number,
+            "phones": _to_json(lot.phones) if lot.phones else None,
+            "is_homeowner": lot.is_homeowner,
+            "is_pro_seller": lot.is_pro_seller,
+            "bargain_allowed": lot.bargain_allowed,
+            "sale_type": lot.sale_type,
+            "metro_stations": _to_json(lot.metro_stations) if lot.metro_stations else None,
+            "agency_name": lot.agency_name,
+            "yandex_offer_id": lot.yandex_offer_id,
+            "predicted_price_rub": lot.predicted_price_rub,
+            "predicted_price_min": lot.predicted_price_min,
+            "predicted_price_max": lot.predicted_price_max,
+            "price_trend": lot.price_trend,
+            "price_previous_rub": lot.price_previous_rub,
+            "geo_precision": lot.geo_precision,
+            "card_hash": card_hash,
+        }
 
+        _upsert_sql = text(
+            """
+            INSERT INTO listings (
+                source, source_url, source_id, dedup_hash,
+                address, lat, lon, region_code,
+                rooms, area_m2, floor, total_floors, year_built,
+                house_type, repair_state, has_balcony,
+                house_source, house_ext_id, house_url, listing_segment,
+                newbuilding_id, newbuilding_url,
+                price_rub, price_per_m2,
+                listing_date, publish_date, days_on_market, description,
+                photo_urls, raw_payload,
+                living_area_m2, bedrooms_count, balconies_count, loggias_count,
+                description_minhash, cadastral_number, building_cadastral_number,
+                phones, is_homeowner, is_pro_seller,
+                bargain_allowed, sale_type, metro_stations, agency_name,
+                yandex_offer_id, predicted_price_rub,
+                predicted_price_min, predicted_price_max,
+                price_trend, price_previous_rub,
+                geo_precision, card_hash,
+                scraped_at, last_seen_at
+            ) VALUES (
+                :source, :source_url, :source_id, :dedup,
+                :address, :lat, :lon, 66,
+                :rooms, :area_m2, :floor, :total_floors, :year_built,
+                :house_type, :repair_state, :has_balcony,
+                :house_source, :house_ext_id, :house_url, :listing_segment,
+                :newbuilding_id, :newbuilding_url,
+                :price_rub, :ppm2,
+                :listing_date, :publish_date, :days_on_market, :description,
+                CAST(:photos AS jsonb),
+                CAST(:raw AS jsonb),
+                :living_area_m2, :bedrooms_count, :balconies_count, :loggias_count,
+                :description_minhash, :cadastral_number, :building_cadastral_number,
+                CAST(:phones AS jsonb), :is_homeowner, :is_pro_seller,
+                :bargain_allowed, :sale_type, CAST(:metro_stations AS jsonb), :agency_name,
+                :yandex_offer_id, :predicted_price_rub,
+                :predicted_price_min, :predicted_price_max,
+                :price_trend, :price_previous_rub,
+                :geo_precision, :card_hash,
+                NOW(), NOW()
+            )
+            -- Конфликт-арбитр — dedup_hash (sha256(source|source_id)).
+            -- Для одного (source, source_id) формула даёт ОДИН dedup_hash,
+            -- поэтому этот апсерт уже не вставит второй раз → согласован с
+            -- частичным UNIQUE (source, source_id) WHERE source_id IS NOT NULL
+            -- (133_listings_uq_source_source_id.sql). Менять арбитр на
+            -- (source, source_id) НЕЛЬЗЯ: yandex/url-only дают source_id=NULL,
+            -- а NULL не годится как conflict-target → дубли вернулись бы (#1773).
+            ON CONFLICT (dedup_hash) DO UPDATE
+                SET last_seen_at = NOW(),
+                    is_active = true,
+                    -- если цена изменилась — обновляем
+                    price_rub = EXCLUDED.price_rub,
+                    price_per_m2 = EXCLUDED.price_per_m2,
+                    -- Cian-specific: обновляем при каждом re-scrape
+                    living_area_m2 = EXCLUDED.living_area_m2,
+                    bedrooms_count = EXCLUDED.bedrooms_count,
+                    balconies_count = EXCLUDED.balconies_count,
+                    loggias_count = EXCLUDED.loggias_count,
+                    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,
+                    listing_date = COALESCE(EXCLUDED.listing_date, listings.listing_date),
+                    area_m2 = COALESCE(EXCLUDED.area_m2, listings.area_m2),
+                    -- 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.
+                    publish_date = COALESCE(EXCLUDED.publish_date, listings.publish_date),
+                    days_on_market = COALESCE(
+                        EXCLUDED.days_on_market, listings.days_on_market
+                    ),
+                    description = COALESCE(EXCLUDED.description, listings.description),
+                    agency_name = COALESCE(EXCLUDED.agency_name, listings.agency_name),
+                    yandex_offer_id = COALESCE(
+                        EXCLUDED.yandex_offer_id, listings.yandex_offer_id
+                    ),
+                    predicted_price_rub = COALESCE(
+                        EXCLUDED.predicted_price_rub, listings.predicted_price_rub
+                    ),
+                    predicted_price_min = COALESCE(
+                        EXCLUDED.predicted_price_min, listings.predicted_price_min
+                    ),
+                    predicted_price_max = COALESCE(
+                        EXCLUDED.predicted_price_max, listings.predicted_price_max
+                    ),
+                    price_trend = COALESCE(EXCLUDED.price_trend, listings.price_trend),
+                    price_previous_rub = COALESCE(
+                        EXCLUDED.price_previous_rub, listings.price_previous_rub
+                    ),
+                    newbuilding_id = COALESCE(
+                        EXCLUDED.newbuilding_id, listings.newbuilding_id
+                    ),
+                    newbuilding_url = COALESCE(
+                        EXCLUDED.newbuilding_url, listings.newbuilding_url
+                    ),
+                    card_hash = EXCLUDED.card_hash
+            RETURNING id, (xmax = 0) AS inserted
+            """
+        )
+
+        # ── Per-lot SAVEPOINT вокруг listings-upsert ────────────────────────
+        # Когда контент существующего ad дрейфит → новый dedup_hash → INSERT
+        # не находит existing row по dedup_hash → пробует fresh INSERT → летит
+        # UniqueViolation по (source, source_id) WHERE source_id IS NOT NULL
+        # (constraint listings_source_source_id_uq). Без SAVEPOINT это переводит
+        # весь Session в InFailedSqlTransaction → весь прогон падает.
+        # С SAVEPOINT ловим IntegrityError/UniqueViolation, делаем rollback,
+        # и применяем UPDATE by (source, source_id) — reconcile drifted row.
+        # source_id IS NULL (yandex) → constraint не фаерит, обычный upsert путь.
+        result = None
         listing_id: int | None = None
-        if result is not None:
-            listing_id = int(result.id)
-            if result.inserted:
-                inserted += 1
+        try:
+            with db.begin_nested():
+                result = db.execute(_upsert_sql, _upsert_params).fetchone()
+        except IntegrityError as ie:
+            # Проверяем что это UniqueViolation именно по (source, source_id).
+            if lot.source_id is not None and isinstance(ie.orig, psycopg.errors.UniqueViolation):
+                logger.warning(
+                    "save_listings:dedup_hash_drift source=%s source_id=%s "
+                    "— reconcile UPDATE by (source, source_id)",
+                    lot.source,
+                    lot.source_id,
+                )
+                # SAVEPOINT уже откачен; session чистая — применяем прямой UPDATE.
+                rec_row = db.execute(
+                    text(
+                        """
+                        UPDATE listings
+                           SET dedup_hash = :dedup,
+                               last_seen_at = NOW(),
+                               is_active = true,
+                               price_rub = :price_rub,
+                               price_per_m2 = :ppm2,
+                               living_area_m2 = :living_area_m2,
+                               bedrooms_count = :bedrooms_count,
+                               balconies_count = :balconies_count,
+                               loggias_count = :loggias_count,
+                               description_minhash = :description_minhash,
+                               cadastral_number = :cadastral_number,
+                               building_cadastral_number = :building_cadastral_number,
+                               phones = CAST(:phones AS jsonb),
+                               is_homeowner = :is_homeowner,
+                               is_pro_seller = :is_pro_seller,
+                               bargain_allowed = :bargain_allowed,
+                               sale_type = :sale_type,
+                               metro_stations = CAST(:metro_stations AS jsonb),
+                               listing_date = COALESCE(:listing_date, listing_date),
+                               area_m2 = COALESCE(:area_m2, area_m2),
+                               publish_date = COALESCE(:publish_date, publish_date),
+                               days_on_market = COALESCE(:days_on_market, days_on_market),
+                               description = COALESCE(:description, description),
+                               agency_name = COALESCE(:agency_name, agency_name),
+                               yandex_offer_id = COALESCE(:yandex_offer_id, yandex_offer_id),
+                               predicted_price_rub = COALESCE(
+                                   :predicted_price_rub, predicted_price_rub
+                               ),
+                               predicted_price_min = COALESCE(
+                                   :predicted_price_min, predicted_price_min
+                               ),
+                               predicted_price_max = COALESCE(
+                                   :predicted_price_max, predicted_price_max
+                               ),
+                               price_trend = COALESCE(:price_trend, price_trend),
+                               price_previous_rub = COALESCE(
+                                   :price_previous_rub, price_previous_rub
+                               ),
+                               newbuilding_id = COALESCE(:newbuilding_id, newbuilding_id),
+                               newbuilding_url = COALESCE(:newbuilding_url, newbuilding_url),
+                               card_hash = :card_hash
+                         WHERE source = :source
+                           AND source_id = :source_id
+                        RETURNING id
+                        """
+                    ),
+                    _upsert_params,
+                ).fetchone()
+                if rec_row is not None:
+                    listing_id = int(rec_row.id)
+                    reconciled += 1
+                else:
+                    logger.error(
+                        "save_listings:reconcile_failed source=%s source_id=%s "
+                        "— UPDATE returned no row (unexpected)",
+                        lot.source,
+                        lot.source_id,
+                    )
             else:
-                updated += 1
+                logger.error(
+                    "save_listings:integrity_error source=%s source_id=%s: %s",
+                    lot.source,
+                    lot.source_id,
+                    ie,
+                )
+                raise
+        else:
+            if result is not None:
+                listing_id = int(result.id)
+                if result.inserted:
+                    inserted += 1
+                else:
+                    updated += 1
 
         # ── Snapshot: point-in-time observation in listings_snapshots ───
         # Fault-tolerant: failure here MUST NOT abort the listings batch.
@@ -554,11 +658,12 @@ def save_listings(
 
     db.commit()
     logger.info(
-        "save_listings: source=%s inserted=%d updated=%d skipped_seen_today=%d "
-        "matched=%d match_failures=%d (total %d)",
+        "save_listings: source=%s inserted=%d updated=%d reconciled=%d "
+        "skipped_seen_today=%d matched=%d match_failures=%d (total %d)",
         lots[0].source if lots else "?",
         inserted,
         updated,
+        reconciled,
         skipped,
         matched,
         match_failures,

tradein-mvp/backend/tests/test_save_listings_upsert_reconcile.py

@@ -0,0 +1,327 @@
+"""Тесты Fix 1: save_listings — SAVEPOINT вокруг INSERT + reconcile по (source,source_id).
+
+Корень бага: при dedup_hash-дрейфе (контент ad изменился, новый source_id не меняется)
+INSERT не находит existing row по dedup_hash → пробует свежий INSERT → UniqueViolation
+по listings_source_source_id_uq (source, source_id) → весь прогон падает.
+
+Фикс: per-lot SAVEPOINT + IntegrityError catch + UPDATE by (source, source_id).
+"""
+
+from __future__ import annotations
+
+import os
+from contextlib import contextmanager
+from unittest.mock import MagicMock, patch
+
+import psycopg.errors
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from app.services.scrapers.base import ScrapedLot, save_listings
+
+# ── Fixtures ─────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture(autouse=True)
+def _patch_matching():
+    """Stub matching + snapshot so tests focus on the upsert/reconcile logic."""
+    with (
+        patch("app.services.scrapers.base.match_or_create_house") as m_house,
+        patch("app.services.scrapers.base.upsert_listing_source") as m_link,
+        patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap,
+    ):
+        m_house.return_value = (101, 1.0, "new")
+        m_link.return_value = None
+        m_snap.return_value = None
+        yield {"house": m_house, "link": m_link, "snap": m_snap}
+
+
+def _base_lot(**overrides) -> ScrapedLot:
+    defaults = {
+        "source": "avito",
+        "source_url": "https://www.avito.ru/ekaterinburg/test/1",
+        "source_id": "7960764619",
+        "price_rub": 5_000_000,
+    }
+    defaults.update(overrides)
+    return ScrapedLot(**defaults)
+
+
+@contextmanager
+def _nested_ctx():
+    yield MagicMock()
+
+
+def _mock_db_normal(inserted: bool = True, listing_id: int = 42) -> MagicMock:
+    """DB mock для нормального пути (нет UniqueViolation)."""
+    insert_row = MagicMock()
+    insert_row.id = listing_id
+    insert_row.inserted = inserted
+
+    db = MagicMock()
+
+    def _execute(sql, params=None):
+        s = str(sql)
+        res = MagicMock()
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
+            res.fetchone.return_value = None  # нет prior row → fresh insert
+        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 _mock_db_with_unique_violation(reconcile_id: int = 77) -> MagicMock:
+    """DB mock: первый INSERT вызывает UniqueViolation (dedup_hash drift).
+
+    Симулирует кейс: INSERT по новому dedup_hash → коллизия по (source, source_id).
+    Reconcile UPDATE by (source, source_id) → возвращает id=reconcile_id.
+    """
+    uv_orig = psycopg.errors.UniqueViolation()
+    integrity_err = IntegrityError(
+        statement="INSERT INTO listings ...",
+        params={},
+        orig=uv_orig,
+    )
+
+    rec_row = MagicMock()
+    rec_row.id = reconcile_id
+
+    db = MagicMock()
+
+    def _execute(sql, params=None):
+        s = str(sql)
+        res = MagicMock()
+        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
+            # prior_row=None → INSERT будет сделан (no prior dedup_hash match)
+            res.fetchone.return_value = None
+        elif "FROM listings_snapshots" in s:
+            res.fetchone.return_value = None
+        elif "INSERT INTO listings (" in s:
+            # Имитируем исключение внутри begin_nested() context
+            raise integrity_err
+        elif "UPDATE listings" in s and "SET dedup_hash" in s:
+            # reconcile UPDATE by (source, source_id)
+            res.fetchone.return_value = rec_row
+        elif "UPDATE listings" in s and "house_id_fk" in s:
+            res.fetchone.return_value = None
+        else:
+            res.fetchone.return_value = None
+        return res
+
+    db.execute.side_effect = _execute
+
+    # begin_nested() нужен для per-lot SAVEPOINT + snapshot + matching hook
+    # Используем реальный contextmanager который НЕ перехватывает IntegrityError
+    # (savepoint rollback симулируется тем, что db.execute после rollback'а
+    # уже вызывает reconcile UPDATE, а не снова INSERT).
+    savepoint_entered = [False]
+
+    @contextmanager
+    def _nested():
+        savepoint_entered[0] = True
+        try:
+            yield MagicMock()
+        except IntegrityError:
+            # SAVEPOINT rollback — пробрасываем наверх (SQLAlchemy делает rollback)
+            raise
+
+    db.begin_nested.side_effect = _nested
+    return db
+
+
+# ── Tests ─────────────────────────────────────────────────────────────────────
+
+
+def test_normal_insert_path_unchanged():
+    """Нормальный путь (нет UniqueViolation) → inserted=1, updated=0."""
+    db = _mock_db_normal(inserted=True, listing_id=10)
+    lot = _base_lot()
+
+    inserted, updated = save_listings(db, [lot])
+
+    assert inserted == 1
+    assert updated == 0
+
+
+def test_normal_update_path_unchanged():
+    """Нормальный update path (ON CONFLICT DO UPDATE) → inserted=0, updated=1."""
+    insert_row = MagicMock()
+    insert_row.id = 42
+    insert_row.inserted = False  # xmax != 0 → update
+
+    db = MagicMock()
+
+    def _execute(sql, params=None):
+        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
+
+    lot = _base_lot(source_id="upd_test")
+    inserted, updated = save_listings(db, [lot])
+
+    assert inserted == 0
+    assert updated == 1
+
+
+def test_null_source_id_no_savepoint_needed():
+    """source_id=None (yandex) → обычный INSERT, UniqueViolation не может возникнуть.
+
+    Проверяем, что нормальный путь не изменился при source_id=None.
+    """
+    db = _mock_db_normal(inserted=True, listing_id=99)
+    lot = _base_lot(
+        source="yandex",
+        source_url="https://realty.yandex.ru/offer/123/",
+        source_id=None,
+    )
+
+    inserted, updated = save_listings(db, [lot])
+
+    assert inserted == 1
+    assert updated == 0
+
+
+def test_unique_violation_triggers_reconcile_update(caplog, _patch_matching):
+    """UniqueViolation по (source, source_id) → SAVEPOINT rollback → UPDATE by (s,sid).
+
+    После reconcile UPDATE: listing_id определён, matching hook вызывается.
+    inserted=0, updated=0 (reconciled не входит в возврат, но лог есть).
+    """
+    import logging
+
+    db = _mock_db_with_unique_violation(reconcile_id=77)
+    lot = _base_lot(source="avito", source_id="7960764619")
+
+    with caplog.at_level(logging.WARNING, logger="app.services.scrapers.base"):
+        inserted, updated = save_listings(db, [lot])
+
+    # Основные counters: reconcile не добавляет к inserted/updated
+    assert inserted == 0
+    assert updated == 0
+
+    # Лог предупреждения о reconcile
+    warns = [r for r in caplog.records if "dedup_hash_drift" in r.getMessage()]
+    assert len(warns) == 1
+    assert "7960764619" in warns[0].getMessage()
+
+    # Matching hook должен вызваться (listing_id=77 из reconcile UPDATE)
+    assert _patch_matching["link"].call_count == 1
+
+
+def test_unique_violation_reconcile_uses_update_by_source_source_id():
+    """Reconcile UPDATE содержит WHERE source = :source AND source_id = :source_id."""
+    db = _mock_db_with_unique_violation(reconcile_id=55)
+    lot = _base_lot(source="cian", source_id="330200428")
+
+    save_listings(db, [lot])
+
+    # Ищем вызов reconcile UPDATE (содержит SET dedup_hash)
+    reconcile_calls = [
+        c
+        for c in db.execute.call_args_list
+        if "UPDATE listings" in str(c.args[0]) and "SET dedup_hash" in str(c.args[0])
+    ]
+    assert len(reconcile_calls) == 1
+    _sql, params = str(reconcile_calls[0].args[0]), reconcile_calls[0].args[1]
+    assert "WHERE source = :source" in _sql
+    assert "AND source_id = :source_id" in _sql
+    assert params["source"] == "cian"
+    assert params["source_id"] == "330200428"
+
+
+def test_reconcile_includes_new_dedup_hash():
+    """Reconcile UPDATE обновляет dedup_hash на новый (дрейфнувший) хэш."""
+    db = _mock_db_with_unique_violation(reconcile_id=66)
+    lot = _base_lot(source="avito", source_id="7960764619", price_rub=5_500_000)
+
+    expected_dedup = lot.compute_dedup_hash()
+    save_listings(db, [lot])
+
+    reconcile_calls = [
+        c
+        for c in db.execute.call_args_list
+        if "UPDATE listings" in str(c.args[0]) and "SET dedup_hash" in str(c.args[0])
+    ]
+    assert len(reconcile_calls) == 1
+    params = reconcile_calls[0].args[1]
+    assert params["dedup"] == expected_dedup
+
+
+def test_reconcile_run_does_not_crash_full_batch(_patch_matching):
+    """Один лот с UniqueViolation в batch из 3 → остальные 2 успешно обрабатываются.
+
+    Ключевая проверка что один битый lot НЕ роняет прогон.
+    """
+    rec_row = MagicMock()
+    rec_row.id = 77
+
+    ok_insert_row = MagicMock()
+    ok_insert_row.id = 88
+    ok_insert_row.inserted = True
+
+    uv_orig = psycopg.errors.UniqueViolation()
+    integrity_err = IntegrityError("INSERT", {}, uv_orig)
+
+    def _execute(sql, params=None):
+        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:
+            if params and params.get("source_id") == "CONFLICT_ID":
+                raise integrity_err
+            res.fetchone.return_value = ok_insert_row
+        elif "UPDATE listings" in s and "SET dedup_hash" in s:
+            res.fetchone.return_value = rec_row
+        elif "UPDATE listings" in s and "house_id_fk" in s:
+            res.fetchone.return_value = None
+        else:
+            res.fetchone.return_value = None
+        return res
+
+    db = MagicMock()
+    db.execute.side_effect = _execute
+
+    @contextmanager
+    def _nested():
+        try:
+            yield MagicMock()
+        except IntegrityError:
+            raise
+
+    db.begin_nested.side_effect = _nested
+
+    lot_conflict = _base_lot(source_id="CONFLICT_ID")
+    lot_ok1 = _base_lot(source_id="OK_1", source_url="https://www.avito.ru/ok1")
+    lot_ok2 = _base_lot(source_id="OK_2", source_url="https://www.avito.ru/ok2")
+
+    inserted, updated = save_listings(db, [lot_conflict, lot_ok1, lot_ok2])
+
+    # 2 нормальных INSERT + 1 reconcile (не считается в inserted/updated)
+    assert inserted == 2
+    assert updated == 0
+    # Все 3 лота дошли до matching hook
+    assert _patch_matching["link"].call_count == 3