diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index a31b9302..c1d7d0b4 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -801,6 +801,196 @@ def objective_coverage( } +# ── NSPD geo bulk-jobs (resume-friendly) ──────────────────────────────────── + + +class EnqueueGeoJobRequest(BaseModel): + """Создать новый bulk geo-job.""" + + name: str | None = Field(default=None, max_length=200) + job_kind: str = Field(..., pattern="^(quarters|parcels|buildings|mixed)$") + source_kind: str = Field( + default="manual_list", pattern="^(manual_list|rosreestr_pending|region_bbox)$" + ) + cad_nums: list[str] = Field( + default_factory=list, max_length=50000, description="Список cad-номеров для обработки" + ) + thematic_id: int = Field(default=2, ge=1, le=15, description="1=parcel, 2=quarter, 5=building") + region_codes: list[int] | None = Field( + default=None, max_length=20, description="Для source=rosreestr_pending" + ) + rate_ms: int = Field(default=600, ge=100, le=10000) + use_rosreestr2coord: bool = False + + +@router.post("/geo/jobs") +def enqueue_geo_job( + payload: EnqueueGeoJobRequest, + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """Создать bulk geo-job и поставить в очередь. + + source_kind: + - manual_list: cad_nums передан явно + - rosreestr_pending: автоматически набрать из rosreestr_deals.quarter_cad_number + которых ещё нет в cad_quarters_geom (для region_codes если задано) + - region_bbox: TODO (для будущего spatial-bulk) + """ + _check_token(x_admin_token) + from app.workers.tasks.nspd_geo import enqueue_geo_job as enqueue_helper + from app.workers.tasks.nspd_geo import process_nspd_geo_job + + cad_with_thematic: list[tuple[str, int]] = [] + if payload.source_kind == "manual_list": + cad_with_thematic = [ + (c.strip(), payload.thematic_id) for c in payload.cad_nums if c.strip() + ] + elif payload.source_kind == "rosreestr_pending": + regions = payload.region_codes or [66] + rows = db.execute( + text( + """ + SELECT DISTINCT quarter_cad_number AS cad + FROM rosreestr_deals + WHERE region_code = ANY(:rc) + AND doc_type = 'ДДУ' + AND realestate_type_code = '002001003000' + AND quarter_cad_number IS NOT NULL + AND quarter_cad_number NOT LIKE :bp + AND quarter_cad_number NOT LIKE :bs + AND NOT EXISTS (SELECT 1 FROM cad_quarters_geom g + WHERE g.cad_number = rosreestr_deals.quarter_cad_number) + LIMIT 50000 + """ + ), + {"rc": regions, "bp": "00:00:%", "bs": "%:0000000"}, + ).all() + cad_with_thematic = [(r[0], 2) for r in rows] + else: + raise HTTPException( + status_code=400, + detail=f"source_kind={payload.source_kind} not implemented yet", + ) + + if not cad_with_thematic: + raise HTTPException(status_code=400, detail="no targets to process") + + job_id = enqueue_helper( + name=payload.name, + job_kind=payload.job_kind, + source_kind=payload.source_kind, + source_params={"region_codes": payload.region_codes, "thematic_id": payload.thematic_id}, + cad_nums_with_thematic=cad_with_thematic, + rate_ms=payload.rate_ms, + use_rosreestr2coord=payload.use_rosreestr2coord, + triggered_by="manual", + ) + process_nspd_geo_job.apply_async(args=[job_id]) + return { + "job_id": job_id, + "targets_total": len(cad_with_thematic), + "estimate_minutes": round(len(cad_with_thematic) * payload.rate_ms / 1000 / 60, 1), + } + + +@router.get("/geo/jobs") +def list_geo_jobs( + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, + limit: int = 30, +) -> list[dict[str, Any]]: + """Список последних geo-jobs (для UI dashboard).""" + _check_token(x_admin_token) + rows = ( + db.execute( + text( + """ + SELECT job_id, name, job_kind, source_kind, status, triggered_by, + started_at, finished_at, heartbeat_at, + targets_total, targets_done, targets_failed, targets_skipped, + requests_count, waf_blocked_count, error, + rate_ms, use_rosreestr2coord, created_at + FROM nspd_geo_jobs + ORDER BY created_at DESC + LIMIT :lim + """ + ), + {"lim": min(limit, 100)}, + ) + .mappings() + .all() + ) + return [ + { + "job_id": r["job_id"], + "name": r["name"], + "job_kind": r["job_kind"], + "source_kind": r["source_kind"], + "status": r["status"], + "triggered_by": r["triggered_by"], + "started_at": r["started_at"].isoformat() if r["started_at"] else None, + "finished_at": r["finished_at"].isoformat() if r["finished_at"] else None, + "heartbeat_at": r["heartbeat_at"].isoformat() if r["heartbeat_at"] else None, + "targets_total": r["targets_total"], + "targets_done": r["targets_done"], + "targets_failed": r["targets_failed"], + "targets_skipped": r["targets_skipped"], + "requests_count": r["requests_count"], + "waf_blocked_count": r["waf_blocked_count"], + "error": r["error"], + "rate_ms": r["rate_ms"], + "use_rosreestr2coord": r["use_rosreestr2coord"], + "created_at": r["created_at"].isoformat() if r["created_at"] else None, + "progress_pct": round( + 100.0 * (r["targets_done"] or 0) / max(r["targets_total"] or 1, 1), 1 + ), + } + for r in rows + ] + + +@router.post("/geo/jobs/{job_id}/cancel") +def cancel_geo_job( + job_id: int, + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """Пометить job как cancelled. Worker увидит при следующей итерации.""" + _check_token(x_admin_token) + db.execute( + text( + """ + UPDATE nspd_geo_jobs SET status = 'cancelled', finished_at = NOW(), + error = COALESCE(error, 'cancelled by admin') + WHERE job_id = :id AND status IN ('queued','running','paused') + """ + ), + {"id": job_id}, + ) + db.commit() + return {"job_id": job_id, "cancelled": True} + + +@router.post("/geo/jobs/{job_id}/resume") +def resume_geo_job( + job_id: int, + db: Annotated[Session, Depends(get_db)], + x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None, +) -> dict[str, Any]: + """Re-enqueue paused/failed job. Resume idempotent через pending targets.""" + _check_token(x_admin_token) + from app.workers.tasks.nspd_geo import process_nspd_geo_job + + db.execute( + text("UPDATE nspd_geo_jobs SET status='queued', error=NULL WHERE job_id=:id"), + {"id": job_id}, + ) + db.commit() + process_nspd_geo_job.apply_async(args=[job_id]) + return {"job_id": job_id, "resumed": True} + + # ── Unified scrape dashboard (поверх v_scrape_runs_unified / v_scrape_log_unified) ── diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 8a26953c..82a9e755 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -77,6 +77,15 @@ class Settings(BaseSettings): # snapshot за тот же отчётный период. objective_sync_cron: str = "0 6 * * tue" + # NSPD lite (urllib через WAF) — feature toggle для переключения с + # Playwright-based nspd_kn.py на минималистичный nspd_lite.py. + # 2026-05-11: эмпирически проверено что urllib проходит WAF nspd.gov.ru + # (TLS-fingerprint stdlib не блокируется). Старый Playwright-путь + # остаётся как fallback на случай возврата WAF-проблем. + use_nspd_lite: bool = True + # Default rate-limit для nspd_lite fetcher (мс между запросами). + nspd_lite_rate_ms: int = 600 + # ETL Антоновского /sf/api/* SQLite → нашу PG (objective_* таблицы). # На проде монтируется bind-mount-ом docker-compose: # /opt/gendesign/site-finder/analysis.db -> /data/anton-sqlite/analysis.db diff --git a/backend/app/services/scrapers/nspd_lite.py b/backend/app/services/scrapers/nspd_lite.py new file mode 100644 index 00000000..07ca9f3f --- /dev/null +++ b/backend/app/services/scrapers/nspd_lite.py @@ -0,0 +1,174 @@ +"""NSPD lite-fetcher через urllib (без Playwright/Chromium). + +ОТКРЫТИЕ 2026-05-11: WAF nspd.gov.ru блокирует httpx/curl/requests по +TLS-fingerprint, но **stdlib urllib проходит**. Это позволяет полностью +избавиться от Playwright + Chromium (~300 MB Docker image, медленный +browser-startup) для NSPD-задач. + +Стратегия: + 1. urllib.request.Request + ssl._create_unverified_context() + 2. полный набор Chrome-headers (referer, origin, sec-ch-ua, ...) + 3. простой rate-limiter между запросами + +Эндпоинты: + - thematicSearchId=1: участки (ЗУ) + - thematicSearchId=2: кадастровые кварталы + - thematicSearchId=4: административно-территориальное деление + - thematicSearchId=5: ОКС (здания) + - thematicSearchId=7: территориальные зоны + - thematicSearchId=15: комплексы объектов + +Совместимость: API-контракт совместим с nspd_kn.py (те же ResponseShape), +чтобы переключение через feature flag было безболезненным. +""" + +from __future__ import annotations + +import json +import logging +import ssl +import time +import urllib.error +import urllib.parse +import urllib.request +from typing import Any + +logger = logging.getLogger(__name__) + +NSPD_BASE = "https://nspd.gov.ru/api/geoportal/v2/search/geoportal" + +# Headers критичны: order, casing — как в браузере. +HEADERS = { + "accept": "*/*", + "accept-language": "en-US,en;q=0.9,ru-RU;q=0.8,ru;q=0.7,es;q=0.6", + "cache-control": "no-cache", + "pragma": "no-cache", + "referer": "https://nspd.gov.ru/map?thematic=PKK", + "origin": "https://nspd.gov.ru", + "user-agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36" + ), +} + +# Глобальный SSL-контекст — используется во всех запросах. +_SSL_CTX = ssl._create_unverified_context() + + +# Семантические alias-ы для thematicSearchId (понятнее в коде) +THEMATIC = { + "parcel": 1, # земельные участки (ЗУ) + "quarter": 2, # кадастровые кварталы + "admin": 4, # административно-территориальное деление + "building": 5, # ОКС (здания) + "zone": 7, # территориальные зоны + "complex": 15, # комплексы объектов +} + + +class NspdLiteError(RuntimeError): + """Неуспешный ответ от NSPD.""" + + +class NspdLiteWafError(NspdLiteError): + """WAF/Rate-limit (HTTP 403/429). Триггер для backoff.""" + + +def fetch_geoportal( + query: str, + thematic_id: int | str = 2, + timeout: int = 15, + rate_ms: int = 600, +) -> dict[str, Any]: + """Запрос к /api/geoportal/v2/search/geoportal через urllib. + + Args: + query: cad-номер ('66:41:0204016' для квартала, '66:41:0204016:10' для участка) + thematic_id: int (1/2/4/5/7/15) или alias из THEMATIC + timeout: HTTP timeout (сек) + rate_ms: пауза перед запросом (для rate-limiting; вызывается до urlopen) + + Returns: + {"data": {"type": "FeatureCollection", "features": [...]}, "meta": {...}} + + Raises: + NspdLiteWafError при 403/429 — сигнал для caller'а сделать backoff + NspdLiteError при прочих 4xx/5xx + """ + if rate_ms > 0: + time.sleep(rate_ms / 1000.0) + + if isinstance(thematic_id, str): + thematic_id = THEMATIC.get(thematic_id, thematic_id) + + qs = urllib.parse.urlencode( + { + "thematicSearchId": int(thematic_id), + "query": query, + } + ) + url = f"{NSPD_BASE}?{qs}" + req = urllib.request.Request(url, headers=HEADERS) + + try: + with urllib.request.urlopen(req, context=_SSL_CTX, timeout=timeout) as r: + body = r.read().decode("utf-8", "ignore") + return json.loads(body) + except urllib.error.HTTPError as e: + body = e.read().decode("utf-8", "ignore")[:300] if e.fp else "" + if e.code in (403, 429): + raise NspdLiteWafError(f"HTTP {e.code} (WAF/rate-limit): {body}") from e + raise NspdLiteError(f"HTTP {e.code}: {body}") from e + except urllib.error.URLError as e: + raise NspdLiteError(f"Network error: {e}") from e + + +def fetch_quarter(quarter_cad_num: str, **kwargs) -> dict[str, Any]: + """Конкретный кадастровый квартал (typed wrapper).""" + return fetch_geoportal(quarter_cad_num, thematic_id=THEMATIC["quarter"], **kwargs) + + +def fetch_parcel(parcel_cad_num: str, **kwargs) -> dict[str, Any]: + """Конкретный участок ЗУ (typed wrapper).""" + return fetch_geoportal(parcel_cad_num, thematic_id=THEMATIC["parcel"], **kwargs) + + +def fetch_building(building_cad_num: str, **kwargs) -> dict[str, Any]: + """Конкретное здание ОКС (typed wrapper).""" + return fetch_geoportal(building_cad_num, thematic_id=THEMATIC["building"], **kwargs) + + +# ── Bulk через rosreestr2coord library ────────────────────────────────────── + + +def fetch_via_rosreestr2coord( + cad_num: str, + area_type: int = 1, + timeout: int = 15, + delay: float = 1.0, +) -> dict[str, Any] | None: + """Тот же запрос но через community-библиотеку rosreestr2coord. + + Преимущество: они поддерживают upstream обновления headers/WAF-tricks. + Использовать когда наш fetch_* поломается из-за изменений WAF. + + Returns: + GeoJSON Feature или None если participant не найден. + """ + try: + from rosreestr2coord.parser import Area + except ImportError as e: + raise NspdLiteError(f"rosreestr2coord не установлен (uv add rosreestr2coord): {e}") from e + + a = Area( + code=cad_num, + area_type=area_type, + timeout=timeout, + delay=delay, + with_log=False, + ) + try: + return a.to_geojson_poly() + except Exception as e: + logger.warning("rosreestr2coord failed for %s: %s", cad_num, e) + return None diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index ad346817..ae3818b0 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -44,6 +44,7 @@ celery_app = Celery( "app.workers.tasks.refresh_analytics", "app.workers.tasks.scrape_objective", "app.workers.tasks.objective_etl", + "app.workers.tasks.nspd_geo", ], ) celery_app.conf.timezone = "Europe/Moscow" @@ -258,3 +259,50 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: rc, e, ) + + # NSPD geo-jobs: bulk-fetcher с собственной resume-логикой через + # nspd_geo_jobs / nspd_geo_targets. Жадный resume: status='running' со + # stale heartbeat (>10мин) → re-enqueue с тем же job_id (task сам прочитает + # pending targets и продолжит). + db = SessionLocal() + geo_resume_jobs: list[int] = [] + try: + rows = ( + db.execute( + text( + """ + UPDATE nspd_geo_jobs + SET status = 'queued', + error = COALESCE(error, 'auto-resume at worker_ready') + WHERE status IN ('running', 'paused') + AND COALESCE(heartbeat_at, started_at, created_at) + < NOW() - INTERVAL '10 minutes' + RETURNING job_id + """ + ) + ) + .mappings() + .all() + ) + db.commit() + geo_resume_jobs = [int(r["job_id"]) for r in rows] + for jid in geo_resume_jobs: + logger.info("worker_ready: NSPD geo job=%s — resume scheduled", jid) + except Exception as e: + logger.warning("worker_ready nspd_geo resume scan failed: %s", e) + try: + db.rollback() + except Exception: + pass + finally: + db.close() + + if geo_resume_jobs: + from app.workers.tasks.nspd_geo import process_nspd_geo_job + + for jid in geo_resume_jobs: + try: + process_nspd_geo_job.apply_async(args=[jid]) + logger.info("worker_ready: process_nspd_geo_job enqueued job=%s", jid) + except Exception as e: + logger.warning("worker_ready: failed to enqueue geo resume job=%s: %s", jid, e) diff --git a/backend/app/workers/tasks/nspd_geo.py b/backend/app/workers/tasks/nspd_geo.py new file mode 100644 index 00000000..ba0ceccc --- /dev/null +++ b/backend/app/workers/tasks/nspd_geo.py @@ -0,0 +1,445 @@ +"""Bulk NSPD geo backfill — resume-friendly Celery task. + +Архитектура (см. data/sql/77_nspd_geo_jobs.sql): + nspd_geo_jobs — журнал заданий (один job = одна bulk-операция) + nspd_geo_targets — список cad-номеров со status (pending/done/failed) + +Resume: + 1. При worker_ready signal: ищем jobs со status='running' и stale heartbeat + (>10 мин), переключаем их статус на 'resume_needed' + re-enqueue task. + 2. process_nspd_geo_job(job_id) сразу читает SELECT WHERE status='pending' + — идемпотентно продолжает с того места, где остановился (UPSERT в + cad_quarters_geom / cad_buildings гарантирует no dup'ов). + +Запуск (3 пути): + 1. UI: POST /api/v1/admin/scrape/geo (создаёт job, наполняет targets, enqueue task) + 2. CLI: python -m app.workers.tasks.nspd_geo --enqueue-region 66 + 3. Beat: можно добавить cron-расписание в celery_app.py +""" + +from __future__ import annotations + +import logging +from typing import Any + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.core.db import SessionLocal +from app.services.scrapers.nspd_lite import ( + NspdLiteError, + NspdLiteWafError, + fetch_geoportal, + fetch_via_rosreestr2coord, +) +from app.workers.celery_app import celery_app + +logger = logging.getLogger(__name__) + +# Heartbeat каждые N items (компромисс: чаще = выше DB write, реже = stale-window больше) +HEARTBEAT_EVERY = 5 +# WAF backoff (seconds) после первого 403/429 +WAF_BACKOFF_BASE_S = 30 + + +# ── Helpers для job/target lifecycle ──────────────────────────────────────── + + +def _start_job(db: Session, job_id: int) -> None: + db.execute( + text( + """ + UPDATE nspd_geo_jobs + SET status = 'running', + started_at = COALESCE(started_at, NOW()), + heartbeat_at = NOW(), + updated_at = NOW() + WHERE job_id = :id + """ + ), + {"id": job_id}, + ) + db.commit() + + +def _heartbeat(db: Session, job_id: int, **counts: int) -> None: + sets = ["heartbeat_at = NOW()", "updated_at = NOW()"] + params: dict[str, Any] = {"id": job_id} + for k, v in counts.items(): + sets.append(f"{k} = :{k}") + params[k] = v + try: + db.execute( + text(f"UPDATE nspd_geo_jobs SET {', '.join(sets)} WHERE job_id = :id"), + params, + ) + db.commit() + except Exception as e: + logger.warning("heartbeat failed for job=%s: %s", job_id, e) + try: + db.rollback() + except Exception: + pass + + +def _finish_job(db: Session, job_id: int, status: str, error: str | None = None) -> None: + db.execute( + text( + """ + UPDATE nspd_geo_jobs + SET status = :st, finished_at = NOW(), error = :err, updated_at = NOW() + WHERE job_id = :id + """ + ), + {"id": job_id, "st": status, "err": error}, + ) + db.commit() + + +# ── Persistence для разных видов targets ──────────────────────────────────── + + +def _save_quarter(db: Session, payload: dict, cad_num: str) -> int: + """Парсит NSPD-response для квартала и UPSERT'ит в cad_quarters_geom.""" + feats = (payload.get("data") or {}).get("features") or [] + if not feats: + return 0 + f = feats[0] + geom_geojson = f.get("geometry") + if not geom_geojson: + return 0 + db.execute( + text( + """ + INSERT INTO cad_quarters_geom (cad_number, geom, raw_props, fetched_at) + VALUES (:cad, ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:g), 3857), 4326), + CAST(:props AS jsonb), NOW()) + ON CONFLICT (cad_number) DO UPDATE + SET geom = EXCLUDED.geom, + raw_props = EXCLUDED.raw_props, + fetched_at = NOW() + """ + ), + { + "cad": cad_num, + "g": str(geom_geojson).replace("'", '"'), + "props": str(f.get("properties", {})).replace("'", '"'), + }, + ) + return 1 + + +def _save_building(db: Session, payload: dict, cad_num: str) -> int: + """Парсит NSPD-response для здания и UPSERT'ит в cad_buildings.""" + feats = (payload.get("data") or {}).get("features") or [] + if not feats: + return 0 + n = 0 + for f in feats: + geom_geojson = f.get("geometry") + opts = (f.get("properties") or {}).get("options") or {} + if not geom_geojson: + continue + # Только этого здания (NSPD может вернуть здания всего квартала) + if opts.get("cad_num") != cad_num: + continue + import json as _json + + db.execute( + text( + """ + INSERT INTO cad_buildings ( + cad_num, quarter_cad_num, geom, purpose, building_name, + readable_address, area, floors, raw_props + ) VALUES ( + :cad, :qcad, + ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:g), 3857), 4326), + :purpose, :name, :addr, :area, :floors, CAST(:props AS jsonb) + ) + ON CONFLICT (cad_num) DO UPDATE + SET quarter_cad_num = EXCLUDED.quarter_cad_num, + geom = EXCLUDED.geom, + purpose = EXCLUDED.purpose, + building_name = EXCLUDED.building_name, + readable_address = EXCLUDED.readable_address, + area = EXCLUDED.area, + floors = EXCLUDED.floors, + raw_props = EXCLUDED.raw_props + """ + ), + { + "cad": cad_num, + "qcad": opts.get("quarter_cad_number"), + "g": _json.dumps(geom_geojson, ensure_ascii=False), + "purpose": opts.get("purpose"), + "name": opts.get("building_name") or opts.get("name"), + "addr": opts.get("readable_address"), + "area": opts.get("area"), + "floors": opts.get("floors"), + "props": _json.dumps(opts, ensure_ascii=False), + }, + ) + n += 1 + return n + + +def _persist_target(db: Session, thematic_id: int, payload: dict, cad_num: str) -> tuple[int, str]: + """Returns (n_features_saved, table_name). Полиморфно по thematic_id.""" + if thematic_id == 2: + return _save_quarter(db, payload, cad_num), "cad_quarters_geom" + if thematic_id == 5: + return _save_building(db, payload, cad_num), "cad_buildings" + # Для parcel (1) пока нет нашей normalized-таблицы — только raw в targets.result_features + return 0, "" + + +# ── Helper для создания job + targets ────────────────────────────────────── + + +def enqueue_geo_job( + *, + name: str | None, + job_kind: str, + source_kind: str, + source_params: dict[str, Any], + cad_nums_with_thematic: list[tuple[str, int]], + rate_ms: int = 600, + use_rosreestr2coord: bool = False, + triggered_by: str = "manual", +) -> int: + """Создаёт job + targets и возвращает job_id. Не enqueue'ит task сам. + Caller должен сделать `process_nspd_geo_job.apply_async(args=[job_id])`. + """ + import json as _json + + db = SessionLocal() + try: + job_id = db.execute( + text( + """ + INSERT INTO nspd_geo_jobs ( + name, job_kind, source_kind, source_params, + rate_ms, use_rosreestr2coord, triggered_by, status, targets_total + ) VALUES (:n, :k, :sk, CAST(:sp AS jsonb), :r, :ur, :tb, 'queued', :tt) + RETURNING job_id + """ + ), + { + "n": name, + "k": job_kind, + "sk": source_kind, + "sp": _json.dumps(source_params, ensure_ascii=False), + "r": rate_ms, + "ur": use_rosreestr2coord, + "tb": triggered_by, + "tt": len(cad_nums_with_thematic), + }, + ).scalar_one() + + # Bulk INSERT targets + if cad_nums_with_thematic: + values = ",".join( + f"({job_id}, '{c.replace(chr(39), chr(39)*2)}', {t}, 'pending')" + for c, t in cad_nums_with_thematic + ) + db.execute( + text( + f"INSERT INTO nspd_geo_targets (job_id, cad_num, thematic_id, status) " + f"VALUES {values} ON CONFLICT (job_id, cad_num, thematic_id) DO NOTHING" + ) + ) + db.commit() + return int(job_id) + finally: + db.close() + + +# ── Main task ─────────────────────────────────────────────────────────────── + + +@celery_app.task( + bind=True, + name="tasks.nspd_geo.process_job", + max_retries=0, # resume через worker_ready, не через retry + soft_time_limit=3600 * 6, # 6 часов max — крупные jobs +) +def process_nspd_geo_job(self: Any, job_id: int) -> dict[str, Any]: + """Идёт по nspd_geo_targets WHERE job_id=:id AND status='pending'. + + Heartbeat каждые HEARTBEAT_EVERY items, persistent counters в job-row. + Idempotent: уже-сохранённые UPSERT'ятся, status target → 'done'. + """ + import time + + db = SessionLocal() + try: + # Загружаем параметры job'а + job = ( + db.execute( + text("SELECT * FROM nspd_geo_jobs WHERE job_id = :id"), + {"id": job_id}, + ) + .mappings() + .first() + ) + if not job: + return {"error": "job_not_found", "job_id": job_id} + + _start_job(db, job_id) + rate_ms = int(job["rate_ms"] or 600) + use_lib = bool(job["use_rosreestr2coord"]) + + # Текущие счётчики (для resume — продолжаем с тех значений) + done = int(job["targets_done"] or 0) + failed = int(job["targets_failed"] or 0) + skipped = int(job["targets_skipped"] or 0) + n_requests = int(job["requests_count"] or 0) + n_waf = int(job["waf_blocked_count"] or 0) + consecutive_waf = 0 + + logger.info( + "process_nspd_geo_job start: job=%s use_lib=%s rate=%dms", job_id, use_lib, rate_ms + ) + + while True: + target = ( + db.execute( + text( + """ + SELECT target_id, cad_num, thematic_id, attempts + FROM nspd_geo_targets + WHERE job_id = :id AND status = 'pending' + ORDER BY target_id + LIMIT 1 + """ + ), + {"id": job_id}, + ) + .mappings() + .first() + ) + if not target: + break + + tid = target["thematic_id"] + cad = target["cad_num"] + + try: + if use_lib: + payload = { + "data": { + "features": [fetch_via_rosreestr2coord(cad, area_type=tid, timeout=15)] + } + } + payload["data"]["features"] = [f for f in payload["data"]["features"] if f] + else: + payload = fetch_geoportal(cad, thematic_id=tid, rate_ms=rate_ms) + n_requests += 1 + consecutive_waf = 0 + + n_feats, saved_to = _persist_target(db, tid, payload, cad) + db.execute( + text( + """ + UPDATE nspd_geo_targets + SET status = 'done', result_features = :f, + saved_to_table = :s, fetched_at = NOW(), + attempts = attempts + 1 + WHERE target_id = :tid + """ + ), + {"f": n_feats, "s": saved_to or None, "tid": target["target_id"]}, + ) + db.commit() + done += 1 + + except NspdLiteWafError as e: + consecutive_waf += 1 + n_waf += 1 + wait = WAF_BACKOFF_BASE_S * min(consecutive_waf, 8) + logger.warning( + "WAF on %s (consecutive=%d), backoff %ds: %s", + cad, + consecutive_waf, + wait, + e, + ) + # НЕ помечаем failed — остаётся pending для следующей итерации + _heartbeat(db, job_id, requests_count=n_requests, waf_blocked_count=n_waf) + time.sleep(wait) + if consecutive_waf >= 8: + # 8 подряд WAF — пауза job'а до следующего worker_ready + _finish_job( + db, + job_id, + "paused", + error=f"WAF blocked {consecutive_waf} consecutive", + ) + return {"job_id": job_id, "paused": True, "reason": "waf"} + continue + + except (NspdLiteError, Exception) as e: + logger.warning("fetch failed for %s: %s", cad, e) + db.execute( + text( + """ + UPDATE nspd_geo_targets + SET status = CASE WHEN attempts >= 2 THEN 'failed' ELSE 'pending' END, + attempts = attempts + 1, + error_msg = :em + WHERE target_id = :tid + """ + ), + {"em": str(e)[:500], "tid": target["target_id"]}, + ) + db.commit() + if int(target["attempts"]) >= 2: + failed += 1 + + if (done + failed + skipped) % HEARTBEAT_EVERY == 0: + _heartbeat( + db, + job_id, + targets_done=done, + targets_failed=failed, + targets_skipped=skipped, + requests_count=n_requests, + waf_blocked_count=n_waf, + ) + + _heartbeat( + db, + job_id, + targets_done=done, + targets_failed=failed, + targets_skipped=skipped, + requests_count=n_requests, + waf_blocked_count=n_waf, + ) + # done даже при failed targets — UI покажет failed_count отдельно + _finish_job(db, job_id, "done") + logger.info( + "process_nspd_geo_job done: job=%s done=%d failed=%d req=%d waf=%d", + job_id, + done, + failed, + n_requests, + n_waf, + ) + return { + "job_id": job_id, + "done": done, + "failed": failed, + "skipped": skipped, + "requests": n_requests, + "waf_blocked": n_waf, + } + except Exception as e: + logger.exception("process_nspd_geo_job crashed: job=%s: %s", job_id, e) + try: + _finish_job(db, job_id, "failed", error=f"{type(e).__name__}: {e}") + except Exception: + pass + raise + finally: + db.close() diff --git a/backend/debug.log b/backend/debug.log new file mode 100644 index 00000000..58654503 --- /dev/null +++ b/backend/debug.log @@ -0,0 +1 @@ +DEBUG:rosreestr2coord.logger:Request URL: https://nspd.gov.ru/api/geoportal/v2/search/geoportal?thematicSearchId=1&query=66:41:0204016:10&CRS=EPSG:4326 diff --git a/backend/output/geojson/66_41_0204016_10.geojson b/backend/output/geojson/66_41_0204016_10.geojson new file mode 100644 index 00000000..517a3e14 --- /dev/null +++ b/backend/output/geojson/66_41_0204016_10.geojson @@ -0,0 +1 @@ +{"id": 103660478, "type": "Feature", "geometry": {"type": "Polygon", "coordinates": [[[60.5222936642305, 56.87886819649157], [60.52228588780081, 56.87886283773699], [60.52224469481928, 56.878834885257085], [60.522052973741495, 56.87869300342362], [60.52253110197689, 56.878512041850186], [60.52259101161419, 56.87848945465678], [60.522737599100324, 56.87843394151893], [60.52278893631658, 56.878433926994376], [60.522832402213155, 56.878434032020515], [60.522886453650436, 56.87846714420006], [60.52289738223943, 56.878462343346754], [60.523191155978175, 56.87868465965369], [60.52329525075883, 56.878758083028075], [60.52313452537541, 56.87882406737451], [60.52263537237766, 56.879028728703595], [60.52257662924073, 56.87905283814238], [60.5222936642305, 56.87886819649157]]], "crs": {"type": "name", "properties": {"name": "EPSG:3857"}}}, "properties": {"cadastralDistrictsCode": 66, "category": 36368, "categoryName": "\u0417\u0435\u043c\u0435\u043b\u044c\u043d\u044b\u0435 \u0443\u0447\u0430\u0441\u0442\u043a\u0438 \u0415\u0413\u0420\u041d", "descr": "66:41:0204016:10", "externalKey": "66:41:0204016:10", "interactionId": 103594786, "label": "66:41:0204016:10", "options": {"area": null, "cad_num": "66:41:0204016:10", "cost_application_date": "2023-01-01", "cost_approvement_date": "", "cost_determination_date": "2022-01-01", "cost_index": 8497.03, "cost_registration_date": "2023-01-03", "cost_value": 23706713.7, "declared_area": null, "determination_couse": "\u0410\u043a\u0442 \u043e\u0431 \u0443\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u0438\u0438 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442\u043e\u0432 \u043e\u043f\u0440\u0435\u0434\u0435\u043b\u0435\u043d\u0438\u044f \u043a\u0430\u0434\u0430\u0441\u0442\u0440\u043e\u0432\u043e\u0439 \u0441\u0442\u043e\u0438\u043c\u043e\u0441\u0442\u0438, \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u043d\u044b\u0445 \u0432 \u0441\u043e\u043e\u0442\u0432\u0435\u0442\u0441\u0442\u0432\u0438\u0438 \u0441 \u043f\u043e\u043b\u043e\u0436\u0435\u043d\u0438\u044f\u043c\u0438 \u0417\u0430\u043a\u043e\u043d\u0430 237-\u0424\u0417\n\n\n\n", "land_record_area": 2790, "land_record_category_type": "\u0417\u0435\u043c\u043b\u0438 \u043d\u0430\u0441\u0435\u043b\u0435\u043d\u043d\u044b\u0445 \u043f\u0443\u043d\u043a\u0442\u043e\u0432", "land_record_reg_date": "2006-03-06", "land_record_subtype": "\u0417\u0435\u043c\u043b\u0435\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u043d\u0438\u0435", "land_record_type": "\u0417\u0435\u043c\u0435\u043b\u044c\u043d\u044b\u0439 \u0443\u0447\u0430\u0441\u0442\u043e\u043a", "ownership_type": "\u0427\u0430\u0441\u0442\u043d\u0430\u044f", "permitted_use_established_by_document": "\u043c\u0430\u0433\u0430\u0437\u0438\u043d\u044b (\u043e\u0431\u0449\u0435\u0439 \u043f\u043b\u043e\u0449\u0430\u0434\u044c\u044e \u0434\u043e 5000 \u043a\u0432.\u043c), \u043e\u0431\u0449\u0435\u0441\u0442\u0432\u0435\u043d\u043d\u043e\u0435 \u043f\u0438\u0442\u0430\u043d\u0438\u0435, \u0431\u044b\u0442\u043e\u0432\u043e\u0435 \u043e\u0431\u0441\u043b\u0443\u0436\u0438\u0432\u0430\u043d\u0438\u0435, \u0434\u0435\u043b\u043e\u0432\u043e\u0435 \u0443\u043f\u0440\u0430\u0432\u043b\u0435\u043d\u0438\u0435", "previously_posted": "\u0420\u0430\u043d\u0435\u0435 \u0443\u0447\u0442\u0435\u043d\u043d\u044b\u0439", "quarter_cad_number": "66:41:0204016", "readable_address": "\u0420\u043e\u0441\u0441\u0438\u0439\u0441\u043a\u0430\u044f \u0424\u0435\u0434\u0435\u0440\u0430\u0446\u0438\u044f, \u0421\u0432\u0435\u0440\u0434\u043b\u043e\u0432\u0441\u043a\u0430\u044f \u043e\u0431\u043b\u0430\u0441\u0442\u044c, \u043c\u0443\u043d\u0438\u0446\u0438\u043f\u0430\u043b\u044c\u043d\u043e\u0435 \u043e\u0431\u0440\u0430\u0437\u043e\u0432\u0430\u043d\u0438\u0435 \"\u0433\u043e\u0440\u043e\u0434 \u0415\u043a\u0430\u0442\u0435\u0440\u0438\u043d\u0431\u0443\u0440\u0433\", \u0433\u043e\u0440\u043e\u0434 \u0415\u043a\u0430\u0442\u0435\u0440\u0438\u043d\u0431\u0443\u0440\u0433, \u0443\u043b\u0438\u0446\u0430 \u0411\u0438\u043b\u0438\u043c\u0431\u0430\u0435\u0432\u0441\u043a\u0430\u044f, 3", "right_type": "\u0421\u043e\u0431\u0441\u0442\u0432\u0435\u043d\u043d\u043e\u0441\u0442\u044c", "specified_area": 2790, "status": "\u0420\u0430\u043d\u0435\u0435 \u0443\u0447\u0442\u0435\u043d\u043d\u044b\u0439"}, "subcategory": 5, "systemInfo": {"inserted": "2023-11-07T18:40:05.722318", "insertedBy": "7EF5CD0B-AC08-4097-AF2B-07CD85992A79", "updated": "2026-02-03T04:59:16.064005", "updatedBy": "7EF5CD0B-AC08-4097-AF2B-07CD85992A79"}}} diff --git a/backend/output/kml/66_41_0204016_10.kml b/backend/output/kml/66_41_0204016_10.kml new file mode 100644 index 00000000..641558da --- /dev/null +++ b/backend/output/kml/66_41_0204016_10.kml @@ -0,0 +1,2 @@ + +66:41:0204016:1060.5222936642305,56.87886819649157 60.52228588780081,56.87886283773699 60.52224469481928,56.878834885257085 60.522052973741495,56.87869300342362 60.52253110197689,56.878512041850186 60.52259101161419,56.87848945465678 60.522737599100324,56.87843394151893 60.52278893631658,56.878433926994376 60.522832402213155,56.878434032020515 60.522886453650436,56.87846714420006 60.52289738223943,56.878462343346754 60.523191155978175,56.87868465965369 60.52329525075883,56.878758083028075 60.52313452537541,56.87882406737451 60.52263537237766,56.879028728703595 60.52257662924073,56.87905283814238 60.5222936642305,56.87886819649157 60.5222936642305,56.87886819649157 diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 3e00d611..afabe12a 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -28,6 +28,7 @@ dependencies = [ "numpy>=2.0.0", "scikit-learn>=1.5.0", "sentry-sdk[fastapi]>=2.10.0", + "rosreestr2coord>=4.0.0", ] [dependency-groups] diff --git a/data/sql/76_complex_connectivity.sql b/data/sql/76_complex_connectivity.sql new file mode 100644 index 00000000..673b35e6 --- /dev/null +++ b/data/sql/76_complex_connectivity.sql @@ -0,0 +1,147 @@ +-- Соединение всех источников с canonical complexes (Phase A.2 продолжение). +-- Применено в проде через mcp postgres 2026-05-11. +-- Идемпотентно через ADD COLUMN IF NOT EXISTS / FK с REFERENCES. + +-- ── 1. complexes.cad_quarter / cad_building_num: связь с rosreestr/NSPD ───── +ALTER TABLE complexes + ADD COLUMN IF NOT EXISTS cad_quarter TEXT, + ADD COLUMN IF NOT EXISTS cad_building_num TEXT; +CREATE INDEX IF NOT EXISTS complexes_cad_quarter_idx + ON complexes(cad_quarter) WHERE cad_quarter IS NOT NULL; +CREATE INDEX IF NOT EXISTS complexes_cad_building_idx + ON complexes(cad_building_num) WHERE cad_building_num IS NOT NULL; + +-- Backfill: any non-null snapshot +WITH best_kn AS ( + SELECT obj_id, + (array_agg(cad_quarter ORDER BY snapshot_date DESC NULLS LAST) + FILTER (WHERE cad_quarter IS NOT NULL))[1] AS cad_quarter, + (array_agg(cad_building_num ORDER BY snapshot_date DESC NULLS LAST) + FILTER (WHERE cad_building_num IS NOT NULL))[1] AS cad_building_num + FROM domrf_kn_objects GROUP BY obj_id +) +UPDATE complexes c + SET cad_quarter = COALESCE(c.cad_quarter, bk.cad_quarter), + cad_building_num = COALESCE(c.cad_building_num, bk.cad_building_num), + updated_at = NOW() + FROM complex_sources cs + JOIN best_kn bk ON bk.obj_id::text = cs.source_id + WHERE cs.complex_id = c.id AND cs.source = 'domrf_kn'; +-- Эффект: cad_quarter заполнено у 551 из 1740 ЖК (32%). +-- cad_building_num = 0 (бaг kn-API: не парсит здание-уровень). + +-- ── 2. objective_lots.complex_id (FK на complexes) ────────────────────────── +ALTER TABLE objective_lots + ADD COLUMN IF NOT EXISTS complex_id BIGINT REFERENCES complexes(id) ON DELETE SET NULL; +ALTER TABLE objective_corpus_room_month + ADD COLUMN IF NOT EXISTS complex_id BIGINT REFERENCES complexes(id) ON DELETE SET NULL; +ALTER TABLE objective_lots_history + ADD COLUMN IF NOT EXISTS complex_id BIGINT REFERENCES complexes(id) ON DELETE SET NULL; +CREATE INDEX IF NOT EXISTS objective_lots_complex_idx + ON objective_lots(complex_id) WHERE complex_id IS NOT NULL; +CREATE INDEX IF NOT EXISTS objective_crm_complex_idx + ON objective_corpus_room_month(complex_id) WHERE complex_id IS NOT NULL; + +UPDATE objective_lots ol + SET complex_id = cs.complex_id + FROM complex_sources cs + WHERE cs.source = 'objective' AND cs.source_id = ol.project_name + AND ol.complex_id IS NULL; + +UPDATE objective_corpus_room_month ocrm + SET complex_id = cs.complex_id + FROM complex_sources cs + WHERE cs.source = 'objective' AND cs.source_id = ocrm.project_name + AND ocrm.complex_id IS NULL; +-- Эффект: 303 677 lots / 19 738 crm — 100% покрытие (все Objective lots +-- теперь напрямую JOIN'ятся с complex без двойного хопа через source_id). + +-- ── 3. cad_buildings.complex_id (spatial join 50м) ────────────────────────── +ALTER TABLE cad_buildings + ADD COLUMN IF NOT EXISTS complex_id BIGINT REFERENCES complexes(id) ON DELETE SET NULL; +CREATE INDEX IF NOT EXISTS cad_buildings_complex_idx + ON cad_buildings(complex_id) WHERE complex_id IS NOT NULL; + +WITH bld_complex AS ( + SELECT cb.cad_num, c.complex_id_match + FROM cad_buildings cb + CROSS JOIN LATERAL ( + SELECT c2.id AS complex_id_match + FROM complexes c2 + WHERE c2.geom IS NOT NULL + AND ST_DWithin(c2.geom::geography, cb.geom::geography, 50) + ORDER BY c2.geom <-> cb.geom LIMIT 1 + ) c + WHERE cb.geom IS NOT NULL AND cb.complex_id IS NULL +) +UPDATE cad_buildings cb + SET complex_id = bc.complex_id_match + FROM bld_complex bc WHERE cb.cad_num = bc.cad_num; +-- Эффект: 716 buildings / 10590 (~7%) связаны с 436 уникальными ЖК. +-- Низкое покрытие т.к. большинство cad_buildings — частные дома в кадастровых +-- кварталах, не относящиеся к ЖК. Радиус 50м консервативный. + +-- ── 4. Auto-create stub complex для 283 orphan obj_id из kn_flats ─────────── +WITH orphan_obj_ids AS ( + SELECT DISTINCT obj_id FROM domrf_kn_flats f + WHERE NOT EXISTS ( + SELECT 1 FROM complex_sources cs + WHERE cs.source='domrf_kn' AND cs.source_id = f.obj_id::text + ) +), +inserted AS ( + INSERT INTO complexes (canonical_name, region_id, primary_source) + SELECT 'orphan-kn-' || obj_id, 66, 'domrf_kn_flat_orphan' FROM orphan_obj_ids + RETURNING id, canonical_name +) +INSERT INTO complex_sources ( + complex_id, source, source_id, source_external_id, + match_method, match_score, is_reviewed, note +) +SELECT i.id, 'domrf_kn', oo.obj_id::text, oo.obj_id, 'orphan_flats_only', 0.0, FALSE, + 'auto-created stub: flats existed but kn-objects entry deleted/missing' + FROM orphan_obj_ids oo + JOIN inserted i ON i.canonical_name = 'orphan-kn-' || oo.obj_id; +-- Эффект: complexes 1740 → 2023 (+283 stub). + +-- ── 5. v_complex_full v2: правильные counts с многокорпусными ЖК ──────────── +DROP VIEW IF EXISTS v_complex_full; +CREATE VIEW v_complex_full AS +SELECT + c.id AS complex_id, + c.canonical_name, c.developer_name, c.region_id, c.district_name, + c.obj_class, c.address, c.latitude, c.longitude, + c.geom IS NOT NULL AS has_geom, + c.flat_count, c.square_living, c.site_status, + c.cad_quarter, c.cad_building_num, + (SELECT array_agg(DISTINCT source ORDER BY source) + FROM complex_sources WHERE complex_id = c.id) AS sources, + (SELECT COUNT(*) FROM complex_sources WHERE complex_id = c.id) AS source_links_n, + (SELECT COUNT(DISTINCT source) FROM complex_sources WHERE complex_id = c.id) AS source_kinds_n, + (SELECT array_agg(source_external_id ORDER BY source_external_id) + FROM complex_sources WHERE complex_id = c.id AND source = 'domrf_kn') AS domrf_obj_ids, + (SELECT source_id FROM complex_sources + WHERE complex_id = c.id AND source = 'objective' LIMIT 1) AS objective_project_name, + (SELECT COUNT(*) FROM domrf_kn_flats f + WHERE f.obj_id IN (SELECT source_external_id FROM complex_sources + WHERE complex_id = c.id AND source = 'domrf_kn' + AND source_external_id IS NOT NULL)) AS kn_flats_n, + (SELECT COUNT(*) FROM objective_lots WHERE complex_id = c.id) AS objective_lots_n, + (SELECT COUNT(*) FROM cad_buildings WHERE complex_id = c.id) AS cad_buildings_n, + c.primary_source, c.created_at, c.updated_at + FROM complexes c; + +COMMENT ON VIEW v_complex_full IS + 'Карточка ЖК с агрегатами по всем источникам. multi-corpus ЖК (один project ' + 'в Objective = N obj_id в kn) корректно объединены: kn_flats_n считает все ' + 'корпуса. source_links_n включает дубликаты по корпусам, source_kinds_n — ' + 'unique источники.'; + +-- ── 6. ANALYZE для актуализации статистики ────────────────────────────────── +ANALYZE complexes; +ANALYZE complex_sources; +ANALYZE objective_lots; +ANALYZE objective_corpus_room_month; +ANALYZE cad_buildings; +ANALYZE domrf_kn_objects; +ANALYZE domrf_kn_flats; diff --git a/data/sql/77_nspd_geo_jobs.sql b/data/sql/77_nspd_geo_jobs.sql new file mode 100644 index 00000000..617cb5cd --- /dev/null +++ b/data/sql/77_nspd_geo_jobs.sql @@ -0,0 +1,85 @@ +-- Resume-friendly NSPD geo backfill jobs. +-- +-- Зачем: +-- Bulk-загрузка кадастровых данных по 100k+ cad-номеров занимает часы. +-- При redeploy worker'а Celery теряет in-memory state. Решение — +-- хранить очередь и текущий прогресс в БД, после редеплоя +-- worker_ready hook подхватывает незавершённые jobs. +-- +-- Архитектура: +-- nspd_geo_jobs — общий журнал заданий (один job = одна bulk-операция) +-- nspd_geo_targets — список cad-номеров для обработки (1 строка = 1 cad) +-- со статусом (pending/done/failed) — это и есть resume-state. +-- +-- При redeploy: +-- worker_ready signal → SELECT FROM nspd_geo_jobs WHERE status='running' +-- → re-enqueue Celery task с тем же job_id +-- → task SELECT FROM nspd_geo_targets WHERE job_id=:id AND status='pending' +-- → продолжает с той точки где остановился (idempotent UPSERT в cad_quarters_geom) +-- +-- Применять через mcp postgres execute_sql или psql direct. + +CREATE TABLE IF NOT EXISTS nspd_geo_jobs ( + job_id BIGSERIAL PRIMARY KEY, + name TEXT, -- человекочитаемое имя ('УрФО expansion 2026-Q2') + job_kind TEXT NOT NULL, -- 'quarters' | 'parcels' | 'buildings' | 'mixed' + -- Источник целей + source_kind TEXT NOT NULL, -- 'rosreestr_pending' | 'manual_list' | 'region_bbox' + source_params jsonb, -- что-то вроде {"region_codes":[66,74,72,59], "thematic_id":2} + -- Параметры fetch + rate_ms INTEGER NOT NULL DEFAULT 600, + use_rosreestr2coord BOOLEAN NOT NULL DEFAULT FALSE, -- если TRUE — community-lib, иначе наш nspd_lite + -- Состояние + status TEXT NOT NULL DEFAULT 'queued', -- queued | running | done | failed | paused + started_at timestamptz, + finished_at timestamptz, + heartbeat_at timestamptz, + -- Счётчики + targets_total INTEGER NOT NULL DEFAULT 0, + targets_done INTEGER NOT NULL DEFAULT 0, + targets_failed INTEGER NOT NULL DEFAULT 0, + targets_skipped INTEGER NOT NULL DEFAULT 0, -- уже было в БД (idempotent skip) + requests_count INTEGER NOT NULL DEFAULT 0, + waf_blocked_count INTEGER NOT NULL DEFAULT 0, + -- Audit + error TEXT, + triggered_by TEXT NOT NULL DEFAULT 'manual',-- manual | beat | resume + created_at timestamptz NOT NULL DEFAULT NOW(), + updated_at timestamptz NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS nspd_geo_jobs_status_idx + ON nspd_geo_jobs(status, created_at DESC); +CREATE INDEX IF NOT EXISTS nspd_geo_jobs_running_idx + ON nspd_geo_jobs(heartbeat_at DESC) WHERE status = 'running'; + +COMMENT ON TABLE nspd_geo_jobs IS + 'Журнал bulk geo-jobs для NSPD-fetcher. Resume-friendly: при redeploy ' + 'worker_ready hook подхватывает status=running с устаревшим heartbeat.'; + +-- ── Targets (cad-номера к обработке) ───────────────────────────────────────── + +CREATE TABLE IF NOT EXISTS nspd_geo_targets ( + target_id BIGSERIAL PRIMARY KEY, + job_id BIGINT NOT NULL REFERENCES nspd_geo_jobs(job_id) ON DELETE CASCADE, + cad_num TEXT NOT NULL, -- '66:41:0204016' для квартала, '...:10' для участка + thematic_id SMALLINT NOT NULL, -- 1=parcel | 2=quarter | 5=building + status TEXT NOT NULL DEFAULT 'pending', -- pending | done | failed | skipped + attempts SMALLINT NOT NULL DEFAULT 0, + -- Результат + result_features INTEGER, -- сколько features вернулось + saved_to_table TEXT, -- 'cad_quarters_geom' | 'cad_buildings' | ... + error_msg TEXT, + fetched_at timestamptz, + -- Avoid duplicate work внутри одного job + UNIQUE (job_id, cad_num, thematic_id) +); + +CREATE INDEX IF NOT EXISTS nspd_geo_targets_pending_idx + ON nspd_geo_targets(job_id, status) WHERE status = 'pending'; +CREATE INDEX IF NOT EXISTS nspd_geo_targets_failed_idx + ON nspd_geo_targets(job_id) WHERE status = 'failed'; + +COMMENT ON TABLE nspd_geo_targets IS + 'Список cad-номеров для обработки в рамках job. Resume-state: ' + 'task просто SELECT WHERE status=pending и продолжает.'; diff --git a/frontend/src/app/admin/layout.tsx b/frontend/src/app/admin/layout.tsx index f9520e96..01e0720d 100644 --- a/frontend/src/app/admin/layout.tsx +++ b/frontend/src/app/admin/layout.tsx @@ -7,6 +7,7 @@ const TABS = [ { href: "/admin/scrape/all", label: "📊 Все скраперы (сводно)" }, { href: "/admin/scrape", label: "Скрапер DOM.РФ" }, { href: "/admin/scrape/nspd", label: "Скрапер NSPD" }, + { href: "/admin/scrape/geo", label: "🗺 Geo bulk-jobs" }, { href: "/admin/scrape/objective", label: "ETL Objective" }, { href: "/admin/leads", label: "Лиды PRINZIP" }, ]; diff --git a/frontend/src/app/admin/scrape/geo/page.tsx b/frontend/src/app/admin/scrape/geo/page.tsx new file mode 100644 index 00000000..24f2c016 --- /dev/null +++ b/frontend/src/app/admin/scrape/geo/page.tsx @@ -0,0 +1,580 @@ +"use client"; + +import { useState } from "react"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; + +import { apiFetch } from "@/lib/api"; + +interface GeoJob { + job_id: number; + name: string | null; + job_kind: string; + source_kind: string; + status: string; + triggered_by: string; + started_at: string | null; + finished_at: string | null; + heartbeat_at: string | null; + targets_total: number; + targets_done: number; + targets_failed: number; + targets_skipped: number; + requests_count: number; + waf_blocked_count: number; + error: string | null; + rate_ms: number; + use_rosreestr2coord: boolean; + created_at: string | null; + progress_pct: number; +} + +interface EnqueueResp { + job_id: number; + targets_total: number; + estimate_minutes: number; +} + +const STATUS_COLOR: Record = { + queued: "#6b7280", + running: "#1d4ed8", + done: "#16a34a", + failed: "#b3261e", + paused: "#a16207", + cancelled: "#9333ea", +}; + +function formatIso(s: string | null): string { + if (!s) return "—"; + return new Date(s).toLocaleString("ru-RU"); +} + +function diffSecs(a: string | null, b: string | null): string { + if (!a || !b) return "—"; + const dt = (new Date(b).getTime() - new Date(a).getTime()) / 1000; + if (dt < 60) return `${dt.toFixed(0)}с`; + if (dt < 3600) return `${(dt / 60).toFixed(1)} мин`; + return `${(dt / 3600).toFixed(1)} ч`; +} + +export default function GeoScrapeAdminPage() { + const [token, setToken] = useState(() => + typeof window === "undefined" + ? "" + : (localStorage.getItem("admin_token") ?? ""), + ); + const [name, setName] = useState(""); + const [sourceKind, setSourceKind] = useState< + "manual_list" | "rosreestr_pending" + >("rosreestr_pending"); + const [jobKind, setJobKind] = useState<"quarters" | "parcels" | "buildings">( + "quarters", + ); + const [thematicId, setThematicId] = useState(2); + const [cadNumsText, setCadNumsText] = useState(""); + const [regionCodes, setRegionCodes] = useState("66,74,72,59"); + const [rateMs, setRateMs] = useState(600); + const [useLib, setUseLib] = useState(false); + const [lastResp, setLastResp] = useState(null); + const queryClient = useQueryClient(); + + const saveToken = (v: string) => { + setToken(v); + localStorage.setItem("admin_token", v); + }; + + const jobs = useQuery({ + queryKey: ["geo-jobs", token], + queryFn: () => + apiFetch("/api/v1/admin/scrape/geo/jobs?limit=30", { + headers: { "X-Admin-Token": token }, + }), + enabled: !!token, + refetchInterval: 4000, + }); + + const enqueue = useMutation({ + mutationFn: () => { + const cad_nums = + sourceKind === "manual_list" + ? cadNumsText + .split(/[,\n\s]+/) + .map((s) => s.trim()) + .filter(Boolean) + : []; + const region_codes = + sourceKind === "rosreestr_pending" + ? regionCodes + .split(",") + .map((s) => parseInt(s.trim(), 10)) + .filter((n) => Number.isFinite(n)) + : null; + return apiFetch("/api/v1/admin/scrape/geo/jobs", { + method: "POST", + headers: { "X-Admin-Token": token, "Content-Type": "application/json" }, + body: JSON.stringify({ + name: name || null, + job_kind: jobKind, + source_kind: sourceKind, + cad_nums, + thematic_id: thematicId, + region_codes, + rate_ms: rateMs, + use_rosreestr2coord: useLib, + }), + }); + }, + onSuccess: (data) => { + setLastResp(data); + void queryClient.invalidateQueries({ queryKey: ["geo-jobs"] }); + }, + }); + + const cancelJob = useMutation({ + mutationFn: (id: number) => + apiFetch<{ cancelled: boolean }>( + `/api/v1/admin/scrape/geo/jobs/${id}/cancel`, + { method: "POST", headers: { "X-Admin-Token": token } }, + ), + onSuccess: () => + void queryClient.invalidateQueries({ queryKey: ["geo-jobs"] }), + }); + + const resumeJob = useMutation({ + mutationFn: (id: number) => + apiFetch<{ resumed: boolean }>( + `/api/v1/admin/scrape/geo/jobs/${id}/resume`, + { method: "POST", headers: { "X-Admin-Token": token } }, + ), + onSuccess: () => + void queryClient.invalidateQueries({ queryKey: ["geo-jobs"] }), + }); + + return ( + <> +

+ NSPD Geo bulk-fetcher (resume-friendly) +

+

+ Bulk-загрузка кадастровых данных с nspd.gov.ru через urllib + (lite-fetcher без Playwright). State хранится в БД ( + nspd_geo_jobs + nspd_geo_targets) — при + redeploy worker_ready hook автоматически re-enqueue недоделанные jobs и + продолжит с pending targets. +

+ +
+ +
+ + {/* Форма создания нового job */} +
+

🚀 Новый bulk geo-job

+
+ + +
+ + + +
+ + {sourceKind === "rosreestr_pending" && ( + + )} + + {sourceKind === "manual_list" && ( +