feat(scrape-kn): parallel Phase B/C + wire obj_checks + documents + is_ekb auto-derive

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 не прерывается).
2026-05-17 22:34:58 +03:00 · 2026-05-17 22:34:58 +03:00 · 1db19b3b8a
commit 1db19b3b8a
parent c80d79e2d4
2 changed files with 408 additions and 55 deletions
--- 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 {
--- a/backend/app/services/scrapers/obj_checks.py
+++ 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