gendesign/backend/app/services/site_finder/cadastre_fetch.py

"""On-demand cadastre geometry fetching for /site-finder.

Issue #93: пользователь вводит cad-номер которого нет в БД (cad_quarters_geom /
cad_buildings / cad_parcels_geom) — раньше получал 404 «Загрузи через NSPD geo»
(dead-end для non-admin). Теперь:

1. `find_or_enqueue_fetch(db, cad_num)` — atomic helper:
   - Если cad есть в БД → return ("ready", None)
   - Если уже есть active on-demand job → return ("fetching", job_id)
   - Иначе создаёт новый NSPD job + enqueues Celery task → return ("fetching", job_id)
2. `fetch_status(db, cad_num)` — polling endpoint helper:
   - Smart status: проверяет и БД (ready), и nspd_geo_jobs (fetching/failed),
     и nspd_geo_targets (per-cad status, distinguishes not_in_nspd vs failed).

Дедупликация: `pg_try_advisory_xact_lock(hashtext(cad_num))` оборачивает
шаги "check active job → enqueue new" в `find_or_enqueue_fetch`. Lock держится
до COMMIT текущей транзакции (xact_lock — transaction-scoped). Параллельные
запросы на тот же cad: второй увидит lock=false и фолбэк-чит активный job уже
от первого транзакта (после первого COMMIT).
"""

from __future__ import annotations

import logging
from typing import Literal

from sqlalchemy import text
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

FetchStatus = Literal["ready", "fetching", "failed", "not_in_nspd", "invalid_format"]

# Source kind для on-demand jobs — отдельный от bulk-операций ("rosreestr_pending",
# "manual_list"), позволяет фильтровать в admin UI и в analytics queries.
ON_DEMAND_SOURCE_KIND = "auto_on_demand"


def detect_thematic_id(cad_num: str) -> int | None:
    """3-segment cad → quarter (thematic_id=2); 4-segment → parcel (thematic_id=1).

    Building cad (5-segment) обычно не вводится пользователем напрямую,
    но если придёт — попробуем как parcel.
    """
    parts = cad_num.split(":")
    n = len(parts)
    if n == 3:
        return 2  # quarter
    if n == 4:
        return 1  # parcel
    if n == 5:
        return 5  # building
    return None


def validate_cad_format(cad_num: str) -> bool:
    """Проверка формата: NN:NN:NNNNNN[:NN[:NN]].

    Каждый сегмент — только цифры (1+).
    """
    parts = cad_num.split(":")
    if not (3 <= len(parts) <= 5):
        return False
    return all(p.isdigit() and len(p) >= 1 for p in parts)


def cad_exists_in_db(db: Session, cad_num: str) -> bool:
    """Проверка: есть ли НЕПУСТАЯ geometry для cad_num в любой из 3 таблиц.

    Использует EXISTS short-circuit — самая дешёвая проверка.

    NB: `geom IS NOT NULL` обязателен — есть ~964 строк в cad_parcels_geom с meta,
    но NULL-геометрией. Без фильтра функция врала бы «geometry есть» → analyze
    fallback решал бы "ready"/не ставил NSPD-fetch → участок навсегда застревал в
    202 (никогда не подтянул бы реальную геометрию). С фильтром null-geom участок
    корректно идёт в fetch → НСПД отдаёт геометрию → analyze работает (или 404).
    """
    row = db.execute(
        text(
            """
            SELECT 1 WHERE EXISTS (
                SELECT 1 FROM cad_quarters_geom g
                 WHERE g.cad_number = :c AND g.geom IS NOT NULL
                UNION ALL
                SELECT 1 FROM cad_buildings b
                 WHERE b.cad_num = :c AND b.geom IS NOT NULL
                UNION ALL
                SELECT 1 FROM cad_parcels_geom p
                 WHERE p.cad_num = :c AND p.geom IS NOT NULL
            ) LIMIT 1
            """
        ),
        {"c": cad_num},
    ).first()
    return row is not None


def find_active_on_demand_job(db: Session, cad_num: str) -> int | None:
    """Найти существующий on-demand job (queued/running) для этого cad.

    Возвращает job_id или None. Если в БД есть FAILED on-demand за последние 60
    секунд — тоже None (чтобы повторно пробовать). Если есть DONE job, но cad
    отсутствует в БД (на NSPD не нашлось) — тоже None, но caller через
    `fetch_status` отличит этот случай как `not_in_nspd`.
    """
    row = db.execute(
        text(
            """
            SELECT j.job_id
              FROM nspd_geo_jobs j
              JOIN nspd_geo_targets t ON t.job_id = j.job_id
             WHERE j.source_kind = :src
               AND j.status IN ('queued', 'running')
               AND t.cad_num = :c
             ORDER BY j.created_at DESC
             LIMIT 1
            """
        ),
        {"src": ON_DEMAND_SOURCE_KIND, "c": cad_num},
    ).first()
    return int(row[0]) if row else None


def find_recent_completed_job(db: Session, cad_num: str) -> dict | None:
    """Last on-demand job (done/failed) для этого cad — для smart status.

    Используется в `fetch_status` чтобы отличить:
    - target.status='done' + cad есть в БД → ready
    - target.status='done' + cad нет в БД → not_in_nspd (НСПД не вернул geometry)
    - target.status='failed' → failed (rate-limit/WAF/timeout)
    """
    row = db.execute(
        text(
            """
            SELECT j.job_id, j.status AS job_status,
                   t.status AS target_status, t.error_msg
              FROM nspd_geo_jobs j
              JOIN nspd_geo_targets t ON t.job_id = j.job_id
             WHERE j.source_kind = :src
               AND t.cad_num = :c
             ORDER BY j.created_at DESC
             LIMIT 1
            """
        ),
        {"src": ON_DEMAND_SOURCE_KIND, "c": cad_num},
    ).first()
    if not row:
        return None
    return {
        "job_id": int(row[0]),
        "job_status": row[1],
        "target_status": row[2],
        "error_msg": row[3],
    }


def enqueue_on_demand_fetch(db: Session, cad_num: str, thematic_id: int) -> int:
    """Создать новый on-demand NSPD job + enqueue Celery task.

    Возвращает job_id. Использует существующий `enqueue_geo_job` helper из
    `app.workers.tasks.nspd_geo` для единообразия с bulk операциями.
    """
    from app.services.job_settings import get_setting_value
    from app.workers.tasks.nspd_geo import enqueue_geo_job, process_nspd_geo_job

    name = f"on-demand: {cad_num}"
    job_kind_map = {1: "parcels", 2: "quarters", 5: "buildings"}
    job_kind = job_kind_map.get(thematic_id, "mixed")

    job_id = enqueue_geo_job(
        name=name,
        job_kind=job_kind,
        source_kind=ON_DEMAND_SOURCE_KIND,
        source_params={"cad_num": cad_num, "trigger": "site-finder"},
        cad_nums_with_thematic=[(cad_num, thematic_id)],
        # On-demand priority — rate_ms ниже чем у bulk (быстрее), чтобы
        # пользователь получал ответ за ~10-20с.
        rate_ms=200,
        triggered_by="auto_on_demand",
    )
    # Enqueue в high-priority queue если есть, иначе fallback "geo"
    geo_queue = get_setting_value("nspd_geo", "queue_name", "geo")
    process_nspd_geo_job.apply_async(args=[job_id], queue=geo_queue, priority=9)

    logger.info(
        "on-demand cadastre fetch enqueued: cad=%s job=%s thematic=%s",
        cad_num,
        job_id,
        thematic_id,
    )
    return job_id


def find_or_enqueue_fetch(db: Session, cad_num: str) -> tuple[FetchStatus, int | None, str | None]:
    """Главный entrypoint для analyze fallback flow.

    Returns: (status, job_id, error_msg).

    - ("ready", None, None) — cad уже в БД, можно делать analyze
    - ("invalid_format", None, msg) — bad format
    - ("fetching", job_id, None) — fetch инициирован или уже идёт
    - ("not_in_nspd", None, msg) — последний fetch вернул empty (НСПД нет cad)
    - ("failed", job_id, msg) — последний fetch упал (rate-limit/WAF)

    Caller (analyze endpoint) использует status чтобы решить:
    - "ready" → продолжить с analyze (path existing)
    - "fetching" → 202 + job_id для polling
    - "not_in_nspd" / "invalid_format" → 404 с понятным сообщением
    - "failed" → 503 с retry-after

    Дедупликация (через pg advisory lock):
      Шаги 3-6 ниже (check active → check recent → enqueue) оборачиваются
      `pg_try_advisory_xact_lock(hashtext(cad_num))`. Lock automatically
      released at xact end. Параллельные запросы на тот же cad:
      - Первый запрос получает lock → выполняет enqueue → COMMIT → release.
      - Второй запрос lock=false → re-проверяет active job (тот что enqueue'нул
        первый и уже видим в DB после COMMIT) → возвращает ("fetching", job_id1).
    """
    # 1) Format check (cheap, lock-free)
    if not validate_cad_format(cad_num):
        return (
            "invalid_format",
            None,
            ("Неверный формат кадастрового номера. Ожидается NN:NN:NNNNNN или NN:NN:NNNNNN:NN"),
        )

    # 2) Already in DB? (lock-free read)
    if cad_exists_in_db(db, cad_num):
        return ("ready", None, None)

    # 3-6) Critical section: advisory lock на hashtext(cad_num) гарантирует
    # один enqueue per cad даже при concurrent requests.
    got_lock = db.execute(
        text("SELECT pg_try_advisory_xact_lock(hashtext(:c))"),
        {"c": cad_num},
    ).scalar()

    # Lock acquired or not — в обоих случаях сначала проверяем active job
    # (если другой xact уже enqueue'нул — мы видим его после его COMMIT).
    active_job_id = find_active_on_demand_job(db, cad_num)
    if active_job_id is not None:
        return ("fetching", active_job_id, None)

    if not got_lock:
        # Lock потерян concurrent xact но active job ещё не закоммичен.
        # Возвращаем "fetching" с None job_id — frontend начнёт polling и
        # подцепит job через /fetch-status (там тоже проверка active job).
        logger.info("advisory lock contention for cad=%s — caller will poll status", cad_num)
        return ("fetching", None, None)

    # 4) Recent completed job? Distinguish not_in_nspd vs failed.
    recent = find_recent_completed_job(db, cad_num)
    if recent:
        if recent["target_status"] == "done":
            # NSPD не вернул geometry (target_status=done but cad не появился в БД)
            return (
                "not_in_nspd",
                recent["job_id"],
                f"Кадастровый номер {cad_num} не найден в НСПД.",
            )
        elif recent["target_status"] == "failed":
            # Re-enqueue новый job — даём ещё попытку
            logger.info("retry on-demand fetch after prior failure: cad=%s", cad_num)
            # fall through to enqueue ниже

    # 5) Determine thematic_id from cad format
    thematic_id = detect_thematic_id(cad_num)
    if thematic_id is None:
        return (
            "invalid_format",
            None,
            f"Cannot determine cadastre type from {cad_num}",
        )

    # 6) Enqueue new on-demand fetch (lock освободится на COMMIT после return)
    try:
        job_id = enqueue_on_demand_fetch(db, cad_num, thematic_id)
        return ("fetching", job_id, None)
    except Exception as e:  # pragma: no cover
        logger.exception("on-demand enqueue failed for cad=%s", cad_num)
        return ("failed", None, f"Не удалось поставить fetch в очередь: {e}")


def fetch_status(db: Session, cad_num: str) -> dict:
    """Polling endpoint helper. Возвращает dict для FastAPI JSON response.

    Контракт для frontend:
    {
        "status": "ready" | "fetching" | "not_in_nspd" | "failed" | "invalid_format",
        "job_id": int | None,
        "error_msg": str | None,
        "eta_seconds": int | None,  # для fetching — оценка ожидания
    }
    """
    if not validate_cad_format(cad_num):
        return {
            "status": "invalid_format",
            "job_id": None,
            "error_msg": "Неверный формат кадастрового номера",
            "eta_seconds": None,
        }

    # Recheck DB first — может быть NSPD job уже отработал
    if cad_exists_in_db(db, cad_num):
        return {
            "status": "ready",
            "job_id": None,
            "error_msg": None,
            "eta_seconds": None,
        }

    # Active in-flight job?
    active_job_id = find_active_on_demand_job(db, cad_num)
    if active_job_id is not None:
        # Estimate ETA — depends on queue depth, simplified to ~15s
        return {
            "status": "fetching",
            "job_id": active_job_id,
            "error_msg": None,
            "eta_seconds": 15,
        }

    # Recent completed job for diagnosis
    recent = find_recent_completed_job(db, cad_num)
    if recent:
        if recent["target_status"] == "done":
            # Job finished but cad still not in DB → NSPD не отдал geometry
            return {
                "status": "not_in_nspd",
                "job_id": recent["job_id"],
                "error_msg": (
                    f"Кадастровый номер {cad_num} не найден в НСПД. "
                    "Проверьте формат: NN:NN:NNNNNN[:NN]"
                ),
                "eta_seconds": None,
            }
        elif recent["target_status"] == "failed":
            return {
                "status": "failed",
                "job_id": recent["job_id"],
                "error_msg": recent["error_msg"]
                or "НСПД временно недоступен. Попробуйте через минуту.",
                "eta_seconds": None,
            }

    # No job at all — first request without prior history
    return {
        "status": "not_in_nspd",
        "job_id": None,
        "error_msg": (
            f"Кадастровый номер {cad_num} не загружен. "
            "Откройте /site-finder и введите cad — он будет загружен автоматически."
        ),
        "eta_seconds": None,
    }