From 13025c81ee93374cac25fb86b0867eea4f101754 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Fri, 15 May 2026 13:03:23 +0300 Subject: [PATCH 1/4] feat(cadastre): bulk_harvest_quarter Celery task + grid-walker + saga state (#168 PR3/5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- backend/app/api/v1/admin_cadastre.py | 355 +++++++ backend/app/main.py | 6 + backend/app/services/cadastre/__init__.py | 0 backend/app/services/cadastre/bulk_harvest.py | 990 ++++++++++++++++++ .../app/services/cadastre/grid_geometry.py | 100 ++ backend/app/workers/celery_app.py | 1 + backend/app/workers/tasks/scrape_cadastre.py | 415 ++++++++ backend/tests/api/__init__.py | 0 backend/tests/api/v1/__init__.py | 0 backend/tests/api/v1/test_admin_cadastre.py | 304 ++++++ backend/tests/services/__init__.py | 0 backend/tests/services/test_cadastre_bulk.py | 506 +++++++++ 12 files changed, 2677 insertions(+) create mode 100644 backend/app/api/v1/admin_cadastre.py create mode 100644 backend/app/services/cadastre/__init__.py create mode 100644 backend/app/services/cadastre/bulk_harvest.py create mode 100644 backend/app/services/cadastre/grid_geometry.py create mode 100644 backend/app/workers/tasks/scrape_cadastre.py create mode 100644 backend/tests/api/__init__.py create mode 100644 backend/tests/api/v1/__init__.py create mode 100644 backend/tests/api/v1/test_admin_cadastre.py create mode 100644 backend/tests/services/__init__.py create mode 100644 backend/tests/services/test_cadastre_bulk.py diff --git a/backend/app/api/v1/admin_cadastre.py b/backend/app/api/v1/admin_cadastre.py new file mode 100644 index 00000000..ff85f0d0 --- /dev/null +++ b/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} diff --git a/backend/app/main.py b/backend/app/main.py index 55ddcd2e..93db3a29 100644 --- a/backend/app/main.py +++ b/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") diff --git a/backend/app/services/cadastre/__init__.py b/backend/app/services/cadastre/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/app/services/cadastre/bulk_harvest.py b/backend/app/services/cadastre/bulk_harvest.py new file mode 100644 index 00000000..55c60fec --- /dev/null +++ b/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 diff --git a/backend/app/services/cadastre/grid_geometry.py b/backend/app/services/cadastre/grid_geometry.py new file mode 100644 index 00000000..75a6960c --- /dev/null +++ b/backend/app/services/cadastre/grid_geometry.py @@ -0,0 +1,100 @@ +"""Геометрические утилиты для 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_width: int = 512, + tile_height: 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_width: ширина виртуального WMS тайла в пикселях. + tile_height: высота виртуального 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_width // 2 + click_y = tile_height // 2 + result.append( + ( + (cell_xmin, cell_ymin, cell_xmax, cell_ymax), + (click_x, click_y), + ) + ) + + return result diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index ecaa911a..9bf49e43 100644 --- a/backend/app/workers/celery_app.py +++ b/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" diff --git a/backend/app/workers/tasks/scrape_cadastre.py b/backend/app/workers/tasks/scrape_cadastre.py new file mode 100644 index 00000000..19cb7699 --- /dev/null +++ b/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} diff --git a/backend/tests/api/__init__.py b/backend/tests/api/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/tests/api/v1/__init__.py b/backend/tests/api/v1/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/tests/api/v1/test_admin_cadastre.py b/backend/tests/api/v1/test_admin_cadastre.py new file mode 100644 index 00000000..8ca81225 --- /dev/null +++ b/backend/tests/api/v1/test_admin_cadastre.py @@ -0,0 +1,304 @@ +"""Тесты для admin_cadastre.py API endpoints. + +Использует FastAPI TestClient с mock DB и mock Celery tasks. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from fastapi.testclient import TestClient + +from app.main import app + +# scrape_cadastre зависит от app.scrapers.nspd_bulk_client (PR 2/5). +# Если ветка ещё не rebased на main — тесты которые импортируют task помечаем skip. +_SCRAPE_CADASTRE_AVAILABLE = True +try: + import app.workers.tasks.scrape_cadastre +except ModuleNotFoundError: + _SCRAPE_CADASTRE_AVAILABLE = False + +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) diff --git a/backend/tests/services/__init__.py b/backend/tests/services/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/tests/services/test_cadastre_bulk.py b/backend/tests/services/test_cadastre_bulk.py new file mode 100644 index 00000000..6f6b296c --- /dev/null +++ b/backend/tests/services/test_cadastre_bulk.py @@ -0,0 +1,506 @@ +"""Тесты для 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 + +# nspd_bulk schemas живут в app.schemas.nspd_bulk (добавлены в PR 2/5). +# importorskip пропустит весь модуль (skip_module=True по умолчанию в pytest 7+) +# если PR 2 ещё не смержен в эту ветку. +pytest.importorskip( + "app.schemas.nspd_bulk", + reason=( + "app.schemas.nspd_bulk отсутствует — нужно смержить main (PR #170). " + "Команда: git rebase origin/main" + ), +) + +from app.schemas.nspd_bulk import ( # type: ignore[import] # noqa: E402 + NSPDBulkFeature, + QuarterSnapshot, +) +from app.services.cadastre.bulk_harvest import ( # type: ignore[import] # noqa: E402 + _derive_quarter_cad, + _safe_int, + _safe_numeric, + upsert_features, +) +from app.services.cadastre.grid_geometry import ( # type: ignore[import] # noqa: E402 + 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 -- 2.45.3 From 60f075cf9ff9337330a5e27630275410dab15fea Mon Sep 17 00:00:00 2001 From: lekss361 Date: Fri, 15 May 2026 13:10:11 +0300 Subject: [PATCH 2/4] =?UTF-8?q?fixup(cadastre):=20drop=20importorskip=20(P?= =?UTF-8?q?R2=20merged)=20=E2=80=94=20imports=20now=20top-level=20(#168=20?= =?UTF-8?q?PR3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/tests/services/test_cadastre_bulk.py | 22 +++----------------- 1 file changed, 3 insertions(+), 19 deletions(-) diff --git a/backend/tests/services/test_cadastre_bulk.py b/backend/tests/services/test_cadastre_bulk.py index 6f6b296c..fc0a7300 100644 --- a/backend/tests/services/test_cadastre_bulk.py +++ b/backend/tests/services/test_cadastre_bulk.py @@ -11,30 +11,14 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -# nspd_bulk schemas живут в app.schemas.nspd_bulk (добавлены в PR 2/5). -# importorskip пропустит весь модуль (skip_module=True по умолчанию в pytest 7+) -# если PR 2 ещё не смержен в эту ветку. -pytest.importorskip( - "app.schemas.nspd_bulk", - reason=( - "app.schemas.nspd_bulk отсутствует — нужно смержить main (PR #170). " - "Команда: git rebase origin/main" - ), -) - -from app.schemas.nspd_bulk import ( # type: ignore[import] # noqa: E402 - NSPDBulkFeature, - QuarterSnapshot, -) -from app.services.cadastre.bulk_harvest import ( # type: ignore[import] # noqa: E402 +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 ( # type: ignore[import] # noqa: E402 - generate_grid_click_points, -) +from app.services.cadastre.grid_geometry import generate_grid_click_points # ── Фикстуры ───────────────────────────────────────────────────────────────── -- 2.45.3 From 32eeda5db13a8a29554a56bec78e3452d740e49e Mon Sep 17 00:00:00 2001 From: lekss361 Date: Fri, 15 May 2026 13:17:48 +0300 Subject: [PATCH 3/4] fixup(cadastre): tile_size unified param + register slow marker (#168 PR3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- backend/app/services/cadastre/grid_geometry.py | 10 ++++------ backend/pyproject.toml | 3 +++ 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/backend/app/services/cadastre/grid_geometry.py b/backend/app/services/cadastre/grid_geometry.py index 75a6960c..1c1d8728 100644 --- a/backend/app/services/cadastre/grid_geometry.py +++ b/backend/app/services/cadastre/grid_geometry.py @@ -57,8 +57,7 @@ def quarter_bbox_3857(db: Session, quarter: str) -> tuple[float, float, float, f def generate_grid_click_points( bbox: tuple[float, float, float, float], grid_size: int = 15, - tile_width: int = 512, - tile_height: int = 512, + tile_size: int = 512, ) -> list[tuple[tuple[float, float, float, float], tuple[int, int]]]: """Генерировать grid_size × grid_size ячеек для bbox квартала. @@ -69,8 +68,7 @@ def generate_grid_click_points( Args: bbox: (xmin, ymin, xmax, ymax) в EPSG:3857. grid_size: количество ячеек по каждой оси (15 → 225 запросов). - tile_width: ширина виртуального WMS тайла в пикселях. - tile_height: высота виртуального WMS тайла в пикселях. + tile_size: размер виртуального WMS тайла в пикселях (квадратный). Returns: Список (sub_bbox, click_xy) — sub_bbox в EPSG:3857, click_xy в пикселях. @@ -88,8 +86,8 @@ def generate_grid_click_points( cell_xmax = cell_xmin + x_step cell_ymax = cell_ymin + y_step # Клик в центр ячейки - click_x = tile_width // 2 - click_y = tile_height // 2 + click_x = tile_size // 2 + click_y = tile_size // 2 result.append( ( (cell_xmin, cell_ymin, cell_xmax, cell_ymax), diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 8df3fa61..89525714 100644 --- a/backend/pyproject.toml +++ b/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')", +] -- 2.45.3 From 7e95c9776f778335b47645debd1393c44ab0edee Mon Sep 17 00:00:00 2001 From: lekss361 Date: Fri, 15 May 2026 13:24:24 +0300 Subject: [PATCH 4/4] fixup(cadastre): use importlib.util.find_spec instead of import (#168 PR3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- backend/tests/api/v1/test_admin_cadastre.py | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/backend/tests/api/v1/test_admin_cadastre.py b/backend/tests/api/v1/test_admin_cadastre.py index 8ca81225..72116c70 100644 --- a/backend/tests/api/v1/test_admin_cadastre.py +++ b/backend/tests/api/v1/test_admin_cadastre.py @@ -5,6 +5,12 @@ 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 @@ -13,13 +19,9 @@ from fastapi.testclient import TestClient from app.main import app -# scrape_cadastre зависит от app.scrapers.nspd_bulk_client (PR 2/5). -# Если ветка ещё не rebased на main — тесты которые импортируют task помечаем skip. -_SCRAPE_CADASTRE_AVAILABLE = True -try: - import app.workers.tasks.scrape_cadastre -except ModuleNotFoundError: - _SCRAPE_CADASTRE_AVAILABLE = False +_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, -- 2.45.3