811 lines
34 KiB
Python
811 lines
34 KiB
Python
"""Avito detail page parser — Stage 2b.
|
||
|
||
Извлекает 30+ полей из страницы объявления:
|
||
- Identity (item_id, title, price, publish_date, views)
|
||
- Apartment params (rooms, area, floor, balcony_loggia, room_layout,
|
||
bathroom_type, windows_view, repair_state, sale_type, mortgage_available)
|
||
- Location (lat, lon, avito_location_id, metro_stations[], address_full)
|
||
- House params (house_type, total_floors_house, lifts, concierge, closed_yard,
|
||
house_catalog_url) — собираются но НЕ сохраняются в БД (Stage 2c)
|
||
- Description
|
||
- Domoteka 5 полей (owners_count, owners_at_least, last_owner_change_date,
|
||
encumbrances_clean, registry_match)
|
||
- Gallery (photo_urls[])
|
||
|
||
Точка входа для batch-обогащения:
|
||
enrichment = await fetch_detail(item_url)
|
||
saved = save_detail_enrichment(db, enrichment)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from datetime import date
|
||
from typing import TYPE_CHECKING, Any
|
||
from urllib.parse import urljoin, urlparse
|
||
|
||
from curl_cffi.requests import AsyncSession
|
||
from selectolax.parser import HTMLParser, Node
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.services.scrapers.avito import _clean_address, _is_firewall_page
|
||
from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError
|
||
from app.services.scrapers.repair_state_normalizer import infer_repair_state_from_text
|
||
|
||
if TYPE_CHECKING:
|
||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
AVITO_BASE = "https://www.avito.ru"
|
||
|
||
# ── Русские месяцы ─────────────────────────────────────────────────────────
|
||
RUS_MONTHS: dict[str, int] = {
|
||
"января": 1,
|
||
"февраля": 2,
|
||
"марта": 3,
|
||
"апреля": 4,
|
||
"мая": 5,
|
||
"июня": 6,
|
||
"июля": 7,
|
||
"августа": 8,
|
||
"сентября": 9,
|
||
"октября": 10,
|
||
"ноября": 11,
|
||
"декабря": 12,
|
||
}
|
||
|
||
# ── Regex ────────────────────────────────────────────────────────────────────
|
||
_ITEM_ID_RE = re.compile(r"№\s*(\d+)")
|
||
_VIEWS_TOTAL_RE = re.compile(r"(\d+)\s+просмотр")
|
||
_VIEWS_TODAY_RE = re.compile(r"\+\s*(\d+)\s+сегодня")
|
||
_PUBLISH_DATE_RE = re.compile(
|
||
r"(\d{1,2})\s+(января|февраля|марта|апреля|мая|июня|"
|
||
r"июля|августа|сентября|октября|ноября|декабря)\s+в",
|
||
re.IGNORECASE,
|
||
)
|
||
_FLOAT_RE = re.compile(r"[\d.,]+")
|
||
_INT_RE = re.compile(r"\d+")
|
||
_FLOOR_SPLIT_RE = re.compile(r"(\d+)\s+из\s+(\d+)")
|
||
|
||
OWNERS_RE = re.compile(r"^(\d+)\s+собственник(?:\s+или\s+больше)?", re.IGNORECASE)
|
||
OWNER_CHANGE_RE = re.compile(r"смена собственника\s+(\d+)\s+(\w+)\s+(\d{4})", re.IGNORECASE)
|
||
ENCUMBRANCES_RE = re.compile(r"Не\s+найдены\s+ограничения", re.IGNORECASE)
|
||
REGISTRY_MATCH_RE = re.compile(r"Совпадают\s+площадь", re.IGNORECASE)
|
||
|
||
METRO_RE = re.compile(
|
||
r"([А-ЯЁ][а-яё]+(?:ская|инская|итская|овская|енская))\s*"
|
||
r"(?:(\d+)[–\-](\d+)|от\s+(\d+))\s*мин",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# ── Маппинг RUS полей params-блока ───────────────────────────────────────────
|
||
# (rus_label) -> (field_name, conversion_type)
|
||
RUS_FIELD_MAP: dict[str, tuple[str, str | None]] = {
|
||
"Количество комнат": ("rooms", "int"),
|
||
"Общая площадь": ("area_m2", "float_m2"),
|
||
"Площадь кухни": ("kitchen_area_m2", "float_m2"),
|
||
"Жилая площадь": ("living_area_m2", "float_m2"),
|
||
"Высота потолков": ("ceiling_height_m", "height_m"),
|
||
"Отделка": ("finishing", "text_lower"),
|
||
"Этаж": ("floor_split", None),
|
||
"Балкон или лоджия": ("balcony_loggia", "balcony"),
|
||
"Тип комнат": ("room_layout", "layout"),
|
||
"Санузел": ("bathroom_type", "bathroom"),
|
||
"Окна": ("windows_view", "windows"),
|
||
"Ремонт": ("repair_state", "repair"),
|
||
"Способ продажи": ("sale_type", "sale_type"),
|
||
"Условия продажи": ("sale_conditions_raw", None),
|
||
"Тип дома": ("house_type", "house_type"),
|
||
"Этажей в доме": ("total_floors_house", "int"),
|
||
"Пассажирский лифт": ("passenger_elevators", "int"),
|
||
"Грузовой лифт": ("cargo_elevators", "int"),
|
||
"В доме": ("in_house_features_raw", None),
|
||
"Двор": ("yard_features_raw", None),
|
||
}
|
||
|
||
# ── Enum-маппинги ─────────────────────────────────────────────────────────────
|
||
_BALCONY_MAP: dict[str, str] = {
|
||
"балкон": "balcony",
|
||
"лоджия": "loggia",
|
||
"балкон, лоджия": "both",
|
||
"лоджия, балкон": "both",
|
||
"нет": "none",
|
||
}
|
||
|
||
_LAYOUT_MAP: dict[str, str] = {
|
||
"изолированные": "isolated",
|
||
"смежные": "adjacent",
|
||
"смешанные": "mixed",
|
||
}
|
||
|
||
_BATHROOM_MAP: dict[str, str] = {
|
||
"совмещенный": "combined",
|
||
"раздельный": "separate",
|
||
"совмещенный, раздельный": "both",
|
||
"раздельный, совмещенный": "both",
|
||
}
|
||
|
||
_WINDOWS_MAP: dict[str, str] = {
|
||
"во двор": "yard",
|
||
"на улицу": "street",
|
||
"во двор и на улицу": "both",
|
||
"на улицу и во двор": "both",
|
||
"в парк": "park",
|
||
}
|
||
|
||
_REPAIR_MAP: dict[str, str] = {
|
||
# Русские значения из Avito HTML → каноничный enum (needs_repair/standard/good/excellent)
|
||
# Маппинг выровнен по repair_state_normalizer.normalize_repair_state().
|
||
"косметический": "standard",
|
||
"евро": "good",
|
||
"дизайнерский": "excellent",
|
||
"требуется": "needs_repair",
|
||
"без ремонта": "needs_repair",
|
||
}
|
||
|
||
_SALE_TYPE_MAP: dict[str, str] = {
|
||
"свободная": "free",
|
||
"альтернатива": "alternative",
|
||
"аукцион": "auction",
|
||
"переуступка": "assignment",
|
||
}
|
||
|
||
_HOUSE_TYPE_MAP: dict[str, str] = {
|
||
"монолитный": "monolith",
|
||
"панельный": "panel",
|
||
"кирпичный": "brick",
|
||
"монолитно-кирпичный": "monolith_brick",
|
||
"блочный": "block",
|
||
"деревянный": "wood",
|
||
}
|
||
|
||
|
||
# ── DetailEnrichment dataclass ────────────────────────────────────────────────
|
||
@dataclass
|
||
class DetailEnrichment:
|
||
"""Результат парсинга detail page Avito.
|
||
|
||
Содержит 30+ полей из страницы объявления.
|
||
save_detail_enrichment() пишет подмножество в listings (house-level поля пропускаются
|
||
— они canonical из Houses Catalog Stage 2c).
|
||
"""
|
||
|
||
# Identity
|
||
item_id: str
|
||
source_url: str
|
||
title: str | None = None
|
||
price_rub: int | None = None
|
||
publish_date: date | None = None
|
||
views_total: int | None = None
|
||
views_today: int | None = None
|
||
|
||
# Apartment params
|
||
rooms: int | None = None
|
||
area_m2: float | None = None
|
||
kitchen_area_m2: float | None = None
|
||
living_area_m2: float | None = None
|
||
ceiling_height_m: float | None = None
|
||
finishing: str | None = None
|
||
floor: int | None = None
|
||
total_floors: int | None = None
|
||
balcony_loggia: str | None = None
|
||
room_layout: str | None = None
|
||
bathroom_type: str | None = None
|
||
windows_view: str | None = None
|
||
repair_state: str | None = None
|
||
sale_type: str | None = None
|
||
mortgage_available: bool | None = None
|
||
|
||
# House params (собираем, но НЕ сохраняем в listings — Stage 2c)
|
||
house_type: str | None = None
|
||
total_floors_house: int | None = None
|
||
passenger_elevators: int | None = None
|
||
cargo_elevators: int | None = None
|
||
has_concierge: bool | None = None
|
||
closed_yard: bool | None = None
|
||
house_catalog_url: str | None = None
|
||
|
||
# Location
|
||
address_full: str | None = None
|
||
lat: float | None = None
|
||
lon: float | None = None
|
||
avito_location_id: int | None = None
|
||
metro_stations: list[dict[str, Any]] = field(default_factory=list)
|
||
|
||
# Description
|
||
description: str | None = None
|
||
|
||
# Domoteka (5 полей)
|
||
owners_count: int | None = None
|
||
owners_at_least: bool | None = None
|
||
last_owner_change_date: date | None = None
|
||
encumbrances_clean: bool | None = None
|
||
registry_match: bool | None = None
|
||
|
||
# Gallery
|
||
photo_urls: list[str] = field(default_factory=list)
|
||
|
||
# Raw для retrofit / debug
|
||
raw_html_meta: dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
# ── fetch_detail ──────────────────────────────────────────────────────────────
|
||
async def fetch_detail(
|
||
item_url: str,
|
||
*,
|
||
cffi_session: AsyncSession | None = None,
|
||
browser_fetcher: BrowserFetcher | None = None,
|
||
) -> DetailEnrichment:
|
||
"""GET <avito>/{item_url} → parse HTML via selectolax → DetailEnrichment.
|
||
|
||
Если browser_fetcher передан — использует браузерный fetch (browser mode).
|
||
Если cffi_session не передана — создаёт новую (impersonate='chrome120').
|
||
Raises:
|
||
httpx.HTTPError — если status != 200 (через raise_for_status-like).
|
||
ValueError — если item_id не извлечён из HTML.
|
||
"""
|
||
if browser_fetcher is not None:
|
||
full_url = item_url if item_url.startswith("http") else urljoin(AVITO_BASE, item_url)
|
||
html = await browser_fetcher.fetch(full_url)
|
||
if _is_firewall_page(html):
|
||
raise AvitoBlockedError(f"Avito detail firewall (browser-mode) for {full_url}")
|
||
return parse_detail_html(html, full_url)
|
||
|
||
own_session = cffi_session is None
|
||
if cffi_session is None:
|
||
# Mobile proxy wiring (#806 follow-up): own-session path (no shared session passed,
|
||
# e.g. admin endpoint). Caller-provided sessions (scrape_pipeline) already have proxy
|
||
# from AvitoScraper.__aenter__ — don't override. proxy=None → прямое подключение.
|
||
from app.core.config import settings
|
||
|
||
_proxy_url = settings.scraper_proxy_url
|
||
_proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None
|
||
session = AsyncSession(
|
||
impersonate="chrome120",
|
||
timeout=25,
|
||
proxies=_proxies,
|
||
headers={
|
||
"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",
|
||
"Cache-Control": "max-age=0",
|
||
"Sec-Fetch-Dest": "document",
|
||
"Sec-Fetch-Mode": "navigate",
|
||
"Sec-Fetch-Site": "none",
|
||
"Sec-Fetch-User": "?1",
|
||
"Upgrade-Insecure-Requests": "1",
|
||
},
|
||
)
|
||
else:
|
||
session = cffi_session
|
||
|
||
full_url = item_url if item_url.startswith("http") else urljoin(AVITO_BASE, item_url)
|
||
try:
|
||
response = await session.get(full_url)
|
||
if response.status_code == 403:
|
||
raise AvitoBlockedError(f"Avito detail HTTP 403 for {full_url}")
|
||
if response.status_code == 429:
|
||
raise AvitoRateLimitedError(f"Avito detail HTTP 429 for {full_url}")
|
||
if response.status_code != 200:
|
||
raise ValueError(f"avito detail HTTP {response.status_code} for {full_url}")
|
||
enrichment = parse_detail_html(response.text, full_url)
|
||
return enrichment
|
||
finally:
|
||
if own_session:
|
||
await session.close()
|
||
|
||
|
||
# ── parse_detail_html ─────────────────────────────────────────────────────────
|
||
def parse_detail_html(html: str, source_url: str) -> DetailEnrichment:
|
||
"""Pure parser — отдельно для тестов на fixtures без сети.
|
||
|
||
Raises:
|
||
ValueError — если item_id не извлечён.
|
||
"""
|
||
tree = HTMLParser(html)
|
||
|
||
# ── Identity ────────────────────────────────────────────────────────────
|
||
item_id = _extract_item_id(tree)
|
||
if not item_id:
|
||
raise ValueError(f"Cannot extract item_id from {source_url}")
|
||
|
||
title = _text(tree, "[data-marker='item-view/title-info']")
|
||
price_rub = _extract_price(tree)
|
||
publish_date, views_total, views_today = _extract_meta(tree)
|
||
|
||
# ── Location ────────────────────────────────────────────────────────────
|
||
map_el = tree.css_first("[data-marker='item-map-wrapper']")
|
||
lat: float | None = None
|
||
lon: float | None = None
|
||
avito_location_id: int | None = None
|
||
if map_el is not None:
|
||
lat = _try_float(map_el.attributes.get("data-map-lat"))
|
||
lon = _try_float(map_el.attributes.get("data-map-lon"))
|
||
avito_location_id = _try_int(map_el.attributes.get("data-location-id"))
|
||
|
||
address_full = _clean_address(_text(tree, "[itemprop='address']"))
|
||
|
||
# ── Description ─────────────────────────────────────────────────────────
|
||
description = _text(tree, "[data-marker='item-view/item-description']")
|
||
|
||
# ── Metro from description ───────────────────────────────────────────────
|
||
metro_stations: list[dict[str, Any]] = []
|
||
if description:
|
||
metro_stations = _extract_metro(description)
|
||
|
||
# ── House catalog link ───────────────────────────────────────────────────
|
||
# Новостройки отдают кнопку ЖК (nd-jk-details-button); вторичка — ссылку
|
||
# «узнать больше о доме» под другим маркером. Fallback на любой /catalog/houses/
|
||
# якорь: verified 2026-05-31 (#871) — эта ссылка есть в SSR detail-страницы,
|
||
# даже когда отсутствует в JS-рендеренной SERP-карточке (откуда мы брали 0).
|
||
house_catalog_url: str | None = None
|
||
house_link = tree.css_first("[data-marker='nd-jk-details-button']")
|
||
if house_link is None:
|
||
house_link = tree.css_first("a[href*='/catalog/houses/']")
|
||
if house_link is not None:
|
||
href = house_link.attributes.get("href", "")
|
||
if href:
|
||
# отбрасываем ?context-трекинг → чистый catalog-путь
|
||
house_catalog_url = urljoin(AVITO_BASE, urlparse(href).path)
|
||
|
||
# ── Photo gallery ────────────────────────────────────────────────────────
|
||
photo_urls: list[str] = []
|
||
for img in tree.css("[data-marker='image-preview/item'] img"):
|
||
src = img.attributes.get("src") or ""
|
||
if src and src not in photo_urls:
|
||
photo_urls.append(src)
|
||
|
||
# ── Params blocks ────────────────────────────────────────────────────────
|
||
params_block = tree.css_first("[data-marker='item-view/item-params']")
|
||
apt_params: dict[str, str] = {}
|
||
house_params: dict[str, str] = {}
|
||
if params_block is not None:
|
||
ul_els = params_block.css("ul")
|
||
if len(ul_els) >= 1:
|
||
apt_params = _parse_params_ul(ul_els[0])
|
||
if len(ul_els) >= 2:
|
||
house_params = _parse_params_ul(ul_els[1])
|
||
|
||
# Merge house_params into apt_params — некоторые страницы совмещают всё в одном ul
|
||
all_params = {**apt_params, **house_params}
|
||
|
||
# ── Parse apartment fields ───────────────────────────────────────────────
|
||
rooms: int | None = None
|
||
area_m2: float | None = None
|
||
kitchen_area_m2: float | None = None
|
||
living_area_m2: float | None = None
|
||
ceiling_height_m: float | None = None
|
||
finishing: str | None = None
|
||
floor: int | None = None
|
||
total_floors: int | None = None
|
||
balcony_loggia: str | None = None
|
||
room_layout: str | None = None
|
||
bathroom_type: str | None = None
|
||
windows_view: str | None = None
|
||
repair_state: str | None = None
|
||
sale_type: str | None = None
|
||
mortgage_available: bool | None = None
|
||
house_type: str | None = None
|
||
total_floors_house: int | None = None
|
||
passenger_elevators: int | None = None
|
||
cargo_elevators: int | None = None
|
||
has_concierge: bool | None = None
|
||
closed_yard: bool | None = None
|
||
|
||
for rus_label, val in all_params.items():
|
||
mapping = RUS_FIELD_MAP.get(rus_label)
|
||
if mapping is None:
|
||
continue
|
||
field_name, _conv = mapping
|
||
|
||
if field_name == "rooms":
|
||
rooms = _try_int(val)
|
||
elif field_name == "area_m2":
|
||
area_m2 = _parse_float_m2(val)
|
||
elif field_name == "kitchen_area_m2":
|
||
kitchen_area_m2 = _parse_float_m2(val)
|
||
elif field_name == "living_area_m2":
|
||
living_area_m2 = _parse_float_m2(val)
|
||
elif field_name == "ceiling_height_m":
|
||
ceiling_height_m = _parse_height_m(val)
|
||
elif field_name == "finishing":
|
||
finishing = val.strip().lower() or None
|
||
elif field_name == "floor_split":
|
||
floor, total_floors = _parse_floor_split(val)
|
||
elif field_name == "balcony_loggia":
|
||
balcony_loggia = _map_lower(val, _BALCONY_MAP)
|
||
elif field_name == "room_layout":
|
||
room_layout = _map_lower(val, _LAYOUT_MAP)
|
||
elif field_name == "bathroom_type":
|
||
bathroom_type = _map_lower(val, _BATHROOM_MAP)
|
||
elif field_name == "windows_view":
|
||
windows_view = _map_lower(val, _WINDOWS_MAP)
|
||
elif field_name == "repair_state":
|
||
repair_state = _map_lower(val, _REPAIR_MAP)
|
||
elif field_name == "sale_type":
|
||
sale_type = _map_lower(val, _SALE_TYPE_MAP)
|
||
elif field_name == "sale_conditions_raw":
|
||
if "возможна ипотека" in val.lower():
|
||
mortgage_available = True
|
||
elif field_name == "house_type":
|
||
house_type = _map_lower(val, _HOUSE_TYPE_MAP)
|
||
elif field_name == "total_floors_house":
|
||
total_floors_house = _try_int(val)
|
||
elif field_name == "passenger_elevators":
|
||
passenger_elevators = _try_int(val)
|
||
elif field_name == "cargo_elevators":
|
||
cargo_elevators = _try_int(val)
|
||
elif field_name == "in_house_features_raw":
|
||
if "консьерж" in val.lower():
|
||
has_concierge = True
|
||
elif field_name == "yard_features_raw":
|
||
if "закрытая территория" in val.lower():
|
||
closed_yard = True
|
||
|
||
# ── Fallback: repair_state из описания, если структурное поле пустое (#622) ──
|
||
if repair_state is None and description:
|
||
repair_state = infer_repair_state_from_text(description)
|
||
|
||
# ── Domoteka ─────────────────────────────────────────────────────────────
|
||
owners_count: int | None = None
|
||
owners_at_least: bool | None = None
|
||
last_owner_change_date: date | None = None
|
||
encumbrances_clean: bool | None = None
|
||
registry_match: bool | None = None
|
||
|
||
domoteka_block = tree.css_first("[data-marker='domoteka-entry-block']")
|
||
if domoteka_block is not None:
|
||
badges = domoteka_block.css("p[data-marker='TeaserData.item']")
|
||
for badge in badges:
|
||
badge_text = badge.text(strip=True)
|
||
if not badge_text:
|
||
continue
|
||
|
||
m_owners = OWNERS_RE.match(badge_text)
|
||
if m_owners:
|
||
owners_count = int(m_owners.group(1))
|
||
owners_at_least = "или больше" in badge_text.lower()
|
||
continue
|
||
|
||
m_change = OWNER_CHANGE_RE.search(badge_text)
|
||
if m_change:
|
||
try:
|
||
day = int(m_change.group(1))
|
||
month_word = m_change.group(2).lower()
|
||
year = int(m_change.group(3))
|
||
month = RUS_MONTHS.get(month_word)
|
||
if month:
|
||
last_owner_change_date = date(year, month, day)
|
||
except (ValueError, KeyError):
|
||
logger.warning("Cannot parse owner change date: %s", badge_text)
|
||
continue
|
||
|
||
if ENCUMBRANCES_RE.search(badge_text):
|
||
encumbrances_clean = True
|
||
continue
|
||
|
||
if REGISTRY_MATCH_RE.search(badge_text):
|
||
registry_match = True
|
||
continue
|
||
|
||
return DetailEnrichment(
|
||
item_id=item_id,
|
||
source_url=source_url,
|
||
title=title,
|
||
price_rub=price_rub,
|
||
publish_date=publish_date,
|
||
views_total=views_total,
|
||
views_today=views_today,
|
||
rooms=rooms,
|
||
area_m2=area_m2,
|
||
kitchen_area_m2=kitchen_area_m2,
|
||
living_area_m2=living_area_m2,
|
||
ceiling_height_m=ceiling_height_m,
|
||
finishing=finishing,
|
||
floor=floor,
|
||
total_floors=total_floors,
|
||
balcony_loggia=balcony_loggia,
|
||
room_layout=room_layout,
|
||
bathroom_type=bathroom_type,
|
||
windows_view=windows_view,
|
||
repair_state=repair_state,
|
||
sale_type=sale_type,
|
||
mortgage_available=mortgage_available,
|
||
house_type=house_type,
|
||
total_floors_house=total_floors_house,
|
||
passenger_elevators=passenger_elevators,
|
||
cargo_elevators=cargo_elevators,
|
||
has_concierge=has_concierge,
|
||
closed_yard=closed_yard,
|
||
house_catalog_url=house_catalog_url,
|
||
address_full=address_full,
|
||
lat=lat,
|
||
lon=lon,
|
||
avito_location_id=avito_location_id,
|
||
metro_stations=metro_stations,
|
||
description=description,
|
||
owners_count=owners_count,
|
||
owners_at_least=owners_at_least,
|
||
last_owner_change_date=last_owner_change_date,
|
||
encumbrances_clean=encumbrances_clean,
|
||
registry_match=registry_match,
|
||
photo_urls=photo_urls,
|
||
)
|
||
|
||
|
||
# ── save_detail_enrichment ────────────────────────────────────────────────────
|
||
def save_detail_enrichment(db: Session, e: DetailEnrichment) -> bool:
|
||
"""UPDATE listings SET <25+ cols> WHERE source='avito' AND source_id=:item_id.
|
||
|
||
Returns True если строка обновлена, False если listing не найден.
|
||
Устанавливает detail_enriched_at=NOW().
|
||
House-level поля (passenger_elevators, has_concierge, closed_yard,
|
||
total_floors_house, house_type) игнорируются — canonical из Houses Catalog (Stage 2c).
|
||
"""
|
||
result = db.execute(
|
||
text("""
|
||
UPDATE listings SET
|
||
kitchen_area_m2 = COALESCE(:kitchen_area_m2, kitchen_area_m2),
|
||
living_area_m2 = COALESCE(:living_area_m2, living_area_m2),
|
||
ceiling_height_m = COALESCE(:ceiling_height_m, ceiling_height_m),
|
||
finishing = COALESCE(:finishing, finishing),
|
||
balcony_loggia = COALESCE(:balcony_loggia, balcony_loggia),
|
||
room_layout = COALESCE(:room_layout, room_layout),
|
||
bathroom_type = COALESCE(:bathroom_type, bathroom_type),
|
||
windows_view = COALESCE(:windows_view, windows_view),
|
||
sale_type = COALESCE(:sale_type, sale_type),
|
||
mortgage_available = COALESCE(:mortgage_available, mortgage_available),
|
||
views_total = COALESCE(:views_total, views_total),
|
||
views_today = COALESCE(:views_today, views_today),
|
||
publish_date = COALESCE(:publish_date, publish_date),
|
||
description = COALESCE(:description, description),
|
||
owners_count = COALESCE(:owners_count, owners_count),
|
||
owners_at_least = COALESCE(:owners_at_least, owners_at_least),
|
||
last_owner_change_date = COALESCE(:last_owner_change_date, last_owner_change_date),
|
||
encumbrances_clean = COALESCE(:encumbrances_clean, encumbrances_clean),
|
||
registry_match = COALESCE(:registry_match, registry_match),
|
||
metro_stations = COALESCE(CAST(:metro_stations AS jsonb), metro_stations),
|
||
avito_location_id = COALESCE(:avito_location_id, avito_location_id),
|
||
lat = COALESCE(:lat, lat),
|
||
lon = COALESCE(:lon, lon),
|
||
address = COALESCE(:address, address),
|
||
rooms = COALESCE(:rooms, rooms),
|
||
area_m2 = COALESCE(:area_m2, area_m2),
|
||
floor = COALESCE(:floor, floor),
|
||
total_floors = COALESCE(:total_floors, total_floors),
|
||
repair_state = COALESCE(:repair_state, repair_state),
|
||
price_rub = COALESCE(:price_rub, price_rub),
|
||
detail_enriched_at = NOW()
|
||
WHERE source = 'avito' AND source_id = :item_id
|
||
"""),
|
||
{
|
||
"item_id": e.item_id,
|
||
"kitchen_area_m2": e.kitchen_area_m2,
|
||
"living_area_m2": e.living_area_m2,
|
||
"ceiling_height_m": e.ceiling_height_m,
|
||
"finishing": e.finishing,
|
||
"balcony_loggia": e.balcony_loggia,
|
||
"room_layout": e.room_layout,
|
||
"bathroom_type": e.bathroom_type,
|
||
"windows_view": e.windows_view,
|
||
"sale_type": e.sale_type,
|
||
"mortgage_available": e.mortgage_available,
|
||
"views_total": e.views_total,
|
||
"views_today": e.views_today,
|
||
"publish_date": e.publish_date,
|
||
"description": e.description,
|
||
"owners_count": e.owners_count,
|
||
"owners_at_least": e.owners_at_least,
|
||
"last_owner_change_date": e.last_owner_change_date,
|
||
"encumbrances_clean": e.encumbrances_clean,
|
||
"registry_match": e.registry_match,
|
||
"metro_stations": (
|
||
json.dumps(e.metro_stations, ensure_ascii=False) if e.metro_stations else None
|
||
),
|
||
"avito_location_id": e.avito_location_id,
|
||
"lat": e.lat,
|
||
"lon": e.lon,
|
||
"address": e.address_full,
|
||
"rooms": e.rooms,
|
||
"area_m2": e.area_m2,
|
||
"floor": e.floor,
|
||
"total_floors": e.total_floors,
|
||
"repair_state": e.repair_state,
|
||
"price_rub": e.price_rub,
|
||
},
|
||
)
|
||
db.commit()
|
||
found = result.rowcount > 0
|
||
if not found:
|
||
logger.warning("save_detail_enrichment: listing not found in DB for item_id=%s", e.item_id)
|
||
else:
|
||
logger.info("save_detail_enrichment: updated listing item_id=%s", e.item_id)
|
||
return found
|
||
|
||
|
||
# ── Internal helpers ─────────────────────────────────────────────────────────
|
||
def _text(tree: HTMLParser, selector: str) -> str | None:
|
||
"""Вернуть stripped text первого совпавшего элемента или None."""
|
||
el = tree.css_first(selector)
|
||
if el is None:
|
||
return None
|
||
t = el.text(strip=True)
|
||
return t or None
|
||
|
||
|
||
def _try_float(val: Any) -> float | None:
|
||
if val is None:
|
||
return None
|
||
try:
|
||
return float(str(val).replace(",", "."))
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
def _try_int(val: Any) -> int | None:
|
||
if val is None:
|
||
return None
|
||
s = str(val).strip()
|
||
m = _INT_RE.search(s)
|
||
if m:
|
||
try:
|
||
return int(m.group())
|
||
except ValueError:
|
||
return None
|
||
return None
|
||
|
||
|
||
def _parse_float_m2(val: str) -> float | None:
|
||
"""'90.1 м²' → 90.1. '75 м²' → 75.0."""
|
||
m = _FLOAT_RE.search(val)
|
||
if m:
|
||
try:
|
||
return float(m.group().replace(",", "."))
|
||
except ValueError:
|
||
return None
|
||
return None
|
||
|
||
|
||
def _parse_height_m(val: str) -> float | None:
|
||
"""'2.7 м' / '2,7\xa0м' / '3 м' → float метры. Junk → None."""
|
||
m = _FLOAT_RE.search(val)
|
||
if m:
|
||
try:
|
||
return float(m.group().replace(",", "."))
|
||
except ValueError:
|
||
return None
|
||
return None
|
||
|
||
|
||
def _parse_floor_split(val: str) -> tuple[int | None, int | None]:
|
||
"""'20 из 26' → (20, 26)."""
|
||
m = _FLOOR_SPLIT_RE.search(val)
|
||
if m:
|
||
return _try_int(m.group(1)), _try_int(m.group(2))
|
||
return None, None
|
||
|
||
|
||
def _map_lower(val: str, mapping: dict[str, str]) -> str | None:
|
||
"""Нижний регистр + strip → lookup в mapping. None если не найдено."""
|
||
key = val.strip().lower()
|
||
return mapping.get(key)
|
||
|
||
|
||
def _extract_item_id(tree: HTMLParser) -> str | None:
|
||
"""Извлечь item_id из [data-marker='item-view/item-id'] text."""
|
||
el = tree.css_first("[data-marker='item-view/item-id']")
|
||
if el is None:
|
||
return None
|
||
text_val = el.text(strip=True)
|
||
m = _ITEM_ID_RE.search(text_val)
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def _extract_price(tree: HTMLParser) -> int | None:
|
||
"""Извлечь price из [itemprop='price'] content attribute."""
|
||
el = tree.css_first("[itemprop='price']")
|
||
if el is None:
|
||
return None
|
||
content = el.attributes.get("content")
|
||
return _try_int(content)
|
||
|
||
|
||
def _extract_meta(tree: HTMLParser) -> tuple[date | None, int | None, int | None]:
|
||
"""Извлечь publish_date, views_total, views_today из item-id блока."""
|
||
el = tree.css_first("[data-marker='item-view/item-id']")
|
||
if el is None:
|
||
return None, None, None
|
||
|
||
# Поиск по всему тексту включая дочерние элементы
|
||
full_text = el.text(strip=False)
|
||
|
||
publish_date: date | None = None
|
||
m_date = _PUBLISH_DATE_RE.search(full_text)
|
||
if m_date:
|
||
try:
|
||
day = int(m_date.group(1))
|
||
month_word = m_date.group(2).lower()
|
||
month = RUS_MONTHS.get(month_word)
|
||
# Год — текущий (Avito не показывает год для свежих объявлений)
|
||
import datetime
|
||
|
||
current_year = datetime.date.today().year
|
||
if month:
|
||
publish_date = date(current_year, month, day)
|
||
except (ValueError, KeyError):
|
||
pass
|
||
|
||
views_total: int | None = None
|
||
m_total = _VIEWS_TOTAL_RE.search(full_text)
|
||
if m_total:
|
||
views_total = int(m_total.group(1))
|
||
|
||
views_today: int | None = None
|
||
m_today = _VIEWS_TODAY_RE.search(full_text)
|
||
if m_today:
|
||
views_today = int(m_today.group(1))
|
||
|
||
# Попробуем найти views в отдельных data-marker элементах (новый Avito layout)
|
||
if views_total is None:
|
||
v_el = tree.css_first("[data-marker='item-view/total-views']")
|
||
if v_el is not None:
|
||
m = _VIEWS_TOTAL_RE.search(v_el.text(strip=True))
|
||
if m:
|
||
views_total = int(m.group(1))
|
||
|
||
if views_today is None:
|
||
t_el = tree.css_first("[data-marker='item-view/today-views']")
|
||
if t_el is not None:
|
||
m = _VIEWS_TODAY_RE.search(t_el.text(strip=True))
|
||
if m:
|
||
views_today = int(m.group(1))
|
||
|
||
return publish_date, views_total, views_today
|
||
|
||
|
||
def _parse_params_ul(ul: Node) -> dict[str, str]:
|
||
"""Парсим <ul> с <li>Label: Value</li> → {label: value} dict."""
|
||
result: dict[str, str] = {}
|
||
for li in ul.css("li"):
|
||
text_val = li.text(strip=True)
|
||
if ":" in text_val:
|
||
label, _, value = text_val.partition(":")
|
||
label = label.strip()
|
||
value = value.strip()
|
||
if label and value:
|
||
result[label] = value
|
||
return result
|
||
|
||
|
||
def _extract_metro(text_val: str) -> list[dict[str, Any]]:
|
||
"""Извлечь станции метро из текста описания.
|
||
|
||
Примеры: 'Чкаловская 11-15 мин пешком', 'от 5 мин пешком'.
|
||
"""
|
||
stations: list[dict[str, Any]] = []
|
||
for m in METRO_RE.finditer(text_val):
|
||
name = m.group(1)
|
||
if m.group(2) and m.group(3):
|
||
# диапазон: 11-15
|
||
min_from = int(m.group(2))
|
||
min_to = int(m.group(3))
|
||
elif m.group(4):
|
||
# "от N мин"
|
||
min_from = None
|
||
min_to = int(m.group(4))
|
||
else:
|
||
continue
|
||
|
||
stations.append(
|
||
{
|
||
"name": name,
|
||
"min_to": min_to,
|
||
"min_from": min_from,
|
||
"mode": "walk",
|
||
}
|
||
)
|
||
return stations
|