Merge pull request 'fix(scrapers): avito_full_load краш (source_id-reconcile) + cian/yandex ban-ротация' (#1921) from fix/scraper-resilience-upsert-rotation into main

Reviewed-on: #1921

tradein-mvp/backend/app/core/config.py

@@ -369,6 +369,9 @@ class Settings(BaseSettings):
     # changeip-ссылка для Cian-прокси (ротация IP при бане/таймауте). Если не задан —
     # fallback на avito_proxy_rotate_url. ENV: CIAN_PROXY_ROTATE_URL.
     cian_proxy_rotate_url: str | None = None
+    # Максимум IP-ротаций для Cian на один sweep-прогон. Аналог avito_proxy_max_rotations.
+    # ENV: CIAN_PROXY_MAX_ROTATIONS.
+    cian_proxy_max_rotations: int = 4
 
     @property
     def cian_proxy_url(self) -> str | None:
@@ -383,6 +386,9 @@ class Settings(BaseSettings):
     # changeip-ссылка для Yandex-прокси (ротация IP при капче/таймауте). Если не задан —
     # fallback на avito_proxy_rotate_url. ENV: YANDEX_PROXY_ROTATE_URL.
     yandex_proxy_rotate_url: str | None = None
+    # Максимум IP-ротаций для Yandex на один sweep-прогон. Аналог avito_proxy_max_rotations.
+    # ENV: YANDEX_PROXY_MAX_ROTATIONS.
+    yandex_proxy_max_rotations: int = 4
 
     @property
     def yandex_proxy_url(self) -> str | None:

tradein-mvp/backend/app/services/scrape_pipeline.py

@@ -53,17 +53,31 @@ from app.services.yandex_price_history import record_yandex_price_history
 logger = logging.getLogger(__name__)
 
 
-async def _rotate_proxy_ip(*, reason: str, rotations_done: int) -> bool:
-    """Вызвать changeip-ссылку mobileproxy и подождать settle (#1790).
+async def _rotate_proxy_ip(*, reason: str, rotations_done: int, source: str = "avito") -> bool:
+    """Вызвать changeip-ссылку mobileproxy и подождать settle (#1790/#1848).
+
+    Provider-aware: выбирает rotate_url и max_rotations по параметру `source`
+    (avito / cian / yandex). Fallback-цепочка для rotate_url:
+      avito: avito_proxy_rotate_url
+      cian:  cian_proxy_rotate_url → avito_proxy_rotate_url
+      yandex: yandex_proxy_rotate_url → avito_proxy_rotate_url
 
     Аналог AvitoScraper._rotate_ip, но на уровне pipeline — используется в
-    enrichment-фазах (houses + detail) где нет доступа к экземпляру scraper'а.
+    enrichment-фазах где нет доступа к экземпляру scraper'а.
 
     Returns True при успешной смене IP, False если rotate_url не задан или ошибка.
-    Логирует каждую ротацию (причина, порядковый номер, остаток лимита).
+    Логирует каждую ротацию (причина, порядковый номер, source, остаток лимита).
     """
-    rotate_url = settings.avito_proxy_rotate_url
-    max_rot = settings.avito_proxy_max_rotations
+    if source == "cian":
+        rotate_url = settings.cian_proxy_rotate_url or settings.avito_proxy_rotate_url
+        max_rot = settings.cian_proxy_max_rotations
+    elif source == "yandex":
+        rotate_url = settings.yandex_proxy_rotate_url or settings.avito_proxy_rotate_url
+        max_rot = settings.yandex_proxy_max_rotations
+    else:
+        # avito (default) — backward compat
+        rotate_url = settings.avito_proxy_rotate_url
+        max_rot = settings.avito_proxy_max_rotations
     if not rotate_url:
         return False
     sep = "&" if "?" in rotate_url else "?"
@@ -78,7 +92,9 @@ async def _rotate_proxy_ip(*, reason: str, rotations_done: int) -> bool:
             pass
         await asyncio.sleep(settings.avito_proxy_rotate_settle_s)
         logger.info(
-            "pipeline: IP rotated via changeip — reason=%s rotation=#%d remaining=%d new_ip=%s",
+            "pipeline: IP rotated via changeip — source=%s reason=%s "
+            "rotation=#%d remaining=%d new_ip=%s",
+            source,
             reason,
             rotations_done + 1,
             max_rot - rotations_done - 1,
@@ -87,7 +103,8 @@ async def _rotate_proxy_ip(*, reason: str, rotations_done: int) -> bool:
         return True
     except Exception:
         logger.warning(
-            "pipeline: IP rotation failed (reason=%s, rotation=#%d)",
+            "pipeline: IP rotation failed (source=%s reason=%s rotation=#%d)",
+            source,
             reason,
             rotations_done + 1,
             exc_info=True,
@@ -759,9 +776,7 @@ async def run_avito_city_sweep(
                         scraper._browser = shared_bf
                         # Shared-browser режим: _cffi=None → curl_cffi-fallback на
                         # firewall пропускается (by design). Логируем для observability.
-                        logger.debug(
-                            "avito pipeline-mode: shared browser only, no cffi fallback"
-                        )
+                        logger.debug("avito pipeline-mode: shared browser only, no cffi fallback")
                     else:
                         scraper._cffi = session
 
@@ -1482,6 +1497,7 @@ async def run_yandex_city_sweep(
     enrich_delay = request_delay_sec if request_delay_sec is not None else 3.0
     _resolved_delay = request_delay_sec if request_delay_sec is not None else 9.0
     consecutive_failures = 0
+    yandex_rotations_done = 0  # #1848: бюджет IP-ротаций на весь sweep
 
     # Вычисляем watchdog-таймаут для combos-режима (центр, anchors=None).
     # В combos-режиме один "anchor" выполняет num_segments × num_combos × max_pages
@@ -1565,7 +1581,7 @@ async def run_yandex_city_sweep(
                 каждого (room×price) combo, сохраняет порцию сразу в БД + обновляет heartbeat.
                 anchor_lots накапливается (через _al) для address-enrich и price-history.
                 """
-                nonlocal consecutive_failures
+                nonlocal consecutive_failures, yandex_rotations_done
 
                 # ── Phase 1+2: SERP + инкрементальный save per-combo ──────
                 def _on_combo(
@@ -1675,19 +1691,25 @@ async def run_yandex_city_sweep(
                 if items:
                     from curl_cffi.requests import AsyncSession as _AsyncSession
 
+                    # #1848: rotation tracking для Yandex address-enrich (stuck-proxy)
+                    _yandex_enrich_consec_fail = 0
+                    _yandex_enrich_abort = 3  # abort address-enrich после N подряд
+
                     async with _AsyncSession(
                         impersonate="chrome120",
                         timeout=30.0,
                         proxies=_proxies,
                         headers={"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8"},
                     ) as enrich_session:
-                        for eidx, item in enumerate(items):
+                        eidx = 0
+                        while eidx < len(items):
+                            item = items[eidx]
                             lid: int = item["id"]
                             old_addr: str = item["address"]
                             url: str = item["source_url"]
                             try:
                                 resp = await enrich_session.get(url, allow_redirects=True)
-                                if resp.status_code != 200:
+                                if resp.status_code not in {200}:
                                     logger.warning(
                                         "yandex-sweep run_id=%d address-enrich:"
                                         " HTTP %d listing_id=%d",
@@ -1696,6 +1718,11 @@ async def run_yandex_city_sweep(
                                         lid,
                                     )
                                     counters.address_failed += 1
+                                    # 503/429 = прокси заблокирован → считаем как failure
+                                    if resp.status_code in {429, 503}:
+                                        _yandex_enrich_consec_fail += 1
+                                    else:
+                                        _yandex_enrich_consec_fail = 0
                                 else:
                                     new_addr = _extract_address_from_title(resp.text)
                                     if (
@@ -1741,17 +1768,53 @@ async def run_yandex_city_sweep(
                                                 db.rollback()
                                             except Exception:
                                                 pass
+                                    _yandex_enrich_consec_fail = 0
                             except Exception as fetch_exc:
                                 counters.address_failed += 1
+                                _yandex_enrich_consec_fail += 1
                                 logger.warning(
                                     "yandex-sweep run_id=%d: address-enrich "
-                                    "fetch failed listing_id=%d: %s",
+                                    "fetch failed listing_id=%d "
+                                    "(consecutive=%d): %s",
                                     run_id,
                                     lid,
+                                    _yandex_enrich_consec_fail,
                                     fetch_exc,
                                 )
+                            # Ротация Yandex-прокси при stuck/ban (#1848)
+                            if _yandex_enrich_consec_fail >= _yandex_enrich_abort:
+                                if yandex_rotations_done < settings.yandex_proxy_max_rotations:
+                                    rotated = await _rotate_proxy_ip(
+                                        reason=(
+                                            f"yandex enrich consecutive="
+                                            f"{_yandex_enrich_consec_fail}"
+                                        ),
+                                        rotations_done=yandex_rotations_done,
+                                        source="yandex",
+                                    )
+                                    yandex_rotations_done += 1
+                                    if rotated:
+                                        _yandex_enrich_consec_fail = 0
+                                        logger.info(
+                                            "yandex-sweep run_id=%d: enrich ROTATE — "
+                                            "IP changed, reset consecutive "
+                                            "(rotation #%d/%d)",
+                                            run_id,
+                                            yandex_rotations_done,
+                                            settings.yandex_proxy_max_rotations,
+                                        )
+                                        # retry текущего item после ротации
+                                        continue
+                                logger.error(
+                                    "yandex-sweep run_id=%d: address-enrich ABORT — "
+                                    "%d consecutive failures (proxy likely stuck/banned)",
+                                    run_id,
+                                    _yandex_enrich_consec_fail,
+                                )
+                                break
                             if eidx < len(items) - 1:
                                 await asyncio.sleep(enrich_delay)
+                            eidx += 1
 
                     logger.info(
                         "yandex-sweep run_id=%d anchor %s: address enrich=%d/%d failed=%d",
@@ -1954,6 +2017,7 @@ async def run_cian_city_sweep(
     _anchors = anchors if anchors is not None else EKB_ANCHORS
     counters = CianCitySweepCounters(anchors_total=len(_anchors))
     consecutive_failures = 0
+    cian_rotations_done = 0  # #1848: бюджет IP-ротаций на весь sweep
 
     try:
         for idx, (lat, lon, name) in enumerate(_anchors, start=1):
@@ -1994,7 +2058,7 @@ async def run_cian_city_sweep(
                 _a_name: str = _c_name,
             ) -> None:
                 """Все фазы одного cian anchor'а (SERP + detail + houses)."""
-                nonlocal anchor_lots, consecutive_failures
+                nonlocal anchor_lots, consecutive_failures, cian_rotations_done
                 from sqlalchemy import text as _text
 
                 # ── Phase 1+2: SERP + save ─────────────────────────────────
@@ -2051,7 +2115,12 @@ async def run_cian_city_sweep(
                         .mappings()
                         .all()
                     )
-                    for didx, row in enumerate(priority_rows):
+                    # #1848: Cian bans/stuck-proxy tracking + rotation (зеркало avito)
+                    _cian_detail_consec_failures = 0
+                    _cian_detail_abort = 3  # abort detail-фазы после N подряд
+                    didx = 0
+                    while didx < len(priority_rows):
+                        row = priority_rows[didx]
                         listing_id: int = row["id"]
                         source_url: str = row["source_url"]
                         counters.detail_attempted += 1
@@ -2069,18 +2138,54 @@ async def run_cian_city_sweep(
                                     listing_id,
                                     source_url,
                                 )
+                            _cian_detail_consec_failures = 0
                         except Exception as exc:
                             counters.detail_failed += 1
                             counters.errors_count += 1
+                            _cian_detail_consec_failures += 1
                             logger.warning(
-                                "cian-sweep run_id=%d: detail failed listing_id=%d: %s",
+                                "cian-sweep run_id=%d: detail failed listing_id=%d "
+                                "(consecutive=%d): %s",
                                 run_id,
                                 listing_id,
+                                _cian_detail_consec_failures,
                                 exc,
                             )
+                            if _cian_detail_consec_failures >= _cian_detail_abort:
+                                # Попытка ротации Cian-прокси перед abort'ом (#1848)
+                                if cian_rotations_done < settings.cian_proxy_max_rotations:
+                                    rotated = await _rotate_proxy_ip(
+                                        reason=(
+                                            f"cian detail consecutive="
+                                            f"{_cian_detail_consec_failures}"
+                                        ),
+                                        rotations_done=cian_rotations_done,
+                                        source="cian",
+                                    )
+                                    cian_rotations_done += 1
+                                    if rotated:
+                                        _cian_detail_consec_failures = 0
+                                        logger.info(
+                                            "cian-sweep run_id=%d: detail ROTATE — "
+                                            "IP changed, reset consecutive "
+                                            "(rotation #%d/%d)",
+                                            run_id,
+                                            cian_rotations_done,
+                                            settings.cian_proxy_max_rotations,
+                                        )
+                                        # retry текущего detail (не инкрементируем didx)
+                                        continue
+                                logger.error(
+                                    "cian-sweep run_id=%d: detail ABORT — "
+                                    "%d consecutive failures (proxy likely banned/stuck)",
+                                    run_id,
+                                    _cian_detail_consec_failures,
+                                )
+                                break
                         if didx < len(priority_rows) - 1:
                             jitter = random.uniform(0.8, 1.2)
                             await asyncio.sleep(request_delay_sec * jitter)
+                        didx += 1
                     logger.info(
                         "cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d",
                         run_id,

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_provider_rotation_1848.py

@@ -0,0 +1,242 @@
+"""Тесты Fix 2: provider-aware _rotate_proxy_ip (#1848).
+
+Тест 1: source='cian' → cian_proxy_rotate_url используется.
+Тест 2: source='yandex' → yandex_proxy_rotate_url используется.
+Тест 3: source='avito' (default) → avito_proxy_rotate_url (backward compat).
+Тест 4: cian без cian url → fallback на avito_proxy_rotate_url.
+Тест 5: rotate_url=None → returns False (no HTTP call).
+Тест 6: network error → returns False (no crash).
+Тест 7: yandex fallback на avito.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from app.services.scrape_pipeline import _rotate_proxy_ip
+
+# ── Helpers ──────────────────────────────────────────────────────────────────
+
+
+def _async_session_ctx(resp_json: dict | None = None):
+    """Возвращает (async_cm_class, session_mock) для curl_cffi AsyncSession."""
+    session = AsyncMock()
+    if resp_json is not None:
+        mock_resp = MagicMock()
+        mock_resp.json.return_value = resp_json
+        session.get = AsyncMock(return_value=mock_resp)
+    else:
+        session.get = AsyncMock(side_effect=RuntimeError("network error"))
+
+    # curl_cffi AsyncSession используется как `async with AsyncSession(...) as rot:`
+    # Нам нужен callable класс-замена, вызов которого возвращает async context manager.
+    class _MockAsyncSession:
+        def __init__(self, timeout=30):
+            pass
+
+        async def __aenter__(self):
+            return session
+
+        async def __aexit__(self, *_):
+            pass
+
+    return _MockAsyncSession, session
+
+
+# ── Тест 1: cian_proxy_rotate_url ────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_cian_uses_cian_url() -> None:
+    """source='cian' → cian_proxy_rotate_url (не avito_proxy_rotate_url)."""
+    mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "1.2.3.4"})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+    ):
+        s.cian_proxy_rotate_url = "http://cian-rotate.test/"
+        s.avito_proxy_rotate_url = "http://avito-rotate.test/"
+        s.yandex_proxy_rotate_url = None
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian")
+
+    assert result is True
+    called_url: str = session.get.call_args[0][0]
+    assert "cian-rotate.test" in called_url
+    assert "avito-rotate.test" not in called_url
+
+
+# ── Тест 2: yandex_proxy_rotate_url ─────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_yandex_uses_yandex_url() -> None:
+    """source='yandex' → yandex_proxy_rotate_url."""
+    mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "5.6.7.8"})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+    ):
+        s.yandex_proxy_rotate_url = "http://yandex-rotate.test/"
+        s.avito_proxy_rotate_url = "http://avito-rotate.test/"
+        s.cian_proxy_rotate_url = None
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        result = await _rotate_proxy_ip(reason="timeout", rotations_done=0, source="yandex")
+
+    assert result is True
+    called_url: str = session.get.call_args[0][0]
+    assert "yandex-rotate.test" in called_url
+    assert "avito-rotate.test" not in called_url
+
+
+# ── Тест 3: avito backward compat ────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_avito_backward_compat() -> None:
+    """source не передан (дефолт 'avito') → avito_proxy_rotate_url."""
+    mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "9.9.9.9"})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+    ):
+        s.avito_proxy_rotate_url = "http://avito-only.test/"
+        s.cian_proxy_rotate_url = None
+        s.yandex_proxy_rotate_url = None
+        s.avito_proxy_max_rotations = 4
+        s.cian_proxy_max_rotations = 4
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        # Вызов без source= — должен работать как раньше
+        result = await _rotate_proxy_ip(reason="block", rotations_done=0)
+
+    assert result is True
+    called_url: str = session.get.call_args[0][0]
+    assert "avito-only.test" in called_url
+
+
+# ── Тест 4: cian fallback на avito когда нет cian url ─────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_cian_fallback_to_avito() -> None:
+    """cian_proxy_rotate_url=None → fallback на avito_proxy_rotate_url."""
+    mock_session_cls, session = _async_session_ctx(resp_json={})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+    ):
+        s.cian_proxy_rotate_url = None
+        s.avito_proxy_rotate_url = "http://avito-fallback.test/"
+        s.yandex_proxy_rotate_url = None
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian")
+
+    assert result is True
+    called_url: str = session.get.call_args[0][0]
+    assert "avito-fallback.test" in called_url
+
+
+# ── Тест 5: rotate_url=None → returns False ──────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_returns_false_when_no_url() -> None:
+    """Никакой URL не задан → False, HTTP-запрос не делается."""
+    mock_session_cls, session = _async_session_ctx(resp_json={})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+    ):
+        s.cian_proxy_rotate_url = None
+        s.avito_proxy_rotate_url = None
+        s.yandex_proxy_rotate_url = None
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.yandex_proxy_max_rotations = 4
+
+        result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian")
+
+    assert result is False
+    # HTTP-запрос не делался
+    session.get.assert_not_called()
+
+
+# ── Тест 6: network error → returns False ────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_returns_false_on_network_error() -> None:
+    """Ошибка сети → returns False (не крашит caller)."""
+    mock_session_cls, _session = _async_session_ctx(resp_json=None)  # raises RuntimeError
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+    ):
+        s.cian_proxy_rotate_url = "http://cian-rotate.test/"
+        s.avito_proxy_rotate_url = None
+        s.yandex_proxy_rotate_url = None
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        result = await _rotate_proxy_ip(reason="ban", rotations_done=0, source="cian")
+
+    assert result is False
+
+
+# ── Тест 7: yandex fallback на avito ─────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_rotate_proxy_ip_yandex_fallback_to_avito() -> None:
+    """yandex_proxy_rotate_url=None → fallback на avito_proxy_rotate_url."""
+    mock_session_cls, session = _async_session_ctx(resp_json={"new_ip": "11.22.33.44"})
+
+    with (
+        patch("app.services.scrape_pipeline.settings") as s,
+        patch("app.services.scrape_pipeline.AsyncSession", mock_session_cls),
+        patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()),
+    ):
+        s.yandex_proxy_rotate_url = None
+        s.avito_proxy_rotate_url = "http://shared-proxy.test/"
+        s.cian_proxy_rotate_url = None
+        s.yandex_proxy_max_rotations = 4
+        s.avito_proxy_max_rotations = 4
+        s.cian_proxy_max_rotations = 4
+        s.avito_proxy_rotate_settle_s = 0.0
+
+        result = await _rotate_proxy_ip(reason="503", rotations_done=0, source="yandex")
+
+    assert result is True
+    called_url: str = session.get.call_args[0][0]
+    assert "shared-proxy.test" in called_url

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