765 lines
38 KiB
Python
765 lines
38 KiB
Python
"""Базовый framework для парсеров объявлений (Avito / Cian / DomKlik / ...).
|
||
|
||
Общие задачи:
|
||
- 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 для регистрации в
|
||
`listing_sources` и привязки к каноническому `houses` через `house_sources`.
|
||
|
||
Каждый конкретный парсер (avito.py, cian.py, ...) наследуется от BaseScraper
|
||
и реализует `_fetch_page()` и `_parse_listings()`.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import logging
|
||
import random
|
||
from abc import ABC, abstractmethod
|
||
from datetime import date, datetime
|
||
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
|
||
|
||
from app.services.matching import match_or_create_house, upsert_listing_source
|
||
from app.services.scrapers.snapshot_writer import upsert_listing_snapshot
|
||
|
||
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)
|
||
|
||
|
||
# ── Унифицированная схема результата ────────────────────────────────────────
|
||
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
|
||
|
||
# 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. Может быть пустым.
|
||
"""
|
||
...
|
||
|
||
|
||
# ── Запись пачки результатов в Postgres ─────────────────────────────────────
|
||
def save_listings(
|
||
db: Session,
|
||
lots: list[ScrapedLot],
|
||
*,
|
||
run_id: int | None = None,
|
||
skip_seen_today: bool = False,
|
||
) -> 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: список распарсенных объявлений.
|
||
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).
|
||
|
||
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
|
||
|
||
today_msk = datetime.now(_MSK).date()
|
||
|
||
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()
|
||
|
||
# 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,
|
||
"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
|
||
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:
|
||
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)
|
||
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 (total %d)",
|
||
lots[0].source if lots else "?",
|
||
inserted,
|
||
updated,
|
||
reconciled,
|
||
skipped,
|
||
matched,
|
||
match_failures,
|
||
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) -> None:
|
||
"""Hook scraped listing into matching service: resolve house, upsert listing_sources.
|
||
|
||
Steps:
|
||
1. 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. 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).
|
||
|
||
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 = 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,
|
||
building_cadastral_number=lot.building_cadastral_number,
|
||
cadastral_number=lot.cadastral_number or lot.kadastr_num,
|
||
source_url=lot.house_url or lot.source_url,
|
||
)
|
||
|
||
# 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},
|
||
)
|
||
|
||
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,
|
||
)
|