From 197068c515689c788866dfa75b050aaabd01903e Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 26 Jun 2026 19:10:12 +0300 Subject: [PATCH] =?UTF-8?q?fix(scrapers):=20save=5Flistings=20=E2=80=94=20?= =?UTF-8?q?UPDATE=20by=20(source,source=5Fid)=20=D0=BF=D1=80=D0=B8=20dedup?= =?UTF-8?q?=5Fhash-=D0=B4=D1=80=D0=B5=D0=B9=D1=84=D0=B5=20(avito=5Ffull=5F?= =?UTF-8?q?load=20=D0=BA=D1=80=D0=B0=D1=88)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../backend/app/services/scrapers/base.py | 443 +++++++++++------- .../test_save_listings_upsert_reconcile.py | 327 +++++++++++++ 2 files changed, 601 insertions(+), 169 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_save_listings_upsert_reconcile.py diff --git a/tradein-mvp/backend/app/services/scrapers/base.py b/tradein-mvp/backend/app/services/scrapers/base.py index cef1c2a6..4cfd5893 100644 --- a/tradein-mvp/backend/app/services/scrapers/base.py +++ b/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, diff --git a/tradein-mvp/backend/tests/test_save_listings_upsert_reconcile.py b/tradein-mvp/backend/tests/test_save_listings_upsert_reconcile.py new file mode 100644 index 00000000..ddcb4385 --- /dev/null +++ b/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