feat(cadastre): bulk_harvest worker + grid-walker + admin API (#168 PR3/5) (#171)

* feat(cadastre): bulk_harvest_quarter Celery task + grid-walker + saga state (#168 PR3/5)

Add bulk cadastre harvest pipeline:
- services/cadastre/bulk_harvest.py: async harvest_quarter() orchestrator (4 phases)
  + 7 upsert helpers (parcels/buildings/constructions/oncs/enks/zouit/quarter_stats)
  using CAST(:x AS jsonb) pattern, begin_nested() SAVEPOINT per grid-walk upsert
- services/cadastre/grid_geometry.py: quarter_bbox_3857 (PostGIS ST_Extent)
  + generate_grid_click_points (15x15 = 225 sub-bbox grid)
- workers/tasks/scrape_cadastre.py: bulk_harvest_quarter_task (acks_late=True,
  dont_autoretry_for NspdBulkWafError), enqueue_cadastre_harvest, cleanup_zombies
- api/v1/admin_cadastre.py: 5 endpoints behind AdminTokenAuth — create/list/get/cancel/resume
- Tests: 13 service unit + 8 API tests

* fixup(cadastre): drop importorskip (PR2 merged) — imports now top-level (#168 PR3)

* fixup(cadastre): tile_size unified param + register slow marker (#168 PR3)

Blocker #1: rename tile_width/tile_height → tile_size in generate_grid_click_points.
Callers in bulk_harvest.py and test_cadastre_bulk.py used tile_size; def had
tile_width+tile_height → TypeError at runtime.

Blocker #2 (10 failures in test_admin_cadastre.py) was side-effect of #1:
TypeError at import chain (app.main → admin_cadastre → bulk_harvest)
broke FastAPI app load → dependency_overrides AttributeError. Now resolved.

Also register `slow` pytest marker in pyproject.toml to suppress
PytestUnknownMarkWarning.

* fixup(cadastre): use importlib.util.find_spec instead of import (#168 PR3)

Root cause Blocker #2: `import app.workers.tasks.scrape_cadastre` at module
top-level REBOUND `app` from FastAPI instance (line 14) to the Python package.
All subsequent `app.dependency_overrides[get_db] = ...` failed with
AttributeError because `app` was now the namespace module.

Fix: probe module availability with importlib.util.find_spec — does not
import or rebind names. FastAPI `app` instance stays intact.

All 28 cadastre tests pass locally.

---------

Co-authored-by: lekss361 <claudestars@proton.me>

backend/app/api/v1/admin_cadastre.py

@@ -0,0 +1,355 @@
+"""Admin API для bulk cadastre harvest (GH issue #168, PR 3/5).
+
+Endpoints:
+  POST /api/v1/admin/cadastre/jobs       — создать job + enqueue
+  GET  /api/v1/admin/cadastre/jobs       — список jobs
+  GET  /api/v1/admin/cadastre/jobs/{id}  — детали job
+  POST /api/v1/admin/cadastre/jobs/{id}/cancel — отменить job
+  POST /api/v1/admin/cadastre/jobs/{id}/resume — возобновить paused job
+
+Scope variants:
+  pilot       — первые 50 кварталов ЕКБ (66:41:%)
+  ekb_full    — все ~2408 кварталов ЕКБ
+  manual_list — явный список quarters в body.quarters
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel, Field
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.db import get_db
+from app.core.deps import AdminTokenAuth
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# Ограничение пилота
+PILOT_LIMIT = 50
+# Регион по умолчанию — ЕКБ (66:41:%)
+EKB_PREFIX = "66:41:%"
+
+
+# ── Request / response schemas ───────────────────────────────────────────────
+
+
+class CreateCadastreJobRequest(BaseModel):
+    """Параметры создания bulk cadastre job."""
+
+    scope: str = Field(
+        ...,
+        pattern="^(pilot|ekb_full|manual_list)$",
+        description="pilot=50 кварталов ЕКБ, ekb_full=все 2408, manual_list=явный список",
+    )
+    quarters: list[str] | None = Field(
+        default=None,
+        max_length=5000,
+        description="Список кварталов для scope=manual_list",
+    )
+    limit: int | None = Field(
+        default=None,
+        ge=1,
+        le=10000,
+        description="Ограничить количество кварталов (override для pilot/ekb_full)",
+    )
+    name: str | None = Field(default=None, max_length=200)
+    rate_ms: int = Field(
+        default=333,
+        ge=50,
+        le=10000,
+        description="Задержка между запросами в мс (333мс = ~3 req/s)",
+    )
+
+
+# ── Helpers ──────────────────────────────────────────────────────────────────
+
+
+def _resolve_quarters(
+    db: Session,
+    scope: str,
+    quarters: list[str] | None,
+    limit: int | None,
+) -> list[str]:
+    """Собрать список кварталов согласно scope."""
+    if scope == "manual_list":
+        if not quarters:
+            raise HTTPException(
+                status_code=400,
+                detail="scope=manual_list требует непустой список quarters",
+            )
+        return [q.strip() for q in quarters if q.strip()]
+
+    cap = limit or (PILOT_LIMIT if scope == "pilot" else 100000)
+
+    rows = db.execute(
+        text("""
+            SELECT cad_number
+            FROM cad_quarters_geom
+            WHERE cad_number LIKE :prefix
+            ORDER BY cad_number
+            LIMIT :cap
+        """),
+        {"prefix": EKB_PREFIX, "cap": cap},
+    ).all()
+
+    if not rows:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"Нет кварталов с префиксом {EKB_PREFIX!r} в cad_quarters_geom. "
+                "Убедитесь что geo-backfill кварталов завершён."
+            ),
+        )
+
+    return [str(r[0]) for r in rows]
+
+
+def _create_job(
+    db: Session,
+    *,
+    name: str | None,
+    scope: str,
+    quarters: list[str],
+    rate_ms: int,
+    triggered_by: str = "manual",
+) -> int:
+    """INSERT в cadastre_jobs, вернуть job_id."""
+    job_id = db.execute(
+        text("""
+            INSERT INTO cadastre_jobs (
+                name, job_kind, scope, status,
+                rate_ms, triggered_by,
+                targets_total, raw_targets
+            ) VALUES (
+                :name, :job_kind, :scope, 'queued',
+                :rate_ms, :triggered_by,
+                :targets_total, CAST(:raw_targets AS jsonb)
+            )
+            RETURNING job_id
+        """),
+        {
+            "name": name,
+            "job_kind": scope,
+            "scope": scope,
+            "rate_ms": rate_ms,
+            "triggered_by": triggered_by,
+            "targets_total": len(quarters),
+            "raw_targets": json.dumps({"quarters": quarters}, ensure_ascii=False),
+        },
+    ).scalar_one()
+    db.commit()
+    return int(job_id)
+
+
+def _serialize_job(row: Any) -> dict[str, Any]:
+    """Сериализовать строку cadastre_jobs в dict для API response."""
+    targets_total = row["targets_total"] or 0
+    targets_done = row["targets_done"] or 0
+
+    return {
+        "job_id": row["job_id"],
+        "name": row["name"],
+        "job_kind": row["job_kind"],
+        "scope": row["scope"],
+        "status": row["status"],
+        "triggered_by": row["triggered_by"],
+        "rate_ms": row["rate_ms"],
+        "created_at": row["created_at"].isoformat() if row["created_at"] else None,
+        "started_at": row["started_at"].isoformat() if row["started_at"] else None,
+        "finished_at": row["finished_at"].isoformat() if row["finished_at"] else None,
+        "heartbeat_at": row["heartbeat_at"].isoformat() if row["heartbeat_at"] else None,
+        "targets_total": targets_total,
+        "targets_done": targets_done,
+        "targets_failed": row["targets_failed"] or 0,
+        "targets_skipped": row["targets_skipped"] or 0,
+        "requests_count": row["requests_count"] or 0,
+        "waf_blocked_count": row["waf_blocked_count"] or 0,
+        "error": row["error"],
+        "phase_state": row["phase_state"],
+        "progress_pct": round(100.0 * targets_done / max(targets_total, 1), 1),
+        "estimate_minutes": round(targets_total * (row["rate_ms"] or 333) / 1000 / 60, 1),
+    }
+
+
+# ── Endpoints ────────────────────────────────────────────────────────────────
+
+
+@router.post("/jobs")
+def create_cadastre_job(
+    body: CreateCadastreJobRequest,
+    db: Annotated[Session, Depends(get_db)],
+    _: AdminTokenAuth,
+) -> dict[str, Any]:
+    """Создать bulk cadastre harvest job и поставить в очередь.
+
+    scope=pilot    → первые 50 ЕКБ кварталов из cad_quarters_geom
+    scope=ekb_full → все ~2408 ЕКБ кварталов
+    scope=manual_list → body.quarters (явный список)
+
+    Возвращает: {job_id, targets_total, estimate_minutes}
+    """
+    from app.workers.tasks.scrape_cadastre import enqueue_cadastre_harvest
+
+    quarters = _resolve_quarters(db, body.scope, body.quarters, body.limit)
+
+    job_id = _create_job(
+        db,
+        name=body.name or f"cadastre_{body.scope}",
+        scope=body.scope,
+        quarters=quarters,
+        rate_ms=body.rate_ms,
+    )
+
+    # Enqueue таск который в свою очередь запустит per-quarter tasks
+    enqueue_cadastre_harvest.apply_async(
+        args=[job_id],
+        queue="geo",
+    )
+
+    estimate_minutes = round(len(quarters) * body.rate_ms / 1000 / 60, 1)
+
+    logger.info(
+        "create_cadastre_job: job_id=%d scope=%s quarters=%d estimate=%.1fmin",
+        job_id,
+        body.scope,
+        len(quarters),
+        estimate_minutes,
+    )
+
+    return {
+        "job_id": job_id,
+        "scope": body.scope,
+        "targets_total": len(quarters),
+        "estimate_minutes": estimate_minutes,
+    }
+
+
+@router.get("/jobs")
+def list_cadastre_jobs(
+    db: Annotated[Session, Depends(get_db)],
+    _: AdminTokenAuth,
+    limit: int = 30,
+) -> list[dict[str, Any]]:
+    """Список последних cadastre harvest jobs."""
+    rows = (
+        db.execute(
+            text("""
+                SELECT job_id, name, job_kind, scope, status, triggered_by,
+                       rate_ms, created_at, started_at, finished_at, heartbeat_at,
+                       targets_total, targets_done, targets_failed, targets_skipped,
+                       requests_count, waf_blocked_count,
+                       error, phase_state
+                FROM cadastre_jobs
+                ORDER BY created_at DESC
+                LIMIT :lim
+            """),
+            {"lim": min(limit, 200)},
+        )
+        .mappings()
+        .all()
+    )
+    return [_serialize_job(r) for r in rows]
+
+
+@router.get("/jobs/{job_id}")
+def get_cadastre_job(
+    job_id: int,
+    db: Annotated[Session, Depends(get_db)],
+    _: AdminTokenAuth,
+) -> dict[str, Any]:
+    """Детали одного cadastre harvest job."""
+    row = (
+        db.execute(
+            text("""
+                SELECT job_id, name, job_kind, scope, status, triggered_by,
+                       rate_ms, created_at, started_at, finished_at, heartbeat_at,
+                       targets_total, targets_done, targets_failed, targets_skipped,
+                       requests_count, waf_blocked_count,
+                       error, phase_state, raw_targets
+                FROM cadastre_jobs
+                WHERE job_id = :id
+            """),
+            {"id": job_id},
+        )
+        .mappings()
+        .first()
+    )
+    if not row:
+        raise HTTPException(status_code=404, detail=f"cadastre_job {job_id} не найден")
+
+    result = _serialize_job(row)
+    # Добавить сводку по raw_targets (количество + первые 10 для UI)
+    raw_targets = row["raw_targets"] or {}
+    quarters = raw_targets.get("quarters", [])
+    result["raw_targets_count"] = len(quarters)
+    result["raw_targets_preview"] = quarters[:10]
+    return result
+
+
+@router.post("/jobs/{job_id}/cancel")
+def cancel_cadastre_job(
+    job_id: int,
+    db: Annotated[Session, Depends(get_db)],
+    _: AdminTokenAuth,
+) -> dict[str, Any]:
+    """Отменить job. Воркеры увидят status='cancelled' при следующей итерации и skip."""
+    result = db.execute(
+        text("""
+            UPDATE cadastre_jobs
+            SET status = 'cancelled',
+                finished_at = COALESCE(finished_at, NOW()),
+                error = COALESCE(error, 'cancelled by admin')
+            WHERE job_id = :id
+              AND status IN ('queued', 'running', 'paused')
+            RETURNING job_id
+        """),
+        {"id": job_id},
+    ).first()
+    db.commit()
+
+    if not result:
+        raise HTTPException(
+            status_code=404,
+            detail=f"cadastre_job {job_id} не найден или уже завершён/отменён",
+        )
+    logger.info("cancel_cadastre_job: job_id=%d cancelled", job_id)
+    return {"job_id": job_id, "cancelled": True}
+
+
+@router.post("/jobs/{job_id}/resume")
+def resume_cadastre_job(
+    job_id: int,
+    db: Annotated[Session, Depends(get_db)],
+    _: AdminTokenAuth,
+) -> dict[str, Any]:
+    """Возобновить paused/failed job. Re-enqueue enqueue_cadastre_harvest."""
+    from app.workers.tasks.scrape_cadastre import enqueue_cadastre_harvest
+
+    result = db.execute(
+        text("""
+            UPDATE cadastre_jobs
+            SET status = 'queued', error = NULL
+            WHERE job_id = :id
+              AND status IN ('paused', 'failed')
+            RETURNING job_id
+        """),
+        {"id": job_id},
+    ).first()
+    db.commit()
+
+    if not result:
+        raise HTTPException(
+            status_code=404,
+            detail=f"cadastre_job {job_id} не найден или не в состоянии paused/failed",
+        )
+
+    enqueue_cadastre_harvest.apply_async(args=[job_id], queue="geo")
+    logger.info("resume_cadastre_job: job_id=%d re-enqueued", job_id)
+    return {"job_id": job_id, "resumed": True}

backend/app/main.py

@@ -6,6 +6,7 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 from app.api.v1 import (
+    admin_cadastre,
     admin_jobs,
     admin_leads,
     admin_scrape,
@@ -52,6 +53,11 @@ app.include_router(
     tags=["admin", "site-finder"],
 )
 app.include_router(photos.router, prefix="/api/v1/photos", tags=["photos"])
+app.include_router(
+    admin_cadastre.router,
+    prefix="/api/v1/admin/cadastre",
+    tags=["admin", "cadastre"],
+)
 
 
 @app.get("/health")

backend/app/services/cadastre/bulk_harvest.py

@@ -0,0 +1,990 @@
+"""Оркестратор bulk cadastre harvest — pure-async, без Celery декораторов.
+
+Вызывается из Celery-таска (scrape_cadastre.py) через asyncio.run().
+
+Фазы harvest_quarter:
+  Phase 1: search_by_quarter → upsert первые 20 per категории
+  Phase 2: определить overflow-категории (meta > 20)
+  Phase 3: grid_walk для overflow 36368 (ЗУ) и 36369 (Здания)
+  Phase 4: upsert_quarter_stats из meta квартального snapshot (36381)
+
+Idempotent: все upsert'ы — ON CONFLICT DO UPDATE.
+Resumable: phase_state в cadastre_jobs показывает прогресс.
+
+Питфоллы (см. CLAUDE.md):
+  - CAST(:x AS jsonb) не :x::jsonb (SQLAlchemy парсит :: как параметр)
+  - psycopg v3 only (не psycopg2)
+  - SAVEPOINT через db.begin_nested() вокруг каждого upsert
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.schemas.nspd_bulk import NSPDBulkFeature, QuarterSnapshot
+from app.scrapers.nspd_bulk_client import NSPDBulkClient
+from app.services.cadastre.grid_geometry import generate_grid_click_points, quarter_bbox_3857
+
+logger = logging.getLogger(__name__)
+
+# Категории NSPD для routing
+CAT_PARCEL = 36368  # ЗУ ЕГРН
+CAT_BUILDING = 36369  # Здания ЕГРН
+CAT_CONSTRUCTION = 36383  # Сооружения
+CAT_ONC = 36384  # ОНС
+CAT_ENK = 39663  # ЕНК
+CAT_QUARTER_STATS = 36381  # Квартальная статистика
+CAT_ZOUIT = frozenset({36940, 469039, 469040, 469042})  # ЗОУИТ
+
+# Категории для grid-walk overflow
+GRID_WALK_CATS = frozenset({CAT_PARCEL, CAT_BUILDING})
+
+
+# ── Result type ──────────────────────────────────────────────────────────────
+
+
+@dataclass
+class HarvestResult:
+    """Итог harvest одного квартала."""
+
+    quarter: str
+    parcels_upserted: int = 0
+    buildings_upserted: int = 0
+    constructions_upserted: int = 0
+    oncs_upserted: int = 0
+    enks_upserted: int = 0
+    zouit_upserted: int = 0
+    grid_walk_requests: int = 0
+    phase_state: dict[str, Any] | None = None
+
+
+# ── Основной оркестратор ─────────────────────────────────────────────────────
+
+
+async def harvest_quarter(
+    db: Session,
+    client: NSPDBulkClient,
+    quarter: str,
+    job_id: int,
+    update_progress: Callable[[dict[str, Any]], None],
+) -> HarvestResult:
+    """Собрать все NSPD данные для квартала в cad_* таблицы.
+
+    Идемпотентен: повторный вызов не создаёт дубликаты (ON CONFLICT DO UPDATE).
+    Возобновляем: проверяет job.phase_state, пропускает завершённые фазы.
+
+    Args:
+        db: SQLAlchemy session. Caller должен управлять commit/rollback.
+        client: NSPDBulkClient (async context manager, уже открытый).
+        quarter: кадастровый номер квартала.
+        job_id: ID записи в cadastre_jobs для обновления прогресса.
+        update_progress: callback для записи phase_state в cadastre_jobs.
+
+    Returns:
+        HarvestResult с счётчиками upserted объектов.
+
+    Raises:
+        NspdBulkWafError: при WAF 403 — caller не retry.
+        NspdBulkRateLimitError: при исчерпании retries — caller может retry.
+        Exception: прочие ошибки — propagate.
+    """
+    result = HarvestResult(quarter=quarter)
+
+    # Читаем текущий phase_state из job (для resumability)
+    job_row = (
+        db.execute(
+            text("SELECT phase_state FROM cadastre_jobs WHERE job_id = :id"),
+            {"id": job_id},
+        )
+        .mappings()
+        .first()
+    )
+    phase_state: dict[str, Any] = {}
+    if job_row and job_row["phase_state"]:
+        phase_state = dict(job_row["phase_state"])
+
+    # Ранний выход если quarter уже полностью обработан
+    if phase_state.get("phase") == "done":
+        logger.info("harvest_quarter: quarter=%s уже done (job_id=%s), skip", quarter, job_id)
+        return result
+
+    # ── Phase 1: search_by_quarter snapshot ──────────────────────────────────
+    update_progress({"phase": "snapshot_started", "quarter": quarter})
+
+    snapshot: QuarterSnapshot = await client.search_by_quarter(quarter)
+
+    snapshot_stats = upsert_features(db, snapshot.features, source="search")
+    result.parcels_upserted += snapshot_stats["parcels"]
+    result.buildings_upserted += snapshot_stats["buildings"]
+    result.constructions_upserted += snapshot_stats["constructions"]
+    result.oncs_upserted += snapshot_stats["oncs"]
+    result.enks_upserted += snapshot_stats["enks"]
+    result.zouit_upserted += snapshot_stats["zouit"]
+    db.commit()
+
+    parcels_meta = snapshot.meta_counts.get(CAT_PARCEL, 0)
+    buildings_meta = snapshot.meta_counts.get(CAT_BUILDING, 0)
+
+    update_progress(
+        {
+            "phase": "snapshot_done",
+            "quarter": quarter,
+            "parcels_total_meta": parcels_meta,
+            "buildings_total_meta": buildings_meta,
+            "overflow_cats": snapshot.overflow_categories,
+        }
+    )
+
+    # ── Phase 2-3: grid-walk для overflow категорий ───────────────────────────
+    overflow_cats = [c for c in snapshot.overflow_categories if c in GRID_WALK_CATS]
+
+    for cat_id in overflow_cats:
+        cat_label = "parcels" if cat_id == CAT_PARCEL else "buildings"
+        update_progress({"phase": f"grid_walk_{cat_label}_started", "quarter": quarter})
+
+        discovered, n_requests = await _grid_walk_category(
+            db=db,
+            client=client,
+            quarter=quarter,
+            layer_id=cat_id,
+        )
+        result.grid_walk_requests += n_requests
+
+        if cat_id == CAT_PARCEL:
+            result.parcels_upserted += discovered
+        else:
+            result.buildings_upserted += discovered
+
+        db.commit()
+        update_progress(
+            {
+                "phase": f"grid_walk_{cat_label}_done",
+                "quarter": quarter,
+                f"{cat_label}_grid_discovered": discovered,
+            }
+        )
+
+    # ── Phase 4: quarter stats из snapshot ───────────────────────────────────
+    stats_features = [f for f in snapshot.features if f.category_id == CAT_QUARTER_STATS]
+    if stats_features:
+        upsert_quarter_stats(db, quarter, stats_features[0])
+        db.commit()
+
+    update_progress({"phase": "done", "quarter": quarter})
+    result.phase_state = {"phase": "done", "quarter": quarter}
+
+    logger.info(
+        "harvest_quarter done: quarter=%s parcels=%d buildings=%d "
+        "constructions=%d zouit=%d grid_requests=%d",
+        quarter,
+        result.parcels_upserted,
+        result.buildings_upserted,
+        result.constructions_upserted,
+        result.zouit_upserted,
+        result.grid_walk_requests,
+    )
+    return result
+
+
+# ── Grid-walk для одной категории ────────────────────────────────────────────
+
+
+async def _grid_walk_category(
+    db: Session,
+    client: NSPDBulkClient,
+    quarter: str,
+    layer_id: int,
+    grid_size: int = 15,
+    tile_size: int = 512,
+) -> tuple[int, int]:
+    """Grid-walk layer_id в bbox квартала.
+
+    Генерирует grid_size×grid_size ячеек, для каждой делает wms_feature_info.
+    Дедуплицирует по cad_num, upsert'ит новые объекты.
+
+    Returns:
+        (upserted_count, requests_count)
+    """
+    bbox = quarter_bbox_3857(db, quarter)
+    if bbox is None:
+        logger.warning(
+            "_grid_walk_category: bbox для квартала %s не найден, пропускаем layer=%d",
+            quarter,
+            layer_id,
+        )
+        return 0, 0
+
+    grid_points = generate_grid_click_points(bbox, grid_size=grid_size, tile_size=tile_size)
+
+    discovered_cads: set[str] = set()
+    upserted = 0
+    requests = 0
+
+    for cell_bbox, click_xy in grid_points:
+        try:
+            features = await client.wms_feature_info(
+                layer_id=layer_id,
+                bbox=cell_bbox,
+                click_xy=click_xy,
+                width=tile_size,
+                height=tile_size,
+            )
+            requests += 1
+        except Exception as e:
+            logger.warning(
+                "_grid_walk_category: wms_feature_info error layer=%d quarter=%s: %s",
+                layer_id,
+                quarter,
+                e,
+            )
+            requests += 1
+            continue
+
+        for feature in features:
+            cad = feature.cad_num
+            if not cad or cad in discovered_cads:
+                continue
+            discovered_cads.add(cad)
+            # Upsert в соответствующую таблицу
+            try:
+                with db.begin_nested():
+                    stats = upsert_features(db, [feature], source="wms_grid_walk")
+                    upserted += sum(stats.values())
+            except Exception as e:
+                logger.warning(
+                    "_grid_walk_category: upsert failed cad=%s layer=%d: %s",
+                    cad,
+                    layer_id,
+                    e,
+                )
+
+    return upserted, requests
+
+
+# ── Routing dispatcher ───────────────────────────────────────────────────────
+
+
+def upsert_features(
+    db: Session,
+    features: list[NSPDBulkFeature],
+    source: str = "search",
+) -> dict[str, int]:
+    """Маршрутизировать features по category_id в специфичные upsert-функции.
+
+    Returns:
+        Словарь {table: count} с количеством upserted строк.
+    """
+    counts: dict[str, int] = {
+        "parcels": 0,
+        "buildings": 0,
+        "constructions": 0,
+        "oncs": 0,
+        "enks": 0,
+        "zouit": 0,
+        "skipped": 0,
+    }
+
+    for feature in features:
+        cat = feature.category_id
+        if cat == CAT_PARCEL:
+            upsert_parcel(db, feature, source=source)
+            counts["parcels"] += 1
+        elif cat == CAT_BUILDING:
+            upsert_building(db, feature, source=source)
+            counts["buildings"] += 1
+        elif cat == CAT_CONSTRUCTION:
+            upsert_construction(db, feature, source=source)
+            counts["constructions"] += 1
+        elif cat == CAT_ONC:
+            upsert_onc(db, feature, source=source)
+            counts["oncs"] += 1
+        elif cat == CAT_ENK:
+            upsert_enk(db, feature, source=source)
+            counts["enks"] += 1
+        elif cat in CAT_ZOUIT:
+            upsert_zouit(db, feature, source=source)
+            counts["zouit"] += 1
+        elif cat == CAT_QUARTER_STATS:
+            pass  # обрабатывается отдельно в upsert_quarter_stats
+        else:
+            if cat is not None:
+                logger.debug("upsert_features: неизвестный category_id=%s, skip", cat)
+            counts["skipped"] += 1
+
+    return counts
+
+
+# ── cad_parcels upsert ───────────────────────────────────────────────────────
+
+
+def upsert_parcel(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None:
+    """Upsert ЗУ ЕГРН в cad_parcels. ON CONFLICT (cad_num) DO UPDATE."""
+    opts = feature.options
+    cad_num = opts.effective_cad_num
+    if not cad_num:
+        logger.debug("upsert_parcel: пропуск — нет cad_num в feature id=%s", feature.id)
+        return
+
+    raw_opts = feature.properties.get("options") or {}
+    geom_json = json.dumps(feature.geometry) if feature.geometry else None
+    quarter_cad = opts.quarter_cad_number or _derive_quarter_cad(cad_num)
+
+    db.execute(
+        text("""
+            INSERT INTO cad_parcels (
+                cad_num, quarter_cad_number, category_id,
+                land_record_area, specified_area, declared_area,
+                land_record_category_type, land_record_subtype, land_record_type,
+                permitted_use_established_by_document,
+                cost_value, cost_index,
+                ownership_type, right_type,
+                readable_address, status, previously_posted,
+                registration_date, subcategory, cadastral_districts_code,
+                geom, raw_props, source, fetched_at, updated_at
+            ) VALUES (
+                :cad_num, :quarter_cad_number, 36368,
+                :land_record_area, :specified_area, :declared_area,
+                :land_record_category_type, :land_record_subtype, :land_record_type,
+                :permitted_use,
+                :cost_value, :cost_index,
+                :ownership_type, :right_type,
+                :readable_address, :status, :previously_posted,
+                CAST(:registration_date AS date),
+                :subcategory, :cadastral_districts_code,
+                CASE WHEN :geom IS NOT NULL THEN
+                    ST_Transform(
+                        ST_SetSRID(ST_GeomFromGeoJSON(:geom), 3857),
+                        4326
+                    )
+                ELSE NULL END,
+                CAST(:raw_props AS jsonb), :source, NOW(), NOW()
+            )
+            ON CONFLICT (cad_num) DO UPDATE SET
+                quarter_cad_number = EXCLUDED.quarter_cad_number,
+                land_record_area = EXCLUDED.land_record_area,
+                specified_area = EXCLUDED.specified_area,
+                declared_area = EXCLUDED.declared_area,
+                land_record_category_type = EXCLUDED.land_record_category_type,
+                land_record_subtype = EXCLUDED.land_record_subtype,
+                land_record_type = EXCLUDED.land_record_type,
+                permitted_use_established_by_document =
+                    EXCLUDED.permitted_use_established_by_document,
+                cost_value = EXCLUDED.cost_value,
+                cost_index = EXCLUDED.cost_index,
+                ownership_type = EXCLUDED.ownership_type,
+                right_type = EXCLUDED.right_type,
+                readable_address = EXCLUDED.readable_address,
+                status = EXCLUDED.status,
+                previously_posted = EXCLUDED.previously_posted,
+                registration_date = EXCLUDED.registration_date,
+                subcategory = EXCLUDED.subcategory,
+                cadastral_districts_code = EXCLUDED.cadastral_districts_code,
+                geom = COALESCE(EXCLUDED.geom, cad_parcels.geom),
+                raw_props = EXCLUDED.raw_props,
+                source = EXCLUDED.source,
+                updated_at = NOW()
+        """),
+        {
+            "cad_num": cad_num,
+            "quarter_cad_number": quarter_cad,
+            "land_record_area": _safe_numeric(raw_opts.get("land_record_area")),
+            "specified_area": _safe_numeric(raw_opts.get("specified_area")),
+            "declared_area": _safe_numeric(raw_opts.get("declared_area")),
+            "land_record_category_type": raw_opts.get("land_record_category_type"),
+            "land_record_subtype": raw_opts.get("land_record_subtype"),
+            "land_record_type": raw_opts.get("land_record_type"),
+            "permitted_use": raw_opts.get("permitted_use_established_by_document"),
+            "cost_value": _safe_numeric(raw_opts.get("cost_value")),
+            "cost_index": _safe_numeric(raw_opts.get("cost_index")),
+            "ownership_type": raw_opts.get("ownership_type"),
+            "right_type": raw_opts.get("right_type"),
+            "readable_address": raw_opts.get("readable_address"),
+            "status": raw_opts.get("status"),
+            "previously_posted": raw_opts.get("previously_posted"),
+            "registration_date": raw_opts.get("registration_date"),
+            "subcategory": _safe_int(raw_opts.get("subcategory")),
+            "cadastral_districts_code": _safe_int(raw_opts.get("cadastral_districts_code")),
+            "geom": geom_json,
+            "raw_props": json.dumps(raw_opts, ensure_ascii=False),
+            "source": source,
+        },
+    )
+
+
+# ── cad_buildings upsert ─────────────────────────────────────────────────────
+
+
+def upsert_building(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None:
+    """Upsert здание ЕГРН в cad_buildings. ON CONFLICT (cad_num) DO UPDATE."""
+    opts = feature.options
+    cad_num = opts.effective_cad_num
+    if not cad_num:
+        logger.debug("upsert_building: пропуск — нет cad_num в feature id=%s", feature.id)
+        return
+
+    raw_opts = feature.properties.get("options") or {}
+    geom_json = json.dumps(feature.geometry) if feature.geometry else None
+    quarter_cad = opts.quarter_cad_number or _derive_quarter_cad(cad_num)
+
+    # floors из NSPD может быть str "5" или int 5
+    floors_val = _safe_int(raw_opts.get("floors"))
+    ugnd_floors_val = _safe_int(raw_opts.get("underground_floors"))
+
+    # cultural_heritage — строка → bool
+    ch_str = raw_opts.get("cultural_heritage")
+    cultural_heritage_bool = bool(ch_str) if ch_str is not None else None
+
+    db.execute(
+        text("""
+            INSERT INTO cad_buildings (
+                cad_num, quarter_cad_number,
+                objdoc_id, registers_id,
+                purpose, building_name,
+                floors, underground_floors,
+                area, build_record_area,
+                cost_value, cost_index,
+                year_built, year_commisioning,
+                build_record_registration_date, registration_date,
+                ownership_type,
+                cultural_heritage_object, cultural_heritage_val,
+                readable_address, status, common_data_status,
+                type, build_record_type_value,
+                geom, raw_props, source, fetched_at, updated_at
+            ) VALUES (
+                :cad_num, :quarter_cad_number,
+                :objdoc_id, :registers_id,
+                :purpose, :building_name,
+                :floors, :underground_floors,
+                :area, :build_record_area,
+                :cost_value, :cost_index,
+                :year_built, :year_commisioning,
+                CAST(:build_record_registration_date AS date),
+                CAST(:registration_date AS date),
+                :ownership_type,
+                :cultural_heritage_object, :cultural_heritage_val,
+                :readable_address, :status, :common_data_status,
+                :type, :build_record_type_value,
+                CASE WHEN :geom IS NOT NULL THEN
+                    ST_Transform(
+                        ST_SetSRID(ST_GeomFromGeoJSON(:geom), 3857),
+                        4326
+                    )
+                ELSE NULL END,
+                CAST(:raw_props AS jsonb), :source, NOW(), NOW()
+            )
+            ON CONFLICT (cad_num) DO UPDATE SET
+                quarter_cad_number = EXCLUDED.quarter_cad_number,
+                objdoc_id = COALESCE(EXCLUDED.objdoc_id, cad_buildings.objdoc_id),
+                registers_id = COALESCE(EXCLUDED.registers_id, cad_buildings.registers_id),
+                purpose = EXCLUDED.purpose,
+                building_name = EXCLUDED.building_name,
+                floors = EXCLUDED.floors,
+                underground_floors = EXCLUDED.underground_floors,
+                area = EXCLUDED.area,
+                build_record_area = EXCLUDED.build_record_area,
+                cost_value = EXCLUDED.cost_value,
+                cost_index = EXCLUDED.cost_index,
+                year_built = EXCLUDED.year_built,
+                year_commisioning = EXCLUDED.year_commisioning,
+                build_record_registration_date = EXCLUDED.build_record_registration_date,
+                registration_date = EXCLUDED.registration_date,
+                ownership_type = EXCLUDED.ownership_type,
+                cultural_heritage_object = EXCLUDED.cultural_heritage_object,
+                cultural_heritage_val = EXCLUDED.cultural_heritage_val,
+                readable_address = EXCLUDED.readable_address,
+                status = EXCLUDED.status,
+                common_data_status = EXCLUDED.common_data_status,
+                type = EXCLUDED.type,
+                build_record_type_value = EXCLUDED.build_record_type_value,
+                geom = COALESCE(EXCLUDED.geom, cad_buildings.geom),
+                raw_props = EXCLUDED.raw_props,
+                source = EXCLUDED.source,
+                updated_at = NOW()
+        """),
+        {
+            "cad_num": cad_num,
+            "quarter_cad_number": quarter_cad,
+            "objdoc_id": _safe_int(raw_opts.get("objdoc_id")),
+            "registers_id": _safe_int(raw_opts.get("registers_id")),
+            "purpose": raw_opts.get("purpose"),
+            "building_name": raw_opts.get("building_name") or raw_opts.get("name"),
+            "floors": floors_val,
+            "underground_floors": ugnd_floors_val,
+            "area": _safe_numeric(raw_opts.get("area")),
+            "build_record_area": _safe_numeric(raw_opts.get("build_record_area")),
+            "cost_value": _safe_numeric(raw_opts.get("cost_value")),
+            "cost_index": _safe_numeric(raw_opts.get("cost_index")),
+            "year_built": _safe_int(raw_opts.get("year_built")),
+            "year_commisioning": _safe_int(raw_opts.get("year_commisioning")),
+            "build_record_registration_date": raw_opts.get("build_record_registration_date"),
+            "registration_date": raw_opts.get("registration_date"),
+            "ownership_type": raw_opts.get("ownership_type"),
+            "cultural_heritage_object": cultural_heritage_bool,
+            "cultural_heritage_val": ch_str,
+            "readable_address": raw_opts.get("readable_address"),
+            "status": raw_opts.get("status"),
+            "common_data_status": raw_opts.get("common_data_status"),
+            "type": raw_opts.get("obj_type") or raw_opts.get("type"),
+            "build_record_type_value": (
+                raw_opts.get("build_record_type_value") or raw_opts.get("build_record_type")
+            ),
+            "geom": geom_json,
+            "raw_props": json.dumps(raw_opts, ensure_ascii=False),
+            "source": source,
+        },
+    )
+
+
+# ── cad_constructions upsert ─────────────────────────────────────────────────
+
+
+def upsert_construction(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None:
+    """Upsert сооружение в cad_constructions. ON CONFLICT (cad_num) DO UPDATE."""
+    raw_props = feature.properties
+    raw_opts = raw_props.get("options") or {}
+    cad_num = raw_opts.get("cad_num") or raw_opts.get("cad_number")
+    if not cad_num:
+        logger.debug("upsert_construction: пропуск — нет cad_num, feature id=%s", feature.id)
+        return
+
+    cat_id = feature.category_id
+    # Определить тип: 36383 = сооружение, у NSPD есть 2 варианта
+    is_egrn = bool(raw_opts.get("is_egrn") or raw_opts.get("registers_id"))
+    category_name = raw_props.get("categoryName") or str(cat_id or "")
+    quarter_cad = raw_opts.get("quarter_cad_number") or _derive_quarter_cad(cad_num)
+    geom_json = json.dumps(feature.geometry) if feature.geometry else None
+
+    # params_* поля из NSPD characteristics
+    params = raw_opts.get("params") or {}
+
+    db.execute(
+        text("""
+            INSERT INTO cad_constructions (
+                cad_num, quarter_cad_number,
+                category_name, is_egrn,
+                params_purpose, params_name,
+                params_area, params_built_up_area,
+                params_year_built, params_year_commisioning,
+                params_height, params_volume, params_floors, params_underground_floors,
+                object_type_value, facility_cad_number,
+                cost_value, cost_index,
+                ownership_type, right_type,
+                readable_address, address_readable_address,
+                status, common_data_status,
+                registration_date, registers_id, united_cad_number,
+                geom, raw_props, source, fetched_at, updated_at
+            ) VALUES (
+                :cad_num, :quarter_cad_number,
+                :category_name, :is_egrn,
+                :params_purpose, :params_name,
+                :params_area, :params_built_up_area,
+                :params_year_built, :params_year_commisioning,
+                :params_height, :params_volume, :params_floors, :params_underground_floors,
+                :object_type_value, :facility_cad_number,
+                :cost_value, :cost_index,
+                :ownership_type, :right_type,
+                :readable_address, :address_readable_address,
+                :status, :common_data_status,
+                CAST(:registration_date AS date), :registers_id, :united_cad_number,
+                CASE WHEN :geom IS NOT NULL THEN
+                    ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom), 3857), 4326)
+                ELSE NULL END,
+                CAST(:raw_props AS jsonb), :source, NOW(), NOW()
+            )
+            ON CONFLICT (cad_num) DO UPDATE SET
+                category_name = EXCLUDED.category_name,
+                is_egrn = EXCLUDED.is_egrn,
+                params_purpose = EXCLUDED.params_purpose,
+                params_name = EXCLUDED.params_name,
+                params_area = EXCLUDED.params_area,
+                params_built_up_area = EXCLUDED.params_built_up_area,
+                params_year_built = EXCLUDED.params_year_built,
+                params_year_commisioning = EXCLUDED.params_year_commisioning,
+                params_height = EXCLUDED.params_height,
+                params_volume = EXCLUDED.params_volume,
+                params_floors = EXCLUDED.params_floors,
+                params_underground_floors = EXCLUDED.params_underground_floors,
+                object_type_value = EXCLUDED.object_type_value,
+                facility_cad_number = EXCLUDED.facility_cad_number,
+                cost_value = EXCLUDED.cost_value,
+                cost_index = EXCLUDED.cost_index,
+                ownership_type = EXCLUDED.ownership_type,
+                right_type = EXCLUDED.right_type,
+                readable_address = EXCLUDED.readable_address,
+                address_readable_address = EXCLUDED.address_readable_address,
+                status = EXCLUDED.status,
+                common_data_status = EXCLUDED.common_data_status,
+                registration_date = EXCLUDED.registration_date,
+                registers_id = EXCLUDED.registers_id,
+                united_cad_number = EXCLUDED.united_cad_number,
+                geom = COALESCE(EXCLUDED.geom, cad_constructions.geom),
+                raw_props = EXCLUDED.raw_props,
+                source = EXCLUDED.source,
+                updated_at = NOW()
+        """),
+        {
+            "cad_num": cad_num,
+            "quarter_cad_number": quarter_cad,
+            "category_name": category_name,
+            "is_egrn": is_egrn,
+            "params_purpose": params.get("purpose") or raw_opts.get("purpose"),
+            "params_name": params.get("name") or raw_opts.get("building_name"),
+            "params_area": _safe_numeric(params.get("area") or raw_opts.get("area")),
+            "params_built_up_area": _safe_numeric(raw_opts.get("built_up_area")),
+            "params_year_built": _safe_int(params.get("year_built") or raw_opts.get("year_built")),
+            "params_year_commisioning": _safe_int(raw_opts.get("year_commisioning")),
+            "params_height": _safe_numeric(raw_opts.get("height")),
+            "params_volume": _safe_numeric(raw_opts.get("volume")),
+            "params_floors": _safe_int(raw_opts.get("floors")),
+            "params_underground_floors": _safe_int(raw_opts.get("underground_floors")),
+            "object_type_value": raw_opts.get("object_type_value"),
+            "facility_cad_number": raw_opts.get("facility_cad_number"),
+            "cost_value": _safe_numeric(raw_opts.get("cost_value")),
+            "cost_index": _safe_numeric(raw_opts.get("cost_index")),
+            "ownership_type": raw_opts.get("ownership_type"),
+            "right_type": raw_opts.get("right_type"),
+            "readable_address": raw_opts.get("readable_address"),
+            "address_readable_address": raw_opts.get("address_readable_address"),
+            "status": raw_opts.get("status"),
+            "common_data_status": raw_opts.get("common_data_status"),
+            "registration_date": raw_opts.get("registration_date"),
+            "registers_id": _safe_int(raw_opts.get("registers_id")),
+            "united_cad_number": raw_opts.get("united_cad_number"),
+            "geom": geom_json,
+            "raw_props": json.dumps(raw_opts, ensure_ascii=False),
+            "source": source,
+        },
+    )
+
+
+# ── cad_oncs upsert ──────────────────────────────────────────────────────────
+
+
+def upsert_onc(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None:
+    """Upsert ОНС в cad_oncs. ON CONFLICT (cad_num) DO UPDATE."""
+    raw_opts = feature.properties.get("options") or {}
+    cad_num = raw_opts.get("cad_num") or raw_opts.get("cad_number")
+    if not cad_num:
+        logger.debug("upsert_onc: пропуск — нет cad_num, feature id=%s", feature.id)
+        return
+
+    quarter_cad = raw_opts.get("quarter_cad_number") or _derive_quarter_cad(cad_num)
+    geom_json = json.dumps(feature.geometry) if feature.geometry else None
+
+    db.execute(
+        text("""
+            INSERT INTO cad_oncs (
+                cad_num, quarter_cad_number,
+                purpose, area, build_record_area,
+                cost_value, status, readable_address,
+                geom, raw_props, source, fetched_at, updated_at
+            ) VALUES (
+                :cad_num, :quarter_cad_number,
+                :purpose, :area, :build_record_area,
+                :cost_value, :status, :readable_address,
+                CASE WHEN :geom IS NOT NULL THEN
+                    ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom), 3857), 4326)
+                ELSE NULL END,
+                CAST(:raw_props AS jsonb), :source, NOW(), NOW()
+            )
+            ON CONFLICT (cad_num) DO UPDATE SET
+                quarter_cad_number = EXCLUDED.quarter_cad_number,
+                purpose = EXCLUDED.purpose,
+                area = EXCLUDED.area,
+                build_record_area = EXCLUDED.build_record_area,
+                cost_value = EXCLUDED.cost_value,
+                status = EXCLUDED.status,
+                readable_address = EXCLUDED.readable_address,
+                geom = COALESCE(EXCLUDED.geom, cad_oncs.geom),
+                raw_props = EXCLUDED.raw_props,
+                source = EXCLUDED.source,
+                updated_at = NOW()
+        """),
+        {
+            "cad_num": cad_num,
+            "quarter_cad_number": quarter_cad,
+            "purpose": raw_opts.get("purpose"),
+            "area": _safe_numeric(raw_opts.get("area")),
+            "build_record_area": _safe_numeric(raw_opts.get("build_record_area")),
+            "cost_value": _safe_numeric(raw_opts.get("cost_value")),
+            "status": raw_opts.get("status"),
+            "readable_address": raw_opts.get("readable_address"),
+            "geom": geom_json,
+            "raw_props": json.dumps(raw_opts, ensure_ascii=False),
+            "source": source,
+        },
+    )
+
+
+# ── cad_enk upsert ───────────────────────────────────────────────────────────
+
+
+def upsert_enk(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None:
+    """Upsert ЕНК в cad_enk. ON CONFLICT (cad_num) DO UPDATE."""
+    raw_opts = feature.properties.get("options") or {}
+    cad_num = raw_opts.get("cad_num") or raw_opts.get("cad_number")
+    if not cad_num:
+        logger.debug("upsert_enk: пропуск — нет cad_num, feature id=%s", feature.id)
+        return
+
+    quarter_cad = raw_opts.get("quarter_cad_number") or _derive_quarter_cad(cad_num)
+    geom_json = json.dumps(feature.geometry) if feature.geometry else None
+
+    db.execute(
+        text("""
+            INSERT INTO cad_enk (
+                cad_num, quarter_cad_number,
+                name, purpose, type,
+                cost_value, facility_cad_number,
+                readable_address, registration_date, registers_id,
+                geom, raw_props, fetched_at, updated_at
+            ) VALUES (
+                :cad_num, :quarter_cad_number,
+                :name, :purpose, :type,
+                :cost_value, :facility_cad_number,
+                :readable_address, CAST(:registration_date AS date), :registers_id,
+                CASE WHEN :geom IS NOT NULL THEN
+                    ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom), 3857), 4326)
+                ELSE NULL END,
+                CAST(:raw_props AS jsonb), NOW(), NOW()
+            )
+            ON CONFLICT (cad_num) DO UPDATE SET
+                quarter_cad_number = EXCLUDED.quarter_cad_number,
+                name = EXCLUDED.name,
+                purpose = EXCLUDED.purpose,
+                type = EXCLUDED.type,
+                cost_value = EXCLUDED.cost_value,
+                facility_cad_number = EXCLUDED.facility_cad_number,
+                readable_address = EXCLUDED.readable_address,
+                registration_date = EXCLUDED.registration_date,
+                registers_id = EXCLUDED.registers_id,
+                geom = COALESCE(EXCLUDED.geom, cad_enk.geom),
+                raw_props = EXCLUDED.raw_props,
+                updated_at = NOW()
+        """),
+        {
+            "cad_num": cad_num,
+            "quarter_cad_number": quarter_cad,
+            "name": raw_opts.get("name"),
+            "purpose": raw_opts.get("purpose"),
+            "type": raw_opts.get("type") or raw_opts.get("object_type"),
+            "cost_value": _safe_numeric(raw_opts.get("cost_value")),
+            "facility_cad_number": raw_opts.get("facility_cad_number"),
+            "readable_address": raw_opts.get("readable_address"),
+            "registration_date": raw_opts.get("registration_date"),
+            "registers_id": _safe_int(raw_opts.get("registers_id")),
+            "geom": geom_json,
+            "raw_props": json.dumps(raw_opts, ensure_ascii=False),
+        },
+    )
+
+
+# ── cad_zouit upsert ─────────────────────────────────────────────────────────
+
+
+def upsert_zouit(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None:
+    """Upsert ЗОУИТ в cad_zouit.
+
+    Deduplicate по (reg_numb_border, category_id).
+    Geom — MultiPolygon; если пришёл Polygon — NSPD иногда так делает.
+    """
+    raw_opts = feature.properties.get("options") or {}
+    cat_id = feature.category_id
+    reg_numb_border = raw_opts.get("reg_numb_border") or raw_opts.get("registration_number")
+    type_zone = raw_opts.get("type_zone") or raw_opts.get("typeZone") or ""
+
+    if not reg_numb_border or cat_id is None:
+        logger.debug(
+            "upsert_zouit: пропуск — нет reg_numb_border или category_id, feature id=%s",
+            feature.id,
+        )
+        return
+
+    geom_json = json.dumps(feature.geometry) if feature.geometry else None
+
+    db.execute(
+        text("""
+            INSERT INTO cad_zouit (
+                reg_numb_border, category_id, category_name,
+                type_zone, type_boundary_value, subcategory,
+                cadastral_district,
+                content_restrict_encumbrances, name_by_doc,
+                legal_act_name, legal_act_date, legal_act_number,
+                legal_act_issuer, registration_date,
+                old_account_number,
+                geom, raw_props, source, fetched_at, updated_at
+            ) VALUES (
+                :reg_numb_border, :category_id, :category_name,
+                :type_zone, :type_boundary_value, :subcategory,
+                :cadastral_district,
+                :content_restrict_encumbrances, :name_by_doc,
+                :legal_act_name, CAST(:legal_act_date AS date), :legal_act_number,
+                :legal_act_issuer, CAST(:registration_date AS date),
+                :old_account_number,
+                CASE WHEN :geom IS NOT NULL THEN
+                    ST_Multi(
+                        ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom), 3857), 4326)
+                    )
+                ELSE NULL END,
+                CAST(:raw_props AS jsonb), :source, NOW(), NOW()
+            )
+            ON CONFLICT (reg_numb_border, category_id) DO UPDATE SET
+                category_name = EXCLUDED.category_name,
+                type_zone = EXCLUDED.type_zone,
+                type_boundary_value = EXCLUDED.type_boundary_value,
+                subcategory = EXCLUDED.subcategory,
+                cadastral_district = EXCLUDED.cadastral_district,
+                content_restrict_encumbrances = EXCLUDED.content_restrict_encumbrances,
+                name_by_doc = EXCLUDED.name_by_doc,
+                legal_act_name = EXCLUDED.legal_act_name,
+                legal_act_date = EXCLUDED.legal_act_date,
+                legal_act_number = EXCLUDED.legal_act_number,
+                legal_act_issuer = EXCLUDED.legal_act_issuer,
+                registration_date = EXCLUDED.registration_date,
+                old_account_number = EXCLUDED.old_account_number,
+                geom = COALESCE(EXCLUDED.geom, cad_zouit.geom),
+                raw_props = EXCLUDED.raw_props,
+                source = EXCLUDED.source,
+                updated_at = NOW()
+        """),
+        {
+            "reg_numb_border": reg_numb_border,
+            "category_id": cat_id,
+            "category_name": feature.properties.get("categoryName") or str(cat_id),
+            "type_zone": type_zone,
+            "type_boundary_value": raw_opts.get("type_boundary_value"),
+            "subcategory": _safe_int(raw_opts.get("subcategory")),
+            "cadastral_district": raw_opts.get("cadastral_district"),
+            "content_restrict_encumbrances": raw_opts.get("content_restrict_encumbrances"),
+            "name_by_doc": raw_opts.get("name_by_doc"),
+            "legal_act_name": raw_opts.get("legal_act_name"),
+            "legal_act_date": raw_opts.get("legal_act_date"),
+            "legal_act_number": raw_opts.get("legal_act_number"),
+            "legal_act_issuer": raw_opts.get("legal_act_issuer"),
+            "registration_date": raw_opts.get("registration_date"),
+            "old_account_number": raw_opts.get("old_account_number"),
+            "geom": geom_json,
+            "raw_props": json.dumps(raw_opts, ensure_ascii=False),
+            "source": source,
+        },
+    )
+
+
+# ── cad_quarter_stats upsert ─────────────────────────────────────────────────
+
+
+def upsert_quarter_stats(
+    db: Session,
+    quarter: str,
+    feature: NSPDBulkFeature,
+) -> None:
+    """Upsert квартальную статистику из feature 36381 в cad_quarter_stats.
+
+    cad_quarter_stats имеет FK на cad_quarters_geom(cad_number).
+    Если квартала ещё нет в cad_quarters_geom — пропускаем тихо
+    (INSERT не нарушит FK, но строка не появится; при следующем
+    проходе после добавления квартала — сработает).
+    """
+    raw_opts = feature.properties.get("options") or {}
+
+    db.execute(
+        text("""
+            INSERT INTO cad_quarter_stats (
+                cad_number,
+                cnt_land, cnt_oks,
+                cnt_land_geom, cnt_land_not_geom,
+                cnt_oks_area, cnt_oks_geom,
+                cnt_enk, cnt_pik,
+                cost_value_total, cost_value_total_geom,
+                sum_land_area, sum_land_geom_area,
+                date_cr, real_srid,
+                raw_props, fetched_at, updated_at
+            )
+            SELECT
+                :cad_number,
+                :cnt_land, :cnt_oks,
+                :cnt_land_geom, :cnt_land_not_geom,
+                :cnt_oks_area, :cnt_oks_geom,
+                :cnt_enk, :cnt_pik,
+                :cost_value_total, :cost_value_total_geom,
+                :sum_land_area, :sum_land_geom_area,
+                CAST(:date_cr AS date), :real_srid,
+                CAST(:raw_props AS jsonb), NOW(), NOW()
+            WHERE EXISTS (
+                SELECT 1 FROM cad_quarters_geom WHERE cad_number = :cad_number
+            )
+            ON CONFLICT (cad_number) DO UPDATE SET
+                cnt_land = EXCLUDED.cnt_land,
+                cnt_oks = EXCLUDED.cnt_oks,
+                cnt_land_geom = EXCLUDED.cnt_land_geom,
+                cnt_land_not_geom = EXCLUDED.cnt_land_not_geom,
+                cnt_oks_area = EXCLUDED.cnt_oks_area,
+                cnt_oks_geom = EXCLUDED.cnt_oks_geom,
+                cnt_enk = EXCLUDED.cnt_enk,
+                cnt_pik = EXCLUDED.cnt_pik,
+                cost_value_total = EXCLUDED.cost_value_total,
+                cost_value_total_geom = EXCLUDED.cost_value_total_geom,
+                sum_land_area = EXCLUDED.sum_land_area,
+                sum_land_geom_area = EXCLUDED.sum_land_geom_area,
+                date_cr = EXCLUDED.date_cr,
+                real_srid = EXCLUDED.real_srid,
+                raw_props = EXCLUDED.raw_props,
+                updated_at = NOW()
+        """),
+        {
+            "cad_number": quarter,
+            "cnt_land": _safe_int(raw_opts.get("cnt_land")),
+            "cnt_oks": _safe_int(raw_opts.get("cnt_oks")),
+            "cnt_land_geom": _safe_int(raw_opts.get("cnt_land_geom")),
+            "cnt_land_not_geom": _safe_int(raw_opts.get("cnt_land_not_geom")),
+            "cnt_oks_area": _safe_int(raw_opts.get("cnt_oks_area")),
+            "cnt_oks_geom": _safe_int(raw_opts.get("cnt_oks_geom")),
+            "cnt_enk": _safe_int(raw_opts.get("cnt_enk")),
+            "cnt_pik": _safe_int(raw_opts.get("cnt_pik")),
+            "cost_value_total": _safe_numeric(raw_opts.get("cost_value_total")),
+            "cost_value_total_geom": _safe_numeric(raw_opts.get("cost_value_total_geom")),
+            "sum_land_area": _safe_numeric(raw_opts.get("sum_land_area")),
+            "sum_land_geom_area": _safe_numeric(raw_opts.get("sum_land_geom_area")),
+            "date_cr": raw_opts.get("date_cr"),
+            "real_srid": _safe_int(raw_opts.get("real_srid")),
+            "raw_props": json.dumps(raw_opts, ensure_ascii=False),
+        },
+    )
+
+
+# ── Утилиты ──────────────────────────────────────────────────────────────────
+
+
+def _derive_quarter_cad(cad_num: str) -> str:
+    """Вывести номер квартала из кад. номера: '66:41:0303161:1' → '66:41:0303161'."""
+    parts = cad_num.split(":")
+    if len(parts) >= 3:
+        return ":".join(parts[:3])
+    return cad_num
+
+
+def _safe_numeric(value: Any) -> float | None:
+    """Конвертировать значение в float или None (безопасно)."""
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (ValueError, TypeError):
+        return None
+
+
+def _safe_int(value: Any) -> int | None:
+    """Конвертировать значение в int или None (безопасно)."""
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except (ValueError, TypeError):
+        return None

backend/app/services/cadastre/grid_geometry.py

@@ -0,0 +1,98 @@
+"""Геометрические утилиты для grid-walk по кадастровому кварталу.
+
+Используется в bulk_harvest.py для вычисления bbox квартала в EPSG:3857
+и генерации сетки точек для WMS GetFeatureInfo.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+
+def quarter_bbox_3857(db: Session, quarter: str) -> tuple[float, float, float, float] | None:
+    """Получить bbox квартала из cad_quarters_geom в EPSG:3857.
+
+    Args:
+        db: SQLAlchemy session.
+        quarter: кадастровый номер квартала, например '66:41:0303161'.
+
+    Returns:
+        (xmin, ymin, xmax, ymax) в метрах EPSG:3857, или None если квартал не найден.
+    """
+    row = (
+        db.execute(
+            text("""
+                SELECT
+                    ST_XMin(b) AS xmin,
+                    ST_YMin(b) AS ymin,
+                    ST_XMax(b) AS xmax,
+                    ST_YMax(b) AS ymax
+                FROM (
+                    SELECT ST_Extent(ST_Transform(geom, 3857)) AS b
+                    FROM cad_quarters_geom
+                    WHERE cad_number = :q
+                ) sub
+            """),
+            {"q": quarter},
+        )
+        .mappings()
+        .first()
+    )
+    if not row or row["xmin"] is None:
+        logger.warning("quarter_bbox_3857: квартал %s не найден в cad_quarters_geom", quarter)
+        return None
+    return (
+        float(row["xmin"]),
+        float(row["ymin"]),
+        float(row["xmax"]),
+        float(row["ymax"]),
+    )
+
+
+def generate_grid_click_points(
+    bbox: tuple[float, float, float, float],
+    grid_size: int = 15,
+    tile_size: int = 512,
+) -> list[tuple[tuple[float, float, float, float], tuple[int, int]]]:
+    """Генерировать grid_size × grid_size ячеек для bbox квартала.
+
+    Каждая ячейка — свой sub-bbox + центральный пиксель (click_xy) для
+    WMS GetFeatureInfo. 15×15 = 225 запросов обеспечивают хорошее покрытие
+    при шаге ~50м для типичного квартала ЕКБ (~750м × ~750м).
+
+    Args:
+        bbox: (xmin, ymin, xmax, ymax) в EPSG:3857.
+        grid_size: количество ячеек по каждой оси (15 → 225 запросов).
+        tile_size: размер виртуального WMS тайла в пикселях (квадратный).
+
+    Returns:
+        Список (sub_bbox, click_xy) — sub_bbox в EPSG:3857, click_xy в пикселях.
+    """
+    xmin, ymin, xmax, ymax = bbox
+    x_step = (xmax - xmin) / grid_size
+    y_step = (ymax - ymin) / grid_size
+
+    result: list[tuple[tuple[float, float, float, float], tuple[int, int]]] = []
+
+    for i in range(grid_size):
+        for j in range(grid_size):
+            cell_xmin = xmin + i * x_step
+            cell_ymin = ymin + j * y_step
+            cell_xmax = cell_xmin + x_step
+            cell_ymax = cell_ymin + y_step
+            # Клик в центр ячейки
+            click_x = tile_size // 2
+            click_y = tile_size // 2
+            result.append(
+                (
+                    (cell_xmin, cell_ymin, cell_xmax, cell_ymax),
+                    (click_x, click_y),
+                )
+            )
+
+    return result

backend/app/workers/celery_app.py

@@ -26,6 +26,7 @@ celery_app = Celery(
         "app.workers.tasks.poi_sync",
         "app.workers.tasks.noise_sync",
         "app.workers.tasks.pzz_sync",
+        "app.workers.tasks.scrape_cadastre",
     ],
 )
 celery_app.conf.timezone = "Europe/Moscow"

backend/app/workers/tasks/scrape_cadastre.py

@@ -0,0 +1,415 @@
+"""Celery tasks для bulk cadastre harvest (GH issue #168, PR 3/5).
+
+Архитектура:
+  - enqueue_cadastre_harvest(job_id) — читает cadastre_jobs.raw_targets,
+    запускает bulk_harvest_quarter_task для каждого квартала
+  - bulk_harvest_quarter_task(quarter, job_id) — per-quarter Celery task,
+    оборачивает async harvest_quarter() через asyncio.run()
+
+Saga state в cadastre_jobs.phase_state JSONB:
+  {phase, quarter, parcels_total_meta, buildings_total_meta, overflow_cats, ...}
+
+acks_late=True + idempotent upsert (ON CONFLICT) → resume после redeploy.
+WAF 403 → no retry, increment waf_blocked_count.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import datetime as _dt
+import json
+import logging
+from typing import Any
+
+from sqlalchemy import text
+
+from app.core.db import SessionLocal
+from app.scrapers.nspd_bulk_client import NspdBulkWafError
+from app.workers.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+# Очередь для cadastre задач
+CADASTRE_QUEUE = "geo"
+
+
+# ── Saga state helpers ───────────────────────────────────────────────────────
+
+
+def _update_job(
+    db: Any,
+    job_id: int,
+    *,
+    status: str | None = None,
+    started_at: _dt.datetime | None = None,
+    finished_at: _dt.datetime | None = None,
+    phase_state_update: dict[str, Any] | None = None,
+    heartbeat: bool = False,
+    targets_done_inc: int = 0,
+    targets_failed_inc: int = 0,
+    requests_count_inc: int = 0,
+    waf_blocked_inc: int = 0,
+    last_error: str | None = None,
+) -> None:
+    """Атомарный UPDATE одной строки cadastre_jobs.
+
+    Все параметры опциональны — только переданные поля обновляются.
+    phase_state_update мержится поверх текущего JSONB.
+    """
+    sets: list[str] = []
+    params: dict[str, Any] = {"id": job_id}
+
+    if heartbeat or heartbeat is True:
+        sets.append("heartbeat_at = NOW()")
+
+    if status is not None:
+        sets.append("status = :status")
+        params["status"] = status
+
+    if started_at is not None:
+        sets.append("started_at = COALESCE(started_at, :started_at)")
+        params["started_at"] = started_at
+
+    if finished_at is not None:
+        sets.append("finished_at = :finished_at")
+        params["finished_at"] = finished_at
+
+    if last_error is not None:
+        sets.append("error = :last_error")
+        params["last_error"] = last_error[:2000]
+
+    if phase_state_update is not None:
+        # Merge JSONB: текущий phase_state || новый partial update
+        sets.append(
+            "phase_state = COALESCE(phase_state, CAST('{}' AS jsonb)) "
+            "|| CAST(:phase_state_update AS jsonb)"
+        )
+        params["phase_state_update"] = json.dumps(phase_state_update, ensure_ascii=False)
+
+    if targets_done_inc:
+        sets.append("targets_done = targets_done + :done_inc")
+        params["done_inc"] = targets_done_inc
+
+    if targets_failed_inc:
+        sets.append("targets_failed = targets_failed + :failed_inc")
+        params["failed_inc"] = targets_failed_inc
+
+    if requests_count_inc:
+        sets.append("requests_count = requests_count + :req_inc")
+        params["req_inc"] = requests_count_inc
+
+    if waf_blocked_inc:
+        sets.append("waf_blocked_count = waf_blocked_count + :waf_inc")
+        params["waf_inc"] = waf_blocked_inc
+
+    if not sets:
+        return
+
+    db.execute(
+        text(f"UPDATE cadastre_jobs SET {', '.join(sets)} WHERE job_id = :id"),
+        params,
+    )
+
+
+def _is_job_cancelled(db: Any, job_id: int) -> bool:
+    """Проверить не отменён ли job (воркер должен stop обработку)."""
+    row = (
+        db.execute(
+            text("SELECT status FROM cadastre_jobs WHERE job_id = :id"),
+            {"id": job_id},
+        )
+        .mappings()
+        .first()
+    )
+    return row is not None and row["status"] == "cancelled"
+
+
+# ── Основной per-quarter task ────────────────────────────────────────────────
+
+
+@celery_app.task(
+    bind=True,
+    name="tasks.cadastre.harvest_quarter",
+    acks_late=True,
+    max_retries=3,
+    autoretry_for=(Exception,),
+    retry_backoff=True,
+    retry_backoff_max=60,
+    dont_autoretry_for=(NspdBulkWafError,),
+    soft_time_limit=600,  # 10 мин: 225 grid-walk запросов × ~0.1с + overhead
+    time_limit=900,
+)
+def bulk_harvest_quarter_task(self: Any, quarter: str, job_id: int) -> dict[str, Any]:
+    """Celery wrapper для harvest_quarter одного кадастрового квартала.
+
+    Sync→async bridge через asyncio.run().
+    Saga state пишется в cadastre_jobs.phase_state через update_progress callback.
+    WAF 403 → не retry, waf_blocked_count++.
+    """
+    from app.services.cadastre.bulk_harvest import harvest_quarter
+
+    db = SessionLocal()
+    try:
+        # Проверить не отменён ли job
+        if _is_job_cancelled(db, job_id):
+            logger.info(
+                "bulk_harvest_quarter_task: job=%d cancelled, skip quarter=%s",
+                job_id,
+                quarter,
+            )
+            db.commit()
+            return {"quarter": quarter, "job_id": job_id, "skipped": "cancelled"}
+
+        # Обновить heartbeat + started_at
+        _update_job(
+            db,
+            job_id,
+            status="running",
+            started_at=_dt.datetime.now(_dt.UTC),
+            heartbeat=True,
+        )
+        db.commit()
+
+        def progress_cb(state: dict[str, Any]) -> None:
+            """Callback: пишет phase_state в DB при каждом шаге фазы."""
+            try:
+                _update_job(db, job_id, phase_state_update=state, heartbeat=True)
+                db.commit()
+            except Exception as cb_err:
+                logger.warning(
+                    "progress_cb failed for job=%d quarter=%s: %s",
+                    job_id,
+                    quarter,
+                    cb_err,
+                )
+                try:
+                    db.rollback()
+                except Exception:
+                    pass
+
+        # Запустить async harvest_quarter в синхронном контексте Celery
+        async def _run() -> dict[str, Any]:
+            from app.scrapers.nspd_bulk_client import NSPDBulkClient
+
+            async with NSPDBulkClient() as client:
+                result = await harvest_quarter(
+                    db=db,
+                    client=client,
+                    quarter=quarter,
+                    job_id=job_id,
+                    update_progress=progress_cb,
+                )
+            return {
+                "quarter": quarter,
+                "job_id": job_id,
+                "parcels": result.parcels_upserted,
+                "buildings": result.buildings_upserted,
+                "constructions": result.constructions_upserted,
+                "oncs": result.oncs_upserted,
+                "enks": result.enks_upserted,
+                "zouit": result.zouit_upserted,
+                "grid_walk_requests": result.grid_walk_requests,
+            }
+
+        result_dict = asyncio.run(_run())
+
+        # Успех: increment targets_done
+        _update_job(
+            db,
+            job_id,
+            targets_done_inc=1,
+            requests_count_inc=result_dict.get("grid_walk_requests", 0),
+            heartbeat=True,
+        )
+        # Проверить не завершён ли job (все targets done)
+        _maybe_finish_job(db, job_id)
+        db.commit()
+
+        logger.info(
+            "bulk_harvest_quarter_task done: quarter=%s job=%d parcels=%d buildings=%d",
+            quarter,
+            job_id,
+            result_dict.get("parcels", 0),
+            result_dict.get("buildings", 0),
+        )
+        return result_dict
+
+    except NspdBulkWafError as e:
+        # WAF 403 — не retry, только счётчик
+        logger.warning(
+            "bulk_harvest_quarter_task WAF block: quarter=%s job=%d: %s",
+            quarter,
+            job_id,
+            e,
+        )
+        _update_job(
+            db,
+            job_id,
+            waf_blocked_inc=1,
+            targets_failed_inc=1,
+            last_error=f"WAF: {e}",
+            heartbeat=True,
+        )
+        db.commit()
+        # NspdBulkWafError в dont_autoretry_for → не retry
+        raise
+
+    except Exception as e:
+        logger.exception(
+            "bulk_harvest_quarter_task failed: quarter=%s job=%d: %s",
+            quarter,
+            job_id,
+            e,
+        )
+        try:
+            _update_job(
+                db,
+                job_id,
+                last_error=f"{type(e).__name__}: {e}"[:500],
+                heartbeat=True,
+            )
+            db.commit()
+        except Exception as inner_e:
+            logger.warning("failed to update job state: %s", inner_e)
+            try:
+                db.rollback()
+            except Exception:
+                pass
+        raise
+
+    finally:
+        db.close()
+
+
+def _maybe_finish_job(db: Any, job_id: int) -> None:
+    """Отметить job как done если targets_done + targets_failed == targets_total."""
+    row = (
+        db.execute(
+            text("""
+                SELECT targets_total, targets_done, targets_failed, status
+                FROM cadastre_jobs WHERE job_id = :id
+            """),
+            {"id": job_id},
+        )
+        .mappings()
+        .first()
+    )
+    if not row:
+        return
+    if row["status"] in ("cancelled", "done", "failed"):
+        return
+    total = row["targets_total"] or 0
+    done = row["targets_done"] or 0
+    failed = row["targets_failed"] or 0
+    if total > 0 and (done + failed) >= total:
+        db.execute(
+            text("""
+                UPDATE cadastre_jobs
+                SET status = 'done', finished_at = NOW()
+                WHERE job_id = :id AND status NOT IN ('cancelled', 'done', 'failed')
+            """),
+            {"id": job_id},
+        )
+        logger.info(
+            "cadastre job %d finished (done=%d failed=%d total=%d)", job_id, done, failed, total
+        )
+
+
+# ── Enqueue task ─────────────────────────────────────────────────────────────
+
+
+@celery_app.task(
+    bind=True,
+    name="tasks.cadastre.enqueue_harvest",
+)
+def enqueue_cadastre_harvest(self: Any, job_id: int) -> dict[str, Any]:
+    """Прочитать cadastre_jobs.raw_targets и запустить per-quarter task'и.
+
+    Идемпотентен: повторный вызов просто повторно enqueue'ит те же quarters.
+    Это OK т.к. bulk_harvest_quarter_task idempotent (ON CONFLICT DO UPDATE +
+    phase_state='done' early exit).
+    """
+    db = SessionLocal()
+    try:
+        job = (
+            db.execute(
+                text("SELECT * FROM cadastre_jobs WHERE job_id = :id"),
+                {"id": job_id},
+            )
+            .mappings()
+            .first()
+        )
+        if not job:
+            logger.error("enqueue_cadastre_harvest: job_id=%d not found", job_id)
+            return {"error": "job_not_found", "job_id": job_id}
+
+        if job["status"] == "cancelled":
+            return {"job_id": job_id, "skipped": "cancelled"}
+
+        raw_targets = job["raw_targets"] or {}
+        quarters: list[str] = raw_targets.get("quarters", [])
+
+        if not quarters:
+            logger.warning("enqueue_cadastre_harvest: job=%d имеет пустой список quarters", job_id)
+            return {"job_id": job_id, "enqueued": 0}
+
+        # Обновить статус на running
+        _update_job(db, job_id, status="running", started_at=_dt.datetime.now(_dt.UTC))
+        db.commit()
+
+        for quarter in quarters:
+            bulk_harvest_quarter_task.apply_async(
+                args=[quarter, job_id],
+                queue=CADASTRE_QUEUE,
+            )
+
+        logger.info(
+            "enqueue_cadastre_harvest: job=%d enqueued %d quarters",
+            job_id,
+            len(quarters),
+        )
+        return {"job_id": job_id, "enqueued": len(quarters)}
+
+    finally:
+        db.close()
+
+
+# ── Zombie cleanup ───────────────────────────────────────────────────────────
+
+
+@celery_app.task(name="tasks.cadastre.cleanup_zombies")
+def cleanup_cadastre_zombies() -> dict[str, Any]:
+    """Periodic: найти stale cadastre_jobs (heartbeat > 5 мин) и переставить в паузу.
+
+    Запускается через beat (beat_schedule.py). Idempotent.
+    """
+    db = SessionLocal()
+    stale_ids: list[int] = []
+    try:
+        rows = (
+            db.execute(
+                text("""
+                    UPDATE cadastre_jobs
+                    SET status = 'paused',
+                        error = COALESCE(error, 'auto-paused: stale heartbeat')
+                    WHERE status = 'running'
+                      AND heartbeat_at < NOW() - INTERVAL '5 minutes'
+                    RETURNING job_id
+                """)
+            )
+            .mappings()
+            .all()
+        )
+        db.commit()
+        stale_ids = [int(r["job_id"]) for r in rows]
+        if stale_ids:
+            logger.warning("cleanup_cadastre_zombies: paused stale jobs: %s", stale_ids)
+    except Exception as e:
+        logger.exception("cleanup_cadastre_zombies failed: %s", e)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+    finally:
+        db.close()
+    return {"stale_paused": stale_ids}

backend/pyproject.toml

@@ -79,3 +79,6 @@ strict = true
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 asyncio_mode = "auto"
+markers = [
+    "slow: marks tests as slow (need real network, deselect with -m 'not slow')",
+]

backend/tests/api/v1/test_admin_cadastre.py

@@ -0,0 +1,306 @@
+"""Тесты для admin_cadastre.py API endpoints.
+
+Использует FastAPI TestClient с mock DB и mock Celery tasks.
+"""
+
+from __future__ import annotations
+
+# scrape_cadastre зависит от app.scrapers.nspd_bulk_client (PR 2/5).
+# Используем importlib.util.find_spec вместо прямого import — иначе
+# `import app.workers...` пересоздаёт `app` как Python package и
+# перебивает FastAPI instance, привязанный в строке выше → AttributeError
+# на app.dependency_overrides.
+import importlib.util
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+_SCRAPE_CADASTRE_AVAILABLE = (
+    importlib.util.find_spec("app.workers.tasks.scrape_cadastre") is not None
+)
+
+requires_scrape_cadastre = pytest.mark.skipif(
+    not _SCRAPE_CADASTRE_AVAILABLE,
+    reason=(
+        "app.scrapers.nspd_bulk_client отсутствует — нужно смержить main (PR #170). "
+        "Команда: git rebase origin/main"
+    ),
+)
+
+# Токен для тестов (не реальный)
+ADMIN_TOKEN = "test-admin-token"
+ADMIN_HEADERS = {"X-Admin-Token": ADMIN_TOKEN}
+
+
+def _make_mock_db(quarters: list[str] | None = None, job_row: dict[str, Any] | None = None):
+    """Создать mock db session с преднастроенными ответами."""
+    db = MagicMock()
+
+    # Результат для _resolve_quarters (SELECT cad_number FROM cad_quarters_geom)
+    quarter_rows = [(q,) for q in (quarters or ["66:41:0303161", "66:41:0303162"])]
+
+    # Mock execute chain для разных запросов
+    def mock_execute(stmt, params=None):
+        stmt_str = str(stmt)
+        result = MagicMock()
+
+        if "cad_quarters_geom" in stmt_str and "cad_number" in stmt_str:
+            # _resolve_quarters query
+            result.all.return_value = quarter_rows
+        elif "cadastre_jobs" in stmt_str and "RETURNING job_id" in stmt_str:
+            # _create_job INSERT
+            result.scalar_one.return_value = 42
+        elif "cadastre_jobs" in stmt_str and "SELECT" in stmt_str:
+            # get/list jobs
+            result.mappings.return_value.all.return_value = [job_row] if job_row else []
+            result.mappings.return_value.first.return_value = job_row
+        elif "RETURNING job_id" in stmt_str:
+            # cancel/resume
+            result.first.return_value = (42,)
+        else:
+            result.all.return_value = []
+            result.mappings.return_value.all.return_value = []
+            result.mappings.return_value.first.return_value = None
+            result.scalar_one.return_value = 1
+
+        return result
+
+    db.execute = mock_execute
+    db.commit = MagicMock()
+    return db
+
+
+def _mock_get_db(db):
+    """FastAPI dependency override factory."""
+
+    def _get_db_override():
+        yield db
+
+    return _get_db_override
+
+
+def _make_sample_job_row() -> dict[str, Any]:
+    import datetime
+
+    return {
+        "job_id": 42,
+        "name": "cadastre_pilot",
+        "job_kind": "pilot",
+        "scope": "pilot",
+        "status": "queued",
+        "triggered_by": "manual",
+        "rate_ms": 333,
+        "created_at": datetime.datetime(2026, 5, 15, 10, 0, 0),
+        "started_at": None,
+        "finished_at": None,
+        "heartbeat_at": None,
+        "targets_total": 50,
+        "targets_done": 0,
+        "targets_failed": 0,
+        "targets_skipped": 0,
+        "requests_count": 0,
+        "waf_blocked_count": 0,
+        "error": None,
+        "phase_state": None,
+        "raw_targets": {"quarters": ["66:41:0303161"]},
+    }
+
+
+# ── Tests ────────────────────────────────────────────────────────────────────
+
+
+@requires_scrape_cadastre
+@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
+def test_create_job_pilot_returns_job_id() -> None:
+    """POST /cadastre/jobs scope=pilot → job_id + targets_total."""
+    quarters_50 = [f"66:41:{i:07d}" for i in range(50)]
+    db = _make_mock_db(quarters=quarters_50)
+
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _mock_get_db(db)
+
+    try:
+        with patch("app.workers.tasks.scrape_cadastre.enqueue_cadastre_harvest") as mock_task:
+            mock_task.apply_async = MagicMock()
+            client = TestClient(app)
+            response = client.post(
+                "/api/v1/admin/cadastre/jobs",
+                json={"scope": "pilot"},
+                headers=ADMIN_HEADERS,
+            )
+
+        assert response.status_code == 200, response.text
+        body = response.json()
+        assert "job_id" in body
+        assert body["scope"] == "pilot"
+        assert body["targets_total"] == 50
+        assert "estimate_minutes" in body
+    finally:
+        app.dependency_overrides.clear()
+
+
+@requires_scrape_cadastre
+@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
+def test_create_job_manual_list() -> None:
+    """POST /cadastre/jobs scope=manual_list с явным списком."""
+    db = _make_mock_db()
+
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _mock_get_db(db)
+
+    try:
+        with patch("app.workers.tasks.scrape_cadastre.enqueue_cadastre_harvest") as mock_task:
+            mock_task.apply_async = MagicMock()
+            client = TestClient(app)
+            response = client.post(
+                "/api/v1/admin/cadastre/jobs",
+                json={
+                    "scope": "manual_list",
+                    "quarters": ["66:41:0303161", "66:41:0303162"],
+                },
+                headers=ADMIN_HEADERS,
+            )
+
+        assert response.status_code == 200, response.text
+        body = response.json()
+        assert body["targets_total"] == 2
+    finally:
+        app.dependency_overrides.clear()
+
+
+@requires_scrape_cadastre
+@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
+def test_create_job_manual_list_empty_quarters_returns_400() -> None:
+    """scope=manual_list без quarters → 400."""
+    db = _make_mock_db()
+
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _mock_get_db(db)
+
+    try:
+        client = TestClient(app)
+        response = client.post(
+            "/api/v1/admin/cadastre/jobs",
+            json={"scope": "manual_list"},
+            headers=ADMIN_HEADERS,
+        )
+        assert response.status_code == 400
+    finally:
+        app.dependency_overrides.clear()
+
+
+@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
+def test_list_jobs_returns_list() -> None:
+    """GET /cadastre/jobs → список jobs."""
+    db = _make_mock_db(job_row=_make_sample_job_row())
+
+    # Переопределяем execute чтобы list query вернул список
+    job_row = _make_sample_job_row()
+    mock_result = MagicMock()
+    mock_result.mappings.return_value.all.return_value = [job_row]
+    db.execute = MagicMock(return_value=mock_result)
+
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _mock_get_db(db)
+
+    try:
+        client = TestClient(app)
+        response = client.get("/api/v1/admin/cadastre/jobs", headers=ADMIN_HEADERS)
+        assert response.status_code == 200, response.text
+        body = response.json()
+        assert isinstance(body, list)
+        assert len(body) == 1
+        assert body[0]["job_id"] == 42
+        assert body[0]["scope"] == "pilot"
+        assert "progress_pct" in body[0]
+    finally:
+        app.dependency_overrides.clear()
+
+
+@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
+def test_get_job_not_found_returns_404() -> None:
+    """GET /cadastre/jobs/9999 → 404."""
+    db = MagicMock()
+    mock_result = MagicMock()
+    mock_result.mappings.return_value.first.return_value = None
+    db.execute = MagicMock(return_value=mock_result)
+
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _mock_get_db(db)
+
+    try:
+        client = TestClient(app)
+        response = client.get("/api/v1/admin/cadastre/jobs/9999", headers=ADMIN_HEADERS)
+        assert response.status_code == 404
+    finally:
+        app.dependency_overrides.clear()
+
+
+@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
+def test_cancel_job_success() -> None:
+    """POST /cadastre/jobs/42/cancel → {cancelled: true}."""
+    db = MagicMock()
+    mock_result = MagicMock()
+    mock_result.first.return_value = (42,)
+    db.execute = MagicMock(return_value=mock_result)
+    db.commit = MagicMock()
+
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _mock_get_db(db)
+
+    try:
+        client = TestClient(app)
+        response = client.post(
+            "/api/v1/admin/cadastre/jobs/42/cancel",
+            headers=ADMIN_HEADERS,
+        )
+        assert response.status_code == 200, response.text
+        body = response.json()
+        assert body["cancelled"] is True
+        assert body["job_id"] == 42
+    finally:
+        app.dependency_overrides.clear()
+
+
+@patch("app.core.config.settings.scrape_admin_token", ADMIN_TOKEN)
+def test_cancel_job_not_found_returns_404() -> None:
+    """POST /cadastre/jobs/9999/cancel когда job не найден → 404."""
+    db = MagicMock()
+    mock_result = MagicMock()
+    mock_result.first.return_value = None  # job не найден / уже cancelled
+    db.execute = MagicMock(return_value=mock_result)
+    db.commit = MagicMock()
+
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _mock_get_db(db)
+
+    try:
+        client = TestClient(app)
+        response = client.post(
+            "/api/v1/admin/cadastre/jobs/9999/cancel",
+            headers=ADMIN_HEADERS,
+        )
+        assert response.status_code == 404
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_create_job_no_token_returns_401() -> None:
+    """Без X-Admin-Token → 401/503."""
+    client = TestClient(app)
+    response = client.post(
+        "/api/v1/admin/cadastre/jobs",
+        json={"scope": "pilot"},
+    )
+    assert response.status_code in (401, 503)

backend/tests/services/test_cadastre_bulk.py

@@ -0,0 +1,490 @@
+"""Тесты для bulk_harvest.py и grid_geometry.py.
+
+Использует mock DB (через unittest.mock) и mock NSPDBulkClient.
+Не требует реального PostgreSQL или NSPD сети.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from app.schemas.nspd_bulk import NSPDBulkFeature, QuarterSnapshot
+from app.services.cadastre.bulk_harvest import (
+    _derive_quarter_cad,
+    _safe_int,
+    _safe_numeric,
+    upsert_features,
+)
+from app.services.cadastre.grid_geometry import generate_grid_click_points
+
+# ── Фикстуры ─────────────────────────────────────────────────────────────────
+
+
+def _make_parcel_feature(cad_num: str = "66:41:0303161:1") -> NSPDBulkFeature:
+    """Создать NSPDBulkFeature для ЗУ (category_id=36368)."""
+    return NSPDBulkFeature.model_validate(
+        {
+            "id": 1,
+            "type": "Feature",
+            "geometry": {
+                "type": "Polygon",
+                "coordinates": [
+                    [
+                        [6090000, 7590000],
+                        [6090100, 7590000],
+                        [6090100, 7590100],
+                        [6090000, 7590100],
+                        [6090000, 7590000],
+                    ]
+                ],
+            },
+            "properties": {
+                "categoryId": 36368,
+                "options": {
+                    "cad_num": cad_num,
+                    "quarter_cad_number": "66:41:0303161",
+                    "land_record_area": 1234.56,
+                    "cost_value": 5000000.0,
+                    "readable_address": "г Екатеринбург, ул Тестовая",
+                    "status": "Учтённый",
+                    "ownership_type": "частная",
+                },
+            },
+        }
+    )
+
+
+def _make_building_feature(cad_num: str = "66:41:0303161:1:1") -> NSPDBulkFeature:
+    """Создать NSPDBulkFeature для здания (category_id=36369)."""
+    return NSPDBulkFeature.model_validate(
+        {
+            "id": 2,
+            "type": "Feature",
+            "geometry": {
+                "type": "Polygon",
+                "coordinates": [
+                    [
+                        [6090000, 7590000],
+                        [6090050, 7590000],
+                        [6090050, 7590050],
+                        [6090000, 7590050],
+                        [6090000, 7590000],
+                    ]
+                ],
+            },
+            "properties": {
+                "categoryId": 36369,
+                "options": {
+                    "cad_num": cad_num,
+                    "quarter_cad_number": "66:41:0303161",
+                    "purpose": "Жилое",
+                    "floors": "9",
+                    "area": 5000.0,
+                    "cost_value": 20000000.0,
+                    "readable_address": "г Екатеринбург, ул Тестовая, д 1",
+                },
+            },
+        }
+    )
+
+
+def _make_zouit_feature(reg_numb: str = "66-0.1-1.1-1234") -> NSPDBulkFeature:
+    """Создать NSPDBulkFeature для ЗОУИТ (category_id=36940)."""
+    return NSPDBulkFeature.model_validate(
+        {
+            "id": 3,
+            "type": "Feature",
+            "geometry": {
+                "type": "MultiPolygon",
+                "coordinates": [
+                    [
+                        [
+                            [6090000, 7590000],
+                            [6091000, 7590000],
+                            [6091000, 7591000],
+                            [6090000, 7591000],
+                            [6090000, 7590000],
+                        ]
+                    ]
+                ],
+            },
+            "properties": {
+                "categoryId": 36940,
+                "categoryName": "ЗОУИТ",
+                "options": {
+                    "reg_numb_border": reg_numb,
+                    "type_zone": "Охранная зона ЛЭП",
+                    "registration_date": "2020-01-01",
+                },
+            },
+        }
+    )
+
+
+# ── Вспомогательные функции ───────────────────────────────────────────────────
+
+
+def test_derive_quarter_cad_4_segments() -> None:
+    assert _derive_quarter_cad("66:41:0303161:1") == "66:41:0303161"
+
+
+def test_derive_quarter_cad_5_segments() -> None:
+    assert _derive_quarter_cad("66:41:0303161:1:1") == "66:41:0303161"
+
+
+def test_derive_quarter_cad_3_segments_unchanged() -> None:
+    assert _derive_quarter_cad("66:41:0303161") == "66:41:0303161"
+
+
+def test_safe_numeric_valid() -> None:
+    assert _safe_numeric("1234.56") == pytest.approx(1234.56)
+    assert _safe_numeric(100) == pytest.approx(100.0)
+
+
+def test_safe_numeric_none_returns_none() -> None:
+    assert _safe_numeric(None) is None
+
+
+def test_safe_numeric_invalid_returns_none() -> None:
+    assert _safe_numeric("не число") is None
+
+
+def test_safe_int_valid() -> None:
+    assert _safe_int("9") == 9
+    assert _safe_int(5.7) == 5
+
+
+def test_safe_int_none_returns_none() -> None:
+    assert _safe_int(None) is None
+
+
+# ── Grid geometry ────────────────────────────────────────────────────────────
+
+
+def test_generate_grid_click_points_count() -> None:
+    """15×15 grid → 225 точек."""
+    bbox = (6090000.0, 7590000.0, 6090750.0, 7590750.0)
+    points = generate_grid_click_points(bbox, grid_size=15)
+    assert len(points) == 225
+
+
+def test_generate_grid_click_points_bbox_coverage() -> None:
+    """Все sub-bbox покрывают исходный bbox без перекрытий."""
+    bbox = (0.0, 0.0, 1500.0, 1500.0)
+    points = generate_grid_click_points(bbox, grid_size=3)
+    # 3×3 = 9 ячеек
+    assert len(points) == 9
+    # Первая ячейка начинается с 0,0
+    first_bbox, first_click = points[0]
+    assert first_bbox[0] == pytest.approx(0.0)
+    assert first_bbox[1] == pytest.approx(0.0)
+    # click_xy в центре тайла (256, 256) для tile_size=512
+    assert first_click == (256, 256)
+
+
+def test_generate_grid_click_points_custom_grid() -> None:
+    bbox = (0.0, 0.0, 1000.0, 1000.0)
+    points = generate_grid_click_points(bbox, grid_size=5, tile_size=256)
+    assert len(points) == 25
+    # Каждая ячейка имеет правильный размер 200м×200м
+    cell_bbox, _ = points[0]
+    assert cell_bbox[2] - cell_bbox[0] == pytest.approx(200.0)
+    assert cell_bbox[3] - cell_bbox[1] == pytest.approx(200.0)
+
+
+# ── upsert_features routing ──────────────────────────────────────────────────
+
+
+def test_upsert_features_routes_parcels_and_buildings() -> None:
+    """upsert_features должен вызывать upsert_parcel и upsert_building."""
+    parcel = _make_parcel_feature()
+    building = _make_building_feature()
+
+    db = MagicMock()
+    db.execute = MagicMock(return_value=MagicMock())
+    db.begin_nested = MagicMock(return_value=MagicMock(__enter__=MagicMock(), __exit__=MagicMock()))
+
+    with (
+        patch("app.services.cadastre.bulk_harvest.upsert_parcel") as mock_parcel,
+        patch("app.services.cadastre.bulk_harvest.upsert_building") as mock_building,
+    ):
+        counts = upsert_features(db, [parcel, building], source="search")
+
+    mock_parcel.assert_called_once_with(db, parcel, source="search")
+    mock_building.assert_called_once_with(db, building, source="search")
+    assert counts["parcels"] == 1
+    assert counts["buildings"] == 1
+
+
+def test_upsert_features_routes_zouit() -> None:
+    """ЗОУИТ (36940) маршрутируется в upsert_zouit."""
+    zouit = _make_zouit_feature()
+    db = MagicMock()
+
+    with patch("app.services.cadastre.bulk_harvest.upsert_zouit") as mock_zouit:
+        counts = upsert_features(db, [zouit], source="search")
+
+    mock_zouit.assert_called_once()
+    assert counts["zouit"] == 1
+
+
+def test_upsert_features_unknown_category_counts_as_skipped() -> None:
+    """Неизвестный category_id → skipped, не raise."""
+    unknown_feature = NSPDBulkFeature.model_validate(
+        {
+            "type": "Feature",
+            "geometry": None,
+            "properties": {"categoryId": 99999, "options": {}},
+        }
+    )
+    db = MagicMock()
+    counts = upsert_features(db, [unknown_feature], source="search")
+    assert counts["skipped"] == 1
+
+
+def test_upsert_features_quarter_stats_not_counted() -> None:
+    """category_id=36381 (quarter stats) — не считается ни в одну таблицу."""
+    stats_feature = NSPDBulkFeature.model_validate(
+        {
+            "type": "Feature",
+            "geometry": None,
+            "properties": {"categoryId": 36381, "options": {"cnt_land": 5}},
+        }
+    )
+    db = MagicMock()
+    counts = upsert_features(db, [stats_feature], source="search")
+    # Все счётчики должны быть 0, skipped тоже (36381 — особый случай)
+    assert counts["parcels"] == 0
+    assert counts["buildings"] == 0
+    assert counts["skipped"] == 0
+
+
+# ── upsert_parcel SQL params ─────────────────────────────────────────────────
+
+
+def test_upsert_parcel_uses_cast_not_double_colon() -> None:
+    """Проверить что SQL использует CAST(:x AS type) а не :x::type (SQLAlchemy ловушка).
+
+    Используем inspect самого SQL через db.execute calls.
+    """
+    from app.services.cadastre.bulk_harvest import upsert_parcel
+
+    db = MagicMock()
+    executed_sqls: list[str] = []
+
+    def capture_execute(stmt: Any, params: Any = None) -> MagicMock:
+        executed_sqls.append(str(stmt))
+        return MagicMock()
+
+    db.execute = capture_execute
+
+    feature = _make_parcel_feature()
+    upsert_parcel(db, feature, source="search")
+
+    assert len(executed_sqls) == 1
+    sql = executed_sqls[0]
+    # Проверяем что нет `:x::type` паттерна (только CAST(:x AS ...))
+    assert "::jsonb" not in sql, "SQL содержит ::jsonb — нужно CAST(:x AS jsonb)"
+    assert "::date" not in sql, "SQL содержит ::date — нужно CAST(:x AS date)"
+    assert "CAST(:raw_props AS jsonb)" in sql
+
+
+# ── harvest_quarter integration (async, mocked) ───────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_harvest_quarter_phase_done_early_return() -> None:
+    """Если phase_state='done' — harvest_quarter возвращается ранним выходом."""
+    from app.services.cadastre.bulk_harvest import harvest_quarter
+
+    db = MagicMock()
+    db.execute = MagicMock(
+        return_value=MagicMock(
+            mappings=MagicMock(
+                return_value=MagicMock(
+                    first=MagicMock(return_value={"phase_state": {"phase": "done"}})
+                )
+            )
+        )
+    )
+
+    client = AsyncMock()
+    progress_calls: list[Any] = []
+
+    await harvest_quarter(
+        db=db,
+        client=client,
+        quarter="66:41:0303161",
+        job_id=1,
+        update_progress=lambda s: progress_calls.append(s),
+    )
+
+    # client.search_by_quarter не должен вызываться
+    client.search_by_quarter.assert_not_called()
+    # Нет progress calls
+    assert len(progress_calls) == 0
+
+
+@pytest.mark.asyncio
+async def test_harvest_quarter_calls_upsert_features() -> None:
+    """harvest_quarter Phase 1: search_by_quarter → upsert_features."""
+    from app.services.cadastre.bulk_harvest import harvest_quarter
+
+    parcel = _make_parcel_feature()
+    snapshot = QuarterSnapshot(
+        quarter="66:41:0303161",
+        fetched_at="2026-05-15T10:00:00+00:00",
+        features=[parcel],
+        meta_counts={36368: 5, 36369: 3},  # нет overflow
+    )
+
+    db = MagicMock()
+    # phase_state = None → начинаем с нуля
+    db.execute = MagicMock(
+        return_value=MagicMock(
+            mappings=MagicMock(
+                return_value=MagicMock(first=MagicMock(return_value={"phase_state": None}))
+            )
+        )
+    )
+    db.commit = MagicMock()
+
+    client = AsyncMock()
+    client.search_by_quarter = AsyncMock(return_value=snapshot)
+
+    progress_states: list[dict[str, Any]] = []
+
+    with patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert:
+        mock_upsert.return_value = {
+            "parcels": 1,
+            "buildings": 0,
+            "constructions": 0,
+            "oncs": 0,
+            "enks": 0,
+            "zouit": 0,
+            "skipped": 0,
+        }
+        result = await harvest_quarter(
+            db=db,
+            client=client,
+            quarter="66:41:0303161",
+            job_id=1,
+            update_progress=lambda s: progress_states.append(s),
+        )
+
+    client.search_by_quarter.assert_awaited_once_with("66:41:0303161")
+    mock_upsert.assert_called()
+    assert result.parcels_upserted == 1
+
+    # Проверить что последний progress = done
+    assert progress_states[-1]["phase"] == "done"
+
+
+# ── Dedup тест для grid-walk ─────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_grid_walk_dedup_same_cad_num() -> None:
+    """Если wms_feature_info возвращает одинаковый cad_num дважды — 1 upsert."""
+    from app.services.cadastre.bulk_harvest import _grid_walk_category
+
+    parcel = _make_parcel_feature("66:41:0303161:1")
+    # Одинаковый feature для каждой grid-точки
+
+    db = MagicMock()
+    db.execute = MagicMock(
+        return_value=MagicMock(
+            mappings=MagicMock(
+                return_value=MagicMock(
+                    first=MagicMock(
+                        return_value={
+                            "xmin": 6090000.0,
+                            "ymin": 7590000.0,
+                            "xmax": 6090750.0,
+                            "ymax": 7590750.0,
+                        }
+                    )
+                )
+            )
+        )
+    )
+    db.begin_nested = MagicMock(
+        return_value=MagicMock(__enter__=MagicMock(return_value=None), __exit__=MagicMock())
+    )
+    db.commit = MagicMock()
+
+    client = AsyncMock()
+    # Каждый wms_feature_info вызов возвращает один и тот же парцел
+    client.wms_feature_info = AsyncMock(return_value=[parcel])
+
+    with patch("app.services.cadastre.bulk_harvest.quarter_bbox_3857") as mock_bbox:
+        mock_bbox.return_value = (6090000.0, 7590000.0, 6090750.0, 7590750.0)
+        with patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert:
+            mock_upsert.return_value = {
+                "parcels": 1,
+                "buildings": 0,
+                "constructions": 0,
+                "oncs": 0,
+                "enks": 0,
+                "zouit": 0,
+                "skipped": 0,
+            }
+
+            # Используем grid_size=2 для скорости теста (4 ячейки)
+            upserted, requests = await _grid_walk_category(
+                db=db,
+                client=client,
+                quarter="66:41:0303161",
+                layer_id=36368,
+                grid_size=2,
+            )
+
+    # Несмотря на 4 запроса, upsert_features вызывается 1 раз (дедуп по cad_num)
+    assert mock_upsert.call_count == 1
+    assert upserted == 1
+    assert requests == 4  # 2×2 grid
+
+
+# ── ЗОУИТ dedup по (reg_numb_border, category_id) ───────────────────────────
+
+
+def test_zouit_dedup_different_category_ids() -> None:
+    """Один reg_numb_border с разными category_id → 2 записи в upsert."""
+    zouit_1 = NSPDBulkFeature.model_validate(
+        {
+            "type": "Feature",
+            "geometry": None,
+            "properties": {
+                "categoryId": 36940,
+                "categoryName": "ЗОУИТ 1",
+                "options": {"reg_numb_border": "66-0.1-1.1-1234", "type_zone": "тип А"},
+            },
+        }
+    )
+    zouit_2 = NSPDBulkFeature.model_validate(
+        {
+            "type": "Feature",
+            "geometry": None,
+            "properties": {
+                "categoryId": 469039,
+                "categoryName": "ЗОУИТ 2",
+                "options": {"reg_numb_border": "66-0.1-1.1-1234", "type_zone": "тип Б"},
+            },
+        }
+    )
+
+    db = MagicMock()
+    calls: list[Any] = []
+    db.execute = MagicMock(
+        side_effect=lambda stmt, params=None: calls.append(params) or MagicMock()
+    )
+
+    with patch("app.services.cadastre.bulk_harvest.upsert_zouit") as mock_zouit:
+        counts = upsert_features(db, [zouit_1, zouit_2], source="search")
+
+    # Обе записи маршрутируются в upsert_zouit
+    assert mock_zouit.call_count == 2
+    assert counts["zouit"] == 2