1294 lines
54 KiB
Python
1294 lines
54 KiB
Python
"""avito_houses.py — Avito Houses Catalog parser (Stage 2c).
|
||
|
||
Парсит страницу /catalog/houses/<slug>/<id>, извлекает window.__preloadedState__
|
||
и разбирает 10 виджетов: housePage, reviews, miniSerp, housePlacementHistory,
|
||
recommendations.
|
||
|
||
Avito (≈2026, MFE) перенёс полезную нагрузку виджетов на уровень глубже —
|
||
под ключ с именем самого виджета (props['reviews']['entries'],
|
||
props['miniSerp']['items'], props['housePlacementHistory']['items'] и т.д.).
|
||
Разворачивание делает _widget_payload с fallback на старую форму (props напрямую).
|
||
|
||
Flow:
|
||
fetch_house_catalog(house_url)
|
||
→ HTTP GET (curl_cffi chrome120)
|
||
→ regex extract __preloadedState__
|
||
→ JS-unescape + json.loads
|
||
→ parse_houses_state(state, house_url)
|
||
→ HouseCatalogEnrichment
|
||
|
||
Таблицы:
|
||
houses (009_houses.sql + 010_houses_alter.sql)
|
||
house_reviews (014_house_reviews.sql)
|
||
sellers (012_sellers.sql)
|
||
listings.seller_id_fk (013_listings_alter_seller.sql)
|
||
house_placement_history (017_house_placement_history.sql)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
import urllib.parse
|
||
from dataclasses import dataclass, field
|
||
from datetime import UTC, date, datetime
|
||
from typing import TYPE_CHECKING, Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.services.matching.houses import match_or_create_house
|
||
from app.services.scrapers.avito import _is_firewall_page
|
||
from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError
|
||
|
||
if TYPE_CHECKING:
|
||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
AVITO_BASE = "https://www.avito.ru"
|
||
|
||
# Regex для извлечения __preloadedState__ из HTML.
|
||
#
|
||
# Avito (≈2026) сменил формат state-блоба: раньше был
|
||
# window.__preloadedState__ = JSON.parse("<js-escaped json>");
|
||
# теперь — URL-encoded строковый литерал:
|
||
# window.__preloadedState__ = "%7B%22abCentral%22%3A..."; (%7B='{', %22='"')
|
||
#
|
||
# PRELOADED_URLENC_RE — текущая форма (декод через urllib.parse.unquote).
|
||
# PRELOADED_JSPARSE_RE — старая форма (оставлена как fallback).
|
||
PRELOADED_URLENC_RE = re.compile(
|
||
r'window\.__preloadedState__\s*=\s*"((?:[^"\\]|\\.)*)"\s*;',
|
||
re.DOTALL,
|
||
)
|
||
PRELOADED_JSPARSE_RE = re.compile(
|
||
r'window\.__preloadedState__\s*=\s*JSON\.parse\("(.*?)"\)\s*;',
|
||
re.DOTALL,
|
||
)
|
||
|
||
# Русские месяцы для парсинга дат отзывов
|
||
RUS_MONTHS = {
|
||
"января": 1,
|
||
"февраля": 2,
|
||
"марта": 3,
|
||
"апреля": 4,
|
||
"мая": 5,
|
||
"июня": 6,
|
||
"июля": 7,
|
||
"августа": 8,
|
||
"сентября": 9,
|
||
"октября": 10,
|
||
"ноября": 11,
|
||
"декабря": 12,
|
||
}
|
||
|
||
# Маппинг expandParams type → (field_name, cast_type)
|
||
EXPAND_PARAMS_MAP: dict[str, tuple[str, str]] = {
|
||
"Год постройки": ("year_built", "int"),
|
||
"Этажей": ("total_floors", "int"),
|
||
"Тип дома": ("house_type", "house_type"),
|
||
"Горячее водоснабжение": ("hot_water", "str"),
|
||
"Пассажирский лифт": ("passenger_elevators", "int"),
|
||
"Грузовой лифт": ("cargo_elevators", "int"),
|
||
"Консьерж": ("has_concierge", "bool_da"),
|
||
"Перекрытия": ("material_floors", "str"),
|
||
"Парковка": ("parking_type", "str"),
|
||
"Детская площадка": ("has_playground", "bool_da"),
|
||
"Закрытый двор": ("closed_yard", "bool_da"),
|
||
# Новостройки (developmentPage) — другие подписи параметров. Лифты во
|
||
# множественном числе; "Этажность" — диапазон ("от 4 до 24"), int-каст
|
||
# вернёт None (хранится только в raw_characteristics).
|
||
"Пассажирских лифтов": ("passenger_elevators", "int"),
|
||
"Грузовых лифтов": ("cargo_elevators", "int"),
|
||
}
|
||
|
||
# URL-маркер новостройки: /catalog/novostroyki/... (vs вторичка /catalog/houses/...).
|
||
NOVOSTROYKA_URL_MARKER = "/catalog/novostroyki/"
|
||
|
||
# Нормализация типа дома из Avito текста в DB enum
|
||
HOUSE_TYPE_MAP: dict[str, str] = {
|
||
"монолитный": "monolith",
|
||
"монолит": "monolith",
|
||
"панельный": "panel",
|
||
"панель": "panel",
|
||
"кирпичный": "brick",
|
||
"кирпич": "brick",
|
||
"монолитно-кирпичный": "monolith_brick",
|
||
"блочный": "block",
|
||
"деревянный": "wood",
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Dataclasses
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class HouseInfo:
|
||
ext_id: int # avitoId (int)
|
||
ext_id_hash: str | None = None # base64 internal ID
|
||
title: str | None = None
|
||
short_address: str | None = None # краткий адрес
|
||
full_address: str | None = None # полный адрес
|
||
lat: float | None = None
|
||
lon: float | None = None
|
||
year_built: int | None = None
|
||
total_floors: int | None = None
|
||
house_type: str | None = None # нормализован: monolith/panel/brick/...
|
||
material_floors: str | None = None
|
||
hot_water: str | None = None
|
||
passenger_elevators: int | None = None
|
||
cargo_elevators: int | None = None
|
||
has_concierge: bool | None = None
|
||
parking_type: str | None = None
|
||
has_playground: bool | None = None
|
||
closed_yard: bool | None = None
|
||
developer_name: str | None = None
|
||
developer_key: str | None = None
|
||
infrastructure_summary: str | None = None
|
||
infrastructure_walk_distance: str | None = None
|
||
rating_score: float | None = None
|
||
rating_string: str | None = None
|
||
reviews_count: int | None = None
|
||
rating_distribution: list[dict[str, Any]] = field(default_factory=list)
|
||
map_pins: list[dict[str, Any]] = field(default_factory=list)
|
||
raw_characteristics: list[dict[str, Any]] = field(default_factory=list)
|
||
|
||
|
||
@dataclass
|
||
class HouseReview:
|
||
ext_review_id: int
|
||
author_name: str | None = None
|
||
review_title: str | None = None
|
||
score: int | None = None
|
||
model_experience: str | None = None
|
||
rated_date: date | None = None
|
||
text_main: str | None = None
|
||
text_pros: str | None = None
|
||
text_cons: str | None = None
|
||
raw_payload: dict[str, Any] | None = None
|
||
|
||
|
||
@dataclass
|
||
class PlacementHistoryItem:
|
||
ext_item_id: str
|
||
title: str | None = None
|
||
start_price: int | None = None
|
||
start_price_date: date | None = None
|
||
last_price: int | None = None
|
||
last_price_date: date | None = None
|
||
exposure_days: int | None = None
|
||
removed_date: date | None = None # ВСЕГДА None для source='avito_widget'
|
||
raw_payload: dict[str, Any] | None = None
|
||
|
||
|
||
@dataclass
|
||
class SellerInfo:
|
||
ext_seller_id: str
|
||
name: str
|
||
seller_type: str | None = None
|
||
on_platform_since: str | None = None
|
||
profile_url: str | None = None
|
||
|
||
|
||
@dataclass
|
||
class MiniSerpListing:
|
||
ext_item_id: int
|
||
title: str | None = None
|
||
price_rub: int | None = None
|
||
description: str | None = None
|
||
url: str = ""
|
||
metro_text: str | None = None
|
||
metro_color: str | None = None
|
||
seller: SellerInfo | None = None
|
||
|
||
|
||
@dataclass
|
||
class RecommendationItem:
|
||
title: str | None = None
|
||
price_text: str | None = None
|
||
address: str | None = None
|
||
url: str = ""
|
||
image_url: str | None = None
|
||
|
||
|
||
@dataclass
|
||
class HouseCatalogEnrichment:
|
||
house_url: str
|
||
house: HouseInfo
|
||
reviews: list[HouseReview] = field(default_factory=list)
|
||
placement_history: list[PlacementHistoryItem] = field(default_factory=list)
|
||
mini_serp: list[MiniSerpListing] = field(default_factory=list)
|
||
recommendations: list[RecommendationItem] = field(default_factory=list)
|
||
reviews_next_page_url: str | None = None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helper functions
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _parse_ru_date(text_val: str | None) -> date | None:
|
||
"""Парсит дату вида "9 октября 2025" → date(2025, 10, 9)."""
|
||
if not text_val:
|
||
return None
|
||
m = re.match(r"(\d+)\s+([а-я]+)\s+(\d{4})", text_val.strip(), re.IGNORECASE)
|
||
if not m:
|
||
return None
|
||
day = int(m.group(1))
|
||
month_word = m.group(2).lower()
|
||
year = int(m.group(3))
|
||
month = RUS_MONTHS.get(month_word)
|
||
return date(year, month, day) if month else None
|
||
|
||
|
||
def _unix_to_date(ts: int | None) -> date | None:
|
||
"""Конвертирует unix timestamp (секунды) → date."""
|
||
if ts is None:
|
||
return None
|
||
try:
|
||
return datetime.fromtimestamp(ts, tz=UTC).date()
|
||
except (OSError, OverflowError, ValueError):
|
||
logger.warning("Не удалось конвертировать unix timestamp: %r", ts)
|
||
return None
|
||
|
||
|
||
def _strip_price(price_str: str | None) -> int | None:
|
||
"""Извлекает целое число из строки "11 990 000 ₽" → 11990000."""
|
||
if not price_str:
|
||
return None
|
||
digits = re.sub(r"[^\d]", "", price_str)
|
||
return int(digits) if digits else None
|
||
|
||
|
||
def _normalize_house_type(raw: str | None) -> str | None:
|
||
"""Нормализует тип дома: "Монолитный" → "monolith"."""
|
||
if not raw:
|
||
return None
|
||
return HOUSE_TYPE_MAP.get(raw.lower(), "other")
|
||
|
||
|
||
def _cast_expand_param(value: str, cast_type: str) -> Any:
|
||
"""Кастует строковое значение expandParams к нужному типу."""
|
||
if cast_type == "int":
|
||
try:
|
||
return int(value)
|
||
except (ValueError, TypeError):
|
||
return None
|
||
if cast_type == "bool_da":
|
||
return value.strip().lower() == "да"
|
||
if cast_type == "house_type":
|
||
return _normalize_house_type(value)
|
||
# str
|
||
return value
|
||
|
||
|
||
def _parse_expand_params(items: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""Извлекает поля из expandParams['items'] по EXPAND_PARAMS_MAP.
|
||
|
||
Возвращает dict с распарсенными полями + raw_characteristics (полный список).
|
||
"""
|
||
result: dict[str, Any] = {}
|
||
for category in items:
|
||
for param in category.get("params", []):
|
||
param_type = param.get("type", "")
|
||
if param_type in EXPAND_PARAMS_MAP:
|
||
field_name, cast_type = EXPAND_PARAMS_MAP[param_type]
|
||
raw_value = param.get("value", "")
|
||
result[field_name] = _cast_expand_param(str(raw_value), cast_type)
|
||
return result
|
||
|
||
|
||
def _flatten_widgets(placeholders: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""Возвращает плоский список виджетов из placeholders.
|
||
|
||
Avito (≈2026) обернул 10 виджетов в контейнер: placeholders теперь
|
||
[{"id": ..., "widgets": [{"type": "housePage", ...}, ...]}]
|
||
раньше виджеты лежали в placeholders напрямую:
|
||
[{"type": "housePage", ...}, {"type": "reviews", ...}, ...]
|
||
|
||
Хелпер раскрывает любой элемент с ключом "widgets" (список) в его виджеты,
|
||
остальные элементы оставляет как есть → поддержка обеих форм.
|
||
"""
|
||
flat: list[dict[str, Any]] = []
|
||
for entry in placeholders:
|
||
if isinstance(entry, dict) and isinstance(entry.get("widgets"), list):
|
||
flat.extend(w for w in entry["widgets"] if isinstance(w, dict))
|
||
elif isinstance(entry, dict):
|
||
flat.append(entry)
|
||
return flat
|
||
|
||
|
||
def _get_widget(placeholders: list[dict[str, Any]], widget_type: str) -> dict[str, Any] | None:
|
||
"""Ищет виджет по type в массиве placeholders (уже flat)."""
|
||
for w in placeholders:
|
||
if w.get("type") == widget_type:
|
||
return w
|
||
return None
|
||
|
||
|
||
def _widget_payload(widget: dict[str, Any], widget_type: str) -> dict[str, Any]:
|
||
"""Возвращает полезную нагрузку виджета, разворачивая type-вложенность.
|
||
|
||
Avito (≈2026, MFE) переместил данные виджета на уровень глубже —
|
||
под ключ с именем самого виджета:
|
||
новый: widget['props'][widget_type] = {... entries / items ...}
|
||
старый: widget['props'] = {... entries / items ...}
|
||
|
||
Хелпер пробует новую форму (props[widget_type]), fallback на старую
|
||
(props напрямую) → поддержка обеих форм.
|
||
"""
|
||
props = widget.get("props", {})
|
||
if not isinstance(props, dict):
|
||
return {}
|
||
nested = props.get(widget_type)
|
||
if isinstance(nested, dict):
|
||
return nested
|
||
return props
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Widget parsers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _parse_house_page(widget: dict[str, Any]) -> HouseInfo:
|
||
"""Парсит виджет housePage → HouseInfo.
|
||
|
||
developmentData — основной объект с данными дома. Avito (≈2026) изменил путь:
|
||
новый: widget['props']['housePage']['developmentData']
|
||
старый: widget['props']['developmentData']
|
||
Аналогично рейтинг переехал в developmentData['ratingPreview']
|
||
(раньше ratingBadge / ratingSummaryStat лежали прямо в developmentData).
|
||
Хелпер пробует новые пути с fallback на старые → поддержка обеих форм.
|
||
"""
|
||
props = widget.get("props", {})
|
||
dd = props.get("housePage", {}).get("developmentData") or props.get("developmentData", {})
|
||
|
||
# expandParams → характеристики дома
|
||
about = dd.get("aboutDevelopment", {})
|
||
expand_params_items: list[dict[str, Any]] = about.get("expandParams", {}).get("items", [])
|
||
parsed_params = _parse_expand_params(expand_params_items)
|
||
|
||
# Рейтинг: новая форма — внутри ratingPreview, старая — прямо в developmentData
|
||
rating_root = dd.get("ratingPreview") or dd
|
||
rating_badge = rating_root.get("ratingBadge", {})
|
||
rating_info = rating_badge.get("info", {})
|
||
|
||
rating_summary = rating_root.get("ratingSummaryStat", {})
|
||
rating_stat: list[dict[str, Any]] = rating_summary.get("ratingStat", [])
|
||
|
||
# Карта/инфраструктура
|
||
map_preview = dd.get("mapPreview", {})
|
||
|
||
# Разработчик
|
||
developer = dd.get("developer", {})
|
||
|
||
# Координаты
|
||
coords = dd.get("coords", {})
|
||
|
||
return HouseInfo(
|
||
ext_id=dd.get("avitoId", 0),
|
||
ext_id_hash=dd.get("id"),
|
||
title=dd.get("title"),
|
||
short_address=dd.get("address"),
|
||
full_address=dd.get("fullAddress"),
|
||
lat=coords.get("lat"),
|
||
lon=coords.get("lng"),
|
||
year_built=parsed_params.get("year_built"),
|
||
total_floors=parsed_params.get("total_floors"),
|
||
house_type=parsed_params.get("house_type"),
|
||
material_floors=parsed_params.get("material_floors"),
|
||
hot_water=parsed_params.get("hot_water"),
|
||
passenger_elevators=parsed_params.get("passenger_elevators"),
|
||
cargo_elevators=parsed_params.get("cargo_elevators"),
|
||
has_concierge=parsed_params.get("has_concierge"),
|
||
parking_type=parsed_params.get("parking_type"),
|
||
has_playground=parsed_params.get("has_playground"),
|
||
closed_yard=parsed_params.get("closed_yard"),
|
||
developer_name=developer.get("name"),
|
||
developer_key=developer.get("key"),
|
||
infrastructure_summary=map_preview.get("objects"),
|
||
infrastructure_walk_distance=map_preview.get("distance"),
|
||
rating_score=rating_info.get("score"),
|
||
rating_string=rating_info.get("scoreString"),
|
||
reviews_count=rating_summary.get("reviewCount"),
|
||
rating_distribution=rating_stat,
|
||
map_pins=map_preview.get("pins", []),
|
||
raw_characteristics=expand_params_items,
|
||
)
|
||
|
||
|
||
def _is_novostroyka_url(house_url: str) -> bool:
|
||
"""True если house_url — страница новостройки (/catalog/novostroyki/...).
|
||
|
||
Структура новостройки отличается от вторички: вместо виджета 'housePage'
|
||
используется 'developmentPage', нет отзывов/истории размещений.
|
||
"""
|
||
return NOVOSTROYKA_URL_MARKER in house_url
|
||
|
||
|
||
def _parse_development_page(widget: dict[str, Any]) -> HouseInfo:
|
||
"""Парсит виджет 'developmentPage' (новостройка) → HouseInfo.
|
||
|
||
Структура отличается от 'housePage' (вторичка):
|
||
- данные под props['developmentPage']['developmentData'] (fallback props['developmentData'])
|
||
- характеристики не в developmentData.aboutDevelopment.expandParams.items,
|
||
а в aboutDevelopment.tabs[*].expandParams.items[*].params (плоский список params).
|
||
- НЕТ рейтинга (ratingBadge/ratingSummaryStat), года постройки (дом строится —
|
||
есть "Срок сдачи"), отзывов, истории размещений.
|
||
Извлекаем то, что есть: id/адрес/координаты/застройщик/инфраструктура + тип дома,
|
||
этажность (диапазон → raw_characteristics), лифты из таблицы "Параметры".
|
||
"""
|
||
props = widget.get("props", {})
|
||
dd = props.get("developmentPage", {}).get("developmentData") or props.get("developmentData", {})
|
||
|
||
# Характеристики новостройки лежат в tabs[*].expandParams.items[*].params.
|
||
# Собираем их в форму, совместимую с _parse_expand_params (category → params).
|
||
char_categories: list[dict[str, Any]] = []
|
||
about = dd.get("aboutDevelopment", {})
|
||
for tab in about.get("tabs", []):
|
||
if not isinstance(tab, dict):
|
||
continue
|
||
items = tab.get("expandParams", {}).get("items", [])
|
||
for item in items:
|
||
if isinstance(item, dict) and isinstance(item.get("params"), list):
|
||
char_categories.append(item)
|
||
parsed_params = _parse_expand_params(char_categories)
|
||
|
||
map_preview = dd.get("mapPreview", {})
|
||
developer = dd.get("developer", {})
|
||
coords = dd.get("coords", {})
|
||
|
||
return HouseInfo(
|
||
ext_id=dd.get("avitoId", 0),
|
||
ext_id_hash=dd.get("id"),
|
||
title=dd.get("title"),
|
||
short_address=dd.get("address"),
|
||
full_address=dd.get("fullAddress") or None,
|
||
lat=coords.get("lat"),
|
||
lon=coords.get("lng"),
|
||
year_built=parsed_params.get("year_built"), # обычно None — дом строится
|
||
total_floors=parsed_params.get("total_floors"),
|
||
house_type=parsed_params.get("house_type"),
|
||
material_floors=parsed_params.get("material_floors"),
|
||
hot_water=parsed_params.get("hot_water"),
|
||
passenger_elevators=parsed_params.get("passenger_elevators"),
|
||
cargo_elevators=parsed_params.get("cargo_elevators"),
|
||
has_concierge=parsed_params.get("has_concierge"),
|
||
parking_type=parsed_params.get("parking_type"),
|
||
has_playground=parsed_params.get("has_playground"),
|
||
closed_yard=parsed_params.get("closed_yard"),
|
||
developer_name=developer.get("name"),
|
||
developer_key=developer.get("key"),
|
||
infrastructure_summary=map_preview.get("objects"),
|
||
infrastructure_walk_distance=map_preview.get("distance"),
|
||
rating_score=None,
|
||
rating_string=None,
|
||
reviews_count=None,
|
||
rating_distribution=[],
|
||
map_pins=map_preview.get("pins", []),
|
||
raw_characteristics=char_categories,
|
||
)
|
||
|
||
|
||
def _parse_single_review(r: dict[str, Any]) -> HouseReview:
|
||
"""Парсит один объект отзыва из entries."""
|
||
text_main: str | None = None
|
||
text_pros: str | None = None
|
||
text_cons: str | None = None
|
||
|
||
for section in r.get("textSections", []):
|
||
title = section.get("title", "")
|
||
section_text = section.get("text")
|
||
if title == "":
|
||
text_main = section_text
|
||
elif title == "Преимущества":
|
||
text_pros = section_text
|
||
elif title == "Недостатки":
|
||
text_cons = section_text
|
||
|
||
author = r.get("author", {})
|
||
return HouseReview(
|
||
ext_review_id=r["id"],
|
||
author_name=author.get("title"),
|
||
review_title=r.get("reviewTitle"),
|
||
score=r.get("score"),
|
||
model_experience=r.get("modelExperience"),
|
||
rated_date=_parse_ru_date(r.get("rated")),
|
||
text_main=text_main,
|
||
text_pros=text_pros,
|
||
text_cons=text_cons,
|
||
raw_payload=r,
|
||
)
|
||
|
||
|
||
def _parse_reviews(widget: dict[str, Any]) -> tuple[list[HouseReview], str | None]:
|
||
"""Парсит виджет reviews → (список HouseReview, следующая страница URL).
|
||
|
||
Entries — смешанный список с type='rating' (отзывы) и type='pages' (пагинация).
|
||
"""
|
||
reviews: list[HouseReview] = []
|
||
next_page_url: str | None = None
|
||
|
||
payload = _widget_payload(widget, "reviews")
|
||
entries = payload.get("entries", [])
|
||
for entry in entries:
|
||
entry_type = entry.get("type")
|
||
value = entry.get("value", {})
|
||
|
||
if entry_type == "rating":
|
||
try:
|
||
reviews.append(_parse_single_review(value))
|
||
except (KeyError, TypeError) as exc:
|
||
logger.warning("Не удалось распарсить отзыв: %r — %s", value.get("id"), exc)
|
||
elif entry_type == "pages":
|
||
next_page_url = value.get("nextPageUrl")
|
||
|
||
return reviews, next_page_url
|
||
|
||
|
||
def _parse_seller(seller_raw: dict[str, Any]) -> SellerInfo | None:
|
||
"""Парсит seller из miniSerp item → SellerInfo."""
|
||
if not seller_raw:
|
||
return None
|
||
name = seller_raw.get("name", "")
|
||
if not name:
|
||
return None
|
||
|
||
profile_url = seller_raw.get("url", "")
|
||
# ext_seller_id — последний сегмент URL: "/brands/domrf66" → "domrf66"
|
||
ext_seller_id = profile_url.strip("/").split("/")[-1] if profile_url else ""
|
||
if not ext_seller_id:
|
||
logger.warning("Не удалось извлечь ext_seller_id из seller URL: %r", profile_url)
|
||
return None
|
||
|
||
return SellerInfo(
|
||
ext_seller_id=ext_seller_id,
|
||
name=name,
|
||
seller_type=seller_raw.get("type"),
|
||
on_platform_since=seller_raw.get("from"),
|
||
profile_url=profile_url or None,
|
||
)
|
||
|
||
|
||
def _parse_mini_serp(widget: dict[str, Any]) -> list[MiniSerpListing]:
|
||
"""Парсит виджет miniSerp → список активных объявлений в доме."""
|
||
listings: list[MiniSerpListing] = []
|
||
|
||
items = _widget_payload(widget, "miniSerp").get("items", [])
|
||
for item in items:
|
||
geo = item.get("geo", {})
|
||
colors: list[str] = geo.get("colors", [])
|
||
|
||
seller_raw = item.get("seller")
|
||
seller = _parse_seller(seller_raw) if seller_raw else None
|
||
|
||
price_raw = item.get("price")
|
||
price_rub = _strip_price(price_raw) if isinstance(price_raw, str) else price_raw
|
||
|
||
listings.append(
|
||
MiniSerpListing(
|
||
ext_item_id=item["id"],
|
||
title=item.get("title"),
|
||
price_rub=price_rub,
|
||
description=item.get("description"),
|
||
url=item.get("url", ""),
|
||
metro_text=geo.get("content"),
|
||
metro_color=colors[0] if colors else None,
|
||
seller=seller,
|
||
)
|
||
)
|
||
|
||
return listings
|
||
|
||
|
||
def _parse_placement_history(widget: dict[str, Any]) -> list[PlacementHistoryItem]:
|
||
"""Парсит виджет housePlacementHistory → список PlacementHistoryItem.
|
||
|
||
ВАЖНО: removed_date ВСЕГДА None — виджет не отдаёт эту дату.
|
||
Точная дата только из IMV API (Stage 2d).
|
||
"""
|
||
result: list[PlacementHistoryItem] = []
|
||
|
||
items = _widget_payload(widget, "housePlacementHistory").get("items", [])
|
||
for item in items:
|
||
item_copy = {k: v for k, v in item.items() if k != "itemImage"}
|
||
result.append(
|
||
PlacementHistoryItem(
|
||
ext_item_id=str(item["id"]),
|
||
title=item.get("title"),
|
||
start_price=item.get("startPrice"),
|
||
start_price_date=_unix_to_date(item.get("startPriceDate")),
|
||
last_price=item.get("lastPrice"),
|
||
last_price_date=_unix_to_date(item.get("lastPriceDate")),
|
||
exposure_days=item.get("exposure"),
|
||
removed_date=None, # widget не отдаёт! IMV API Stage 2d
|
||
raw_payload=item_copy,
|
||
)
|
||
)
|
||
|
||
return result
|
||
|
||
|
||
def _parse_recommendations(widget: dict[str, Any]) -> list[RecommendationItem]:
|
||
"""Парсит виджет recommendations → lightweight список похожих объявлений."""
|
||
result: list[RecommendationItem] = []
|
||
|
||
items = _widget_payload(widget, "recommendations").get("items", [])
|
||
for item in items:
|
||
additional_info: list[str] = item.get("additionalInfo", [])
|
||
address = additional_info[0] if additional_info else None
|
||
|
||
images: list[dict[str, Any]] = item.get("images", [])
|
||
image_url: str | None = None
|
||
if images:
|
||
image_url = images[0].get("208x156")
|
||
|
||
result.append(
|
||
RecommendationItem(
|
||
title=item.get("title"),
|
||
price_text=item.get("priceRange"),
|
||
address=address,
|
||
url=item.get("url", ""),
|
||
image_url=image_url,
|
||
)
|
||
)
|
||
|
||
return result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Core parser (pure, no network)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _extract_preloaded_state(html: str) -> dict[str, Any] | None:
|
||
"""Извлекает window.__preloadedState__ из HTML → распарсенный dict.
|
||
|
||
Pure функция (нет I/O) — тестируется на фикстуре, переиспользуется из
|
||
fetch_house_catalog.
|
||
|
||
Две формы (см. PRELOADED_*_RE):
|
||
1. URL-encoded строковый литерал (текущая, ≈2026):
|
||
window.__preloadedState__ = "%7B%22...%22%7D";
|
||
→ urllib.parse.unquote + json.loads
|
||
2. JSON.parse с js-экранированием (старая, fallback):
|
||
window.__preloadedState__ = JSON.parse("{\\"...\\"}");
|
||
→ ручной unescape + json.loads
|
||
|
||
Возвращает dict или None, если маркер не найден / не распарсился.
|
||
"""
|
||
# Форма 1 — URL-encoded литерал (приоритет)
|
||
m = PRELOADED_URLENC_RE.search(html)
|
||
if m is not None:
|
||
try:
|
||
return json.loads(urllib.parse.unquote(m.group(1)))
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.warning("URL-encoded __preloadedState__ не распарсился: %s", exc)
|
||
|
||
# Форма 2 — старый JSON.parse("...") (fallback)
|
||
m = PRELOADED_JSPARSE_RE.search(html)
|
||
if m is not None:
|
||
# JS-стиль unescaping: \\" → " , \\\\ → \ , \\/ → /
|
||
# НЕ .decode("unicode_escape") — ломает многобайтный UTF-8 (кириллица).
|
||
raw = m.group(1).replace('\\"', '"').replace("\\\\", "\\").replace("\\/", "/")
|
||
try:
|
||
return json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
logger.warning("JSON.parse-форма с replace-unescape не удалась, пробуем unicode_escape")
|
||
try:
|
||
return json.loads(bytes(raw, "utf-8").decode("unicode_escape"))
|
||
except (json.JSONDecodeError, UnicodeDecodeError) as exc2:
|
||
logger.warning("JSON.parse-форма __preloadedState__ не распарсилась: %s", exc2)
|
||
|
||
return None
|
||
|
||
|
||
def _parse_development_or_skip(
|
||
placeholders: list[dict[str, Any]], house_url: str
|
||
) -> HouseCatalogEnrichment:
|
||
"""Обрабатывает страницу без виджета 'housePage' (новостройка или неизвестная форма).
|
||
|
||
- Есть виджет 'developmentPage' → парсим его (partial: дом/застройщик/тип/лифты,
|
||
без отзывов/истории/рейтинга).
|
||
- Нет developmentPage, но URL — новостройка → graceful skip: warning + минимальное
|
||
обогащение (ext_id из URL, без характеристик). НЕ кидаем ValueError, чтобы один
|
||
незнакомый шейп новостройки не валил весь enrichment-loop pipeline.
|
||
- Иначе (вторичка без housePage) → ValueError (регрессия шейпа вторички).
|
||
"""
|
||
dev_widget = _get_widget(placeholders, "developmentPage")
|
||
if dev_widget is not None:
|
||
house_info = _parse_development_page(dev_widget)
|
||
logger.info(
|
||
"developmentPage (новостройка) распарсен: ext_id=%s type=%s url=%s",
|
||
house_info.ext_id,
|
||
house_info.house_type,
|
||
house_url,
|
||
)
|
||
return HouseCatalogEnrichment(house_url=house_url, house=house_info)
|
||
|
||
if _is_novostroyka_url(house_url):
|
||
logger.warning(
|
||
"Новостройка без виджета 'developmentPage'/'housePage' (%s) — "
|
||
"graceful skip, обогащение пропущено",
|
||
house_url,
|
||
)
|
||
ext_id = _ext_id_from_url(house_url)
|
||
return HouseCatalogEnrichment(house_url=house_url, house=HouseInfo(ext_id=ext_id))
|
||
|
||
raise ValueError("Виджет 'housePage' не найден в placeholders")
|
||
|
||
|
||
def _ext_id_from_url(house_url: str) -> int:
|
||
"""Извлекает числовой avitoId из хвоста house_url (.../<slug>/<id>). 0 если нет."""
|
||
m = re.search(r"/(\d+)/?(?:\?|$)", house_url)
|
||
return int(m.group(1)) if m else 0
|
||
|
||
|
||
def parse_houses_state(state: dict[str, Any], house_url: str) -> HouseCatalogEnrichment:
|
||
"""Парсит уже разобранный __preloadedState__ dict → HouseCatalogEnrichment.
|
||
|
||
Pure функция: нет HTTP, нет I/O. Принимает state dict и house_url (для поля
|
||
HouseCatalogEnrichment.house_url). Используется в тестах и из fetch_house_catalog.
|
||
|
||
Новостройки (/catalog/novostroyki/...) не имеют виджета 'housePage' — вместо него
|
||
'developmentPage'. Для них парсим что есть (id/адрес/застройщик/тип дома/лифты) и
|
||
возвращаем partial-обогащение без отзывов/истории. Если структура совсем не
|
||
распознана — graceful: warning + минимальное обогащение (НЕ ValueError).
|
||
|
||
Raises:
|
||
ValueError если placeholders не массив, или (только для вторички) housePage
|
||
виджет не найден.
|
||
"""
|
||
try:
|
||
raw_placeholders: list[dict[str, Any]] = state["data"]["data"]["page"]["placeholders"]
|
||
except (KeyError, TypeError) as exc:
|
||
raise ValueError(f"Не удалось найти placeholders в __preloadedState__: {exc}") from exc
|
||
|
||
if not isinstance(raw_placeholders, list):
|
||
raise ValueError(
|
||
f"placeholders должен быть массивом, получен: {type(raw_placeholders).__name__}"
|
||
)
|
||
|
||
# Avito (≈2026) оборачивает виджеты в контейнер {id, widgets:[...]} —
|
||
# раскрываем в плоский список (поддержка старой и новой формы).
|
||
placeholders = _flatten_widgets(raw_placeholders)
|
||
|
||
# housePage — обязательный виджет для вторички. Новостройки используют
|
||
# 'developmentPage' и не имеют housePage — обрабатываем отдельно (graceful).
|
||
house_page_widget = _get_widget(placeholders, "housePage")
|
||
if house_page_widget is None:
|
||
return _parse_development_or_skip(placeholders, house_url)
|
||
|
||
house_info = _parse_house_page(house_page_widget)
|
||
|
||
# reviews — опциональный
|
||
reviews: list[HouseReview] = []
|
||
next_page_url: str | None = None
|
||
reviews_widget = _get_widget(placeholders, "reviews")
|
||
if reviews_widget is not None:
|
||
reviews, next_page_url = _parse_reviews(reviews_widget)
|
||
if not reviews:
|
||
logger.warning("Виджет 'reviews' найден, но 0 отзывов распарсено — проверь шейп props")
|
||
else:
|
||
logger.debug("Виджет 'reviews' не найден — пропускаем")
|
||
|
||
# miniSerp — опциональный
|
||
mini_serp: list[MiniSerpListing] = []
|
||
mini_serp_widget = _get_widget(placeholders, "miniSerp")
|
||
if mini_serp_widget is not None:
|
||
mini_serp = _parse_mini_serp(mini_serp_widget)
|
||
else:
|
||
logger.debug("Виджет 'miniSerp' не найден — пропускаем")
|
||
|
||
# housePlacementHistory — опциональный
|
||
placement_history: list[PlacementHistoryItem] = []
|
||
history_widget = _get_widget(placeholders, "housePlacementHistory")
|
||
if history_widget is not None:
|
||
placement_history = _parse_placement_history(history_widget)
|
||
if not placement_history:
|
||
logger.warning(
|
||
"Виджет 'housePlacementHistory' найден, но 0 items распарсено — проверь шейп props"
|
||
)
|
||
else:
|
||
logger.debug("Виджет 'housePlacementHistory' не найден — пропускаем")
|
||
|
||
# recommendations — опциональный
|
||
recommendations: list[RecommendationItem] = []
|
||
recs_widget = _get_widget(placeholders, "recommendations")
|
||
if recs_widget is not None:
|
||
recommendations = _parse_recommendations(recs_widget)
|
||
else:
|
||
logger.debug("Виджет 'recommendations' не найден — пропускаем")
|
||
|
||
return HouseCatalogEnrichment(
|
||
house_url=house_url,
|
||
house=house_info,
|
||
reviews=reviews,
|
||
placement_history=placement_history,
|
||
mini_serp=mini_serp,
|
||
recommendations=recommendations,
|
||
reviews_next_page_url=next_page_url,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# HTTP fetch (thin wrapper, requires curl_cffi)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def fetch_house_catalog(
|
||
house_url: str,
|
||
*,
|
||
cffi_session: Any | None = None,
|
||
browser_fetcher: BrowserFetcher | None = None,
|
||
) -> HouseCatalogEnrichment:
|
||
"""GET {AVITO_BASE}{house_url} → extract window.__preloadedState__ → parse 10 виджетов.
|
||
|
||
Параметры:
|
||
house_url — относительный путь:
|
||
/catalog/houses/ekaterinburg/ul_postovskogo/3171365
|
||
cffi_session — опциональная curl_cffi AsyncSession (для переиспользования пула соединений).
|
||
Если None — создаётся временная сессия.
|
||
browser_fetcher — если передан, используется браузерный fetch вместо curl_cffi.
|
||
|
||
Raises:
|
||
RuntimeError если curl_cffi не установлен (в curl-режиме).
|
||
httpx.HTTPStatusError / curl_cffi аналог если status != 200.
|
||
ValueError если __preloadedState__ не найден или placeholders не массив.
|
||
"""
|
||
url = f"{AVITO_BASE}{house_url}"
|
||
logger.info("Houses Catalog fetch: %s", url)
|
||
|
||
if browser_fetcher is not None:
|
||
html = await browser_fetcher.fetch(url)
|
||
if _is_firewall_page(html):
|
||
raise AvitoBlockedError(f"Avito houses firewall (browser-mode) for {url}")
|
||
state = _extract_preloaded_state(html)
|
||
if state is None:
|
||
raise ValueError(
|
||
f"window.__preloadedState__ не найден / не распарсился в HTML страницы: {url}"
|
||
)
|
||
logger.info("Houses Catalog state extracted (browser-mode), parsing widgets")
|
||
return parse_houses_state(state, house_url)
|
||
|
||
try:
|
||
from curl_cffi.requests import AsyncSession as CffiAsyncSession
|
||
except ImportError as exc:
|
||
raise RuntimeError(
|
||
"curl_cffi не установлен. Добавь 'curl-cffi>=0.7.0' в pyproject.toml."
|
||
) from exc
|
||
|
||
_own_session = False
|
||
if cffi_session is None:
|
||
cffi_session = CffiAsyncSession(impersonate="chrome120")
|
||
_own_session = True
|
||
|
||
try:
|
||
resp = await cffi_session.get(url)
|
||
if resp.status_code == 403:
|
||
raise AvitoBlockedError(f"Avito houses HTTP 403 for {url}")
|
||
if resp.status_code == 429:
|
||
raise AvitoRateLimitedError(f"Avito houses HTTP 429 for {url}")
|
||
resp.raise_for_status()
|
||
html = resp.text
|
||
|
||
# Извлекаем + декодируем window.__preloadedState__ (URL-encoded или JSON.parse).
|
||
state = _extract_preloaded_state(html)
|
||
if state is None:
|
||
raise ValueError(
|
||
f"window.__preloadedState__ не найден / не распарсился в HTML страницы: {url}"
|
||
)
|
||
|
||
logger.info("Houses Catalog state extracted, parsing widgets")
|
||
return parse_houses_state(state, house_url)
|
||
|
||
finally:
|
||
if _own_session:
|
||
await cffi_session.close()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Save functions
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def upsert_house(db: Session, h: HouseInfo) -> int:
|
||
"""Match-or-create канонический дом через match_or_create_house, затем обновить enrichment.
|
||
|
||
Source='avito'. Возвращает houses.id.
|
||
Делегирует в _persist_house (без url — url пуст как раньше для этого пути).
|
||
|
||
Маппинг полей HouseInfo → DDL колонки (009 + 010):
|
||
ext_id → ext_house_id (text)
|
||
ext_id_hash → avito_id_hash
|
||
short_address → short_address (010)
|
||
full_address → full_address (010)
|
||
address (DDL) → используем short_address как fallback
|
||
rating_score → rating_score (010, numeric(4,2))
|
||
rating (DDL) → округлённый numeric(2,1) из rating_score
|
||
"""
|
||
return _persist_house(db, h, "")
|
||
|
||
|
||
def _upsert_house_with_url(db: Session, h: HouseInfo, house_url: str) -> int:
|
||
"""Внутренний вариант upsert_house с правильным URL.
|
||
|
||
Делегирует в _persist_house — используй его вместо прямого вызова.
|
||
"""
|
||
return _persist_house(db, h, house_url)
|
||
|
||
|
||
def _persist_house(db: Session, h: HouseInfo, house_url: str) -> int:
|
||
"""Канонический путь сохранения дома-avito через match_or_create_house.
|
||
|
||
Шаги:
|
||
1. match_or_create_house — находит или создаёт канонический houses-ряд
|
||
(3-tier: source_exact → fingerprint → geo → new), защищён advisory lock.
|
||
2. UPDATE houses SET ... — записывает все avito-enrichment поля в найденный/новый дом.
|
||
- Avito-авторитетные поля (rating*, developer*, технические хар-ки, url,
|
||
avito_id_hash) перезаписываются напрямую.
|
||
- address/lat/lon/year_built — через COALESCE чтобы не затирать данные
|
||
канонического дома от cian/кадастра.
|
||
3. Не вызывает db.commit() — коммитит вызывающий save_house_catalog_enrichment
|
||
(advisory lock xact-scoped, должен держаться до commit).
|
||
|
||
Returns:
|
||
houses.id (int) канонического дома.
|
||
"""
|
||
house_id, _conf, _method = match_or_create_house(
|
||
db,
|
||
ext_source="avito",
|
||
ext_id=str(h.ext_id),
|
||
# short preferred: aliases have no city/region prefix; full is last resort
|
||
address=h.short_address or h.full_address,
|
||
lat=h.lat,
|
||
lon=h.lon,
|
||
year_built=h.year_built,
|
||
source_url=house_url or None,
|
||
)
|
||
|
||
db.execute(
|
||
text("""
|
||
UPDATE houses SET
|
||
url = CAST(:url AS text),
|
||
short_address = CAST(:short_address AS text),
|
||
full_address = CAST(:full_address AS text),
|
||
avito_id_hash = CAST(:avito_id_hash AS text),
|
||
house_type = CAST(:house_type AS text),
|
||
total_floors = CAST(:total_floors AS int),
|
||
passenger_elevators = CAST(:passenger_elevators AS int),
|
||
cargo_elevators = CAST(:cargo_elevators AS int),
|
||
has_concierge = CAST(:has_concierge AS boolean),
|
||
closed_yard = CAST(:closed_yard AS boolean),
|
||
material_floors = CAST(:material_floors AS text),
|
||
hot_water = CAST(:hot_water AS text),
|
||
parking_type = CAST(:parking_type AS text),
|
||
has_playground = CAST(:has_playground AS boolean),
|
||
developer_name = CAST(:developer_name AS text),
|
||
developer_key = CAST(:developer_key AS text),
|
||
infrastructure_summary = CAST(:infrastructure_summary AS text),
|
||
infrastructure_walk_distance = CAST(:infrastructure_walk_distance AS text),
|
||
rating = CAST(:rating AS numeric),
|
||
rating_score = CAST(:rating_score AS numeric),
|
||
rating_string = CAST(:rating_string AS text),
|
||
reviews_count = CAST(:reviews_count AS int),
|
||
rating_distribution = CAST(:rating_distribution AS jsonb),
|
||
map_pins = CAST(:map_pins AS jsonb),
|
||
raw_characteristics = CAST(:raw_characteristics AS jsonb),
|
||
address = COALESCE(houses.address, CAST(:address AS text)),
|
||
lat = COALESCE(houses.lat, CAST(:lat AS double precision)),
|
||
lon = COALESCE(houses.lon, CAST(:lon AS double precision)),
|
||
year_built = COALESCE(houses.year_built, CAST(:year_built AS int)),
|
||
last_scraped_at = NOW(),
|
||
avito_validated_at = NOW()
|
||
WHERE id = CAST(:house_id AS bigint)
|
||
"""),
|
||
{
|
||
"house_id": house_id,
|
||
"url": house_url or "",
|
||
"short_address": h.short_address,
|
||
"full_address": h.full_address,
|
||
"avito_id_hash": h.ext_id_hash,
|
||
"house_type": h.house_type,
|
||
"total_floors": h.total_floors,
|
||
"passenger_elevators": h.passenger_elevators,
|
||
"cargo_elevators": h.cargo_elevators,
|
||
"has_concierge": h.has_concierge,
|
||
"closed_yard": h.closed_yard,
|
||
"material_floors": h.material_floors,
|
||
"hot_water": h.hot_water,
|
||
"parking_type": h.parking_type,
|
||
"has_playground": h.has_playground,
|
||
"developer_name": h.developer_name,
|
||
"developer_key": h.developer_key,
|
||
"infrastructure_summary": h.infrastructure_summary,
|
||
"infrastructure_walk_distance": h.infrastructure_walk_distance,
|
||
# rating (numeric(2,1)) — округлённое значение для совместимости с 009 DDL
|
||
"rating": round(h.rating_score, 1) if h.rating_score is not None else None,
|
||
"rating_score": h.rating_score,
|
||
"rating_string": h.rating_string,
|
||
"reviews_count": h.reviews_count,
|
||
"rating_distribution": json.dumps(h.rating_distribution, ensure_ascii=False),
|
||
"map_pins": json.dumps(h.map_pins, ensure_ascii=False),
|
||
"raw_characteristics": json.dumps(h.raw_characteristics, ensure_ascii=False),
|
||
"address": h.short_address,
|
||
"lat": h.lat,
|
||
"lon": h.lon,
|
||
"year_built": h.year_built,
|
||
},
|
||
)
|
||
|
||
logger.info(
|
||
"_persist_house: ext_id=%s url=%s → houses.id=%s (method=%s)",
|
||
h.ext_id,
|
||
house_url,
|
||
house_id,
|
||
_method,
|
||
)
|
||
return int(house_id)
|
||
|
||
|
||
def save_house_reviews(db: Session, house_id: int, reviews: list[HouseReview]) -> int:
|
||
"""INSERT INTO house_reviews ON CONFLICT (source, ext_review_id) DO UPDATE.
|
||
|
||
Source='avito'. Возвращает количество вставленных/обновлённых строк.
|
||
"""
|
||
if not reviews:
|
||
return 0
|
||
|
||
count = 0
|
||
for r in reviews:
|
||
db.execute(
|
||
text("""
|
||
INSERT INTO house_reviews (
|
||
house_id, source, ext_review_id,
|
||
author_name, review_title, score,
|
||
model_experience, rated_date,
|
||
text_main, text_pros, text_cons,
|
||
raw_payload, scraped_at
|
||
) VALUES (
|
||
CAST(:house_id AS bigint), 'avito', CAST(:ext_review_id AS bigint),
|
||
CAST(:author_name AS text), CAST(:review_title AS text), CAST(:score AS int),
|
||
CAST(:model_experience AS text), CAST(:rated_date AS date),
|
||
CAST(:text_main AS text), CAST(:text_pros AS text), CAST(:text_cons AS text),
|
||
CAST(:raw_payload AS jsonb), NOW()
|
||
)
|
||
ON CONFLICT (source, ext_review_id) DO UPDATE SET
|
||
author_name = EXCLUDED.author_name,
|
||
review_title = EXCLUDED.review_title,
|
||
score = EXCLUDED.score,
|
||
model_experience = EXCLUDED.model_experience,
|
||
rated_date = EXCLUDED.rated_date,
|
||
text_main = EXCLUDED.text_main,
|
||
text_pros = EXCLUDED.text_pros,
|
||
text_cons = EXCLUDED.text_cons,
|
||
raw_payload = EXCLUDED.raw_payload,
|
||
scraped_at = NOW()
|
||
"""),
|
||
{
|
||
"house_id": house_id,
|
||
"ext_review_id": r.ext_review_id,
|
||
"author_name": r.author_name,
|
||
"review_title": r.review_title,
|
||
"score": r.score,
|
||
"model_experience": r.model_experience,
|
||
"rated_date": r.rated_date,
|
||
"text_main": r.text_main,
|
||
"text_pros": r.text_pros,
|
||
"text_cons": r.text_cons,
|
||
"raw_payload": (
|
||
json.dumps(r.raw_payload, ensure_ascii=False) if r.raw_payload else None
|
||
),
|
||
},
|
||
)
|
||
count += 1
|
||
|
||
logger.info("save_house_reviews: house_id=%s, saved=%d", house_id, count)
|
||
return count
|
||
|
||
|
||
def save_sellers(db: Session, sellers: list[SellerInfo]) -> dict[str, int]:
|
||
"""INSERT INTO sellers ON CONFLICT (source, ext_seller_id) DO UPDATE.
|
||
|
||
Source='avito'. Возвращает {ext_seller_id: sellers.id}.
|
||
"""
|
||
result: dict[str, int] = {}
|
||
for s in sellers:
|
||
row = db.execute(
|
||
text("""
|
||
INSERT INTO sellers (
|
||
source, ext_seller_id,
|
||
name, seller_type, on_platform_since, profile_url,
|
||
last_seen_at
|
||
) VALUES (
|
||
'avito', CAST(:ext_seller_id AS text),
|
||
CAST(:name AS text), CAST(:seller_type AS text),
|
||
CAST(:on_platform_since AS text), CAST(:profile_url AS text),
|
||
NOW()
|
||
)
|
||
ON CONFLICT (source, ext_seller_id) DO UPDATE SET
|
||
name = EXCLUDED.name,
|
||
seller_type = EXCLUDED.seller_type,
|
||
on_platform_since = EXCLUDED.on_platform_since,
|
||
profile_url = EXCLUDED.profile_url,
|
||
last_seen_at = NOW()
|
||
RETURNING id
|
||
"""),
|
||
{
|
||
"ext_seller_id": s.ext_seller_id,
|
||
"name": s.name,
|
||
"seller_type": s.seller_type,
|
||
"on_platform_since": s.on_platform_since,
|
||
"profile_url": s.profile_url,
|
||
},
|
||
)
|
||
seller_db_id: int = row.scalar_one()
|
||
result[s.ext_seller_id] = seller_db_id
|
||
|
||
logger.info("save_sellers: saved=%d", len(result))
|
||
return result
|
||
|
||
|
||
def link_listings_to_seller(db: Session, ext_item_id_to_seller_db_id: dict[int, int]) -> int:
|
||
"""UPDATE listings SET seller_id_fk = :seller_id WHERE source='avito' AND source_id = :item_id.
|
||
|
||
Возвращает количество обновлённых строк.
|
||
"""
|
||
if not ext_item_id_to_seller_db_id:
|
||
return 0
|
||
|
||
total = 0
|
||
for item_id, seller_db_id in ext_item_id_to_seller_db_id.items():
|
||
result = db.execute(
|
||
text("""
|
||
UPDATE listings
|
||
SET seller_id_fk = CAST(:seller_id AS bigint)
|
||
WHERE source = 'avito'
|
||
AND source_id = CAST(:item_id AS text)
|
||
AND seller_id_fk IS DISTINCT FROM CAST(:seller_id AS bigint)
|
||
"""),
|
||
{"seller_id": seller_db_id, "item_id": str(item_id)},
|
||
)
|
||
total += result.rowcount
|
||
|
||
logger.info("link_listings_to_seller: updated=%d listings", total)
|
||
return total
|
||
|
||
|
||
def save_placement_history(db: Session, items: list[PlacementHistoryItem]) -> int:
|
||
"""INSERT INTO house_placement_history с source='avito_widget' и removed_date=NULL.
|
||
|
||
ON CONFLICT (source, ext_item_id) DO UPDATE last_price/last_price_date/exposure_days.
|
||
Возвращает количество строк.
|
||
"""
|
||
if not items:
|
||
return 0
|
||
|
||
count = 0
|
||
for item in items:
|
||
db.execute(
|
||
text("""
|
||
INSERT INTO house_placement_history (
|
||
source, ext_item_id,
|
||
title,
|
||
start_price, start_price_date,
|
||
last_price, last_price_date,
|
||
exposure_days,
|
||
removed_date,
|
||
raw_payload, scraped_at
|
||
) VALUES (
|
||
'avito_widget', CAST(:ext_item_id AS text),
|
||
CAST(:title AS text),
|
||
CAST(:start_price AS bigint), CAST(:start_price_date AS date),
|
||
CAST(:last_price AS bigint), CAST(:last_price_date AS date),
|
||
CAST(:exposure_days AS int),
|
||
NULL,
|
||
CAST(:raw_payload AS jsonb), NOW()
|
||
)
|
||
ON CONFLICT (source, ext_item_id) DO UPDATE SET
|
||
title = EXCLUDED.title,
|
||
last_price = EXCLUDED.last_price,
|
||
last_price_date = EXCLUDED.last_price_date,
|
||
exposure_days = EXCLUDED.exposure_days,
|
||
raw_payload = EXCLUDED.raw_payload,
|
||
scraped_at = NOW()
|
||
"""),
|
||
{
|
||
"ext_item_id": item.ext_item_id,
|
||
"title": item.title,
|
||
"start_price": item.start_price,
|
||
"start_price_date": item.start_price_date,
|
||
"last_price": item.last_price,
|
||
"last_price_date": item.last_price_date,
|
||
"exposure_days": item.exposure_days,
|
||
"raw_payload": (
|
||
json.dumps(item.raw_payload, ensure_ascii=False) if item.raw_payload else None
|
||
),
|
||
},
|
||
)
|
||
count += 1
|
||
|
||
logger.info("save_placement_history: saved=%d items (source=avito_widget)", count)
|
||
return count
|
||
|
||
|
||
def save_house_catalog_enrichment(
|
||
db: Session,
|
||
e: HouseCatalogEnrichment,
|
||
) -> dict[str, int]:
|
||
"""Orchestrator — полный pipeline сохранения одного обогащения.
|
||
|
||
Операции (в порядке):
|
||
1. upsert_house → houses.id
|
||
2. save_house_reviews(house_id, reviews)
|
||
3. save_sellers([m.seller for m in mini_serp if m.seller]) → {ext_seller_id: db_id}
|
||
4. link_listings_to_seller({item_id: seller_db_id})
|
||
5. save_placement_history(placement_history)
|
||
|
||
Возвращает:
|
||
{'house_id': N, 'reviews': N, 'sellers': N, 'listings_linked': N, 'placement_history': N}
|
||
"""
|
||
# 1. Upsert дом с правильным URL из enrichment
|
||
house_id = _upsert_house_with_url(db, e.house, e.house_url)
|
||
|
||
# 2. Сохранить отзывы
|
||
reviews_count = save_house_reviews(db, house_id, e.reviews)
|
||
|
||
# 3. Сохранить продавцов (дедуп по ext_seller_id)
|
||
sellers_to_save = [m.seller for m in e.mini_serp if m.seller is not None]
|
||
seen_seller_ids: set[str] = set()
|
||
unique_sellers: list[SellerInfo] = []
|
||
for s in sellers_to_save:
|
||
if s.ext_seller_id not in seen_seller_ids:
|
||
unique_sellers.append(s)
|
||
seen_seller_ids.add(s.ext_seller_id)
|
||
|
||
seller_id_map = save_sellers(db, unique_sellers)
|
||
|
||
# 4. Слинковать объявления с продавцами
|
||
# Маппинг: ext_item_id (int) → seller db_id
|
||
item_to_seller: dict[int, int] = {}
|
||
for listing in e.mini_serp:
|
||
if listing.seller and listing.seller.ext_seller_id in seller_id_map:
|
||
item_to_seller[listing.ext_item_id] = seller_id_map[listing.seller.ext_seller_id]
|
||
|
||
listings_linked = link_listings_to_seller(db, item_to_seller)
|
||
|
||
# 5. Сохранить историю размещений
|
||
history_count = save_placement_history(db, e.placement_history)
|
||
|
||
db.commit()
|
||
|
||
result = {
|
||
"house_id": house_id,
|
||
"reviews": reviews_count,
|
||
"sellers": len(seller_id_map),
|
||
"listings_linked": listings_linked,
|
||
"placement_history": history_count,
|
||
}
|
||
logger.info("save_house_catalog_enrichment: %s", result)
|
||
return result
|