"""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 asyncio
import json
import logging
import random
import re
from dataclasses import dataclass, field
from datetime import date
from typing import TYPE_CHECKING, Any
from urllib.parse import quote, urljoin, urlparse
from curl_cffi.requests import AsyncSession
from selectolax.parser import HTMLParser, Node
from sqlalchemy import text
from sqlalchemy.orm import Session
from scraper_kit.avito_exceptions import (
AvitoBlockedError,
AvitoListingGoneError,
AvitoRateLimitedError,
)
from scraper_kit.geo import is_within_ekb_bbox_wide
from scraper_kit.providers.avito.serp import _clean_address, _is_firewall_page
from scraper_kit.providers.avito.shared import RUS_MONTHS
from scraper_kit.repair_state_normalizer import infer_repair_state_from_text
if TYPE_CHECKING:
from scraper_kit.browser_fetcher import BrowserFetcher
from scraper_kit.contracts import ScraperConfig
logger = logging.getLogger(__name__)
AVITO_BASE = "https://www.avito.ru"
# Живой прогрев сессии (#1551): GET Yandex SERP (органический referer-источник) -> GET
# Avito EKB search-страница (Referer=yandex) сеет антибот-куки, после чего прогретая
# curl_cffi-сессия держит батч detail-GET'ов без 403 (доказано пробами на проде: 22/22
# карточек подряд, 0 блоков). referer на detail-GET = URL avito-search страницы.
_AVITO_WARM_SEARCH_URL = "https://www.avito.ru/ekaterinburg/kvartiry/prodam?q=" + quote(
"купить квартиру"
)
_AVITO_WARM_YANDEX_URL = "https://yandex.ru/search/?text=" + quote(
"купить квартиру екатеринбург авито"
)
_AVITO_WARM_YANDEX_REFERER = "https://yandex.ru/"
# Reconnect-retry на detail-странице под backconnect-прокси — зеркало SERP-фикса
# (#1765/#1769, avito.py): 403/firewall = залочен лишь текущий exit-IP backconnect-прокси;
# пересоздание curl_cffi-сессии = новый CONNECT-туннель = свежий exit-IP → escape.
# 429 = transient conn-limit: сначала короткий retry на ТОЙ ЖЕ сессии. Но keep-alive
# detail-fetch переиспользует ОДИН backconnect exit-IP на последовательные запросы →
# avito лочит этот IP, и short-retry той же сессии не помогает (run 217: 5 подряд 429,
# abort на 19 enriched). Поэтому при исчерпании short-retry под backconnect делаем ещё
# _AVITO_DETAIL_429_RECONNECT_RETRIES попыток через эфемерную сессию (свежий exit-IP).
_AVITO_DETAIL_403_MAX_RETRIES = 5
_AVITO_DETAIL_403_BACKOFF_SEC = 2.0
_AVITO_DETAIL_429_MAX_RETRIES = 8
_AVITO_DETAIL_429_BACKOFF_SEC = 1.5
_AVITO_DETAIL_429_RECONNECT_RETRIES = 3
# ── 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)
def _build_detail_session(config: ScraperConfig | None = None) -> AsyncSession:
"""Создать curl_cffi-сессию для detail-fetch (impersonate chrome120, через прокси
если задан config.scraper_proxy_url).
Вынесено из own-session-ветки fetch_detail чтобы reconnect-retry мог пересоздать
эфемерную сессию (новый CONNECT-туннель = свежий backconnect exit-IP) под 403.
proxy=None → прямое подключение.
Strangler-инжекция (#2133): прокси приходит из инжектируемого `ScraperConfig`
(scraper_kit.contracts) вместо прямого импорта app.core.config.settings.
"""
proxy_url = config.scraper_proxy_url if config is not None else None
proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None
return 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",
},
)
async def warm_up_session(session: AsyncSession) -> bool:
"""Органический прогрев сессии: Yandex SERP (best-effort) -> Avito EKB search
(Referer=yandex). Сеет антибот-куки, после чего сессия держит батч detail-GET'ов
без 403. Возвращает True если search-страница загрузилась (не firewall).
"""
try:
await session.get(_AVITO_WARM_YANDEX_URL) # органический referer-источник, best-effort
except Exception as exc:
logger.debug("avito warm-up: yandex GET failed (non-fatal): %s", exc)
await asyncio.sleep(random.uniform(2.0, 3.5))
try:
r = await session.get(
_AVITO_WARM_SEARCH_URL, headers={"Referer": _AVITO_WARM_YANDEX_REFERER}
)
except Exception as exc:
logger.warning("avito warm-up: search fetch error: %s", exc)
return False
ok = r.status_code == 200 and not _is_firewall_page(r.text)
if not ok:
logger.warning("avito warm-up: search page blocked sc=%s", r.status_code)
return ok
async def research_in_session(session: AsyncSession) -> bool:
"""Лёгкий in-session перепоиск: re-GET avito search-страницы (Referer=yandex) на
ТЕКУЩЕЙ сессии — освежает антибот-куки БЕЗ смены exit-IP. Best-effort (search-страница
иногда 429, это не ломает detail-фетчи). Возвращает True если страница не firewall.
"""
try:
r = await session.get(
_AVITO_WARM_SEARCH_URL, headers={"Referer": _AVITO_WARM_YANDEX_REFERER}
)
except Exception as exc:
logger.warning("avito re-search: fetch error: %s", exc)
return False
ok = r.status_code == 200 and not _is_firewall_page(r.text)
if not ok:
logger.info("avito re-search: search page sc=%s (non-fatal)", r.status_code)
return ok
async def build_warmed_session() -> AsyncSession:
"""Построить detail-сессию (_build_detail_session) и прогреть её. Возвращает
прогретую сессию (даже если warm-up soft-failed — caller может всё равно пробовать).
"""
session = _build_detail_session()
await warm_up_session(session)
return session
# ── Soft-block detection (#1967 / #1950) ──────────────────────────────────────
# Под нагрузкой Avito иногда вместо detail-страницы отдаёт HTTP 200 с ОБЩЕЙ
# витриной (Авито — Объявления на сайте Авито) вместо запрошенного
# объявления — мягкий анти-бот deflect. У такой страницы НЕТ ни firewall-маркеров
# (_is_firewall_page=False), ни item-view → parse_detail_html падал
# ValueError("Cannot extract item_id") → generic failed, IP НЕ ротировался, enriched
# каскадно падал к 0 (прод runs 431→437: 115→82→8→2→1→0 при blocked=0). Детектим
# как блок → reconnect свежей exit-IP сессией / loop-ротация.
_DETAIL_SOFT_BLOCK_TITLE_MARKERS = ("объявления на сайте авито",)
def _is_detail_soft_block(html: str) -> bool:
"""True если detail-fetch вернул HTTP 200, но это анти-бот deflect — общая
витрина Авито вместо объявления.
Сигнатура строго специфична deflect-витрине, чтобы НЕ задевать легитимные
не-блок ответы:
* нормальное объявление содержит item-view-маркеры → False;
* снятое объявление Avito рендерит как item-view + баннер «снято» → False;
* 404 / не-200 обрабатывается выше (ValueError → failed), сюда не попадает.
Положительный сигнал: НЕТ item-view-структуры И generic-витринный .
"""
if "item-view/item-id" in html or "item-view/title-info" in html:
return False
head = html[:16384].lower()
return any(m in head for m in _DETAIL_SOFT_BLOCK_TITLE_MARKERS)
def _is_detail_not_found(html: str) -> bool:
"""True if the page is Avito's 404 / removed-listing page (not an anti-bot deflect).
Guard: a real listing has item-view → False. Keyed on the specific 404 title so it
does NOT catch the soft-block deflect ("объявления на сайте авито" without 404 text)."""
if "item-view/item-id" in html or "item-view/title-info" in html:
return False
head = html[:16384].lower()
return "страница не найдена" in head or "ошибка 404" in head
# ── fetch_detail ──────────────────────────────────────────────────────────────
async def fetch_detail(
item_url: str,
*,
cffi_session: AsyncSession | None = None,
browser_fetcher: BrowserFetcher | None = None,
referer: str | None = None,
reconnect_on_block: bool = True,
config: ScraperConfig | None = None,
) -> DetailEnrichment:
"""GET <avito>/{item_url} → parse HTML via selectolax → DetailEnrichment.
Если browser_fetcher передан — использует браузерный fetch (browser mode).
Если cffi_session не передана — создаёт новую (impersonate='chrome120').
referer — если задан, шлётся в Referer-заголовке detail-GET'а (warm-batch #1551:
referer = URL avito-search страницы, на которой прогрелась сессия).
reconnect_on_block — если False, 403/firewall поднимает AvitoBlockedError СРАЗУ
(без холодных reconnect'ов — ими займётся outer-loop через rebuild+rewarm прогретой
сессии); 429 после short-retry поднимает AvitoRateLimitedError. Дефолт True
сохраняет старое поведение own-session / scrape_pipeline путей.
Raises:
httpx.HTTPError — если status != 200 (через raise_for_status-like).
ValueError — если item_id не извлечён из HTML.
"""
if browser_fetcher is not None:
# Browser-only режим (#1814): когда передан browser_fetcher, fetch идёт
# СТРОГО через camoufox-сервис. Никакой curl_cffi-сессии и никакого
# curl-fallback — avito-антибот банит curl-фингерпринт под нагрузкой
# (429 CONNECT tunnel failed), а browser проходит. Любой firewall/ошибка
# браузера → AvitoBlockedError; caller (backfill) пометит failed и пойдёт
# дальше, НЕ трогая curl.
full_url = item_url if item_url.startswith("http") else urljoin(AVITO_BASE, item_url)
try:
html = await browser_fetcher.fetch(full_url)
except (AvitoBlockedError, AvitoRateLimitedError):
raise
except Exception as exc:
# browser-сервис вернул HTTPError / ConnectError / 502 (firewall, dead
# listing, redirect). НЕ откатываемся в curl — поднимаем block, чтобы
# caller пометил listing и продолжил без curl-фингерпринта.
raise AvitoBlockedError(
f"Avito detail browser fetch failed for {full_url}: {exc}"
) from exc
# 404/removed-listing (#2034): мёртвый листинг рендерит «Ошибка 404» без
# item-view. Это НЕ анти-бот deflect — отдельный сигнал, чтобы backfill
# пометил листинг is_active=FALSE и НЕ крутил consecutive-block breaker
# (run 458: 5 dead подряд → abort до live-листингов). Проверяем ДО
# firewall/soft-block (иначе 404-страница ловится как soft-block).
if _is_detail_not_found(html):
raise AvitoListingGoneError(
f"Avito detail 404/listing-gone (browser-mode) for {full_url}"
)
if _is_firewall_page(html) or _is_detail_soft_block(html):
raise AvitoBlockedError(
f"Avito detail firewall/soft-block (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 → прямое подключение.
session = _build_detail_session(config)
else:
session = cffi_session
full_url = item_url if item_url.startswith("http") else urljoin(AVITO_BASE, item_url)
# curl_cffi egress всегда через scraper_proxy_url (backconnect mproxy). 403/firewall =
# залочен лишь текущий exit-IP → reconnect новой сессией = свежий IP. Гейтим reconnect
# ТОЛЬКО на наличие proxy (зеркало SERP-фикса #1765). Без proxy — старое поведение
# (raise на первом 403). Caller-provided сессию НЕ трогаем: reconnect использует
# эфемерную retry_session, которую сами закрываем в finally.
# reconnect_on_block=False (warm-batch #1551): не делаем холодных per-card
# reconnect'ов — блок поднимает AvitoBlockedError сразу, outer-loop пересоздаёт+
# прогревает сессию (свежий exit-IP). reconnect_on_block=True (own-session /
# scrape_pipeline) — старое поведение: backconnect-reconnect эфемерной сессией.
backconnect = (
bool(config.scraper_proxy_url if config is not None else None) and reconnect_on_block
)
req_headers = {"Referer": referer} if referer else None
r403 = 0
r429 = 0
r429_reconnect = 0
attempt_session = session
retry_session: AsyncSession | None = None
try:
while True:
response = await attempt_session.get(full_url, headers=req_headers)
sc = response.status_code
is_firewall = sc == 200 and (
_is_firewall_page(response.text) or _is_detail_soft_block(response.text)
)
# 429 через backconnect = transient conn-limit, НЕ IP-ban: короткий retry той
# же сессии без reconnect (зеркало SERP). Только при исчерпании — raise.
if sc == 429 and r429 < _AVITO_DETAIL_429_MAX_RETRIES:
r429 += 1
logger.info(
"avito detail HTTP 429 (backconnect conn-limit) — short retry %d/%d",
r429,
_AVITO_DETAIL_429_MAX_RETRIES,
)
await asyncio.sleep(_AVITO_DETAIL_429_BACKOFF_SEC)
continue
if sc == 403 or is_firewall:
if backconnect and r403 < _AVITO_DETAIL_403_MAX_RETRIES:
r403 += 1
# Эфемерная свежая сессия (новый CONNECT-туннель = свежий exit-IP).
# Caller-provided/own session НЕ пересоздаём — только retry_session.
if retry_session is not None:
try:
await retry_session.close()
except Exception:
pass
retry_session = _build_detail_session()
attempt_session = retry_session
logger.info(
"avito detail 403/firewall/soft-block %s — backconnect reconnect "
"retry %d/%d",
full_url,
r403,
_AVITO_DETAIL_403_MAX_RETRIES,
)
await asyncio.sleep(_AVITO_DETAIL_403_BACKOFF_SEC)
continue
raise AvitoBlockedError(f"Avito detail HTTP 403 for {full_url}")
if sc == 429:
# short-retry (та же сессия) исчерпан. Под backconnect — ещё несколько
# попыток через эфемерную сессию (свежий exit-IP): keep-alive держал ОДИН
# залоченный IP, новый CONNECT-туннель уходит на другой. Делит ту же
# retry_session-машинерию что и 403-путь (close-prev → rebuild → retry,
# cleanup в finally). Только потом raise.
if backconnect and r429_reconnect < _AVITO_DETAIL_429_RECONNECT_RETRIES:
r429_reconnect += 1
# Caller-provided/own session НЕ пересоздаём — только retry_session.
if retry_session is not None:
try:
await retry_session.close()
except Exception:
pass
retry_session = _build_detail_session()
attempt_session = retry_session
logger.info(
"avito detail HTTP 429 short-retry exhausted %s — "
"backconnect reconnect retry %d/%d",
full_url,
r429_reconnect,
_AVITO_DETAIL_429_RECONNECT_RETRIES,
)
await asyncio.sleep(_AVITO_DETAIL_403_BACKOFF_SEC)
continue
raise AvitoRateLimitedError(f"Avito detail HTTP 429 for {full_url}")
if sc != 200:
raise ValueError(f"avito detail HTTP {sc} for {full_url}")
return parse_detail_html(response.text, full_url)
finally:
if own_session:
await session.close()
if retry_session is not None and retry_session is not session:
try:
await retry_session.close()
except Exception:
pass
# ── 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"))
# bbox-guard (#1871): detail-страница иногда отдаёт координаты вне ЕКБ
# (Питер/Тюмень/Уфа — кросс-региональные дубли объявлений). Эти координаты
# попадают в UPDATE COALESCE без bbox-валидации (в отличие от geocoder.py,
# где bbox-фильтр везде). Сбрасываем в None → листинг уходит в geocode-cron
# путь, который сам отфильтрует по адресу. SERP-ingest координат не даёт вовсе.
if lat is not None and lon is not None and not is_within_ekb_bbox_wide(lat, lon):
logger.warning(
"avito_detail #1871: dropped non-EKB coords item=%s lat=%s lon=%s",
item_id,
lat,
lon,
)
lat = lon = None
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_type СОХРАНЯЕТСЯ (уже нормализован парсером → канон panel/brick/monolith/
monolith_brick/block/wood) — нужен estimator soft-penalty по типу дома.
house_catalog_url СОХРАНЯЕТСЯ в listings.house_url (линковка дома); отдельной
house_catalog_url колонки нет. house_type — COALESCE existing-first (не затираем
canonical из Houses Catalog Stage 2c); house_url — new-first additive (avito-owned
поле, нет конкурирующего canonical-писателя).
Остальные house-level поля (passenger_elevators, cargo_elevators, has_concierge,
closed_yard, total_floors_house) по-прежнему опускаются — нет колонок в listings,
canonical приходит из Houses Catalog Stage 2c (отдельный follow-up).
"""
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),
house_type = COALESCE(house_type, :house_type),
house_url = COALESCE(:house_catalog_url, house_url),
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,
"house_type": e.house_type,
"house_catalog_url": e.house_catalog_url,
},
)
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