Light1YT
b16935e679
listings_snapshots не имел ни одного writer'а — таблица оставалась пустой. offer_price_history уже писался в save_detail_enrichment, но ON CONFLICT DO NOTHING не указывал имя constraint'а (добавлено в 047_cian_history_sanitize.sql), что делало де-дупликацию неэффективной. Добавлено: - snapshot_writer.py: upsert_listing_snapshot() — атомарный INSERT ... ON CONFLICT DO UPDATE на (listing_id, snapshot_date), без commit (commit у caller'а). - base.py save_listings(): вызывает upsert_listing_snapshot после каждого INSERT/UPDATE listings; принимает опциональный run_id; ошибка snapshot не прерывает батч. - cian_detail.py save_detail_enrichment(): SELECT price_rub/price_per_m2 из listings, затем upsert_listing_snapshot; ON CONFLICT для offer_price_history теперь ссылается на offer_price_history_listing_change_uq (именованный constraint из мигр. 047). - test_snapshot_writer.py: 17 unit-тестов покрывают SQL-форму, params, идемпотентность, интеграцию с save_listings и save_detail_enrichment, fault-tolerance.
476 lines
21 KiB
Python
476 lines
21 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
|
||
from typing import Any
|
||
|
||
import httpx
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import text
|
||
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__)
|
||
|
||
|
||
# ── 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'
|
||
|
||
# Цена (обязательно)
|
||
price_rub: int = Field(gt=0)
|
||
price_per_m2: int | None = None
|
||
|
||
# Метаданные
|
||
listing_date: date | None = None
|
||
days_on_market: int | 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)
|
||
|
||
def compute_dedup_hash(self) -> str:
|
||
"""SHA256(source + (source_id или source_url) + price_rub) — стабильный uniqueness key.
|
||
|
||
Cian URL содержит random `?context=...` токены — поэтому ключ берём
|
||
source_id (offer_id 330200428), а не source_url. Если source_id нет —
|
||
откатываемся на source_url (Yandex, Avito с offerId).
|
||
"""
|
||
h = hashlib.sha256()
|
||
h.update(self.source.encode())
|
||
h.update(b"|")
|
||
key = self.source_id if self.source_id else self.source_url
|
||
h.update((key or "").encode())
|
||
h.update(b"|")
|
||
h.update(str(self.price_rub).encode())
|
||
return h.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,
|
||
) -> 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 вызовах).
|
||
|
||
Returns:
|
||
(inserted, updated) — counters для логов.
|
||
"""
|
||
if not lots:
|
||
return 0, 0
|
||
|
||
inserted = 0
|
||
updated = 0
|
||
matched = 0
|
||
match_failures = 0
|
||
|
||
for lot in lots:
|
||
ppm2 = lot.price_per_m2 or lot.compute_price_per_m2()
|
||
dedup = lot.compute_dedup_hash()
|
||
|
||
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, kadastr_num,
|
||
house_source, house_ext_id, house_url, listing_segment,
|
||
price_rub, price_per_m2,
|
||
listing_date, days_on_market, 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,
|
||
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, :kadastr,
|
||
:house_source, :house_ext_id, :house_url, :listing_segment,
|
||
:price_rub, :ppm2,
|
||
:listing_date, :days_on_market,
|
||
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),
|
||
NOW(), NOW()
|
||
)
|
||
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
|
||
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,
|
||
"kadastr": lot.kadastr_num,
|
||
"house_source": lot.house_source,
|
||
"house_ext_id": lot.house_ext_id,
|
||
"house_url": lot.house_url,
|
||
"listing_segment": lot.listing_segment,
|
||
"price_rub": lot.price_rub,
|
||
"ppm2": ppm2,
|
||
"listing_date": lot.listing_date,
|
||
"days_on_market": lot.days_on_market,
|
||
"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,
|
||
},
|
||
).fetchone()
|
||
|
||
listing_id: int | None = None
|
||
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 за день.
|
||
if listing_id is not None:
|
||
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 matched=%d "
|
||
"match_failures=%d (total %d)",
|
||
lots[0].source if lots else "?",
|
||
inserted,
|
||
updated,
|
||
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
|
||
try:
|
||
with db.begin_nested():
|
||
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,
|
||
)
|
||
except Exception as e:
|
||
# Fall through to listing-only upsert: listing_sources row still useful
|
||
# even without house linkage (e.g. for later backfill).
|
||
logger.warning(
|
||
"save_listings:house_match_failed source=%s ext_id=%s: %s",
|
||
lot.source, ext_id, e,
|
||
)
|
||
house_id = None
|
||
|
||
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,
|
||
)
|