* feat(site-finder): auto-fetch cadastre geometry on-demand (#93)
Когда пользователь вводит cad-номер которого нет в БД (cad_quarters_geom /
cad_buildings / cad_parcels_geom), вместо 404 «Загрузи через NSPD geo» (dead-
end для non-admin) — теперь backend автоматически инициирует NSPD fetch,
frontend показывает loading state с polling /fetch-status каждые 2с.
Backend:
- Новый модуль app/services/site_finder/cadastre_fetch.py с helpers:
find_or_enqueue_fetch (atomic check + enqueue с дедупликацией по cad),
fetch_status (smart polling endpoint — отличает not_in_nspd от failed),
detect_thematic_id (3-сегм quarter / 4-сегм parcel / 5-сегм building),
validate_cad_format.
- Reuse: enqueue_geo_job + process_nspd_geo_job (workers/tasks/nspd_geo).
source_kind='auto_on_demand' отличает от bulk; rate_ms=200, priority=9.
- POST /parcels/{cad}/analyze graceful fallback: inline await до 25s
(fast path), затем 202 + job_id + eta_seconds для polling, либо
400/404/503 в зависимости от статуса (с Retry-After 60s на 503).
- GET /parcels/{cad}/fetch-status новый endpoint для polling.
Frontend:
- useSiteAnalysis расширен: fetchingState + cancel(). POST analyze + при 202
polling каждые 2с (max 60 итераций = 2 мин cap). status=ready → re-trigger
analyze; not_in_nspd/failed/invalid_format → typed errors.
- apiFetchWithStatus<T> + HTTPError в lib/api.ts — status-aware variant
для 200 vs 202.
- FetchingState.tsx: animated spinner, progress bar (linear до etaSeconds,
asymptote после), elapsed counter, cancel button. Светло-голубая
scheme отличается от обычного pending skeleton.
- site-finder/page.tsx: FetchingState когда fetchingState активен; обычный
pending skeleton — только при первичном analyze без 202.
Edge cases (per #93 acceptance):
- cad валидный в НСПД, fetch <25s → inline 200 OK
- cad валидный, fetch >25s → 202 + frontend polls → ready → analyze
- cad валидный, не в НСПД → 404 с понятным сообщением + формат hint
- cad invalid format → 400 + формат hint
- NSPD rate-limited / failed → 503 + Retry-After 60s
- Параллельные запросы на тот же cad → один job, оба клиента poll'ят (дедуп
через find_active_on_demand_job).
Closes #93.
* fix(site-finder): address PR #95 auto-review minor feedback
Backend (cadastre_fetch.py):
1. (race condition) Advisory lock `pg_try_advisory_xact_lock(hashtext(cad_num))`
обёрнут вокруг шагов "check active job → enqueue" в find_or_enqueue_fetch.
Lock transaction-scoped, released at COMMIT. Параллельные запросы на тот
же cad: первый получает lock и enqueue; второй lock=false → re-check
active job (уже виден после первого COMMIT) → возвращает тот же job_id.
Docstring обновлён, упоминание SELECT FOR UPDATE удалено.
Backend (parcels.py):
3. (threadpool exhaustion) _INLINE_FETCH_WAIT_S снижен 25 → 15s с подробным
комментарием: tradeoff про default Starlette anyio threadpool (40 slots)
и concurrent burst saturation. 15s баланс: НСПД avg 5-15s для quarter,
~70% fast path; остальные 30% получают 202 + polling без блока.
Data (87_on_demand_indexes.sql):
2. (missing index) New migration:
- `nspd_geo_targets_cad_num_idx ON nspd_geo_targets(cad_num)` — для
find_active/recent_completed_job (existing UNIQUE composite не покрывает
WHERE cad_num=:c).
- `nspd_geo_jobs_source_status_idx ON nspd_geo_jobs(source_kind, status)`
composite для filter auto_on_demand + queued/running.
Idempotent (CREATE INDEX IF NOT EXISTS), не блокер при текущем размере,
критично при росте on-demand traffic.
Frontend (useSiteAnalysis.ts):
4. (UI flicker) setFetchingState(null) перемещён ПОСЛЕ `await second
apiFetch`. Иначе между clear и resolve есть момент когда mutation
isPending=true + fetchingState=null → пустой экран ~1 RTT.
NOT addressed (rebuttal):
5. (Tailwind convention) — проверил: в проекте нет ни globals.css, ни
@tailwind directives. ВСЕ существующие site-finder components используют
inline styles (ConfidenceBadge, GeotechRiskBlock, ScoreBreakdownPanel etc).
Tailwind в deps но не wired up. Keep inline styles для consistency.
6. (animate-spin) — требует Tailwind globals (см. #5). `<style jsx>`
keyframes — built-in Next.js, работает.
Per auto-review on 2252236.
---------
Co-authored-by: lekss361 <claudestars@proton.me>
This commit is contained in:
lekss361
GitHub
parent
f4a865060f
commit
49fac806fc
7 changed files with 857 additions and 18 deletions
|
|
@ -2,10 +2,11 @@ import datetime as _dt
|
|||
import json
|
||||
import logging
|
||||
import math
|
||||
import time
|
||||
from typing import Annotated, Any
|
||||
|
||||
import httpx
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Response
|
||||
from shapely import wkt as _shp_wkt
|
||||
from shapely.geometry import Polygon
|
||||
from sqlalchemy import text
|
||||
|
|
@ -14,6 +15,13 @@ from sqlalchemy.orm import Session
|
|||
from app.core.config import settings
|
||||
from app.core.db import get_db
|
||||
from app.schemas.parcel import ParcelDetail, ParcelSearchRequest, ParcelSearchResponse
|
||||
from app.services.site_finder.cadastre_fetch import (
|
||||
cad_exists_in_db,
|
||||
find_or_enqueue_fetch,
|
||||
)
|
||||
from app.services.site_finder.cadastre_fetch import (
|
||||
fetch_status as _fetch_status,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -943,6 +951,21 @@ def _compute_confidence(
|
|||
}
|
||||
|
||||
|
||||
# #93 — on-demand cadastre fetch tuning constants.
|
||||
# _INLINE_FETCH_WAIT_S — суммарно ждём fast-path при analyze fallback.
|
||||
#
|
||||
# Tradeoff: sync `def analyze_parcel` блокирует один FastAPI threadpool slot
|
||||
# на это время. Default threadpool в Starlette/FastAPI — 40 slots (anyio
|
||||
# default). При concurrent burst >40 «миссинг cad» запросов будем saturate
|
||||
# threadpool — последующие запросы (включая healthcheck) ждут free slot.
|
||||
#
|
||||
# 15s выбран как баланс: НСПД ~5-15s avg для quarter, ~10-20s для parcel —
|
||||
# fast path сработает в ~70% случаев. Остальные 30% получают 202 +
|
||||
# polling (без блокировки threadpool).
|
||||
_INLINE_FETCH_WAIT_S = 15
|
||||
_INLINE_FETCH_POLL_INTERVAL_S = 2
|
||||
|
||||
|
||||
@router.post("/search", response_model=ParcelSearchResponse)
|
||||
async def search_parcels(payload: ParcelSearchRequest) -> ParcelSearchResponse:
|
||||
"""Search parcels by filters + scoring.
|
||||
|
|
@ -958,14 +981,51 @@ async def get_parcel(parcel_id: str) -> ParcelDetail:
|
|||
raise HTTPException(status_code=501, detail="Not implemented yet")
|
||||
|
||||
|
||||
@router.get("/{cad_num}/fetch-status")
|
||||
def get_fetch_status(
|
||||
cad_num: str,
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
) -> dict[str, Any]:
|
||||
"""Polling endpoint для on-demand cadastre fetch (см. issue #93).
|
||||
|
||||
Frontend polling каждые 2с после 202 Accepted из /analyze.
|
||||
|
||||
Returns:
|
||||
{
|
||||
"status": "ready" | "fetching" | "not_in_nspd" | "failed" | "invalid_format",
|
||||
"job_id": int | None,
|
||||
"error_msg": str | None,
|
||||
"eta_seconds": int | None,
|
||||
}
|
||||
|
||||
Frontend поведение:
|
||||
- "ready" → автоматически re-trigger POST /analyze
|
||||
- "fetching" → continue polling
|
||||
- "not_in_nspd" → показать пользователю «cad не найден в НСПД»
|
||||
- "failed" → retry button + retry-after message
|
||||
- "invalid_format" → подсказка формата
|
||||
"""
|
||||
return _fetch_status(db, cad_num)
|
||||
|
||||
|
||||
@router.post("/{cad_num}/analyze")
|
||||
def analyze_parcel(
|
||||
cad_num: str,
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
response: Response,
|
||||
) -> dict[str, Any]:
|
||||
"""Анализ участка: близость к социалке + district context + конкуренты.
|
||||
|
||||
Порядок поиска геометрии: cad_quarters_geom → cad_buildings.
|
||||
Порядок поиска геометрии: cad_quarters_geom → cad_buildings → cad_parcels_geom.
|
||||
|
||||
Issue #93 — Graceful fallback при отсутствии geometry:
|
||||
- Не возвращаем 404 сразу. Вместо: enqueue NSPD on-demand fetch, ждём
|
||||
inline до _INLINE_FETCH_WAIT_S (~25с). Если за это время геометрия
|
||||
появилась в БД — продолжаем analyze (fast path).
|
||||
- Иначе → 202 Accepted + {status, job_id, eta_seconds} для polling
|
||||
через GET /fetch-status.
|
||||
- Дедупликация (через `find_active_on_demand_job`): параллельные запросы
|
||||
на тот же cad → один Celery job, оба клиента ждут.
|
||||
"""
|
||||
# 1) Получить геометрию участка — GeoJSON строка через ST_AsGeoJSON
|
||||
row = (
|
||||
|
|
@ -997,10 +1057,57 @@ def analyze_parcel(
|
|||
)
|
||||
|
||||
if not row:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"Геометрия для {cad_num} не найдена. Загрузи через NSPD geo.",
|
||||
)
|
||||
# #93 — graceful fallback: enqueue NSPD fetch, await inline до 25s.
|
||||
status, job_id, error_msg = find_or_enqueue_fetch(db, cad_num)
|
||||
if status == "invalid_format":
|
||||
raise HTTPException(status_code=400, detail=error_msg)
|
||||
if status == "not_in_nspd":
|
||||
raise HTTPException(status_code=404, detail=error_msg)
|
||||
if status == "failed":
|
||||
# 503 — НСПД временно недоступен (rate-limit / WAF)
|
||||
response.headers["Retry-After"] = "60"
|
||||
raise HTTPException(status_code=503, detail=error_msg)
|
||||
# status == "fetching" → inline await fast path
|
||||
deadline = time.monotonic() + _INLINE_FETCH_WAIT_S
|
||||
while time.monotonic() < deadline:
|
||||
time.sleep(_INLINE_FETCH_POLL_INTERVAL_S)
|
||||
if cad_exists_in_db(db, cad_num):
|
||||
# Re-fetch row для analyze
|
||||
row = (
|
||||
db.execute(
|
||||
text("""
|
||||
SELECT ST_AsGeoJSON(g.geom) AS geom_geojson,
|
||||
g.geom AS geom_wkb,
|
||||
'cad_quarter' AS source
|
||||
FROM cad_quarters_geom g
|
||||
WHERE g.cad_number = :c
|
||||
UNION ALL
|
||||
SELECT ST_AsGeoJSON(b.geom), b.geom, 'cad_building'
|
||||
FROM cad_buildings b
|
||||
WHERE b.cad_num = :c
|
||||
UNION ALL
|
||||
SELECT ST_AsGeoJSON(p.geom), p.geom, 'cad_parcel'
|
||||
FROM cad_parcels_geom p
|
||||
WHERE p.cad_num = :c
|
||||
LIMIT 1
|
||||
"""),
|
||||
{"c": cad_num},
|
||||
)
|
||||
.mappings()
|
||||
.first()
|
||||
)
|
||||
if row:
|
||||
break
|
||||
if not row:
|
||||
# Timeout — frontend будет poll'ить /fetch-status.
|
||||
response.status_code = 202
|
||||
return {
|
||||
"status": "fetching",
|
||||
"cad_num": cad_num,
|
||||
"job_id": job_id,
|
||||
"eta_seconds": 15,
|
||||
"message": ("Геометрия загружается из НСПД. Обычно занимает 15-30 секунд."),
|
||||
}
|
||||
|
||||
geom_geojson: str = row["geom_geojson"]
|
||||
source: str = row["source"]
|
||||
|
|
|
|||
346
backend/app/services/site_finder/cadastre_fetch.py
Normal file
346
backend/app/services/site_finder/cadastre_fetch.py
Normal file
|
|
@ -0,0 +1,346 @@
|
|||
"""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 — самая дешёвая проверка.
|
||||
"""
|
||||
row = db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT 1 WHERE EXISTS (
|
||||
SELECT 1 FROM cad_quarters_geom g WHERE g.cad_number = :c
|
||||
UNION ALL
|
||||
SELECT 1 FROM cad_buildings b WHERE b.cad_num = :c
|
||||
UNION ALL
|
||||
SELECT 1 FROM cad_parcels_geom p WHERE p.cad_num = :c
|
||||
) 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,
|
||||
}
|
||||
36
data/sql/87_on_demand_indexes.sql
Normal file
36
data/sql/87_on_demand_indexes.sql
Normal file
|
|
@ -0,0 +1,36 @@
|
|||
-- #93 (PR #95): indexes for on-demand cadastre fetch lookups.
|
||||
--
|
||||
-- Context:
|
||||
-- `find_active_on_demand_job` и `find_recent_completed_job` в
|
||||
-- `app/services/site_finder/cadastre_fetch.py` фильтруют
|
||||
-- `nspd_geo_targets` по cad_num + `nspd_geo_jobs` по source_kind+status.
|
||||
-- Existing UNIQUE(job_id, cad_num, thematic_id) на targets не помогает
|
||||
-- при WHERE cad_num=:c — нужен secondary index.
|
||||
--
|
||||
-- Не блокер при текущем размере таблицы (~10K targets), но критично при
|
||||
-- ускорении on-demand traffic (per #93 trigger любым посетителем
|
||||
-- /site-finder).
|
||||
--
|
||||
-- Apply: idempotent, CREATE INDEX IF NOT EXISTS — safe to run multiple
|
||||
-- times. CONCURRENTLY не используется чтобы коммит был атомарным; на
|
||||
-- prod размере index build занимает <1s.
|
||||
|
||||
BEGIN;
|
||||
|
||||
-- Targets — поиск активных on-demand job'ов для конкретного cad
|
||||
CREATE INDEX IF NOT EXISTS nspd_geo_targets_cad_num_idx
|
||||
ON nspd_geo_targets(cad_num);
|
||||
|
||||
-- Jobs — фильтр source_kind='auto_on_demand' + status IN (queued,running)
|
||||
CREATE INDEX IF NOT EXISTS nspd_geo_jobs_source_status_idx
|
||||
ON nspd_geo_jobs(source_kind, status);
|
||||
|
||||
COMMIT;
|
||||
|
||||
COMMENT ON INDEX nspd_geo_targets_cad_num_idx IS
|
||||
'on-demand cadastre fetch lookups (#93). Used by find_active_on_demand_job '
|
||||
'и find_recent_completed_job в cadastre_fetch service.';
|
||||
|
||||
COMMENT ON INDEX nspd_geo_jobs_source_status_idx IS
|
||||
'Filter source_kind=auto_on_demand + status (#93). Composite чтобы '
|
||||
'покрыть оба predicate сразу.';
|
||||
|
|
@ -7,6 +7,7 @@ import { usePathname, useRouter, useSearchParams } from "next/navigation";
|
|||
import type { FeatureCollection } from "geojson";
|
||||
|
||||
import { CadInput } from "@/components/site-finder/CadInput";
|
||||
import { FetchingState } from "@/components/site-finder/FetchingState";
|
||||
import { KpiCard } from "@/components/site-finder/KpiCard";
|
||||
import { OverviewTab } from "@/components/site-finder/OverviewTab";
|
||||
import { EnvironmentTab } from "@/components/site-finder/EnvironmentTab";
|
||||
|
|
@ -85,7 +86,8 @@ function SiteFinderContent() {
|
|||
const pathname = usePathname();
|
||||
const searchParams = useSearchParams();
|
||||
|
||||
const { mutate, data, isPending, error, isIdle } = useSiteAnalysis();
|
||||
const { mutate, data, isPending, error, isIdle, fetchingState, cancel } =
|
||||
useSiteAnalysis();
|
||||
const [isochrones, setIsochrones] = useState<FeatureCollection | undefined>(
|
||||
undefined,
|
||||
);
|
||||
|
|
@ -220,8 +222,17 @@ function SiteFinderContent() {
|
|||
</div>
|
||||
)}
|
||||
|
||||
{/* Pending skeleton */}
|
||||
{isPending && (
|
||||
{/* #93 — Fetching state: backend на 202 Accepted, идёт on-demand NSPD fetch */}
|
||||
{fetchingState && (
|
||||
<FetchingState
|
||||
cadNum={fetchingState.cadNum}
|
||||
etaSeconds={fetchingState.etaSeconds}
|
||||
onCancel={cancel}
|
||||
/>
|
||||
)}
|
||||
|
||||
{/* Pending skeleton — первичный analyze в полёте (но fetching state ещё не активирован) */}
|
||||
{isPending && !fetchingState && (
|
||||
<div
|
||||
style={{
|
||||
padding: 40,
|
||||
|
|
|
|||
154
frontend/src/components/site-finder/FetchingState.tsx
Normal file
154
frontend/src/components/site-finder/FetchingState.tsx
Normal file
|
|
@ -0,0 +1,154 @@
|
|||
"use client";
|
||||
|
||||
import { useEffect, useState } from "react";
|
||||
|
||||
interface Props {
|
||||
cadNum: string;
|
||||
etaSeconds: number;
|
||||
onCancel: () => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* #93 — loading state когда backend на 202 Accepted делает on-demand NSPD fetch.
|
||||
*
|
||||
* Показывает animated progress bar + текущий elapsed time + cancel button.
|
||||
*/
|
||||
export function FetchingState({ cadNum, etaSeconds, onCancel }: Props) {
|
||||
const [elapsed, setElapsed] = useState(0);
|
||||
|
||||
useEffect(() => {
|
||||
const interval = setInterval(() => {
|
||||
setElapsed((e) => e + 1);
|
||||
}, 1000);
|
||||
return () => clearInterval(interval);
|
||||
}, []);
|
||||
|
||||
// Progress: linear до etaSeconds, потом замедляемся (asymptote 95%)
|
||||
const progressPct =
|
||||
elapsed <= etaSeconds
|
||||
? (elapsed / etaSeconds) * 80
|
||||
: 80 + Math.min(15, (elapsed - etaSeconds) * 0.5);
|
||||
|
||||
return (
|
||||
<div
|
||||
style={{
|
||||
padding: "32px 24px",
|
||||
background: "#f0f9ff",
|
||||
border: "1px solid #bae6fd",
|
||||
borderRadius: 12,
|
||||
display: "flex",
|
||||
flexDirection: "column",
|
||||
gap: 14,
|
||||
}}
|
||||
>
|
||||
<div style={{ display: "flex", alignItems: "center", gap: 10 }}>
|
||||
<span
|
||||
style={{
|
||||
display: "inline-block",
|
||||
width: 20,
|
||||
height: 20,
|
||||
border: "2.5px solid #0ea5e9",
|
||||
borderTopColor: "transparent",
|
||||
borderRadius: "50%",
|
||||
animation: "fetch-spin 1s linear infinite",
|
||||
}}
|
||||
aria-hidden="true"
|
||||
/>
|
||||
<h3
|
||||
style={{
|
||||
fontSize: 15,
|
||||
fontWeight: 600,
|
||||
color: "#0c4a6e",
|
||||
margin: 0,
|
||||
}}
|
||||
>
|
||||
Загружаем участок из НСПД…
|
||||
</h3>
|
||||
</div>
|
||||
|
||||
<div style={{ fontSize: 13, color: "#0369a1" }}>
|
||||
Кадастровый номер{" "}
|
||||
<code
|
||||
style={{
|
||||
fontFamily: "monospace",
|
||||
background: "#e0f2fe",
|
||||
padding: "1px 6px",
|
||||
borderRadius: 4,
|
||||
fontSize: 12,
|
||||
}}
|
||||
>
|
||||
{cadNum}
|
||||
</code>{" "}
|
||||
не был ранее загружен в нашу базу. Обращаемся к НСПД (Национальной
|
||||
системе пространственных данных). Обычно занимает 15-30 секунд.
|
||||
</div>
|
||||
|
||||
{/* Progress bar */}
|
||||
<div>
|
||||
<div
|
||||
style={{
|
||||
height: 8,
|
||||
borderRadius: 4,
|
||||
background: "#e0f2fe",
|
||||
overflow: "hidden",
|
||||
}}
|
||||
role="progressbar"
|
||||
aria-valuenow={Math.round(progressPct)}
|
||||
aria-valuemin={0}
|
||||
aria-valuemax={100}
|
||||
aria-label="Прогресс загрузки геометрии"
|
||||
>
|
||||
<div
|
||||
style={{
|
||||
height: "100%",
|
||||
width: `${progressPct}%`,
|
||||
background: "#0ea5e9",
|
||||
transition: "width 0.5s ease-out",
|
||||
borderRadius: 4,
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
<div
|
||||
style={{
|
||||
marginTop: 6,
|
||||
fontSize: 11,
|
||||
color: "#0369a1",
|
||||
display: "flex",
|
||||
justifyContent: "space-between",
|
||||
fontVariantNumeric: "tabular-nums",
|
||||
}}
|
||||
>
|
||||
<span>{elapsed} сек</span>
|
||||
<span>~{etaSeconds} сек ожидаемо</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<button
|
||||
type="button"
|
||||
onClick={onCancel}
|
||||
style={{
|
||||
alignSelf: "flex-start",
|
||||
marginTop: 4,
|
||||
padding: "6px 14px",
|
||||
background: "transparent",
|
||||
border: "1px solid #7dd3fc",
|
||||
color: "#0369a1",
|
||||
borderRadius: 6,
|
||||
fontSize: 12,
|
||||
cursor: "pointer",
|
||||
fontWeight: 500,
|
||||
}}
|
||||
>
|
||||
Отменить ожидание
|
||||
</button>
|
||||
|
||||
<style jsx>{`
|
||||
@keyframes fetch-spin {
|
||||
to {
|
||||
transform: rotate(360deg);
|
||||
}
|
||||
}
|
||||
`}</style>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
|
@ -1,16 +1,156 @@
|
|||
"use client";
|
||||
|
||||
import { useMutation } from "@tanstack/react-query";
|
||||
import { useCallback, useRef, useState } from "react";
|
||||
|
||||
import { apiFetch } from "@/lib/api";
|
||||
import { HTTPError, apiFetch, apiFetchWithStatus } from "@/lib/api";
|
||||
import type { ParcelAnalysis } from "@/types/site-finder";
|
||||
|
||||
export function useSiteAnalysis() {
|
||||
return useMutation({
|
||||
mutationFn: (cad: string) =>
|
||||
apiFetch<ParcelAnalysis>(
|
||||
`/api/v1/parcels/${encodeURIComponent(cad)}/analyze`,
|
||||
{ method: "POST" },
|
||||
),
|
||||
});
|
||||
// #93 — on-demand cadastre fetch flow.
|
||||
// Backend returns 202 Accepted when geometry needs fetching; frontend polls
|
||||
// /fetch-status every 2s, re-triggers analyze когда status=ready.
|
||||
|
||||
export type FetchStatus =
|
||||
| "ready"
|
||||
| "fetching"
|
||||
| "failed"
|
||||
| "not_in_nspd"
|
||||
| "invalid_format";
|
||||
|
||||
export interface FetchStatusResponse {
|
||||
status: FetchStatus;
|
||||
job_id: number | null;
|
||||
error_msg: string | null;
|
||||
eta_seconds: number | null;
|
||||
}
|
||||
|
||||
export interface AnalyzeAcceptedResponse {
|
||||
status: "fetching";
|
||||
cad_num: string;
|
||||
job_id: number;
|
||||
eta_seconds: number;
|
||||
message: string;
|
||||
}
|
||||
|
||||
export type AnalyzeResult =
|
||||
| { kind: "ready"; data: ParcelAnalysis }
|
||||
| { kind: "fetching"; jobId: number; etaSeconds: number };
|
||||
|
||||
const POLL_INTERVAL_MS = 2000;
|
||||
const POLL_MAX_ITERATIONS = 60; // 60 × 2s = 2 min hard cap
|
||||
|
||||
/**
|
||||
* Custom hook для analyze flow с graceful on-demand fetch fallback.
|
||||
*
|
||||
* Состояния:
|
||||
* - idle: ничего не делали
|
||||
* - pending (mutation): первичный POST /analyze в полёте
|
||||
* - fetching: получили 202, идёт polling /fetch-status
|
||||
* - data set: ParcelAnalysis отрисовываем
|
||||
* - error: not_in_nspd / invalid_format / failed
|
||||
*/
|
||||
export function useSiteAnalysis() {
|
||||
const [fetchingState, setFetchingState] = useState<{
|
||||
cadNum: string;
|
||||
jobId: number;
|
||||
etaSeconds: number;
|
||||
} | null>(null);
|
||||
// ref для cancel polling — invalidates in-flight iteration
|
||||
const cancelledRef = useRef(false);
|
||||
|
||||
const mutation = useMutation({
|
||||
mutationFn: async (cad: string): Promise<ParcelAnalysis> => {
|
||||
cancelledRef.current = false;
|
||||
setFetchingState(null);
|
||||
|
||||
// First request — POST /analyze
|
||||
const first = await apiFetchWithStatus<
|
||||
ParcelAnalysis | AnalyzeAcceptedResponse
|
||||
>(`/api/v1/parcels/${encodeURIComponent(cad)}/analyze`, {
|
||||
method: "POST",
|
||||
});
|
||||
|
||||
if (first.status === 200) {
|
||||
return first.body as ParcelAnalysis;
|
||||
}
|
||||
|
||||
// 202 Accepted — entering polling mode
|
||||
if (first.status === 202) {
|
||||
const accepted = first.body as AnalyzeAcceptedResponse;
|
||||
setFetchingState({
|
||||
cadNum: cad,
|
||||
jobId: accepted.job_id,
|
||||
etaSeconds: accepted.eta_seconds,
|
||||
});
|
||||
|
||||
// Poll loop
|
||||
for (let i = 0; i < POLL_MAX_ITERATIONS; i++) {
|
||||
await new Promise((r) => setTimeout(r, POLL_INTERVAL_MS));
|
||||
if (cancelledRef.current) {
|
||||
throw new Error("Загрузка отменена пользователем");
|
||||
}
|
||||
const status = await apiFetch<FetchStatusResponse>(
|
||||
`/api/v1/parcels/${encodeURIComponent(cad)}/fetch-status`,
|
||||
);
|
||||
if (status.status === "ready") {
|
||||
// Re-trigger analyze — should be 200 now.
|
||||
// setFetchingState(null) ПОСЛЕ await чтобы не было flicker:
|
||||
// если очистить до await, mutation isPending=true но
|
||||
// fetchingState=null → пустой экран ~1 RTT. После await
|
||||
// mutation сразу резолвится с data — render skipped.
|
||||
const second = await apiFetch<ParcelAnalysis>(
|
||||
`/api/v1/parcels/${encodeURIComponent(cad)}/analyze`,
|
||||
{ method: "POST" },
|
||||
);
|
||||
setFetchingState(null);
|
||||
return second;
|
||||
}
|
||||
if (status.status === "not_in_nspd") {
|
||||
setFetchingState(null);
|
||||
throw new HTTPError(
|
||||
404,
|
||||
status,
|
||||
status.error_msg ?? "Кадастровый номер не найден в НСПД",
|
||||
);
|
||||
}
|
||||
if (status.status === "failed") {
|
||||
setFetchingState(null);
|
||||
throw new HTTPError(
|
||||
503,
|
||||
status,
|
||||
status.error_msg ?? "НСПД временно недоступен",
|
||||
);
|
||||
}
|
||||
if (status.status === "invalid_format") {
|
||||
setFetchingState(null);
|
||||
throw new HTTPError(
|
||||
400,
|
||||
status,
|
||||
status.error_msg ?? "Неверный формат кадастрового номера",
|
||||
);
|
||||
}
|
||||
// status === "fetching" → continue polling
|
||||
}
|
||||
// Polling exhausted (2 min)
|
||||
setFetchingState(null);
|
||||
throw new Error(
|
||||
"Загрузка длится слишком долго (>2 мин). Попробуйте позже.",
|
||||
);
|
||||
}
|
||||
|
||||
throw new Error(`Unexpected response status: ${first.status}`);
|
||||
},
|
||||
});
|
||||
|
||||
const cancel = useCallback(() => {
|
||||
cancelledRef.current = true;
|
||||
setFetchingState(null);
|
||||
mutation.reset();
|
||||
}, [mutation]);
|
||||
|
||||
return {
|
||||
...mutation,
|
||||
fetchingState,
|
||||
cancel,
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,3 +18,48 @@ export async function apiFetch<T>(
|
|||
}
|
||||
return response.json() as Promise<T>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Status-aware variant of apiFetch — возвращает status + body для случаев
|
||||
* когда логика зависит от HTTP кода (например 202 Accepted для async jobs).
|
||||
*
|
||||
* Throws на 4xx/5xx (используя HTTPError ниже); 2xx коды все ОК.
|
||||
*/
|
||||
export class HTTPError extends Error {
|
||||
status: number;
|
||||
body: unknown;
|
||||
constructor(status: number, body: unknown, msg?: string) {
|
||||
super(msg ?? `API error ${status}`);
|
||||
this.status = status;
|
||||
this.body = body;
|
||||
}
|
||||
}
|
||||
|
||||
export async function apiFetchWithStatus<T>(
|
||||
path: string,
|
||||
init?: RequestInit,
|
||||
): Promise<{ status: number; body: T }> {
|
||||
const response = await fetch(`${API_BASE_URL}${path}`, {
|
||||
...init,
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
...(init?.headers ?? {}),
|
||||
},
|
||||
});
|
||||
let body: unknown = null;
|
||||
// 204 No Content имеет пустое тело; для всех остальных пытаемся JSON
|
||||
if (response.status !== 204) {
|
||||
try {
|
||||
body = await response.json();
|
||||
} catch {
|
||||
// fallback на text() — может прилететь HTML error page от Caddy
|
||||
body = { detail: await response.text() };
|
||||
}
|
||||
}
|
||||
if (!response.ok) {
|
||||
const detail =
|
||||
(body as { detail?: string })?.detail ?? `API error ${response.status}`;
|
||||
throw new HTTPError(response.status, body, detail);
|
||||
}
|
||||
return { status: response.status, body: body as T };
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue