bot-backend
4f0cff92ec
Some checks failed
CI / changes (push) Successful in 8s
CI / changes (pull_request) Successful in 7s
CI / backend-tests (push) Has been skipped
CI / openapi-codegen-check (push) Has been skipped
CI / frontend-tests (push) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Successful in 1m39s
CI / backend-tests (pull_request) Failing after 8m48s
Replace whole-HTML re.search for status with a 3-level block-scoped strategy: 1. CSS badge classes (status/badge/tag/chip/label) — highest precision. 2. Proximity to block labelled «Статус» via _find_text_near. 3. Full blocks scan where sold/reserved keywords always beat free — preventing «в продаже» from nav/similar-flats sections from misclassifying sold flats as free. Add _STATUS_KW_RE and _classify_status_kw at module level with full morphological coverage: продан/продана/продано, реализован[аоы]?, забронирован[аоы]?, свободн[аоы]?. Add 31 tests in test_domrf_catalog_parse.py covering all three extraction levels plus regression for fem. word-form «Квартира продана».
694 lines
30 KiB
Python
694 lines
30 KiB
Python
"""DOM.РФ catalog-квартир HTML scraper (issue #297 22d).
|
||
|
||
kn-API не возвращает цену для большинства квартир (91.5% NULL). Цены живут на
|
||
отдельной странице каталога:
|
||
https://наш.дом.рф/сервисы/каталог-квартир/квартира/{catalog_url_hash}
|
||
|
||
Этот модуль:
|
||
1. Строит URL каталога по `catalog_url_hash` (колонка появляется после миграции 22b).
|
||
2. Получает SSR-HTML через BrowserSession (Playwright, anti-bot — тот же паттерн
|
||
что и get_json, но возвращает HTML text вместо JSON).
|
||
3. Извлекает price_rub, status, finishing_type, ceiling_height_m, section_no,
|
||
catalog_updated_at из HTML с помощью stdlib `html.parser` + regex.
|
||
4. Пишет только catalog-derived поля через UPDATE ... WHERE ods_id = :ods_id —
|
||
НЕ перетирает kn-API метаданные (total_area, rooms и т.д.).
|
||
|
||
Зависимости: нет новых. Использует `html.parser` из stdlib + `re`.
|
||
NOTE: beautifulsoup4 НЕ установлен (нет в pyproject.toml). Если потребуется
|
||
структурированный парсинг — добавить `beautifulsoup4>=4.12` в pyproject.toml
|
||
и заменить _HtmlTextExtractor на `BeautifulSoup(html, "html.parser")`.
|
||
|
||
Контекст Roadmap:
|
||
- Phase 5 (22d) — catalog scraper для цен
|
||
- Согласно update 2026-05-17 (Objective goldmine): Objective уже содержит 81.4%
|
||
цен. Для `domrf_kn_flats` этот scraper остаётся полезен для полей:
|
||
finishing_type, ceiling_height_m, section_no, catalog_updated_at, catalog_url_hash.
|
||
Price coverage через Objective (OBJ-3) — приоритетнее.
|
||
|
||
Wiring (отдельный PR):
|
||
- Celery task: `backend/app/workers/tasks/scrape_catalog.py`
|
||
- Beat schedule: кварть + `catalog_updated_at < NOW() - INTERVAL '30 days'`
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import re
|
||
from datetime import date
|
||
from html.parser import HTMLParser
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.services.scrapers.stealth import BASE_URL, BrowserSession, jitter_sleep
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Per-flat catalog page URL template (IDN encoded — same as what browsers send).
|
||
# Человекочитаемый вид: https://наш.дом.рф/сервисы/каталог-квартир/квартира/{hash}
|
||
CATALOG_FLAT_PATH = "/сервисы/каталог-квартир/квартира/{catalog_url_hash}"
|
||
|
||
# JS snippet: выполняется внутри живой Playwright-страницы.
|
||
# Возвращает HTML текст страницы (text/html).
|
||
# Это аналог _FETCH_JS из stealth.py, но для text/html вместо application/json.
|
||
_FETCH_HTML_JS = """
|
||
async ({url}) => {
|
||
try {
|
||
const r = await fetch(url, {credentials: 'include'});
|
||
const ctype = r.headers.get('content-type') || '';
|
||
const body = await r.text();
|
||
return {ok: r.ok, status: r.status, body, contentType: ctype};
|
||
} catch (e) {
|
||
return {ok: false, status: 0, body: String(e), contentType: ''};
|
||
}
|
||
}
|
||
"""
|
||
|
||
# Нормализованные значения статуса продажи.
|
||
STATUS_FREE = "free"
|
||
STATUS_SOLD = "sold"
|
||
STATUS_RESERVED = "reserved"
|
||
|
||
# Паттерны для извлечения статуса (issue #1609).
|
||
# Все морфоварианты: продан/продана/продано, забронирован[аоы]?, реализован[аоы]?.
|
||
_STATUS_KW_RE = re.compile(
|
||
r"(в\s*продаже|свободн[аоы]?|free"
|
||
r"|продан[аоы]?|реализован[аоы]?|sold"
|
||
r"|забронирован[аоы]?|бронь|reserved)",
|
||
re.IGNORECASE | re.UNICODE,
|
||
)
|
||
# CSS-классы, характерные для статус-бейджей в Next.js/Tailwind.
|
||
_STATUS_BADGE_CLS_RE = re.compile(r"status|badge|tag|chip|label", re.IGNORECASE)
|
||
|
||
|
||
def _classify_status_kw(matched_text: str) -> str | None:
|
||
"""Классифицировать совпавшее ключевое слово в STATUS_*.
|
||
|
||
sold/reserved проверяются ПЕРВЫМИ — они специфичнее "в продаже" и никогда
|
||
не появляются в навигационных блоках страницы, в отличие от «в продаже».
|
||
"""
|
||
s = matched_text.lower()
|
||
if any(kw in s for kw in ("продан", "реализован", "sold")):
|
||
return STATUS_SOLD
|
||
if any(kw in s for kw in ("бронь", "забронирован", "reserved")):
|
||
return STATUS_RESERVED
|
||
if any(kw in s for kw in ("продаже", "свободн", "free")):
|
||
return STATUS_FREE
|
||
return None
|
||
|
||
|
||
# ── HTML fetching ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
async def fetch_catalog_html(session: BrowserSession, catalog_url_hash: str) -> str:
|
||
"""Получить SSR-HTML страницы квартиры в каталоге DOM.РФ.
|
||
|
||
Использует тот же паттерн что get_json(): fetch() внутри живой Playwright-страницы.
|
||
Так WAF-fingerprint идентичен браузеру, cookies проброшены автоматически.
|
||
|
||
Raises:
|
||
RuntimeError: при транзиентной ошибке после 5 попыток.
|
||
WafBlockedError: (из stealth) если вернулся JS-challenge вместо HTML.
|
||
"""
|
||
if session._page is None:
|
||
raise RuntimeError("BrowserSession not bootstrapped")
|
||
|
||
url = BASE_URL + CATALOG_FLAT_PATH.format(catalog_url_hash=catalog_url_hash)
|
||
last_err: Exception | None = None
|
||
|
||
for attempt in range(5):
|
||
async with session._sem:
|
||
await jitter_sleep()
|
||
try:
|
||
session._request_count += 1
|
||
result = await session._page.evaluate(_FETCH_HTML_JS, {"url": url})
|
||
except Exception as exc:
|
||
last_err = exc
|
||
logger.warning(
|
||
"catalog html evaluate err attempt=%d hash=%s: %r",
|
||
attempt,
|
||
catalog_url_hash,
|
||
exc,
|
||
)
|
||
await asyncio.sleep(2**attempt)
|
||
continue
|
||
|
||
status: int = result.get("status", 0)
|
||
body: str = result.get("body", "")
|
||
ctype: str = result.get("contentType", "")
|
||
|
||
if status in (429,) or status >= 500 or status == 0:
|
||
last_err = RuntimeError(f"transient status={status}")
|
||
logger.warning(
|
||
"catalog html transient status=%d attempt=%d hash=%s, backing off",
|
||
status,
|
||
attempt,
|
||
catalog_url_hash,
|
||
)
|
||
await asyncio.sleep(2**attempt)
|
||
continue
|
||
|
||
if status == 404:
|
||
raise RuntimeError(f"catalog 404 for hash={catalog_url_hash}")
|
||
|
||
if status != 200:
|
||
raise RuntimeError(f"catalog http {status}: {body[:200]} hash={catalog_url_hash}")
|
||
|
||
# Успех — но нужно проверить что не пришёл WAF JS-challenge (нет text/html)
|
||
# Страница каталога — SSR, всегда text/html. Если что-то другое — WAF.
|
||
if body and "text/html" not in ctype and "<!doctype" not in body[:100].lower():
|
||
logger.warning(
|
||
"catalog unexpected content-type=%r for hash=%s, treating as HTML anyway",
|
||
ctype,
|
||
catalog_url_hash,
|
||
)
|
||
|
||
if not body:
|
||
raise RuntimeError(f"catalog empty body for hash={catalog_url_hash}")
|
||
|
||
return body
|
||
|
||
raise RuntimeError(f"catalog html max retries exhausted hash={catalog_url_hash}: {last_err!r}")
|
||
|
||
|
||
# ── HTML parsing ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
class _TextCollector(HTMLParser):
|
||
"""Минимальный HTMLParser: собирает текстовые блоки с сохранением атрибутов class/data-*.
|
||
|
||
Вместо полного DOM строим плоский список (tag, attrs_dict, text) для regex-поиска.
|
||
Это сознательный trade-off: не нужен beautifulsoup4 — stdlib достаточно для
|
||
extraction известных структур страницы каталога.
|
||
"""
|
||
|
||
# HTML5 void-элементы: не имеют закрывающего тега → handle_endtag не вызывается.
|
||
# Если пушить их в стек/буфер, чужой handle_endtag поп'ает чужой фрейм →
|
||
# рассинхрон стека, текст блоков теряется (issue #1608).
|
||
_VOID_TAGS = frozenset(
|
||
{
|
||
"area",
|
||
"base",
|
||
"br",
|
||
"col",
|
||
"embed",
|
||
"hr",
|
||
"img",
|
||
"input",
|
||
"link",
|
||
"meta",
|
||
"param",
|
||
"source",
|
||
"track",
|
||
"wbr",
|
||
}
|
||
)
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self._stack: list[tuple[str, dict[str, str]]] = []
|
||
# Список записей: (class_hint, full_text)
|
||
self.blocks: list[tuple[str, str]] = []
|
||
self._buf: list[str] = []
|
||
|
||
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
||
if tag in self._VOID_TAGS:
|
||
return # void-теги не имеют endtag — не открываем фрейм (issue #1608)
|
||
attr_dict = {k: (v or "") for k, v in attrs}
|
||
self._stack.append((tag, attr_dict))
|
||
self._buf.append("") # начало нового буфера для этого тега
|
||
|
||
def handle_endtag(self, _tag: str) -> None:
|
||
if _tag in self._VOID_TAGS:
|
||
return # void-теги фрейм не открывали — нечего поп'ать (issue #1608)
|
||
if not self._stack:
|
||
return
|
||
tag, attr_dict = self._stack.pop()
|
||
text = (self._buf.pop() if self._buf else "").strip()
|
||
cls = attr_dict.get("class", "")
|
||
if text and (cls or tag in ("h1", "h2", "h3", "p", "span", "div", "li")):
|
||
self.blocks.append((cls, text))
|
||
# Propagate accumulated text up to parent buffer
|
||
if self._buf:
|
||
self._buf[-1] += " " + text
|
||
|
||
def handle_data(self, data: str) -> None:
|
||
if self._buf:
|
||
self._buf[-1] += data
|
||
|
||
|
||
def _find_text_near(
|
||
blocks: list[tuple[str, str]], label_pattern: str, value_pattern: str | None = None
|
||
) -> str | None:
|
||
"""Найти текстовое значение рядом с блоком, matching label_pattern.
|
||
|
||
Стратегия: ищем блок где text матчит label_pattern — берём следующий блок
|
||
как значение (value_pattern если задан).
|
||
"""
|
||
label_re = re.compile(label_pattern, re.IGNORECASE)
|
||
for i, (_cls, block_text) in enumerate(blocks):
|
||
if label_re.search(block_text):
|
||
# Попробовать next block
|
||
for j in range(i + 1, min(i + 4, len(blocks))):
|
||
candidate = blocks[j][1].strip()
|
||
if candidate:
|
||
if value_pattern is None:
|
||
return candidate
|
||
if re.search(value_pattern, candidate, re.IGNORECASE):
|
||
return candidate
|
||
return None
|
||
|
||
|
||
def parse_catalog_flat(html: str) -> dict[str, Any]:
|
||
"""Извлечь поля из SSR-HTML страницы квартиры DOM.РФ.
|
||
|
||
Возвращаемые поля (None если не найдено):
|
||
- price_rub (int) Цена квартиры в рублях
|
||
- price_per_m2 (float) Цена за м² (если указана отдельно)
|
||
- status (str) 'free' | 'sold' | 'reserved'
|
||
- finishing_type (str) Тип отделки (Предчистовая, Чистовая, Без отделки, ...)
|
||
- ceiling_height_m (float) Высота потолков в метрах
|
||
- section_no (int) Номер подъезда / секции
|
||
- catalog_updated_at (date) Дата обновления информации на странице
|
||
|
||
Парсинг хрупкий по природе (SSR HTML DOM.РФ меняется без уведомлений).
|
||
Все extraction best-effort — KeyError/AttributeError обёрнуты внутри.
|
||
"""
|
||
result: dict[str, Any] = {}
|
||
|
||
# ── Шаг 1: собрать все текстовые блоки через HTMLParser ──────────────────
|
||
collector = _TextCollector()
|
||
try:
|
||
collector.feed(html)
|
||
except Exception as exc:
|
||
logger.warning("html parse error (non-fatal): %s", exc)
|
||
|
||
blocks = collector.blocks
|
||
|
||
# ── Шаг 2: regex-extraction из полного HTML текста ────────────────────────
|
||
# Страница DOM.РФ SSR встраивает данные и в мета-тегах и в JSON-LD.
|
||
# Ищем в сыром HTML — надёжнее чем DOM-обход для хрупкой структуры.
|
||
|
||
# Price: "7 890 000 ₽" или "7 890 000 руб"
|
||
# Negative lookahead (?!\s*[/⁄]) исключает цену за м² ("217 835 ₽/м²"),
|
||
# которая на странице обычно выше полной стоимости и иначе матчилась бы
|
||
# первой → отбрасывалась санити-границей → price_rub=NULL (issue #1645).
|
||
price_match = re.search(
|
||
r"([\d][\d\s]{3,12}[\d])\s*(?:₽|руб)(?!\s*[/⁄])",
|
||
html,
|
||
re.UNICODE,
|
||
)
|
||
if price_match:
|
||
raw_price = re.sub(r"\s+", "", price_match.group(1))
|
||
try:
|
||
price_val = int(raw_price)
|
||
# Санity: цена квартиры в ЕКБ от 1 до 500 млн
|
||
if 1_000_000 <= price_val <= 500_000_000:
|
||
result["price_rub"] = price_val
|
||
except ValueError:
|
||
pass
|
||
|
||
# Price per m²: "217 835 ₽/м²"
|
||
ppm2_match = re.search(
|
||
r"([\d][\d\s]{2,9}[\d])\s*(?:₽|руб)[/⁄](?:м²|кв\.?\s*м)",
|
||
html,
|
||
re.UNICODE,
|
||
)
|
||
if ppm2_match:
|
||
raw_ppm2 = re.sub(r"\s+", "", ppm2_match.group(1))
|
||
try:
|
||
result["price_per_m2"] = float(raw_ppm2)
|
||
except ValueError:
|
||
pass
|
||
|
||
# Status extraction (issue #1609 fix).
|
||
#
|
||
# Проблема: re.search по всему HTML находит "в продаже" из навигации
|
||
# ("другие квартиры в продаже") РАНЬШЕ, чем реальный статус-бейдж →
|
||
# проданная/забронированная квартира ошибочно маркируется как free.
|
||
#
|
||
# Стратегия (3 уровня, от точного к широкому):
|
||
#
|
||
# 1. Блоки с CSS-классом, содержащим "status"/"badge"/"tag"/"chip"/"label" —
|
||
# типовые имена для статус-бейджей в Next.js/Tailwind.
|
||
# 2. Ближайший блок ПОСЛЕ лейбла "Статус" (proximity-поиск в blocks).
|
||
# 3. Полный перебор всех blocks: sold/reserved побеждают free
|
||
# (они никогда не появляются в навигации, в отличие от "в продаже").
|
||
#
|
||
# Морфоварианты: продан / продана / продано / продан → продан[аоы]?
|
||
# Аналогично: забронирован[аоы]? / реализован[аоы]? (синоним для sold).
|
||
|
||
# Уровень 1: блок с CSS-классом типа badge/status/tag
|
||
status_from_badge: str | None = None
|
||
for cls, block_text in blocks:
|
||
if _STATUS_BADGE_CLS_RE.search(cls):
|
||
m = _STATUS_KW_RE.search(block_text)
|
||
if m:
|
||
status_from_badge = _classify_status_kw(m.group(1))
|
||
if status_from_badge:
|
||
break
|
||
|
||
if status_from_badge:
|
||
result["status"] = status_from_badge
|
||
else:
|
||
# Уровень 2: блок сразу после лейбла "Статус"
|
||
status_label_value = _find_text_near(blocks, r"^статус$")
|
||
status_from_label: str | None = None
|
||
if status_label_value:
|
||
m2 = _STATUS_KW_RE.search(status_label_value)
|
||
if m2:
|
||
status_from_label = _classify_status_kw(m2.group(1))
|
||
|
||
if status_from_label:
|
||
result["status"] = status_from_label
|
||
else:
|
||
# Уровень 3: перебор всех blocks — sold/reserved побеждают free.
|
||
# Если встретили sold/reserved — сразу break (не появляются в nav).
|
||
# Если только free — запоминаем как кандидата (может быть перебит).
|
||
free_candidate: bool = False
|
||
sold_reserved_found: str | None = None
|
||
for _cls, block_text in blocks:
|
||
m3 = _STATUS_KW_RE.search(block_text)
|
||
if not m3:
|
||
continue
|
||
classified = _classify_status_kw(m3.group(1))
|
||
if classified in (STATUS_SOLD, STATUS_RESERVED):
|
||
sold_reserved_found = classified
|
||
break # точнее nav-текстов, дальше не ищем
|
||
if classified == STATUS_FREE:
|
||
free_candidate = True # продолжаем — вдруг sold встретится позже
|
||
if sold_reserved_found:
|
||
result["status"] = sold_reserved_found
|
||
elif free_candidate:
|
||
result["status"] = STATUS_FREE
|
||
|
||
# Finishing type: "Предчистовая", "Чистовая", "Без отделки", "Под ключ"
|
||
finishing_match = re.search(
|
||
r"(предчистовая|чистовая|без\s+отделки|под\s+ключ|white\s+box|whitebox)",
|
||
html,
|
||
re.IGNORECASE | re.UNICODE,
|
||
)
|
||
if finishing_match:
|
||
result["finishing_type"] = finishing_match.group(1).strip().capitalize()
|
||
|
||
# Ceiling height: "2,7 м" или "2.7 м" или "высота потолков 2,7"
|
||
ceiling_match = re.search(
|
||
r"(?:высота\s*потолков?|потолки?)\D{0,20}?([\d][,.][\d])\s*м",
|
||
html,
|
||
re.IGNORECASE | re.UNICODE,
|
||
)
|
||
if not ceiling_match:
|
||
# Fallback: просто "2,7 м" в характеристиках квартиры (диапазон 2.0–4.5 м)
|
||
ceiling_match = re.search(
|
||
r"\b([2-4][,.][\d])\s*м\b",
|
||
html,
|
||
re.UNICODE,
|
||
)
|
||
if ceiling_match:
|
||
raw_ceil = ceiling_match.group(1).replace(",", ".")
|
||
try:
|
||
ceil_val = float(raw_ceil)
|
||
if 2.0 <= ceil_val <= 6.0:
|
||
result["ceiling_height_m"] = ceil_val
|
||
except ValueError:
|
||
pass
|
||
|
||
# Section (подъезд): "Подъезд 1", "Секция 3", "Подъезд №2"
|
||
section_match = re.search(
|
||
r"(?:подъезд|секция)\s*[№#]?\s*(\d+)",
|
||
html,
|
||
re.IGNORECASE | re.UNICODE,
|
||
)
|
||
if section_match:
|
||
try:
|
||
result["section_no"] = int(section_match.group(1))
|
||
except ValueError:
|
||
pass
|
||
|
||
# catalog_updated_at: "Информация обновлена 17.04.2026" или "Обновлено 17.04.2026"
|
||
updated_match = re.search(
|
||
r"(?:информация\s+обновлена|обновлено|обновлён?а?)\D{0,10}?(\d{1,2}[./]\d{1,2}[./]\d{4})",
|
||
html,
|
||
re.IGNORECASE | re.UNICODE,
|
||
)
|
||
if updated_match:
|
||
raw_dt = updated_match.group(1).replace("/", ".")
|
||
try:
|
||
parts = raw_dt.split(".")
|
||
if len(parts) == 3:
|
||
result["catalog_updated_at"] = date(int(parts[2]), int(parts[1]), int(parts[0]))
|
||
except (ValueError, IndexError):
|
||
pass
|
||
|
||
# ── Шаг 3: блочный fallback для ceiling / section (если regex не нашёл) ──
|
||
if "ceiling_height_m" not in result:
|
||
candidate = _find_text_near(blocks, r"потолк|высота", r"[23][,.][\d]")
|
||
if candidate:
|
||
m = re.search(r"([23][,.][\d])", candidate)
|
||
if m:
|
||
try:
|
||
v = float(m.group(1).replace(",", "."))
|
||
if 2.0 <= v <= 6.0:
|
||
result["ceiling_height_m"] = v
|
||
except ValueError:
|
||
pass
|
||
|
||
if "section_no" not in result:
|
||
candidate = _find_text_near(blocks, r"подъезд|секция", r"^\d+$")
|
||
if candidate:
|
||
try:
|
||
result["section_no"] = int(candidate.strip())
|
||
except ValueError:
|
||
pass
|
||
|
||
logger.debug(
|
||
"parse_catalog_flat: extracted fields=%s",
|
||
list(result.keys()),
|
||
)
|
||
return result
|
||
|
||
|
||
# ── DB writes ─────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def upsert_catalog_data(
|
||
db: Session, ods_id: str, catalog_url_hash: str, data: dict[str, Any]
|
||
) -> bool:
|
||
"""UPDATE catalog-derived поля в domrf_kn_flats для конкретной квартиры.
|
||
|
||
Обновляет ТОЛЬКО catalog-only колонки:
|
||
price_rub, price_per_m2, status, finishing_type, ceiling_height_m,
|
||
section_no, catalog_updated_at, catalog_url_hash.
|
||
|
||
НЕ трогает: total_area, rooms, floor, num_floors, flat_type, obj_id и
|
||
другие kn-API метаданные.
|
||
|
||
Использует COALESCE: если новое значение NULL — старое сохраняется.
|
||
Это позволяет повторно запускать scraper не затирая частично заполненные поля.
|
||
|
||
ВАЖНО: колонки section_no, finishing_type, ceiling_height_m,
|
||
catalog_updated_at, catalog_url_hash должны существовать в таблице.
|
||
Они появляются после миграции 22b. Если таблица старая — UPDATE упадёт
|
||
с 'column does not exist'. Решение: сначала выполнить data/sql/NN_22b_flats_cols.sql.
|
||
|
||
Возвращает True если строка найдена и обновлена, False если ods_id не найден.
|
||
"""
|
||
params: dict[str, Any] = {
|
||
"ods_id": ods_id,
|
||
"catalog_url_hash": catalog_url_hash,
|
||
"price_rub": data.get("price_rub"),
|
||
"price_per_m2": data.get("price_per_m2"),
|
||
"status": data.get("status"),
|
||
"finishing_type": data.get("finishing_type"),
|
||
"ceiling_height_m": data.get("ceiling_height_m"),
|
||
"section_no": data.get("section_no"),
|
||
"catalog_updated_at": data.get("catalog_updated_at"),
|
||
}
|
||
|
||
try:
|
||
with db.begin_nested():
|
||
result = db.execute(
|
||
text(
|
||
"""
|
||
UPDATE domrf_kn_flats SET
|
||
catalog_url_hash = :catalog_url_hash,
|
||
price_rub = COALESCE(:price_rub, price_rub),
|
||
price_per_m2 = COALESCE(:price_per_m2, price_per_m2),
|
||
status = COALESCE(:status, status),
|
||
finishing_type = COALESCE(:finishing_type, finishing_type),
|
||
ceiling_height_m = COALESCE(:ceiling_height_m, ceiling_height_m),
|
||
section_no = COALESCE(:section_no, section_no),
|
||
catalog_updated_at = COALESCE(:catalog_updated_at, catalog_updated_at)
|
||
WHERE ods_id = :ods_id
|
||
"""
|
||
),
|
||
params,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("upsert_catalog_data ods_id=%s failed: %s", ods_id, exc)
|
||
return False
|
||
|
||
rows_affected: int = result.rowcount or 0
|
||
if rows_affected == 0:
|
||
logger.warning("upsert_catalog_data: ods_id=%s not found in domrf_kn_flats", ods_id)
|
||
return rows_affected > 0
|
||
|
||
|
||
# ── Per-flat scrape orchestration ─────────────────────────────────────────────
|
||
|
||
|
||
async def scrape_one_flat(
|
||
session: BrowserSession,
|
||
db: Session,
|
||
ods_id: str,
|
||
catalog_url_hash: str,
|
||
) -> dict[str, Any]:
|
||
"""Scrape одной квартиры: fetch HTML → parse → upsert.
|
||
|
||
Возвращает dict с результатом: {ods_id, success, fields_extracted, updated}.
|
||
Ошибки fetch/parse логируются, не бросаются — caller обрабатывает результат.
|
||
"""
|
||
outcome: dict[str, Any] = {
|
||
"ods_id": ods_id,
|
||
"catalog_url_hash": catalog_url_hash,
|
||
"success": False,
|
||
"fields_extracted": 0,
|
||
"updated": False,
|
||
"error": None,
|
||
}
|
||
|
||
try:
|
||
html = await fetch_catalog_html(session, catalog_url_hash)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"catalog fetch failed ods_id=%s hash=%s: %s",
|
||
ods_id,
|
||
catalog_url_hash,
|
||
exc,
|
||
)
|
||
outcome["error"] = str(exc)[:500]
|
||
return outcome
|
||
|
||
try:
|
||
data = parse_catalog_flat(html)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"catalog parse failed ods_id=%s hash=%s: %s",
|
||
ods_id,
|
||
catalog_url_hash,
|
||
exc,
|
||
)
|
||
outcome["error"] = f"parse: {exc!s}"[:500]
|
||
return outcome
|
||
|
||
outcome["fields_extracted"] = len([v for v in data.values() if v is not None])
|
||
outcome["updated"] = upsert_catalog_data(db, ods_id, catalog_url_hash, data)
|
||
# success отражает прохождение пайплайна fetch+parse без исключения; реально
|
||
# ли затронута строка в БД — см. outcome['updated']. Батч-статистика считает
|
||
# отдельный stats['updated'], чтобы не рапортовать ложно высокий success при
|
||
# ненайденном ods_id / пустом парсе (fields_extracted==0) — issue #1610.
|
||
outcome["success"] = True
|
||
if not outcome["updated"]:
|
||
logger.warning(
|
||
"catalog scrape ods_id=%s: fetched+parsed but DB row NOT updated "
|
||
"(fields=%d, ods_id missing or all-NULL parse)",
|
||
ods_id,
|
||
outcome["fields_extracted"],
|
||
)
|
||
logger.info(
|
||
"catalog scrape ods_id=%s: fields=%d updated=%s",
|
||
ods_id,
|
||
outcome["fields_extracted"],
|
||
outcome["updated"],
|
||
)
|
||
return outcome
|
||
|
||
|
||
async def scrape_catalog_batch(
|
||
db: Session,
|
||
flats: list[dict[str, Any]],
|
||
region_code: int = 66,
|
||
headed: bool = False,
|
||
load_state: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Scrape пачки квартир каталога DOM.РФ.
|
||
|
||
`flats` — список dict'ов с ключами {ods_id, catalog_url_hash}.
|
||
Типовой источник: SELECT ods_id, catalog_url_hash FROM domrf_kn_flats
|
||
WHERE catalog_url_hash IS NOT NULL
|
||
AND (catalog_updated_at IS NULL OR catalog_updated_at < NOW() - INTERVAL '30 days')
|
||
LIMIT :batch_size.
|
||
|
||
Использует один BrowserSession на весь пакет (bootstrapped 1 раз).
|
||
jitter_sleep между запросами встроен в fetch_catalog_html (через BrowserSession._sem).
|
||
|
||
Returns:
|
||
{total, success, updated, failed, fields_total}
|
||
- success: прошли fetch+parse без исключения
|
||
- updated: реально затронули строку в БД (issue #1610) — отражает
|
||
фактическое число записанных квартир, в отличие от success
|
||
"""
|
||
stats: dict[str, Any] = {
|
||
"total": len(flats),
|
||
"success": 0,
|
||
"updated": 0,
|
||
"failed": 0,
|
||
"fields_total": 0,
|
||
}
|
||
|
||
if not flats:
|
||
logger.info("scrape_catalog_batch: empty batch, nothing to do")
|
||
return stats
|
||
|
||
logger.info(
|
||
"scrape_catalog_batch: starting %d flats region=%d",
|
||
len(flats),
|
||
region_code,
|
||
)
|
||
|
||
async with BrowserSession(
|
||
region_code=region_code,
|
||
headed=headed,
|
||
load_state=load_state,
|
||
# auth=None — страницы каталога публичные, Basic auth не нужен
|
||
auth=None,
|
||
) as session:
|
||
for flat in flats:
|
||
ods_id = flat.get("ods_id", "")
|
||
catalog_url_hash = flat.get("catalog_url_hash", "")
|
||
if not ods_id or not catalog_url_hash:
|
||
logger.warning("scrape_catalog_batch: skip flat with missing ods_id/hash: %r", flat)
|
||
stats["failed"] += 1
|
||
continue
|
||
|
||
outcome = await scrape_one_flat(session, db, ods_id, catalog_url_hash)
|
||
if outcome["success"]:
|
||
stats["success"] += 1
|
||
stats["fields_total"] += outcome["fields_extracted"]
|
||
if outcome["updated"]:
|
||
stats["updated"] += 1
|
||
else:
|
||
stats["failed"] += 1
|
||
|
||
# Commit outer transaction: SAVEPOINT (`begin_nested`) в upsert_catalog_data
|
||
# лишь RELEASE'ит внутренний SAVEPOINT, но outer tx остаётся autobegin'd —
|
||
# без commit() все UPDATE'ы (цены, отделка, catalog_updated_at) откатятся
|
||
# при db.close() в Celery task. Зеркалит scrape_catalog_objects в
|
||
# domrf_catalog_object.py:460. Issue #1227.
|
||
try:
|
||
db.commit()
|
||
except Exception:
|
||
db.rollback()
|
||
raise
|
||
|
||
logger.info(
|
||
"scrape_catalog_batch done: total=%d success=%d updated=%d failed=%d fields_total=%d",
|
||
stats["total"],
|
||
stats["success"],
|
||
stats["updated"],
|
||
stats["failed"],
|
||
stats["fields_total"],
|
||
)
|
||
return stats
|