81cf274e80
1. _acquire_harvest_lock: Settings() instance → singleton 'settings' import from app.core.config — соответствует convention (scrape_kn.py, objective.py, celery_app.py все используют singleton). 2. _trigger_harvest docstring: убрана incorrect фраза «остальные ... видят harvest_eta_seconds от первого запроса» (код возвращает eta=None для denied-lock callers). Заменена на actual behavior: graceful no-data UI с ПКК fallback на FE. PR #250 re-review followup (issues #3 + docstring sync).
1254 lines
50 KiB
Python
1254 lines
50 KiB
Python
"""Quarter dump lookup helper for analyze_parcel.
|
||
|
||
Sprint 1.1 item #4 (feat/analyze-uses-quarter-dump):
|
||
Читает nspd_quarter_dumps кеш и извлекает:
|
||
- nspd_zoning — территориальная зона ПЗЗ (G1) по centroid участка
|
||
- nspd_zouit_overlaps — список ЗОУИТ (G3) которые пересекаются с участком
|
||
- nspd_engineering_nearby — инженерные сооружения в 200м (I3)
|
||
- nspd_dump — freshness metadata (доступность, возраст, trigger флаг)
|
||
|
||
Если дамп отсутствует или устарел (>180 дней) — fire-and-forget
|
||
harvest_quarter.apply_async() и продолжает без dump-derived полей.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import math
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.exc import OperationalError, ProgrammingError
|
||
from sqlalchemy.orm import Session
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Порог свежести дампа — 180 дней. Совпадает с beat-расписанием (раз в квартал
|
||
# на практике, но с запасом для редко запрашиваемых кварталов).
|
||
_DUMP_MAX_AGE_DAYS = 180
|
||
|
||
# Радиус поиска инженерных сооружений (метры) — договорённость #44 I3.
|
||
_ENGINEERING_RADIUS_M = 200
|
||
|
||
# Issue #234: typical NSPD harvest duration. Frontend использует для countdown +
|
||
# auto-stop polling после ETA*1.5. Замерено по логам harvest_quarter task.
|
||
_HARVEST_ETA_SECONDS = 60
|
||
|
||
# Issue #234: SETNX lock TTL (секунды). Защищает от burst N concurrent analyze
|
||
# на один свежетриггеренный квартал — N одинаковых harvest task в очередь.
|
||
# TTL > _HARVEST_ETA_SECONDS чтобы lock жил пока task реально работает.
|
||
_HARVEST_LOCK_TTL_SECONDS = 120
|
||
|
||
# Sentinel для isinstance-проверок и read-only fallback в parcels.py try/except.
|
||
# НИКОГДА не мутировать — использовать make_empty_result() для новых dict.
|
||
EMPTY_DUMP_RESULT: dict[str, Any] = {
|
||
"nspd_zoning": None,
|
||
"nspd_zouit_overlaps": [],
|
||
"nspd_engineering_nearby": [],
|
||
"nspd_risk_zones": [],
|
||
"nspd_opportunity_parcels": [],
|
||
"nspd_red_lines": [],
|
||
"nspd_dump": {
|
||
"available": False,
|
||
"fetched_at_utc": None,
|
||
"stale": False,
|
||
"harvest_triggered": False,
|
||
"total_features": None,
|
||
"harvest_eta_seconds": None,
|
||
},
|
||
}
|
||
|
||
|
||
def make_empty_result(
|
||
*,
|
||
fetched_at_utc: str | None = None,
|
||
stale: bool = False,
|
||
harvest_triggered: bool = False,
|
||
total_features: int | None = None,
|
||
harvest_eta_seconds: int | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Создаёт свежую копию empty-dump result с возможностью переопределить поля.
|
||
|
||
Вызывать вместо EMPTY_DUMP_RESULT напрямую — чтобы не мутировать singleton.
|
||
|
||
`harvest_eta_seconds` (issue #234): передавать когда harvest_triggered=True,
|
||
чтобы фронт показал countdown вместо бесконечного спиннера и остановил
|
||
re-poll после ETA*1.5.
|
||
"""
|
||
return {
|
||
"nspd_zoning": None,
|
||
"nspd_zouit_overlaps": [],
|
||
"nspd_engineering_nearby": [],
|
||
"nspd_risk_zones": [],
|
||
"nspd_opportunity_parcels": [],
|
||
"nspd_red_lines": [],
|
||
"nspd_dump": {
|
||
"available": False,
|
||
"fetched_at_utc": fetched_at_utc,
|
||
"stale": stale,
|
||
"harvest_triggered": harvest_triggered,
|
||
"total_features": total_features,
|
||
"harvest_eta_seconds": harvest_eta_seconds,
|
||
},
|
||
}
|
||
|
||
|
||
def derive_quarter_cad(cad_num: str) -> str | None:
|
||
"""3-сегментный кадастровый квартал из любого кадастрового номера.
|
||
|
||
- 3-сегмент (квартал) '66:41:0204016' → '66:41:0204016'
|
||
- 4-сегмент (участок) '66:41:0204016:10' → '66:41:0204016'
|
||
- 5-сегмент (здание) '66:41:0204016:10:1' → '66:41:0204016'
|
||
- невалидный формат → None
|
||
"""
|
||
parts = cad_num.strip().split(":")
|
||
if len(parts) < 3:
|
||
return None
|
||
quarter = ":".join(parts[:3])
|
||
# Каждый сегмент — только цифры
|
||
if not all(p.isdigit() and len(p) >= 1 for p in parts[:3]):
|
||
return None
|
||
return quarter
|
||
|
||
|
||
def get_quarter_dump_data(
|
||
db: Session,
|
||
cad_num: str,
|
||
parcel_wkt: str | None,
|
||
) -> dict[str, Any]:
|
||
"""Читает quarter dump для квартала cad_num и возвращает НСПД-контекст.
|
||
|
||
Returns dict с ключами:
|
||
- nspd_zoning: dict | None — зона ПЗЗ из territorial_zones
|
||
- nspd_zouit_overlaps: list[dict] — ЗОУИТ пересечения
|
||
- nspd_engineering_nearby: list[dict] — инженерные сооружения в 200м
|
||
- nspd_dump: dict — freshness metadata
|
||
|
||
Если дамп отсутствует или устарел — вызывает harvest_quarter.apply_async()
|
||
(non-blocking) и возвращает пустые spatial поля + nspd_dump.available=False.
|
||
Если parcel_wkt=None — возвращает только freshness metadata (нет geom для
|
||
spatial queries).
|
||
"""
|
||
quarter = derive_quarter_cad(cad_num)
|
||
if quarter is None:
|
||
logger.warning("quarter_dump_lookup: cannot derive quarter from cad=%s", cad_num)
|
||
return make_empty_result()
|
||
|
||
# Читаем строку дампа из БД. Денормализованные счётчики слоёв используются
|
||
# для early-exit в spatial helpers (M2 mitigation).
|
||
row = db.execute(
|
||
text(
|
||
"""
|
||
SELECT quarter_cad,
|
||
fetched_at_utc,
|
||
total_features,
|
||
harvest_error,
|
||
territorial_zones_count,
|
||
zouit_count,
|
||
engineering_count,
|
||
risks_count,
|
||
COALESCE(opportunity_count, 0) AS opportunity_count,
|
||
COALESCE(red_lines_count, 0) AS red_lines_count
|
||
FROM nspd_quarter_dumps
|
||
WHERE quarter_cad = :q
|
||
"""
|
||
),
|
||
{"q": quarter},
|
||
).first()
|
||
|
||
now = datetime.now(UTC)
|
||
max_age = timedelta(days=_DUMP_MAX_AGE_DAYS)
|
||
|
||
if row is None:
|
||
# Дампа нет — ставим harvest в очередь и делаем fallback на cad_zouit (#243).
|
||
harvest_triggered = _trigger_harvest(quarter)
|
||
cad_zouit_overlaps: list[dict[str, Any]] = (
|
||
_get_cad_zouit_overlaps(db, parcel_wkt) if parcel_wkt is not None else []
|
||
)
|
||
result = make_empty_result(
|
||
harvest_triggered=harvest_triggered,
|
||
harvest_eta_seconds=_HARVEST_ETA_SECONDS if harvest_triggered else None,
|
||
)
|
||
result["nspd_zouit_overlaps"] = cad_zouit_overlaps
|
||
return result
|
||
|
||
fetched_at: datetime = row[1]
|
||
# Убедимся что timezone-aware для корректного сравнения
|
||
if fetched_at.tzinfo is None:
|
||
fetched_at = fetched_at.replace(tzinfo=UTC)
|
||
|
||
total_features: int | None = row[2]
|
||
harvest_error: str | None = row[3]
|
||
territorial_zones_count: int = row[4] or 0
|
||
zouit_count: int = row[5] or 0
|
||
engineering_count: int = row[6] or 0
|
||
risks_count: int = row[7] or 0
|
||
opportunity_count: int = row[8] or 0
|
||
red_lines_count: int = row[9] or 0
|
||
|
||
is_stale = (now - fetched_at) > max_age
|
||
has_error = harvest_error is not None
|
||
|
||
# Устаревший или с ошибкой — триггерим повторный harvest
|
||
if is_stale or has_error:
|
||
harvest_triggered = _trigger_harvest(quarter)
|
||
return make_empty_result(
|
||
fetched_at_utc=fetched_at.isoformat(),
|
||
stale=is_stale,
|
||
harvest_triggered=harvest_triggered,
|
||
total_features=total_features,
|
||
harvest_eta_seconds=_HARVEST_ETA_SECONDS if harvest_triggered else None,
|
||
)
|
||
|
||
# Свежий дамп без ошибок — извлекаем spatial данные
|
||
dump_meta: dict[str, Any] = {
|
||
"available": True,
|
||
"fetched_at_utc": fetched_at.isoformat(),
|
||
"stale": False,
|
||
"harvest_triggered": False,
|
||
"total_features": total_features,
|
||
"harvest_eta_seconds": None,
|
||
}
|
||
|
||
if parcel_wkt is None:
|
||
# Нет геометрии участка — возвращаем только метаданные
|
||
return {
|
||
"nspd_zoning": None,
|
||
"nspd_zouit_overlaps": [],
|
||
"nspd_engineering_nearby": [],
|
||
"nspd_risk_zones": [],
|
||
"nspd_opportunity_parcels": [],
|
||
"nspd_red_lines": [],
|
||
"nspd_dump": dump_meta,
|
||
}
|
||
|
||
layer_counts = {
|
||
"territorial_zones_count": territorial_zones_count,
|
||
"zouit_count": zouit_count,
|
||
"engineering_count": engineering_count,
|
||
"risks_count": risks_count,
|
||
"opportunity_count": opportunity_count,
|
||
"red_lines_count": red_lines_count,
|
||
}
|
||
nspd_zoning = _get_zoning(db, quarter, parcel_wkt, layer_counts)
|
||
nspd_zouit = _get_zouit_overlaps(db, quarter, parcel_wkt, layer_counts)
|
||
nspd_engineering = _get_engineering_nearby(db, quarter, parcel_wkt, layer_counts)
|
||
nspd_risk_zones = _get_risk_zones(db, quarter, parcel_wkt, layer_counts)
|
||
nspd_opportunity = _get_opportunity_parcels(db, quarter, parcel_wkt, layer_counts)
|
||
nspd_red_lines = _get_red_lines(db, quarter, parcel_wkt, layer_counts)
|
||
|
||
return {
|
||
"nspd_zoning": nspd_zoning,
|
||
"nspd_zouit_overlaps": nspd_zouit,
|
||
"nspd_engineering_nearby": nspd_engineering,
|
||
"nspd_risk_zones": nspd_risk_zones,
|
||
"nspd_opportunity_parcels": nspd_opportunity,
|
||
"nspd_red_lines": nspd_red_lines,
|
||
"nspd_dump": dump_meta,
|
||
}
|
||
|
||
|
||
# ── Spatial helpers ───────────────────────────────────────────────────────────
|
||
|
||
|
||
def _get_zoning(
|
||
db: Session,
|
||
quarter: str,
|
||
parcel_wkt: str,
|
||
layer_counts: dict[str, int] | None = None,
|
||
) -> dict[str, Any] | None:
|
||
"""G1: ПЗЗ территориальная зона по centroid участка из dump.
|
||
|
||
Geometry в features_json в EPSG:3857 — ST_Transform на чтении → 4326.
|
||
layer_counts — денормализованные счётчики из строки дампа. Если
|
||
territorial_zones_count == 0 — пропускаем heavy jsonb_array_elements scan.
|
||
"""
|
||
if layer_counts is not None and layer_counts.get("territorial_zones_count", 1) == 0:
|
||
return None
|
||
try:
|
||
row = db.execute(
|
||
text(
|
||
"""
|
||
SELECT feat.value->'properties' AS zone_props
|
||
FROM nspd_quarter_dumps d,
|
||
jsonb_array_elements(d.features_json) AS feat(value)
|
||
WHERE d.quarter_cad = :q
|
||
AND feat.value->>'layer' = 'territorial_zones'
|
||
AND (feat.value->'geometry') IS NOT NULL
|
||
AND feat.value->>'geometry' != 'null'
|
||
AND ST_Intersects(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
),
|
||
ST_Centroid(ST_GeomFromText(:wkt, 4326))
|
||
)
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"q": quarter, "wkt": parcel_wkt},
|
||
).first()
|
||
|
||
if row is None:
|
||
return None
|
||
|
||
props: dict[str, Any] = row[0] if isinstance(row[0], dict) else {}
|
||
zone_code = props.get("reg_numb_border") or props.get("zone_code") or props.get("name")
|
||
zone_name = props.get("type_zone") or props.get("zone_name") or props.get("name")
|
||
return {
|
||
"zone_code": zone_code,
|
||
"zone_name": zone_name,
|
||
"source": "nspd-quarter-dump",
|
||
"raw_props": props,
|
||
}
|
||
except Exception as e:
|
||
logger.warning("nspd zoning query failed for quarter=%s: %s", quarter, e)
|
||
return None
|
||
|
||
|
||
def _get_zouit_overlaps(
|
||
db: Session,
|
||
quarter: str,
|
||
parcel_wkt: str,
|
||
layer_counts: dict[str, int] | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
"""G3: ЗОУИТ которые пересекают участок.
|
||
|
||
Проверяем 5 групп: zouit_okn, zouit_engineering, zouit_natural,
|
||
zouit_protected, zouit_other.
|
||
layer_counts — денормализованные счётчики. Если zouit_count == 0 —
|
||
пропускаем nspd dump scan и делаем fallback на cad_zouit (#232).
|
||
"""
|
||
nspd_has_zouit = layer_counts is None or layer_counts.get("zouit_count", 1) != 0
|
||
if nspd_has_zouit:
|
||
try:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT feat.value->>'layer' AS layer,
|
||
feat.value->'properties' AS props
|
||
FROM nspd_quarter_dumps d,
|
||
jsonb_array_elements(d.features_json) AS feat(value)
|
||
WHERE d.quarter_cad = :q
|
||
AND feat.value->>'layer' LIKE 'zouit_%'
|
||
AND (feat.value->'geometry') IS NOT NULL
|
||
AND feat.value->>'geometry' != 'null'
|
||
AND ST_Intersects(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
),
|
||
ST_GeomFromText(:wkt, 4326)
|
||
)
|
||
"""
|
||
),
|
||
{"q": quarter, "wkt": parcel_wkt},
|
||
).fetchall()
|
||
|
||
result: list[dict[str, Any]] = []
|
||
for r in rows:
|
||
layer: str = r[0] or ""
|
||
props: dict[str, Any] = r[1] if isinstance(r[1], dict) else {}
|
||
group_key = layer.removeprefix("zouit_")
|
||
result.append(
|
||
{
|
||
"group_key": group_key,
|
||
"layer": layer,
|
||
"subcategory": props.get("subcategory") or props.get("type_zone"),
|
||
"name": props.get("name") or props.get("object_name"),
|
||
"raw_props": props,
|
||
"source": "nspd-quarter-dump",
|
||
}
|
||
)
|
||
return result
|
||
except Exception as e:
|
||
logger.warning("nspd zouit query failed for quarter=%s: %s", quarter, e)
|
||
return []
|
||
|
||
# Fallback: cad_zouit (#232) — nspd dump пуст по ЗОУИТ.
|
||
# cad_zouit.subcategory = 100% NULL (verified), используем type_zone keywords.
|
||
return _get_cad_zouit_overlaps(db, parcel_wkt)
|
||
|
||
|
||
def _get_cad_zouit_overlaps(db: Session, parcel_wkt: str) -> list[dict[str, Any]]:
|
||
"""Fallback: читать ЗОУИТ из cad_zouit когда nspd_quarter_dumps пуст (#232).
|
||
|
||
Использует GIST index cad_zouit_geom_gist через ST_Intersects.
|
||
Возвращает тот же формат что nspd-dump path (с доп. полем source='cad_zouit').
|
||
"""
|
||
try:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT type_zone,
|
||
category_name,
|
||
name_by_doc AS name,
|
||
reg_numb_border,
|
||
id AS zouit_id
|
||
FROM cad_zouit
|
||
WHERE ST_Intersects(geom, ST_GeomFromText(:wkt, 4326))
|
||
ORDER BY id
|
||
"""
|
||
),
|
||
{"wkt": parcel_wkt},
|
||
).fetchall()
|
||
|
||
result: list[dict[str, Any]] = []
|
||
for r in rows:
|
||
type_zone: str = r[0] or ""
|
||
result.append(
|
||
{
|
||
"group_key": "cad_zouit",
|
||
"layer": type_zone,
|
||
"subcategory": None, # subcategory = 100% NULL в cad_zouit
|
||
"name": r[2],
|
||
"raw_props": {
|
||
"type_zone": type_zone,
|
||
"category_name": r[1],
|
||
"reg_numb_border": r[3],
|
||
"zouit_id": r[4],
|
||
},
|
||
"source": "cad_zouit",
|
||
"type_zone": type_zone,
|
||
}
|
||
)
|
||
logger.info("cad_zouit fallback: found %d overlaps for wkt prefix", len(result))
|
||
return result
|
||
except Exception as e:
|
||
logger.warning("cad_zouit fallback query failed: %s", e)
|
||
return []
|
||
|
||
|
||
def _get_engineering_nearby(
|
||
db: Session,
|
||
quarter: str,
|
||
parcel_wkt: str,
|
||
layer_counts: dict[str, int] | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
"""I3: Инженерные сооружения в радиусе _ENGINEERING_RADIUS_M от centroid участка.
|
||
|
||
layer_counts — денормализованные счётчики. Если engineering_count == 0 —
|
||
пропускаем heavy jsonb_array_elements + ST_DWithin scan.
|
||
"""
|
||
if layer_counts is not None and layer_counts.get("engineering_count", 1) == 0:
|
||
return []
|
||
try:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT feat.value->'properties' AS props,
|
||
ST_Distance(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)::geography,
|
||
ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
|
||
) AS distance_m
|
||
FROM nspd_quarter_dumps d,
|
||
jsonb_array_elements(d.features_json) AS feat(value)
|
||
WHERE d.quarter_cad = :q
|
||
AND feat.value->>'layer' = 'engineering_structures'
|
||
AND (feat.value->'geometry') IS NOT NULL
|
||
AND feat.value->>'geometry' != 'null'
|
||
AND ST_DWithin(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)::geography,
|
||
ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
|
||
:radius_m
|
||
)
|
||
ORDER BY distance_m ASC
|
||
LIMIT 20
|
||
"""
|
||
),
|
||
{"q": quarter, "wkt": parcel_wkt, "radius_m": _ENGINEERING_RADIUS_M},
|
||
).fetchall()
|
||
|
||
result: list[dict[str, Any]] = []
|
||
for r in rows:
|
||
props: dict[str, Any] = r[0] if isinstance(r[0], dict) else {}
|
||
distance_m = float(r[1]) if r[1] is not None else None
|
||
name = props.get("name") or props.get("object_name")
|
||
obj_type = props.get("object_type") or props.get("type_zone")
|
||
result.append(
|
||
{
|
||
"name": name,
|
||
"type": obj_type,
|
||
"distance_m": round(distance_m) if distance_m is not None else None,
|
||
"raw_props": props,
|
||
}
|
||
)
|
||
return result
|
||
except Exception as e:
|
||
logger.warning("nspd engineering query failed for quarter=%s: %s", quarter, e)
|
||
return []
|
||
|
||
|
||
# ── Risk zones (issue #94 TIER 3) ────────────────────────────────────────────
|
||
|
||
# Human-readable subtype labels for risk layer keys.
|
||
# Keys match NSPDClient.QUARTER_RISK_LAYERS short_names.
|
||
_RISK_SUBTYPE_LABELS: dict[str, str] = {
|
||
"flooding_underground": "Подтопление",
|
||
"flooding": "Затопление",
|
||
"swampification": "Заболачивание",
|
||
"landslide": "Обвально-осыпные процессы",
|
||
"abrasion": "Абразия",
|
||
"erosion_water": "Водная эрозия",
|
||
"erosion_linear": "Линейная эрозия",
|
||
"erosion_wind": "Ветровая эрозия",
|
||
"desertification": "Опустынивание",
|
||
"clutter": "Захламление",
|
||
"burns": "Гари",
|
||
}
|
||
|
||
|
||
def _extract_features_by_layer(
|
||
features: list[dict[str, Any]],
|
||
layer_prefix: str,
|
||
) -> list[dict[str, Any]]:
|
||
"""Generic filter: вернуть features у которых layer начинается с layer_prefix.
|
||
|
||
Предназначен для re-use в PR 2+ (opportunity layers, etc.).
|
||
|
||
Args:
|
||
features: list of feature dicts из features_json (уже загруженных в Python).
|
||
layer_prefix: префикс для фильтрации, e.g. 'risk_', 'zouit_'.
|
||
|
||
Returns:
|
||
Подсписок features где feat['layer'].startswith(layer_prefix).
|
||
"""
|
||
return [
|
||
f
|
||
for f in features
|
||
if isinstance(f.get("layer"), str) and f["layer"].startswith(layer_prefix)
|
||
]
|
||
|
||
|
||
def _get_risk_zones(
|
||
db: Session,
|
||
quarter: str,
|
||
parcel_wkt: str,
|
||
layer_counts: dict[str, int] | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
"""TIER 3: риск-зоны НСПД (11 слоёв), пересекающие участок.
|
||
|
||
Возвращает список риск-зон с полями:
|
||
layer, subtype, geom_wkt, intersection_area_sqm.
|
||
|
||
layer_counts — денормализованные счётчики. Если risks_count == 0 —
|
||
пропускаем heavy jsonb_array_elements scan.
|
||
"""
|
||
if layer_counts is not None and layer_counts.get("risks_count", 1) == 0:
|
||
return []
|
||
try:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT feat.value->>'layer' AS layer,
|
||
feat.value->'properties' AS props,
|
||
ST_AsText(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)
|
||
) AS geom_wkt,
|
||
ST_Area(
|
||
ST_Intersection(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)::geography,
|
||
ST_GeomFromText(:wkt, 4326)::geography
|
||
)
|
||
) AS intersection_area_sqm
|
||
FROM nspd_quarter_dumps d,
|
||
jsonb_array_elements(d.features_json) AS feat(value)
|
||
WHERE d.quarter_cad = :q
|
||
AND feat.value->>'layer' LIKE 'risk_%'
|
||
AND (feat.value->'geometry') IS NOT NULL
|
||
AND feat.value->>'geometry' != 'null'
|
||
AND ST_Intersects(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
),
|
||
ST_GeomFromText(:wkt, 4326)
|
||
)
|
||
ORDER BY layer, intersection_area_sqm DESC NULLS LAST
|
||
"""
|
||
),
|
||
{"q": quarter, "wkt": parcel_wkt},
|
||
).fetchall()
|
||
|
||
result: list[dict[str, Any]] = []
|
||
for r in rows:
|
||
layer: str = r[0] or ""
|
||
props: dict[str, Any] = r[1] if isinstance(r[1], dict) else {}
|
||
geom_wkt_val: str | None = r[2]
|
||
raw_area: Any = r[3]
|
||
intersection_area_sqm: float | None = None
|
||
if raw_area is not None:
|
||
try:
|
||
val = float(raw_area)
|
||
intersection_area_sqm = round(val, 1) if not math.isnan(val) else None
|
||
except (TypeError, ValueError):
|
||
pass
|
||
|
||
# Извлечь short_name из "risk_<short_name>" для человекочитаемого subtype
|
||
short_name = layer.removeprefix("risk_")
|
||
# Попытка взять subtype из properties, затем из mapping
|
||
subtype = (
|
||
props.get("type_zone")
|
||
or props.get("subcategory")
|
||
or props.get("name")
|
||
or _RISK_SUBTYPE_LABELS.get(short_name)
|
||
)
|
||
if subtype is not None:
|
||
subtype = str(subtype)
|
||
|
||
result.append(
|
||
{
|
||
"layer": layer,
|
||
"subtype": subtype,
|
||
"geom_wkt": geom_wkt_val,
|
||
"intersection_area_sqm": intersection_area_sqm,
|
||
}
|
||
)
|
||
return result
|
||
except Exception as e:
|
||
logger.warning("nspd risk zones query failed for quarter=%s: %s", quarter, e)
|
||
return []
|
||
|
||
|
||
# ── Opportunity parcels (issue #94 TIER 4) ───────────────────────────────────
|
||
|
||
# Human-readable type labels for opportunity layer short_names.
|
||
_OPPORTUNITY_TYPE_LABELS: dict[str, str] = {
|
||
"auction_parcels": "auction_parcels",
|
||
"scheme_parcels": "scheme_parcels",
|
||
"free_parcels": "free_parcels",
|
||
"future_parcels": "future_parcels",
|
||
"oopt": "oopt",
|
||
}
|
||
|
||
# Radius (m) for opportunity parcel proximity search from parcel centroid.
|
||
_OPPORTUNITY_RADIUS_M = 500
|
||
|
||
|
||
def _get_opportunity_parcels(
|
||
db: Session,
|
||
quarter: str,
|
||
parcel_wkt: str,
|
||
layer_counts: dict[str, int] | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
"""TIER 4: opportunity parcels вблизи участка (auction, scheme, free, future, oopt).
|
||
|
||
Возвращает список ближайших opportunity ЗУ с полями:
|
||
layer, cad_num, distance_m, geom_wkt.
|
||
|
||
layer_counts — денормализованные счётчики. Если opportunity_count == 0 —
|
||
пропускаем heavy jsonb_array_elements scan (early-exit pattern как у risks).
|
||
"""
|
||
if layer_counts is not None and layer_counts.get("opportunity_count", 1) == 0:
|
||
return []
|
||
try:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT feat.value->>'layer' AS layer,
|
||
feat.value->'properties' AS props,
|
||
ST_AsText(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)
|
||
) AS geom_wkt,
|
||
ST_Distance(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)::geography,
|
||
ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
|
||
) AS distance_m
|
||
FROM nspd_quarter_dumps d,
|
||
jsonb_array_elements(d.features_json) AS feat(value)
|
||
WHERE d.quarter_cad = :q
|
||
AND feat.value->>'layer' LIKE 'opportunity_%'
|
||
AND (feat.value->'geometry') IS NOT NULL
|
||
AND feat.value->>'geometry' != 'null'
|
||
AND ST_DWithin(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)::geography,
|
||
ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
|
||
:radius_m
|
||
)
|
||
ORDER BY distance_m ASC
|
||
LIMIT 30
|
||
"""
|
||
),
|
||
{"q": quarter, "wkt": parcel_wkt, "radius_m": _OPPORTUNITY_RADIUS_M},
|
||
).fetchall()
|
||
|
||
result: list[dict[str, Any]] = []
|
||
for r in rows:
|
||
raw_layer: str = r[0] or ""
|
||
props: dict[str, Any] = r[1] if isinstance(r[1], dict) else {}
|
||
geom_wkt_val: str | None = r[2]
|
||
distance_m = float(r[3]) if r[3] is not None else None
|
||
|
||
# Strip "opportunity_" prefix to get short_name → type label
|
||
short_name = raw_layer.removeprefix("opportunity_")
|
||
layer_type = _OPPORTUNITY_TYPE_LABELS.get(short_name, short_name)
|
||
|
||
cad_num = props.get("cad_num") or props.get("cadastral_number")
|
||
|
||
result.append(
|
||
{
|
||
"layer": layer_type,
|
||
"cad_num": cad_num,
|
||
"distance_m": round(distance_m, 1) if distance_m is not None else None,
|
||
"geom_wkt": geom_wkt_val,
|
||
}
|
||
)
|
||
return result
|
||
except Exception as e:
|
||
logger.warning("nspd opportunity parcels query failed for quarter=%s: %s", quarter, e)
|
||
return []
|
||
|
||
|
||
# Radius (m) for red lines proximity search — beyond intersect check.
|
||
_RED_LINES_NEARBY_M = 200
|
||
|
||
|
||
def _get_red_lines(
|
||
db: Session,
|
||
quarter: str,
|
||
parcel_wkt: str,
|
||
layer_counts: dict[str, int] | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
"""TIER 4: красные линии застройки (layer 879243).
|
||
|
||
Возвращает red lines пересекающие участок ИЛИ ближайшие (до 200м) с полями:
|
||
geom_wkt, intersection_length_m, distance_m.
|
||
|
||
Логика:
|
||
- intersecting: intersection_length_m >= 0, distance_m = None
|
||
- nearby only: intersection_length_m = None, distance_m = расстояние
|
||
|
||
ST_Intersection выполняется в planar EPSG:4326, затем результат кастуется
|
||
в ::geography для ST_Length — это избегает PostGIS 3.4 tolerance bug
|
||
(ST_Intersection geography × geography на LINESTRING бросает transform error).
|
||
|
||
layer_counts — денормализованные счётчики. Если red_lines_count == 0 —
|
||
пропускаем heavy jsonb_array_elements scan.
|
||
"""
|
||
if layer_counts is not None and layer_counts.get("red_lines_count", 1) == 0:
|
||
return []
|
||
try:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT ST_AsText(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)
|
||
) AS geom_wkt,
|
||
ST_Intersects(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
),
|
||
ST_GeomFromText(:wkt, 4326)
|
||
) AS does_intersect,
|
||
ST_Length(
|
||
ST_Intersection(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
),
|
||
ST_GeomFromText(:wkt, 4326)
|
||
)::geography
|
||
) AS intersection_length_m,
|
||
ST_Distance(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)::geography,
|
||
ST_GeomFromText(:wkt, 4326)::geography
|
||
) AS distance_m
|
||
FROM nspd_quarter_dumps d,
|
||
jsonb_array_elements(d.features_json) AS feat(value)
|
||
WHERE d.quarter_cad = :q
|
||
AND feat.value->>'layer' = 'red_lines'
|
||
AND (feat.value->'geometry') IS NOT NULL
|
||
AND feat.value->>'geometry' != 'null'
|
||
AND ST_DWithin(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)::geography,
|
||
ST_GeomFromText(:wkt, 4326)::geography,
|
||
:nearby_m
|
||
)
|
||
ORDER BY distance_m ASC
|
||
LIMIT 50
|
||
"""
|
||
),
|
||
{"q": quarter, "wkt": parcel_wkt, "nearby_m": _RED_LINES_NEARBY_M},
|
||
).fetchall()
|
||
|
||
result: list[dict[str, Any]] = []
|
||
for r in rows:
|
||
geom_wkt_val: str | None = r[0]
|
||
does_intersect: bool = bool(r[1]) if r[1] is not None else False
|
||
raw_length: Any = r[2]
|
||
raw_dist: Any = r[3]
|
||
|
||
intersection_length_m: float | None = None
|
||
if does_intersect and raw_length is not None:
|
||
try:
|
||
val = float(raw_length)
|
||
intersection_length_m = round(val, 1) if not math.isnan(val) else None
|
||
except (TypeError, ValueError):
|
||
pass
|
||
|
||
distance_m: float | None = None
|
||
if not does_intersect and raw_dist is not None:
|
||
try:
|
||
distance_m = round(float(raw_dist), 1)
|
||
except (TypeError, ValueError):
|
||
pass
|
||
|
||
result.append(
|
||
{
|
||
"geom_wkt": geom_wkt_val,
|
||
"intersection_length_m": intersection_length_m,
|
||
"distance_m": distance_m,
|
||
}
|
||
)
|
||
return result
|
||
except Exception as e:
|
||
logger.warning("nspd red lines query failed for quarter=%s: %s", quarter, e)
|
||
return []
|
||
|
||
|
||
# ── Connection-points lookup (issue #115) ────────────────────────────────────
|
||
|
||
|
||
def get_connection_points(db: Session, cad_num: str, radius_m: int = 500) -> dict[str, Any]:
|
||
"""Получить точки инженерных подключений в radius_m от boundary участка.
|
||
|
||
Источник: `nspd_quarter_dumps.features_json` для квартала cad_num —
|
||
layers `engineering_structures` (NSPD cat 36328, ТП/ЦТП/насосные/опоры ЛЭП)
|
||
и `zouit_engineering` (NSPD cat 37578 — охранные зоны инжен.коммуникаций).
|
||
|
||
Args:
|
||
db: SQLAlchemy session.
|
||
cad_num: кадастровый номер участка (e.g. '66:41:0204016:10').
|
||
radius_m: радиус поиска в метрах от boundary участка (default 500).
|
||
|
||
Returns:
|
||
{
|
||
"engineering_structures": [...],
|
||
"zouit_engineering_overlaps": [...],
|
||
"summary": {...},
|
||
"dump_available": bool,
|
||
"dump_fetched_at": str | None,
|
||
}
|
||
|
||
Raises:
|
||
ValueError: если parcel не найден в cad_parcels_geom / cad_quarters_geom.
|
||
"""
|
||
quarter = derive_quarter_cad(cad_num)
|
||
if quarter is None:
|
||
raise ValueError(f"Невалидный формат кадастрового номера: {cad_num!r}")
|
||
|
||
# Получаем WKT геометрию участка (boundary, не centroid)
|
||
parcel_wkt = _get_parcel_wkt(db, cad_num)
|
||
if parcel_wkt is None:
|
||
raise ValueError(f"Участок {cad_num!r} не найден в БД")
|
||
|
||
# Проверяем наличие дампа
|
||
dump_row = db.execute(
|
||
text(
|
||
"""
|
||
SELECT fetched_at_utc, total_features
|
||
FROM nspd_quarter_dumps
|
||
WHERE quarter_cad = :q
|
||
ORDER BY fetched_at_utc DESC
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"q": quarter},
|
||
).first()
|
||
|
||
if dump_row is None:
|
||
_trigger_harvest(quarter)
|
||
return {
|
||
"engineering_structures": [],
|
||
"zouit_engineering_overlaps": [],
|
||
"summary": {
|
||
"nearest_structure_distance_m": None,
|
||
"in_protection_zone": False,
|
||
"protection_zones_intersecting": 0,
|
||
"total_structures_in_radius": 0,
|
||
},
|
||
"dump_available": False,
|
||
"dump_fetched_at": None,
|
||
}
|
||
|
||
fetched_at = dump_row[0]
|
||
if fetched_at is not None and getattr(fetched_at, "isoformat", None):
|
||
dump_fetched_at: str | None = fetched_at.isoformat()
|
||
else:
|
||
dump_fetched_at = str(fetched_at) if fetched_at is not None else None
|
||
|
||
structures = _get_engineering_structures_by_boundary(db, quarter, parcel_wkt, radius_m)
|
||
zouit_overlaps = _get_zouit_engineering_overlaps(db, quarter, parcel_wkt)
|
||
|
||
nearest_dist: float | None = structures[0]["distance_to_boundary_m"] if structures else None
|
||
protection_count = len(zouit_overlaps)
|
||
|
||
return {
|
||
"engineering_structures": structures,
|
||
"zouit_engineering_overlaps": zouit_overlaps,
|
||
"summary": {
|
||
"nearest_structure_distance_m": nearest_dist,
|
||
"in_protection_zone": protection_count > 0,
|
||
"protection_zones_intersecting": protection_count,
|
||
"total_structures_in_radius": len(structures),
|
||
},
|
||
"dump_available": True,
|
||
"dump_fetched_at": dump_fetched_at,
|
||
}
|
||
|
||
|
||
def _get_parcel_wkt(db: Session, cad_num: str) -> str | None:
|
||
"""Получить WKT геометрию участка из cad_parcels_geom или fallback источников.
|
||
|
||
Игнорирует строки с geom IS NULL (data quality) — иначе ST_AsText(NULL)
|
||
вернёт SQL NULL → str(None) = "None" → ST_GeomFromText упадёт, ошибка
|
||
тихо проглотится в caller'е, клиент получит пустые массивы без причины.
|
||
"""
|
||
row = db.execute(
|
||
text(
|
||
"""
|
||
SELECT ST_AsText(geom) AS wkt
|
||
FROM cad_parcels_geom
|
||
WHERE cad_num = :c AND geom IS NOT NULL
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"c": cad_num},
|
||
).first()
|
||
if row is not None and row[0] is not None:
|
||
return str(row[0])
|
||
|
||
# Fallback: кварталы (более крупный объект — менее точно, но лучше чем ничего).
|
||
# cad_quarters_geom хранит 3-сегментные ключи (66:41:0204016), не 4-сегментные.
|
||
quarter_key = derive_quarter_cad(cad_num)
|
||
row = db.execute(
|
||
text(
|
||
"""
|
||
SELECT ST_AsText(geom) AS wkt
|
||
FROM cad_quarters_geom
|
||
WHERE cad_number = :c AND geom IS NOT NULL
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"c": quarter_key},
|
||
).first()
|
||
if row is not None and row[0] is not None:
|
||
return str(row[0])
|
||
|
||
return None
|
||
|
||
|
||
def _get_engineering_structures_by_boundary(
|
||
db: Session,
|
||
quarter: str,
|
||
parcel_wkt: str,
|
||
radius_m: int,
|
||
) -> list[dict[str, Any]]:
|
||
"""Engineering structures из dump в radius_m от boundary участка.
|
||
|
||
Использует ST_Distance к boundary (не centroid) для корректного расстояния.
|
||
Geometry трансформируется из EPSG:3857 (хранение dump) → 4326.
|
||
"""
|
||
try:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT feat.value->'properties' AS props,
|
||
feat.value->>'geometry' AS geom_json,
|
||
ST_Distance(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)::geography,
|
||
ST_GeomFromText(:wkt, 4326)::geography
|
||
) AS distance_m
|
||
FROM nspd_quarter_dumps d,
|
||
jsonb_array_elements(d.features_json) AS feat(value)
|
||
WHERE d.quarter_cad = :q
|
||
AND feat.value->>'layer' = 'engineering_structures'
|
||
AND (feat.value->'geometry') IS NOT NULL
|
||
AND feat.value->>'geometry' != 'null'
|
||
AND ST_DWithin(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
)::geography,
|
||
ST_GeomFromText(:wkt, 4326)::geography,
|
||
:radius_m
|
||
)
|
||
ORDER BY distance_m ASC
|
||
LIMIT 50
|
||
"""
|
||
),
|
||
{"q": quarter, "wkt": parcel_wkt, "radius_m": radius_m},
|
||
).fetchall()
|
||
except (OperationalError, ProgrammingError) as e:
|
||
logger.warning(
|
||
"engineering_structures query failed for quarter=%s: %s",
|
||
quarter,
|
||
e,
|
||
)
|
||
return []
|
||
|
||
result: list[dict[str, Any]] = []
|
||
for r in rows:
|
||
props: dict[str, Any] = r[0] if isinstance(r[0], dict) else {}
|
||
geom_raw: str | None = r[1]
|
||
distance_m = float(r[2]) if r[2] is not None else 0.0
|
||
|
||
# Попытка распарсить geometry как dict для GeoJSON поля
|
||
geom_dict: dict[str, Any] = {}
|
||
if geom_raw:
|
||
try:
|
||
geom_dict = json.loads(geom_raw)
|
||
except (ValueError, json.JSONDecodeError):
|
||
geom_dict = {}
|
||
|
||
result.append(
|
||
{
|
||
"name": props.get("name") or props.get("object_name"),
|
||
"type": props.get("purpose") or props.get("object_type") or props.get("type_zone"),
|
||
"cad_num": props.get("cad_num") or props.get("cadastral_number"),
|
||
"distance_to_boundary_m": round(distance_m, 1),
|
||
"geometry_geojson": geom_dict,
|
||
"readable_address": props.get("readable_address") or props.get("address"),
|
||
"raw_props": props,
|
||
"source": "nspd_36328",
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def _get_zouit_engineering_overlaps(
|
||
db: Session,
|
||
quarter: str,
|
||
parcel_wkt: str,
|
||
) -> list[dict[str, Any]]:
|
||
"""ZOUIT engineering (cat 37578) — охранные зоны, пересекающие участок.
|
||
|
||
Использует слой 'zouit_engineering' из dump.
|
||
"""
|
||
try:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT feat.value->'properties' AS props,
|
||
feat.value->>'geometry' AS geom_json
|
||
FROM nspd_quarter_dumps d,
|
||
jsonb_array_elements(d.features_json) AS feat(value)
|
||
WHERE d.quarter_cad = :q
|
||
AND feat.value->>'layer' = 'zouit_engineering'
|
||
AND (feat.value->'geometry') IS NOT NULL
|
||
AND feat.value->>'geometry' != 'null'
|
||
AND ST_Intersects(
|
||
ST_Transform(
|
||
ST_SetSRID(
|
||
ST_GeomFromGeoJSON(feat.value->>'geometry'),
|
||
3857
|
||
),
|
||
4326
|
||
),
|
||
ST_GeomFromText(:wkt, 4326)
|
||
)
|
||
LIMIT 100
|
||
"""
|
||
),
|
||
{"q": quarter, "wkt": parcel_wkt},
|
||
).fetchall()
|
||
except (OperationalError, ProgrammingError) as e:
|
||
logger.warning(
|
||
"zouit_engineering query failed for quarter=%s: %s",
|
||
quarter,
|
||
e,
|
||
)
|
||
return []
|
||
|
||
result: list[dict[str, Any]] = []
|
||
for r in rows:
|
||
props: dict[str, Any] = r[0] if isinstance(r[0], dict) else {}
|
||
geom_raw: str | None = r[1]
|
||
|
||
geom_dict: dict[str, Any] = {}
|
||
if geom_raw:
|
||
try:
|
||
geom_dict = json.loads(geom_raw)
|
||
except (ValueError, json.JSONDecodeError):
|
||
geom_dict = {}
|
||
|
||
subcategory_raw = props.get("subcategory")
|
||
subcategory: int | None = None
|
||
if subcategory_raw is not None:
|
||
try:
|
||
subcategory = int(subcategory_raw)
|
||
except (ValueError, TypeError):
|
||
subcategory = None
|
||
|
||
result.append(
|
||
{
|
||
"reg_numb_border": props.get("reg_numb_border"),
|
||
"type_zone": props.get("type_zone") or props.get("zone_name"),
|
||
"subcategory": subcategory,
|
||
"intersects_parcel": True,
|
||
"geometry_geojson": geom_dict,
|
||
"raw_props": props,
|
||
"source": "nspd_37578",
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
# ── Harvest trigger ───────────────────────────────────────────────────────────
|
||
|
||
|
||
def _acquire_harvest_lock(quarter: str) -> bool:
|
||
"""Redis SETNX lock на quarter с TTL=_HARVEST_LOCK_TTL_SECONDS.
|
||
|
||
Issue #234: защита от burst N concurrent analyze_parcel запросов — без lock
|
||
каждый из них вызывал бы harvest_quarter.apply_async() и слал бы N
|
||
идентичных задач в Celery, тратя WAF traffic впустую (UPSERT идемпотентен,
|
||
но трафик не вернёшь).
|
||
|
||
Returns:
|
||
True — lock acquired (caller должен enqueue task)
|
||
False — lock уже взят другим запросом ИЛИ redis недоступен (graceful
|
||
degradation: пропускаем дедуп, возвращаем False чтобы caller тоже не
|
||
триггерил — иначе одиночный redis-fail вернул бы поведение до фикса).
|
||
"""
|
||
try:
|
||
import redis
|
||
|
||
from app.core.config import settings
|
||
|
||
client = redis.Redis.from_url(settings.redis_url)
|
||
# SET key value NX EX — атомарно: «установи если не существует, TTL N сек».
|
||
acquired = client.set(
|
||
f"nspd_harvest_lock:{quarter}",
|
||
"1",
|
||
nx=True,
|
||
ex=_HARVEST_LOCK_TTL_SECONDS,
|
||
)
|
||
return bool(acquired)
|
||
except Exception as e:
|
||
# Redis недоступен — лучше не запускать дубль, чем нагрузить WAF.
|
||
logger.warning("redis SETNX lock failed for quarter=%s: %s", quarter, e)
|
||
return False
|
||
|
||
|
||
def _trigger_harvest(quarter: str) -> bool:
|
||
"""Fire-and-forget harvest_quarter.apply_async() с Redis SETNX dedupe.
|
||
|
||
Ленивый импорт чтобы избежать circular import (tasks → services → tasks).
|
||
|
||
Issue #234: burst N concurrent analyze дедуплицируется через
|
||
`_acquire_harvest_lock` — только первый запрос ставит task, остальные
|
||
получают `False` (т.е. harvest_triggered=False, harvest_eta_seconds=None).
|
||
FE при таком ответе показывает graceful no-data UI с ПКК fallback link
|
||
(не infinite spinner). Trade-off: Request B не видит countdown, но и не
|
||
сбивает работу Request A. Это намеренный design choice — см. vault
|
||
fixes/Bug_NSPD_WMS_NotBulk_2026_May14.md.
|
||
"""
|
||
if not _acquire_harvest_lock(quarter):
|
||
logger.info("quarter=%s harvest skipped (lock held by another request)", quarter)
|
||
return False
|
||
|
||
try:
|
||
from app.workers.tasks.nspd_sync import harvest_quarter
|
||
|
||
harvest_quarter.apply_async(
|
||
args=[quarter],
|
||
kwargs={
|
||
"region_code": 66,
|
||
"include_zouit": True,
|
||
"include_risks": True,
|
||
"include_opportunity": True,
|
||
},
|
||
)
|
||
logger.info("quarter dump harvest triggered for quarter=%s", quarter)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("failed to trigger harvest for quarter=%s: %s", quarter, e)
|
||
return False
|