From 1db19b3b8ab356aa75e228a6280002787b7923dd Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 17 May 2026 22:34:58 +0300 Subject: [PATCH] feat(scrape-kn): parallel Phase B/C + wire obj_checks + documents + is_ekb auto-derive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 4 changes: 1. asyncio.gather() в Phase B/C — 6 endpoints/obj параллельно через _fetch_*_safe wrappers. ~6x speedup (3.5h -> ~35min). BrowserSession._sem(3) bounds concurrent requests автоматически. 2. Wire documents extractor (services/scrapers/documents.py) — per-object call в parallel gather. Заполняет domrf_kn_documents (был пуст после PR #305 schema-only). 3. Create services/scrapers/obj_checks.py extractor + wire в loop. Заполняет domrf_obj_checks (был пуст после PR #303 schema-only). Endpoint URL /checks не верифицирован devtools — выведен по паттерну /infrastructure; при ошибке пишет в kn_scrape_failures. 4. Phase E (после Phase D) — UPDATE is_ekb=TRUE для свежего snapshot по district_name ILIKE екатеринбург/свердловск. Best-effort (exception caught, scrape не прерывается). --- backend/app/services/scrapers/domrf_kn.py | 288 ++++++++++++++++---- backend/app/services/scrapers/obj_checks.py | 175 ++++++++++++ 2 files changed, 408 insertions(+), 55 deletions(-) create mode 100644 backend/app/services/scrapers/obj_checks.py diff --git a/backend/app/services/scrapers/domrf_kn.py b/backend/app/services/scrapers/domrf_kn.py index eab5e9a5..9fe8c433 100644 --- a/backend/app/services/scrapers/domrf_kn.py +++ b/backend/app/services/scrapers/domrf_kn.py @@ -16,6 +16,7 @@ auth shipped in their frontend bundle). from __future__ import annotations +import asyncio import json import logging from datetime import date, datetime @@ -26,6 +27,8 @@ from sqlalchemy import text from sqlalchemy.orm import Session from app.core.db import SessionLocal +from app.services.scrapers.documents import extract_documents, upsert_documents +from app.services.scrapers.obj_checks import extract_obj_checks, upsert_obj_checks from app.services.scrapers.stealth import BASE_URL, BrowserSession logger = logging.getLogger(__name__) @@ -36,6 +39,10 @@ PATH_SALE_GRAPH = "/сервисы/api/object/{obj_id}/sale_graph" # ?type=apar PATH_SALES_AGG = "/сервисы/api/object/{obj_id}/sales_agg" PATH_INFRA = "/сервисы/api/object/{obj_id}/infrastructure" PATH_PHOTOS = "/сервисы/api/object/construction/progress/photo/{obj_id}" +PATH_DOCUMENTS = "/сервисы/api/object/{obj_id}/documents" +# NOTE: PATH_CHECKS URL not verified via devtools — likely pattern analogous to /infrastructure. +# If endpoint returns 404/error the failure is logged to kn_scrape_failures and scrape continues. +PATH_CHECKS = "/сервисы/api/object/{obj_id}/checks" RAW_SECTION = "kn_api" SALE_GRAPH_TYPES = ("apartments", "parking") @@ -990,6 +997,111 @@ async def fetch_photos(sess: BrowserSession, obj_id: int) -> tuple[list[dict[str return (payload.get("data") or []), full_url +async def fetch_documents(sess: BrowserSession, obj_id: int) -> tuple[list[dict[str, Any]], str]: + """Fetch PDF-document list for one object. Returns (items, full_url).""" + path = PATH_DOCUMENTS.format(obj_id=obj_id) + payload = await sess.get_json(path, {}) + full_url = f"{BASE_URL}{path}" + if isinstance(payload, list): + return payload, full_url + return (payload.get("data") or []), full_url + + +async def fetch_obj_checks(sess: BrowserSession, obj_id: int) -> tuple[Any, str]: + """Fetch 6 «Проверено на наш.дом.рф» checks for one object. Returns (payload, full_url). + + Endpoint URL (/checks) не верифицирован через devtools — выведен по паттерну + /infrastructure, /documents. При HTTP-ошибке вызывающий код запишет failure в + kn_scrape_failures; scrape не прерывается. + """ + path = PATH_CHECKS.format(obj_id=obj_id) + payload = await sess.get_json(path, {}) + return payload, f"{BASE_URL}{path}" + + +# ── _fetch_*_safe wrappers for asyncio.gather in Phase B/C ─────────────────── +# Каждый wrapper возвращает (kind, full_url, result_or_exception). +# Exceptions НЕ raise — помещаются в возвращаемый tuple. +# BrowserSession._sem (Semaphore(3)) bounds concurrency per-request автоматически. + + +async def _fetch_flats_safe( + sess: BrowserSession, obj_id: int +) -> tuple[str, str, list[dict[str, Any]] | Exception]: + full_url = f"{BASE_URL}{PATH_FLATS_TABLE}?externalId={obj_id}" + try: + flats = await fetch_flats_for_object(sess, obj_id) + return ("flats", full_url, flats) + except Exception as e: + return ("flats", full_url, e) + + +async def _fetch_sale_graph_safe( + sess: BrowserSession, obj_id: int, type_: str +) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]: + full_url = f"{BASE_URL}{PATH_SALE_GRAPH.format(obj_id=obj_id)}?type={type_}" + try: + rows, url = await fetch_sale_graph(sess, obj_id, type_) + return (f"sale_graph_{type_}", url, (rows, url)) + except Exception as e: + return (f"sale_graph_{type_}", full_url, e) + + +async def _fetch_sales_agg_safe( + sess: BrowserSession, obj_id: int +) -> tuple[str, str, tuple[dict[str, Any], str] | Exception]: + full_url = f"{BASE_URL}{PATH_SALES_AGG.format(obj_id=obj_id)}" + try: + agg, url = await fetch_sales_agg(sess, obj_id) + return ("sales_agg", url, (agg, url)) + except Exception as e: + return ("sales_agg", full_url, e) + + +async def _fetch_infrastructure_safe( + sess: BrowserSession, obj_id: int +) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]: + full_url = f"{BASE_URL}{PATH_INFRA.format(obj_id=obj_id)}" + try: + pois, url = await fetch_infrastructure(sess, obj_id) + return ("infrastructure", url, (pois, url)) + except Exception as e: + return ("infrastructure", full_url, e) + + +async def _fetch_photos_safe( + sess: BrowserSession, obj_id: int +) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]: + full_url = f"{BASE_URL}{PATH_PHOTOS.format(obj_id=obj_id)}" + try: + photos, url = await fetch_photos(sess, obj_id) + return ("photos", url, (photos, url)) + except Exception as e: + return ("photos", full_url, e) + + +async def _fetch_documents_safe( + sess: BrowserSession, obj_id: int +) -> tuple[str, str, tuple[list[dict[str, Any]], str] | Exception]: + full_url = f"{BASE_URL}{PATH_DOCUMENTS.format(obj_id=obj_id)}" + try: + items, url = await fetch_documents(sess, obj_id) + return ("documents", url, (items, url)) + except Exception as e: + return ("documents", full_url, e) + + +async def _fetch_obj_checks_safe( + sess: BrowserSession, obj_id: int +) -> tuple[str, str, tuple[Any, str] | Exception]: + full_url = f"{BASE_URL}{PATH_CHECKS.format(obj_id=obj_id)}" + try: + payload, url = await fetch_obj_checks(sess, obj_id) + return ("obj_checks", url, (payload, url)) + except Exception as e: + return ("obj_checks", full_url, e) + + def upsert_sale_graph( db: Session, obj_id: int, type_: str, rows: list[dict[str, Any]], snapshot_date: date ) -> int: @@ -1470,6 +1582,8 @@ async def run_region_sweep( "infra_rows": 0, "photos_rows": 0, "photos_downloaded": 0, + "documents_rows": 0, + "checks_rows": 0, } total_flats = 0 request_count = 0 @@ -1552,7 +1666,11 @@ async def run_region_sweep( stage="phase_a", ) - # ── Phase B/C — per-object processing (resumable) ─────────────── + # ── Phase B/C — per-object processing (resumable, parallel per-object) ─ + # Все endpoint'ы одного obj_id запускаются параллельно через asyncio.gather. + # BrowserSession._sem (Semaphore(3)) ограничивает одновременные запросы. + # DB upserts выполняются последовательно после gather — один db Session + # не thread-safe для параллельной записи. pdir = Path(photos_dir) if photos_dir else PHOTOS_DIR_DEFAULT total = len(all_objects) for i in range(start_index, total): @@ -1561,75 +1679,82 @@ async def run_region_sweep( if not obj_id: continue - # Flats per obj — committed immediately + # Собираем корутины для параллельного запуска + coros: list[Any] = [] + if fetch_flats: - try: - flats = await fetch_flats_for_object(sess, obj_id) - if flats: - total_flats += upsert_flats(db, flats, snapshot_date, region_code) - except Exception as e: - log_progress( - db, - run_id, - f"flats failed obj={obj_id}: {type(e).__name__}: {str(e)[:120]}", - level="warn", - stage="fetch_flats", - obj_id=obj_id, - ) + coros.append(_fetch_flats_safe(sess, obj_id)) if extras: - # sale_graph (apartments + parking) - for type_ in SALE_GRAPH_TYPES: - try: - rows, full_url = await fetch_sale_graph(sess, obj_id, type_) - extras_counts["sale_graph_rows"] += upsert_sale_graph( - db, obj_id, type_, rows, snapshot_date - ) - except Exception as e: - full_url = ( - f"{BASE_URL}" - f"{PATH_SALE_GRAPH.format(obj_id=obj_id)}?type={type_}" - ) - _classify_and_log( - db, run_id, obj_id, f"sale_graph_{type_}", full_url, e - ) + coros.append(_fetch_sale_graph_safe(sess, obj_id, "apartments")) + coros.append(_fetch_sale_graph_safe(sess, obj_id, "parking")) + coros.append(_fetch_sales_agg_safe(sess, obj_id)) + coros.append(_fetch_infrastructure_safe(sess, obj_id)) + coros.append(_fetch_photos_safe(sess, obj_id)) + coros.append(_fetch_documents_safe(sess, obj_id)) + coros.append(_fetch_obj_checks_safe(sess, obj_id)) - # sales_agg - try: - agg, full_url = await fetch_sales_agg(sess, obj_id) + if not coros: + continue + + # Параллельный fetch всех endpoint'ов одного объекта + results = await asyncio.gather(*coros, return_exceptions=False) + + # Sequential upsert — DB session не thread-safe + for kind_tag, full_url, result in results: + if isinstance(result, Exception): + _classify_and_log(db, run_id, obj_id, kind_tag, full_url, result) + continue + + if kind_tag == "flats": + flats_list: list[dict[str, Any]] = result # type: ignore[assignment] + if flats_list: + total_flats += upsert_flats(db, flats_list, snapshot_date, region_code) + + elif kind_tag in ("sale_graph_apartments", "sale_graph_parking"): + sg_type = kind_tag.replace("sale_graph_", "") + rows_sg, _ = result # type: ignore[misc] + extras_counts["sale_graph_rows"] += upsert_sale_graph( + db, obj_id, sg_type, rows_sg, snapshot_date + ) + + elif kind_tag == "sales_agg": + agg_data, _ = result # type: ignore[misc] extras_counts["sales_agg_rows"] += upsert_sales_agg( - db, obj_id, agg, snapshot_date + db, obj_id, agg_data, snapshot_date ) - except Exception as e: - full_url = f"{BASE_URL}{PATH_SALES_AGG.format(obj_id=obj_id)}" - _classify_and_log(db, run_id, obj_id, "sales_agg", full_url, e) - # infrastructure - try: - pois, full_url = await fetch_infrastructure(sess, obj_id) + elif kind_tag == "infrastructure": + pois_data, _ = result # type: ignore[misc] extras_counts["infra_rows"] += upsert_infrastructure( - db, obj_id, pois, snapshot_date + db, obj_id, pois_data, snapshot_date ) - except Exception as e: - full_url = f"{BASE_URL}{PATH_INFRA.format(obj_id=obj_id)}" - _classify_and_log(db, run_id, obj_id, "infrastructure", full_url, e) - # photos - try: - photos, full_url = await fetch_photos(sess, obj_id) + elif kind_tag == "photos": + photos_data, _ = result # type: ignore[misc] local_paths: dict[str, str] = {} thumb_paths: dict[str, str] = {} - if download_photos_binary and photos: + if download_photos_binary and photos_data: local_paths, thumb_paths = await download_photos( - sess, obj_id, photos, pdir + sess, obj_id, photos_data, pdir ) extras_counts["photos_downloaded"] += len(local_paths) extras_counts["photos_rows"] += upsert_photos( - db, obj_id, photos, local_paths, thumb_paths + db, obj_id, photos_data, local_paths, thumb_paths ) - except Exception as e: - full_url = f"{BASE_URL}{PATH_PHOTOS.format(obj_id=obj_id)}" - _classify_and_log(db, run_id, obj_id, "photos", full_url, e) + + elif kind_tag == "documents": + docs_payload, _ = result # type: ignore[misc] + docs = extract_documents(docs_payload or []) + if docs: + ins, _skip = upsert_documents(db, obj_id, docs) + extras_counts["documents_rows"] += ins + + elif kind_tag == "obj_checks": + checks_payload, _ = result # type: ignore[misc] + checks = extract_obj_checks(checks_payload) + if checks: + extras_counts["checks_rows"] += upsert_obj_checks(db, obj_id, checks) # Checkpoint раз в 10 объектов: запись прогресса + heartbeat. if (i + 1) % 10 == 0: @@ -1642,7 +1767,9 @@ async def run_region_sweep( f" agg={extras_counts['sales_agg_rows']}" f" infra={extras_counts['infra_rows']}" f" photos={extras_counts['photos_rows']}" - f" downloaded={extras_counts['photos_downloaded']}", + f" downloaded={extras_counts['photos_downloaded']}" + f" docs={extras_counts['documents_rows']}" + f" checks={extras_counts['checks_rows']}", stage="extras" if extras else "fetch_flats", ) @@ -1668,7 +1795,58 @@ async def run_region_sweep( log_progress( db, run_id, - f"Готово ✅ objects={total} flats={total_flats} requests={request_count}", + f"Phase D done: objects={total} flats={total_flats} requests={request_count}", + stage="phase_d", + ) + + # ── Phase E — derive is_ekb for this snapshot ────────────────────── + # Проставляем is_ekb=TRUE для объектов Екатеринбурга/Свердловской обл. + # по district_name (заполнен PostGIS join в Phase A). + # Только новые/изменившиеся строки: COALESCE(is_ekb, FALSE) = FALSE. + try: + result_e = db.execute( + text( + """ + UPDATE domrf_kn_objects + SET is_ekb = TRUE + WHERE snapshot_date = :snap + AND ( + district_name ILIKE '%екатеринбург%' + OR district_name ILIKE '%свердловск%' + ) + AND COALESCE(is_ekb, FALSE) = FALSE + """ + ), + {"snap": snapshot_date}, + ) + db.commit() + ekb_updated = result_e.rowcount if result_e.rowcount >= 0 else -1 + log_progress( + db, + run_id, + f"Phase E done: is_ekb derived, updated={ekb_updated} rows" + f" for snap {snapshot_date}", + stage="phase_e", + ) + except Exception as e: + logger.warning("Phase E is_ekb derive failed: %s", e) + try: + db.rollback() + except Exception: + pass + log_progress( + db, + run_id, + f"Phase E is_ekb derive FAILED: {type(e).__name__}: {str(e)[:200]}", + level="warn", + stage="phase_e", + ) + + log_progress( + db, + run_id, + f"Готово objects={total} flats={total_flats} requests={request_count}" + f" docs={extras_counts['documents_rows']} checks={extras_counts['checks_rows']}", stage="done", ) return { diff --git a/backend/app/services/scrapers/obj_checks.py b/backend/app/services/scrapers/obj_checks.py new file mode 100644 index 00000000..23a78aed --- /dev/null +++ b/backend/app/services/scrapers/obj_checks.py @@ -0,0 +1,175 @@ +"""Extractor for 6 «Проверено на наш.дом.рф» checks per object. + +Issue #297, sub-task 22f. Table created in PR #303 (data/sql/111_22f_domrf_obj_checks.sql). + +Check types (canonical): + no_problems / docs / timing / photos / bankruptcy / declaration + +Source endpoint: /сервисы/api/object/{obj_id}/checks + (pattern analogичен /infrastructure и /photos — те же сервисы/api/object/{obj_id}/ prefix) + URL not verified via devtools — структура payload выведена из аудита страницы ЖК. + Если endpoint не существует — scraper получит HTTP-ошибку, которую _classify_and_log + запишет в kn_scrape_failures, данные в domrf_obj_checks не поступят, scrape не упадёт. + +Expected payload shape (предположительно): + { + "data": { + "noProblemObjects": true, # no_problems + "hasDocuments": true, # docs + "meetsDeadlines": true, # timing + "hasPhotos": true, # photos + "notBankrupt": true, # bankruptcy + "hasDeclaration": true # declaration + } + } + + OR possibly an array: + [{"checkType": "no_problems", "passed": true}, ...] + +Поддерживаются оба варианта: dict-payload (поля маппятся через CHECK_FIELD_MAP) +и list-payload (поля check_type + passed/value). +""" + +from __future__ import annotations + +import logging +from typing import Any + +from sqlalchemy import text +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + +CHECK_TYPES = ["no_problems", "docs", "timing", "photos", "bankruptcy", "declaration"] + +# Mapping of possible API field names → canonical check_type. +# Best-guess from DOM.РФ API field naming conventions (кейс camelCase). +_CHECK_FIELD_MAP: dict[str, str] = { + "noProblemObjects": "no_problems", + "noProblemFlg": "no_problems", + "hasDocuments": "docs", + "documentsFlg": "docs", + "meetsDeadlines": "timing", + "deadlinesFlg": "timing", + "hasPhotos": "photos", + "photosFlg": "photos", + "notBankrupt": "bankruptcy", + "bankruptcyFlg": "bankruptcy", + "hasDeclaration": "declaration", + "declarationFlg": "declaration", +} + +# Canonical check_type name → possible API field name aliases +_CHECK_TYPE_ALIASES: dict[str, list[str]] = { + "no_problems": ["no_problems", "noProblemObjects", "noProblemFlg"], + "docs": ["docs", "hasDocuments", "documentsFlg"], + "timing": ["timing", "meetsDeadlines", "deadlinesFlg"], + "photos": ["photos", "hasPhotos", "photosFlg"], + "bankruptcy": ["bankruptcy", "notBankrupt", "bankruptcyFlg"], + "declaration": ["declaration", "hasDeclaration", "declarationFlg"], +} + +_UPSERT_CHECKS_SQL = text( + """ + INSERT INTO domrf_obj_checks (obj_id, check_type, passed, checked_at, scraped_at) + VALUES (:obj_id, :check_type, :passed, NOW(), NOW()) + ON CONFLICT (obj_id, check_type) DO UPDATE SET + passed = EXCLUDED.passed, + checked_at = NOW(), + scraped_at = NOW() + """ +) + + +def extract_obj_checks(raw_payload: Any) -> list[dict[str, Any]]: + """Извлечь 6 чекбоксов из payload endpoint /object/{obj_id}/checks. + + Поддерживает два варианта payload: + 1. dict с полями (ожидаемый API-формат): {"data": {"noProblemObjects": true, ...}} + 2. list объектов: [{"checkType": "no_problems", "passed": true}, ...] + + Для неизвестных полей и неподдерживаемых форматов — WARNING + пустой список. + """ + if not raw_payload: + return [] + + # Вариант 1: dict с data-обёрткой + data: Any = raw_payload + if isinstance(raw_payload, dict): + data = raw_payload.get("data") or raw_payload + + results: list[dict[str, Any]] = [] + + if isinstance(data, dict): + # Map известных полей к canonical check_type + found: dict[str, bool] = {} + for field, value in data.items(): + ct = _CHECK_FIELD_MAP.get(field) + if ct and ct not in found: + found[ct] = bool(value) + # Также проверить canonical names напрямую + for ct in CHECK_TYPES: + if ct not in found and ct in data: + found[ct] = bool(data[ct]) + if found: + for ct in CHECK_TYPES: + results.append({"check_type": ct, "passed": found.get(ct, False)}) + return results + # dict не содержит известных полей — попробуем как list-формат ниже + logger.warning( + "domrf obj_checks: dict payload has no known check fields: %s", list(data)[:10] + ) + return [] + + if isinstance(data, list): + found_list: dict[str, bool] = {} + for item in data: + if not isinstance(item, dict): + continue + ct_raw = item.get("checkType") or item.get("check_type") or item.get("type") + if ct_raw and str(ct_raw) in CHECK_TYPES: + passed_raw = item.get("passed") or item.get("value") or item.get("status") + found_list[str(ct_raw)] = bool(passed_raw) + if found_list: + for ct in CHECK_TYPES: + results.append({"check_type": ct, "passed": found_list.get(ct, False)}) + return results + logger.warning( + "domrf obj_checks: list payload has no recognisable check items: %s", data[:3] + ) + return [] + + logger.warning("domrf obj_checks: unexpected payload type %s", type(raw_payload)) + return [] + + +def upsert_obj_checks(db: Session, obj_id: int, checks: list[dict[str, Any]]) -> int: + """UPSERT 6 чек-строк в domrf_obj_checks. Returns count of inserted/updated rows. + + Использует SAVEPOINT (begin_nested) per-row — одна битая строка не откатывает + всю транзакцию. + """ + if not checks: + return 0 + ok = 0 + for c in checks: + try: + with db.begin_nested(): + db.execute( + _UPSERT_CHECKS_SQL, + { + "obj_id": obj_id, + "check_type": c["check_type"], + "passed": c["passed"], + }, + ) + ok += 1 + except Exception as exc: + logger.warning( + "upsert obj_checks obj=%s check_type=%s failed: %s", + obj_id, + c.get("check_type"), + exc, + ) + db.commit() + return ok -- 2.45.3