"""Базовый framework для парсеров объявлений (Avito / Cian / DomKlik / ...). scraper_kit-копия (strangler #2132) базового слоя `app.services.scrapers.base`. Развязана от `app.*` через Protocol-инжекцию (см. `scraper_kit.contracts`): `save_listings` принимает `matcher: HouseMatcher` вместо прямого импорта `app.services.matching`, а `region_code` передаётся параметром (хардкод `66` убран — дефолт задаёт вызывающая продуктовая сторона). Пока НИКТО из боевого кода это не импортирует; старый `app.services.scrapers.base` остаётся боевым. Общие задачи: - httpx.AsyncClient с realistic browser headers + UA rotation - tenacity retry с exp backoff - Sleep между запросами (anti-ban) - ScrapedLot — единая Pydantic схема всех источников - Запись в `listings` Postgres с дедупом по dedup_hash - После INSERT/UPDATE — hook в matching service (инжектируемый `HouseMatcher`) для регистрации в `listing_sources` и привязки к каноническому `houses`. Каждый конкретный парсер (avito.py, cian.py, ...) наследуется от BaseScraper и реализует `_fetch_page()` и `_parse_listings()`. """ from __future__ import annotations import asyncio import hashlib import logging import math import random from abc import ABC, abstractmethod from datetime import date, datetime, timedelta from typing import TYPE_CHECKING, 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 from scraper_kit.ceiling_height import plausible_ceiling_m from scraper_kit.snapshot_writer import upsert_listing_snapshot if TYPE_CHECKING: from scraper_kit.contracts import HouseMatcher logger = logging.getLogger(__name__) _MSK = ZoneInfo("Europe/Moscow") # ── Realistic browser User-Agents (Apr 2026 versions) ────────────────────── _USER_AGENTS: list[str] = [ # Chrome on macOS "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36", # noqa: E501 # Chrome on Windows "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36", # noqa: E501 # Firefox on macOS "Mozilla/5.0 (Macintosh; Intel Mac OS X 14.5; rv:127.0) Gecko/20100101 Firefox/127.0", # Safari on macOS "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Safari/605.1.15", # noqa: E501 # Edge on Windows "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36 Edg/132.0.0.0", # noqa: E501 ] def random_user_agent() -> str: return random.choice(_USER_AGENTS) # ── Правдоподобность listing_date/publish_date ────────────────────────────── _MIN_PLAUSIBLE_LISTING_DATE = date(2010, 1, 1) def _clamp_plausible_date(d: date | None) -> date | None: """Reject implausible listing/publish dates (epoch-sentinel, far-future, ancient). Providers occasionally emit 1970-01-01 (Avito 0-epoch), near-future timestamps, or garbled years. Anything outside [2010-01-01, today+2d] → None (column is nullable). Single choke point so every provider (avito/cian/yandex/domclick) is covered by one guard. LOW audit R2 (#6). """ if d is None: return None if d < _MIN_PLAUSIBLE_LISTING_DATE or d > date.today() + timedelta(days=2): return None return d # ── Унифицированная схема результата ──────────────────────────────────────── class ScrapedLot(BaseModel): """Результат парсинга одного объявления. Все источники приводятся к этому виду перед записью в Postgres. Поля optional если источник их не отдаёт. """ source: str # 'avito' / 'cian' / 'domklik' / ... source_url: str # ссылка на оригинал — кликается из UI source_id: str | None = None # внутренний id источника (для update) # Локация — должна быть хотя бы address ИЛИ (lat, lon) address: str | None = None lat: float | None = None lon: float | None = None # Параметры квартиры rooms: int | None = None # 0 = студия area_m2: float | None = None floor: int | None = None total_floors: int | None = None year_built: int | None = None house_type: str | None = None repair_state: str | None = None has_balcony: bool | None = None kadastr_num: str | None = None # Detail-поля кухни/потолка (#2007). Обычно из detail-страницы (avito_detail, # cian_detail, yandex_detail), но yandex SERP отдаёт их прямо в карточке — # промоутим из raw_payload в колонки для дашборда покрытия и matching. kitchen_area_m2: float | None = None ceiling_height_m: float | None = None # Class B promote-поля (#2008, cian SERP). Для дашборда покрытия / matching, # НЕ для valuation — estimator их пока НЕ читает (follow-up #2012). cian отдаёт # их прямо в карточке; раньше лежали только в raw_payload. mortgage_available: bool | None = None # продавец допускает ипотеку is_apartments: bool | None = None # апартаменты (НЕ квартира — иной правовой статус) is_rosreestr_checked: bool | None = None # объявление сверено с ЕГРН # Avito house linking (Stage 2a — search → houses) house_source: str | None = None # 'avito' / 'cian' house_ext_id: str | None = None # Avito's '3171365' house_url: str | None = None # full URL of /catalog/houses/... listing_segment: str | None = None # 'vtorichka' / 'novostroyki' # Newbuilding (ЖК) linking — slug-id + canonical ЖК-subdomain URL newbuilding_id: str | None = None # e.g. 'federatsiya-ekaterinburg' newbuilding_url: str | None = None # e.g. https://zhk-federatsiya-ekaterinburg.avito.ru # Цена (обязательно) price_rub: int = Field(gt=0) price_per_m2: int | None = None # Геокодинг # None → точность неизвестна (координаты пришли из скрейпера напрямую). # 'city' → геокодер вернул city-centroid (нет номера дома в адресе) — # листинг исключается из radius-аналогов estimator'а. geo_precision: str | None = None # Метаданные listing_date: date | None = None publish_date: date | None = None days_on_market: int | None = None description: str | None = None photo_urls: list[str] = Field(default_factory=list) raw_payload: dict[str, Any] | None = None # Cian-specific extensions (Stage 2 of CianScraper v1) living_area_m2: float | None = None bedrooms_count: int | None = None balconies_count: int | None = None loggias_count: int | None = None description_minhash: str | None = None cadastral_number: str | None = None building_cadastral_number: str | None = None phones: list[dict] = Field(default_factory=list) is_homeowner: bool | None = None is_pro_seller: bool | None = None bargain_allowed: bool | None = None sale_type: str | None = None metro_stations: list[dict] = Field(default_factory=list) agency_name: str | None = None # Yandex gate-API rich fields yandex_offer_id: str | None = None # numeric string offerId (kept as text) predicted_price_rub: int | None = None # Yandex per-offer valuation (predictedPrice.value) predicted_price_min: int | None = None predicted_price_max: int | None = None price_trend: str | None = None # INCREASED / DECREASED / UNCHANGED / ... price_previous_rub: int | None = None # price.previous def compute_dedup_hash(self) -> str: """SHA256(source + stable_key) — стабильный uniqueness key. stable_key = source_id (Cian offer_id 330200428, Avito data-item-id) если есть; иначе source_url БЕЗ query-string. Cian/Avito URL несут волатильный `?context=...` токен — без strip каждый ре-скрейп давал бы новый hash. #753: price_rub в ключ НЕ входит. Смена цены того же объявления — это UPDATE существующей строки (`ON CONFLICT (dedup_hash) DO UPDATE`), а не новый объект. Раньше цена была частью ключа → каждое изменение цены порождало дубль активного listing → искажение asking_to_sold_ratio / sales-vs-listings / активного инвентаря. Смена формулы разово даёт новый hash существующим строкам — следующий скрейп upsert'нет их по новому ключу (всплеск разовый). """ h = hashlib.sha256() h.update(self.source.encode()) h.update(b"|") key = self.source_id if self.source_id else (self.source_url or "").split("?")[0] h.update((key or "").encode()) return h.hexdigest() def compute_card_hash(self) -> str: """SHA-256 over the volatile SERP-card fields. Used to detect whether a listing's card content changed between scrapes (snapshot-dedup, detail-refresh gate). Excludes last_seen/scrape timestamps and detail-only fields. Хэшируем детерминированную нормализованную репрезентацию полей карточки: price_rub, area_m2, floor, total_floors, rooms, address (strip+lower, None→""), len(photo_urls). Порядок частей фиксирован (нет зависимости от порядка итерации dict) → стабильный hash при неизменной карточке. """ address = (self.address or "").strip().lower() parts = [ str(self.price_rub), str(self.area_m2), str(self.floor), str(self.total_floors), str(self.rooms), address, str(len(self.photo_urls)), ] return hashlib.sha256("|".join(parts).encode()).hexdigest() def compute_price_per_m2(self) -> int | None: """price_per_m2 = price_rub / area_m2 если area задана.""" if self.area_m2 and self.area_m2 > 0: return int(self.price_rub / self.area_m2) return None # ── BaseScraper ───────────────────────────────────────────────────────────── class BaseScraper(ABC): """Базовый класс — наследовать для каждого источника. Subclass должен реализовать: - `name` (class attr) — 'avito' / 'cian' / ... - `_fetch_page(client, ...)` — async GET к источнику - `_parse_listings(html_or_json, ...)` — html → list[ScrapedLot] """ name: str = "base" base_url: str = "" request_delay_sec: float = 5.0 # sleep между запросами (anti-ban) def __init__(self) -> None: self._client: httpx.AsyncClient | None = None async def __aenter__(self) -> BaseScraper: self._client = httpx.AsyncClient( timeout=20.0, follow_redirects=True, headers={ "User-Agent": random_user_agent(), "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1", }, ) return self async def __aexit__(self, *args: Any) -> None: if self._client is not None: await self._client.aclose() @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=2, min=2, max=30)) async def _http_get(self, url: str, **kwargs: Any) -> httpx.Response: """GET с retry. Subclass-ы используют это вместо raw httpx.""" assert self._client is not None, "Use scraper as async context manager" response = await self._client.get(url, **kwargs) # 403/429 — ретрай поможет (рандомный UA в новом контексте), 5xx тоже if response.status_code in {403, 429, 502, 503, 504}: response.raise_for_status() return response async def sleep_between_requests(self) -> None: """Случайный jitter в районе self.request_delay_sec — чтобы парсер не выглядел как detstvo-bot.""" jitter = random.uniform(0.7, 1.5) await asyncio.sleep(self.request_delay_sec * jitter) @abstractmethod async def fetch_around(self, lat: float, lon: float, radius_m: int = 1000) -> list[ScrapedLot]: """Главный метод — собрать объявления вокруг (lat, lon) в радиусе radius_m. Returns: Список ScrapedLot. Может быть пустым. """ ... def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float: """Ортодромическое расстояние (км) между двумя точками (сферическая Земля). Используется гео-guard'ом `save_listings(..., city_anchor=..., city_radius_km=...)` (соседний-город-в-развёртке): сверяет физические координаты лота с anchor'ом города-цели без ST_DWithin/PostGIS round-trip — чистая математика, лот уже в памяти (lat/lon — Python float на ScrapedLot). """ r_earth_km = 6371.0 p1, p2 = math.radians(lat1), math.radians(lat2) d_phi = math.radians(lat2 - lat1) d_lmb = math.radians(lon2 - lon1) a = math.sin(d_phi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(d_lmb / 2) ** 2 return 2 * r_earth_km * math.asin(math.sqrt(a)) # ── Запись пачки результатов в Postgres ───────────────────────────────────── def save_listings( db: Session, lots: list[ScrapedLot], *, matcher: HouseMatcher, region_code: int, run_id: int | None = None, skip_seen_today: bool = False, city: str | None = None, city_anchor: tuple[float, float] | None = None, city_radius_km: float | None = None, ) -> tuple[int, int]: """Пишем list[ScrapedLot] в `listings` с upsert по dedup_hash. Дополнительно пишет point-in-time snapshot в listings_snapshots (ON CONFLICT DO UPDATE — берётся последний за день). Ошибка записи snapshot НЕ прерывает основной INSERT. Args: db: SQLAlchemy session. lots: список распарсенных объявлений. matcher: инжектируемый `HouseMatcher` (scraper_kit.contracts) — кросс- источниковый матчинг домов/листингов. Заменяет прямой импорт `app.services.matching` (strangler-развязка). Продуктовая сторона передаёт `RealMatcherAdapter` (см. app.services.scraper_adapters). region_code: код региона для колонки `listings.region_code` (ранее хардкод `66` в SQL). Дефолт задаёт вызывающая сторона, не kit. run_id: FK scrape_runs.id текущего run'а (опционально; None при ad-hoc вызовах). skip_seen_today: если True и листинг уже обновлялся сегодня по МСК (last_seen_at >= начало сегодняшнего дня), то пропустить upsert и snapshot. Используется full_load для экономии redundant upsert + price-trigger churn при повторном прогоне в тот же день. Новые листинги (prior_row=None) всегда вставляются. False = старое поведение (всегда upsert). city: человекочитаемое имя города-цели ЭТОГО batch'а (#2594), например "Нижний Тагил"/"Екатеринбург". Развёртка знает город из своего контекста (city_slug) — ОДИН на весь вызов save_listings (все lots одного batch'а принадлежат одному city-sweep run'у), поэтому это kwarg, а НЕ поле ScrapedLot. None (default) — вызывающая сторона город не знает (ad-hoc admin/manual пути) — колонка остаётся NULL, backward-compatible. ON CONFLICT — COALESCE (новое значение НЕ затирает уже известный город NULL'ом, если какой-то caller ещё не передаёт city). city_anchor: (lat, lon) референсной точки города-цели этого batch'а — гео-guard (соседний-город-в-развёртке, замер на проде: 97% лотов, помеченных "Верхняя Пышма" из yandex-развёртки, физически лежат в Екатеринбурге — anchor города-цели всего ~15км от центра ЕКБ, а провайдер (см. `providers/yandex/serp.py:fetch_around`) скоупит выдачу city-scoped rgid, НЕ радиусом (`lat/lon/radius_m` для yandex gate-API инертны) — rgid захватывает весь ЕКБ вместе с ближним спутником). Если задан ВМЕСТЕ с `city_radius_km` — для каждого лота С координатами (lot.lat/lot.lon НЕ None) считаем haversine- расстояние до `city_anchor`; лот ДАЛЬШЕ `city_radius_km` НЕ получает `city` этого batch'а (пишется NULL, а не угадывается чужой город). Лоты БЕЗ координат — city проставляется как обычно (нечем сверить; провайдер уже скоупил SERP/API-запрос на этот город через city_slug/rgid/region_id — см. resolve_city_name в orchestration/pipeline.py). None/None (default, backward-compatible) — guard выключен, старое поведение (ОДИН city на весь batch без проверки координат) — так вызывается EKB-развёртка (нет в регионе города КРУПНЕЕ ЕКБ, чей SERP мог бы её "поглотить"). city_radius_km: см. `city_anchor` — оба параметра включают guard ТОЛЬКО вместе. Returns: (inserted, updated) — counters для логов. """ if not lots: return 0, 0 inserted = 0 updated = 0 skipped = 0 reconciled = 0 # UPDATE by (source,source_id) при dedup_hash-дрейфе matched = 0 match_failures = 0 geo_guard_dropped = 0 # city NULL'ен из-за geo-guard (лот вне city_radius_km от anchor'а) today_msk = datetime.now(_MSK).date() _geo_guard_active = city is not None and city_anchor is not None and city_radius_km is not None for lot in lots: ppm2 = lot.price_per_m2 or lot.compute_price_per_m2() dedup = lot.compute_dedup_hash() card_hash = lot.compute_card_hash() # ── Гео-guard: соседний-город-в-развёртке (см. docstring city_anchor) ── # Лот С координатами дальше city_radius_km от anchor'а — НЕ доверяем city # этого batch'а (пишем NULL, не текущий-но-неверный город). Лоты БЕЗ координат # проверить нечем — city проставляется как обычно (см. docstring). lot_city = city if _geo_guard_active and lot.lat is not None and lot.lon is not None: assert city_anchor is not None and city_radius_km is not None # narrow for mypy dist_km = _haversine_km(lot.lat, lot.lon, city_anchor[0], city_anchor[1]) if dist_km > city_radius_km: lot_city = None geo_guard_dropped += 1 logger.debug( "save_listings:geo_guard_dropped source=%s dedup=%s dist_km=%.1f " "> city_radius_km=%.1f (target_city=%s anchor=%s)", lot.source, dedup, dist_km, city_radius_km, city, city_anchor, ) # Pre-read the existing row's card_hash and last_seen_at (keyed by # dedup_hash) BEFORE the upsert — needed to know the *prior* card # content and to implement skip_seen_today logic. # The upsert's RETURNING can only expose the NEW values (EXCLUDED.*), # never the old ones, so a lightweight SELECT is the clean way. # None → row did not exist yet (fresh insert) → always process. prior_row = db.execute( text("SELECT card_hash, last_seen_at FROM listings WHERE dedup_hash = :dedup"), {"dedup": dedup}, ).fetchone() prior_card_hash = prior_row.card_hash if prior_row is not None else None # skip_seen_today: пропускаем листинги уже обновлённые сегодня по МСК. # Только для existing строк (prior_row is not None); новые всегда вставляются. if skip_seen_today and prior_row is not None and prior_row.last_seen_at is not None: ls_msk = prior_row.last_seen_at.astimezone(_MSK).date() if ls_msk == today_msk: skipped += 1 continue # ── 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, "region_code": region_code, "address": lot.address, "city": lot_city, "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, "kitchen_area_m2": lot.kitchen_area_m2, # #2699: единый гейт правдоподобия на границе записи — какой бы # провайдер ни наполнил lot, невозможная высота в БД не попадёт. "ceiling_height_m": plausible_ceiling_m(lot.ceiling_height_m), "mortgage_available": lot.mortgage_available, "is_apartments": lot.is_apartments, "is_rosreestr_checked": lot.is_rosreestr_checked, "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, # Clamp implausible dates (Avito 1970-epoch, near-future garbage) — LOW audit R2 (#6) "listing_date": _clamp_plausible_date(lot.listing_date), "publish_date": _clamp_plausible_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, city, lat, lon, region_code, rooms, area_m2, floor, total_floors, year_built, house_type, repair_state, has_balcony, kitchen_area_m2, ceiling_height_m, mortgage_available, is_apartments, is_rosreestr_checked, 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, :city, :lat, :lon, :region_code, :rooms, :area_m2, :floor, :total_floors, :year_built, :house_type, :repair_state, :has_balcony, -- ceiling: одна колонка ceiling_height_m (#2699). До этого тот же -- param писался ещё и в ceiling_height (019) — дубль, из-за которого -- источники разъехались по двум колонкам, а эстиматор читал одну. :kitchen_area_m2, :ceiling_height_m, :mortgage_available, :is_apartments, :is_rosreestr_checked, :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, -- scraped_at / last_seen_at — statement_timestamp(), НЕ NOW() (#2731). -- NOW() == transaction_timestamp() замерзает на старте транзакции, а -- save_listings коммитит один раз в конце всего batch'а → все строки -- прогона несли ОДНУ метку (прод 2026-08-06: 219 строк — 1 метка). -- statement_timestamp() двигается построчно (каждый upsert — свой -- statement) и СТАБИЛЕН внутри statement'а, поэтому обе колонки -- получают одно и то же значение. clock_timestamp() здесь нельзя: -- он вычисляется на каждый вызов отдельно и развёл бы scraped_at и -- last_seen_at на микросекунды — расхождение там, где его нет. statement_timestamp(), statement_timestamp() ) -- Конфликт-арбитр — 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 = statement_timestamp(), -- #2206: ре-подтверждение живым = свежий скрейп; без bump'а -- эстиматор (scraped_at > NOW()-14d) терял ~55% живого инвентаря. -- #2731: обе метки — statement_timestamp() (см. VALUES выше); -- их равенство сохраняется, потому что оно стабильно в statement'е. scraped_at = statement_timestamp(), 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), -- #2777: ДОзаполнение адреса — порядок аргументов обратный остальным, -- существующее значение выигрывает. Адрес не обновлялся при конфликте -- вообще: строка, вставленная без адреса (SERP-вариант его не дал), -- оставалась безадресной НАВСЕГДА, даже когда следующий скрейп адрес -- приносил. Прод 2026-08-10: 862 строки с address IS NULL, у 95 из них -- при этом ЕСТЬ house_id_fk — матчинг в тот раз получил адрес и сматчил -- корректно (fingerprint/new без адреса невозможны), в колонке же -- остался NULL, и он же кормит геокодер мусором. Перезаписывать НЕЛЬЗЯ: -- миграции 062/108/124 чистят listings.address, свежий сырой адрес от -- площадки молча откатил бы эту чистку. address = COALESCE(listings.address, EXCLUDED.address), -- #2594: город развёртки — COALESCE, чтобы caller без city (ad-hoc -- admin/manual пути, city=None) не затирал уже известный город. city = COALESCE(EXCLUDED.city, listings.city), -- kitchen/ceiling (#2007): COALESCE — SERP re-scrape источника без -- этих полей (avito SERP → NULL) НЕ затирает detail-enriched значение -- (avito_detail пишет ceiling_height_m отдельным UPDATE). kitchen_area_m2 = COALESCE( EXCLUDED.kitchen_area_m2, listings.kitchen_area_m2 ), ceiling_height_m = COALESCE( EXCLUDED.ceiling_height_m, listings.ceiling_height_m ), -- Class B promote (#2008, cian SERP): COALESCE — re-scrape -- источника без поля (avito/yandex SERP → NULL) НЕ затирает -- значение, ранее записанное cian-ом / detail-ом. mortgage_available = COALESCE( EXCLUDED.mortgage_available, listings.mortgage_available ), is_apartments = COALESCE( EXCLUDED.is_apartments, listings.is_apartments ), is_rosreestr_checked = COALESCE( EXCLUDED.is_rosreestr_checked, listings.is_rosreestr_checked ), -- 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 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 = statement_timestamp(), -- #2206: ре-подтверждение живым = свежий скрейп; -- без bump'а эстиматор терял ~55% живого инвентаря. -- #2731: statement_timestamp() — построчная метка, -- одинаковая для обеих колонок (см. INSERT выше). scraped_at = statement_timestamp(), 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), -- #2777: см. ON CONFLICT выше — дозаполняем адрес, -- существующее значение выигрывает. address = COALESCE(address, :address), city = COALESCE(:city, city), kitchen_area_m2 = COALESCE(:kitchen_area_m2, kitchen_area_m2), ceiling_height_m = COALESCE(:ceiling_height_m, ceiling_height_m), mortgage_available = COALESCE( :mortgage_available, mortgage_available ), is_apartments = COALESCE(:is_apartments, is_apartments), is_rosreestr_checked = COALESCE( :is_rosreestr_checked, is_rosreestr_checked ), 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: 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. # ON CONFLICT DO UPDATE — самый последний snapshot за день. # # Snapshot-dedup: пропускаем запись только когда МЫ УВЕРЕНЫ что карточка # не изменилась (prior_card_hash == card_hash, row существовал) И snapshot # за сегодня уже есть. Принцип "when in doubt — write": лишний snapshot # неизменной карточки безвреден, ПРОПУЩЕННЫЙ snapshot реального изменения — # недопустим. Поэтому equality-проверки мало (первый скрейп за день мог # ещё не записать сегодняшний snapshot) — дополнительно подтверждаем # наличие сегодняшней строки лёгким EXISTS. if listing_id is not None: skip_snapshot = False if prior_card_hash is not None and prior_card_hash == card_hash: today_row = db.execute( text( "SELECT 1 FROM listings_snapshots " "WHERE listing_id = CAST(:lid AS bigint) " " AND snapshot_date = CURRENT_DATE" ), {"lid": listing_id}, ).fetchone() skip_snapshot = today_row is not None if skip_snapshot: logger.debug( "save_listings:snapshot_skipped (card unchanged) " "source=%s listing_id=%s", lot.source, listing_id, ) if listing_id is not None and not skip_snapshot: try: with db.begin_nested(): upsert_listing_snapshot( db, listing_id=listing_id, price_rub=lot.price_rub, price_per_m2=ppm2, run_id=run_id, status="active", ) except Exception as e: logger.warning( "save_listings:snapshot_failed source=%s listing_id=%s: %s", lot.source, listing_id, e, ) # ── Hook: link listing → house via matching service ───────────── # Idempotent (UPSERT on listing_sources UNIQUE(ext_source, ext_id)) # and fault-tolerant: failure here MUST NOT abort the listings batch. # Wrapped in SAVEPOINT so a failed match only rolls back its own work, # not the surrounding INSERT (see .claude/rules/backend.md SAVEPOINT). if listing_id is not None: try: with db.begin_nested(): _link_listing_to_house(db, listing_id, lot, matcher, city=lot_city) matched += 1 except Exception as e: # Best-effort hook: log and continue so the listings batch isn't aborted. match_failures += 1 logger.warning( "save_listings:match_failed source=%s source_id=%s listing_id=%s: %s", lot.source, lot.source_id, listing_id, e, ) db.commit() logger.info( "save_listings: source=%s inserted=%d updated=%d reconciled=%d " "skipped_seen_today=%d matched=%d match_failures=%d geo_guard_dropped=%d (total %d)", lots[0].source if lots else "?", inserted, updated, reconciled, skipped, matched, match_failures, geo_guard_dropped, len(lots), ) return inserted, updated def _to_json(value: Any) -> str: """JSON serialization helper — для jsonb колонок.""" import json return json.dumps(value, ensure_ascii=False, default=str) def _link_listing_to_house( db: Session, listing_id: int, lot: ScrapedLot, matcher: HouseMatcher, *, city: str | None = None ) -> None: """Hook scraped listing into matching service: resolve house, upsert listing_sources. Steps: 1. matcher.match_or_create_house() — find or create canonical houses row for this listing's address/coords. Uses Tier 0-3 matching (cadastr → source_exact → fingerprint → geo_proximity → new). 2. matcher.upsert_listing_source() — register (ext_source, ext_id) → listing_id in listing_sources. Idempotent via UNIQUE (ext_source, ext_id). The matching method recorded is 'source_link': scraper already deduped by `dedup_hash` (INSERT … ON CONFLICT DO UPDATE), so this is a direct registration rather than a fuzzy listing-level match. ext_id source: lot.source_id if present, else dedup_hash (Yandex without stable source_id falls back to URL-based dedup_hash — same hash on re-scrape). `city` — город-цель ЭТОГО batch'а после гео-guard'а (`lot_city` в save_listings, он же попадает в `listings.city`). Отдаётся матчеру как независимое от строки адреса наблюдение города (#2777): бескоординатный ключ Tier-2a вырождается в один нормализованный адрес, а областной формат Avito SERP («ул. Кирова,4») города не называет — без этого признака карточка из Серова матчится в дом Каменска-Уральского. Skips silently if: - lot has no source_id AND no address/lat/lon (cannot match house anyway) Raises on DB errors — caller wraps in try/except + SAVEPOINT. """ ext_id = lot.source_id or lot.compute_dedup_hash() # House resolution: needs at least address or (lat, lon). Skip otherwise — # listing_sources requires listing_id but not house linkage, so still upsert. house_id: int | None = None if lot.address or (lot.lat is not None and lot.lon is not None): # Use house_source/house_ext_id when scraper extracted them (Avito Houses # catalog link, Cian newbuilding). Falls back to per-listing identity # so each unique listing creates its own row if no canonical house exists. h_src = lot.house_source or lot.source h_ext = lot.house_ext_id or ext_id # No inner begin_nested here — the caller (save_listings) already wraps this # entire function in a per-row SAVEPOINT (backend.md SAVEPOINT pattern). # A second nested SAVEPOINT inside would be redundant: if match_or_create_house # raises a DB error, that error propagates out of _link_listing_to_house, # the caller's SAVEPOINT rolls back only the hook work (not the listings INSERT), # and the caller logs a match_failed warning. One SAVEPOINT level is sufficient. house_id, _conf, _method = matcher.match_or_create_house( db, ext_source=h_src, ext_id=h_ext, address=lot.address, lat=lot.lat, lon=lot.lon, year_built=lot.year_built, # Кадастр КВАРТИРЫ (lot.cadastral_number / lot.kadastr_num) сюда БОЛЬШЕ НЕ идёт # (#2674) — ключ дома строится только на кадастре здания. В listings оба поля # по-прежнему пишутся save_listings'ом, теряется только ложная идентичность. building_cadastral_number=lot.building_cadastral_number, source_url=lot.house_url or lot.source_url, city=city, ) # Mirror the resolved house into listings.house_id_fk so direct # `JOIN houses ON house_id_fk = id` keeps working on freshly scraped / # re-scraped active rows. Without this the FK is populated only by the # offline backfill (063_*.sql / backfill_listing_sources.py), leaving # new active listings unlinked (avito 0/444, cian ~7%). Mirrors the # backfill script's column/param style. IS DISTINCT FROM avoids a no-op # write when the linkage is already correct. if house_id is not None: db.execute( text( "UPDATE listings " " SET house_id_fk = CAST(:hid AS bigint) " " WHERE id = CAST(:lid AS bigint) " " AND house_id_fk IS DISTINCT FROM CAST(:hid AS bigint)" ), {"hid": house_id, "lid": listing_id}, ) matcher.upsert_listing_source( db, listing_id=listing_id, ext_source=lot.source, ext_id=str(ext_id), method="source_link", confidence=1.0, price_rub=lot.price_rub, area_m2=lot.area_m2, floor=lot.floor, rooms_count=lot.rooms, source_url=lot.source_url, source_data={"house_id": house_id} if house_id else None, )