Light1YT
d11c8d47e0
harvest_gknspecial_zones делал единственный commit в конце всех 8 prefix- циклов: WorkerLostError (OOM / рестарт хоста) в середине прогона → ~20k записанных строк откатывались, прогресс ноль. Mirror ird_harvest.py:216 (per-quarter commit): db.commit() после финального logger.info внутри for-prefix loop. SAVEPOINT-per-row уже изолирует битые фичи; per-prefix commit добавляет durability между префиксами. WFS-fail на префиксе → continue → commit не делается; предыдущие префиксы уже durable. +2 теста (durability, WFS-fail skip). 29 gknspecial тестов зелёные. Closes #1116
292 lines
13 KiB
Python
292 lines
13 KiB
Python
"""Celery task: harvest ЗОУИТ ЕКБ-геопортала (gknspecial_zone) → ``ird_overlays`` (м.132).
|
||
|
||
#1067 GG-форсайт: закрывает ИРД-пункты:
|
||
#1063 — сервитут (Зона публичного сервитута)
|
||
#1064 — водоохранные зоны (Водоохранная зона / Прибрежная защитная полоса /
|
||
Зона санитарной охраны / Санитарно-защитная зона)
|
||
#1065 — приаэродромная территория (Приаэродромная территория — ЕГРН)
|
||
#1066 — ОКН-зоны/ЗРЗ (Зона охраны объекта культурного / Зона регулирования застройки)
|
||
|
||
Источник: ЕКБ-геопортал, WFS-слой ``portal:portal_geo_gknspecial_zone`` (ключ 'zouit').
|
||
Геометрия уже в EPSG:4326 — ST_Transform НЕ нужен.
|
||
|
||
Sentinel source_layer_id'ы (вне диапазона НСПД): 990001–990008 + 990099 (unmatched).
|
||
|
||
Conventions (зеркало ird_harvest.py):
|
||
• SessionLocal() + try/finally close.
|
||
• SAVEPOINT per-row (with db.begin_nested():) — одна битая фича не валит батч.
|
||
• db.commit() per-prefix (durability) — WorkerLost в середине прогона не теряет
|
||
уже записанные префиксы (#1116).
|
||
• CAST(:x AS type) — НИКОГДА :x::type (psycopg v3).
|
||
• ON CONFLICT (source_layer_id, geom_data_id) DO UPDATE.
|
||
• content_* пишем мы же (геопортал как источник); content НЕ трогается НСПД-harvest'ом.
|
||
• Кириллица в combined-CQL WFS ломает запрос → фильтр zone_state_v client-side.
|
||
|
||
Расписание: beat_schedule.py → gknspecial-harvest-weekly (пн 05:30 МСК).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.scrapers.ekb_geoportal_client import EKBGeoportalClient
|
||
from app.workers.celery_app import celery_app
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Mapping: full_name-префикс → (layer_kind, sentinel source_layer_id) ─────
|
||
# Sentinels вне диапазона НСПД (990001+), чтобы не конфликтовать с реальными layer_id.
|
||
_PREFIX_MAP: list[tuple[str, str, int]] = [
|
||
("Зона публичного сервитута", "servitude", 990001), # #1063
|
||
("Водоохранная зона", "water_protection", 990002), # #1064
|
||
("Прибрежная защитная полоса", "coastal_strip", 990003), # #1064
|
||
("Зона санитарной охраны", "water_sanitary", 990004), # #1064
|
||
("Санитарно-защитная зона", "sanitary_protection", 990005), # #1064
|
||
("Зона охраны объекта культурного", "okn_protection_zone", 990006), # #1066
|
||
("Зона регулирования застройки", "zrz", 990007), # #1066
|
||
("Приаэродромная территория", "airfield", 990008), # #1065
|
||
]
|
||
|
||
# Sentinel для фич, чей full_name не попал ни под один префикс (defensive).
|
||
_UNMATCHED_LAYER_KIND = "gknspecial_unmatched"
|
||
_UNMATCHED_SOURCE_LAYER_ID = 990099
|
||
|
||
# Regex для извлечения reg_numb_border из full_name (формат «66:41-6.4125»).
|
||
_REG_NUMB_RE = re.compile(r"\d{2}:\d{2}-\d+\.\d+")
|
||
|
||
# UPSERT: геометрия уже EPSG:4326 → ST_SetSRID(ST_GeomFromGeoJSON, 4326), без ST_Transform.
|
||
# content_* поля обновляются в DO UPDATE (геопортал — их источник).
|
||
_UPSERT_SQL = text(
|
||
"""
|
||
INSERT INTO ird_overlays (
|
||
source_layer_id, layer_kind, geom_data_id, reg_numb_border,
|
||
category, subcategory, geom, raw_props,
|
||
zone_type_name, content, content_source, content_fetched_at, fetched_at
|
||
) VALUES (
|
||
CAST(:source_layer_id AS integer),
|
||
CAST(:layer_kind AS text),
|
||
CAST(:geom_data_id AS bigint),
|
||
CAST(:reg_numb_border AS text),
|
||
CAST(:category AS integer),
|
||
CAST(:subcategory AS integer),
|
||
ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geojson AS text)), 4326),
|
||
CAST(:raw_props AS jsonb),
|
||
CAST(:zone_type_name AS text),
|
||
CAST(:content AS jsonb),
|
||
CAST(:content_source AS text),
|
||
CAST(:content_fetched_at AS timestamptz),
|
||
CAST(:fetched_at AS timestamptz)
|
||
)
|
||
ON CONFLICT (source_layer_id, geom_data_id) DO UPDATE SET
|
||
layer_kind = EXCLUDED.layer_kind,
|
||
reg_numb_border = EXCLUDED.reg_numb_border,
|
||
geom = EXCLUDED.geom,
|
||
raw_props = EXCLUDED.raw_props,
|
||
zone_type_name = EXCLUDED.zone_type_name,
|
||
content = EXCLUDED.content,
|
||
content_source = EXCLUDED.content_source,
|
||
content_fetched_at = EXCLUDED.content_fetched_at,
|
||
fetched_at = EXCLUDED.fetched_at
|
||
"""
|
||
)
|
||
|
||
|
||
def _classify_full_name(full_name: str) -> tuple[str, int] | None:
|
||
"""Сопоставить full_name с записью _PREFIX_MAP → (layer_kind, source_layer_id) или None."""
|
||
for prefix, layer_kind, source_layer_id in _PREFIX_MAP:
|
||
if full_name.startswith(prefix):
|
||
return layer_kind, source_layer_id
|
||
return None
|
||
|
||
|
||
def _extract_reg_numb(full_name: str) -> str | None:
|
||
"""Извлечь reg_numb_border из full_name по паттерну \\d{2}:\\d{2}-\\d+\\.\\d+."""
|
||
m = _REG_NUMB_RE.search(full_name)
|
||
return m.group(0) if m else None
|
||
|
||
|
||
def _extract_geom_data_id(feature_id: str | None) -> int | None:
|
||
"""Trailing int из WFS feature_id (формат 'portal_geo_gknspecial_zone.NNNN') или None."""
|
||
if not feature_id:
|
||
return None
|
||
parts = feature_id.split(".")
|
||
if len(parts) < 2:
|
||
return None
|
||
try:
|
||
return int(parts[-1])
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
def _build_content(props: dict[str, Any]) -> str:
|
||
"""Собрать content-jsonb из свойств фичи (только присутствующие ключи, ensure_ascii=False)."""
|
||
mapping: dict[str, str] = {
|
||
"restrictions": "gkn_specialz_contentrestrictions",
|
||
"protected_object": "gkn_specialz_protectedobject",
|
||
"authority_decision": "authority_decision",
|
||
"document": "gkn_document",
|
||
"registration_date": "registration_date",
|
||
"area": "area",
|
||
"zone_state": "zone_state_v",
|
||
}
|
||
content: dict[str, Any] = {}
|
||
for out_key, prop_key in mapping.items():
|
||
val = props.get(prop_key)
|
||
if val is not None:
|
||
content[out_key] = val
|
||
return json.dumps(content, ensure_ascii=False)
|
||
|
||
|
||
def _upsert_feature(
|
||
db: Session,
|
||
*,
|
||
source_layer_id: int,
|
||
layer_kind: str,
|
||
geom_data_id: int,
|
||
props: dict[str, Any],
|
||
geom: dict[str, Any],
|
||
fetched_at: str,
|
||
) -> bool:
|
||
"""UPSERT одной фичи в ird_overlays под SAVEPOINT. True если записана."""
|
||
full_name: str = str(props.get("full_name") or "")
|
||
reg_numb = _extract_reg_numb(full_name)
|
||
content_json = _build_content(props)
|
||
raw_props_json = json.dumps(props, ensure_ascii=False)
|
||
try:
|
||
with db.begin_nested():
|
||
db.execute(
|
||
_UPSERT_SQL,
|
||
{
|
||
"source_layer_id": source_layer_id,
|
||
"layer_kind": layer_kind,
|
||
"geom_data_id": geom_data_id,
|
||
"reg_numb_border": reg_numb,
|
||
"category": None,
|
||
"subcategory": None,
|
||
"geojson": json.dumps(geom),
|
||
"raw_props": raw_props_json,
|
||
"zone_type_name": full_name or None,
|
||
"content": content_json,
|
||
"content_source": "ekb_geoportal",
|
||
"content_fetched_at": fetched_at,
|
||
"fetched_at": fetched_at,
|
||
},
|
||
)
|
||
return True
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"gknspecial_harvest: upsert failed layer=%s geom_data_id=%s: %s",
|
||
layer_kind,
|
||
geom_data_id,
|
||
exc,
|
||
)
|
||
return False
|
||
|
||
|
||
@celery_app.task(name="tasks.gknspecial_harvest.harvest_gknspecial_zones")
|
||
def harvest_gknspecial_zones() -> dict[str, int]:
|
||
"""Harvest ЗОУИТ-слоя ЕКБ-геопортала (gknspecial_zone) → ird_overlays.
|
||
|
||
Для каждого full_name-префикса запрашивает WFS с CQL ``full_name LIKE 'X%'``,
|
||
фильтрует архивные записи client-side (zone_state_v != 'Действующий'), UPSERT-ит
|
||
в ird_overlays под SAVEPOINT per-row.
|
||
|
||
Фичи, чей full_name не попал ни под один известный префикс, пишутся под
|
||
layer_kind='gknspecial_unmatched' (source_layer_id=990099) — defensive, чтобы
|
||
новые типы зон не терялись молча.
|
||
|
||
Returns:
|
||
{'types': N_prefixes_fetched, 'features': M_upserted}
|
||
"""
|
||
fetched_at = datetime.now(UTC).isoformat()
|
||
client = EKBGeoportalClient()
|
||
db = SessionLocal()
|
||
n_types = 0
|
||
n_features = 0
|
||
try:
|
||
for prefix, expected_layer_kind, _ in _PREFIX_MAP:
|
||
cql = f"full_name LIKE '{prefix}%'"
|
||
try:
|
||
feats = client.features_by_cql("zouit", cql)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"gknspecial_harvest: WFS failed prefix=%r layer_kind=%s: %s",
|
||
prefix,
|
||
expected_layer_kind,
|
||
exc,
|
||
)
|
||
continue
|
||
n_types += 1
|
||
n_written = 0
|
||
for feat in feats:
|
||
props = feat.properties or {}
|
||
# Фильтр активности client-side: пропускаем архивные зоны.
|
||
# Если поля нет вообще — не скипаем (неизвестное = считаем активным).
|
||
zone_state = props.get("zone_state_v")
|
||
if zone_state is not None and zone_state != "Действующий":
|
||
continue
|
||
geom = feat.geometry
|
||
if not geom:
|
||
logger.debug(
|
||
"gknspecial_harvest: нет геометрии layer=%s feature_id=%s, skip",
|
||
expected_layer_kind,
|
||
feat.feature_id,
|
||
)
|
||
continue
|
||
geom_data_id = _extract_geom_data_id(feat.feature_id)
|
||
if geom_data_id is None:
|
||
logger.debug(
|
||
"gknspecial_harvest: нет geom_data_id feature_id=%s, skip",
|
||
feat.feature_id,
|
||
)
|
||
continue
|
||
# Роутинг: классифицируем по full_name (единый источник истины).
|
||
# Defensive: WFS мог вернуть фичу с неожиданным full_name — пишем в unmatched.
|
||
full_name: str = str(props.get("full_name") or "")
|
||
cls = _classify_full_name(full_name)
|
||
if cls is None:
|
||
logger.warning(
|
||
"gknspecial_harvest: unmatched full_name=%r geom_data_id=%s, "
|
||
"пишем в unmatched",
|
||
full_name,
|
||
geom_data_id,
|
||
)
|
||
actual_layer_kind = _UNMATCHED_LAYER_KIND
|
||
actual_source_layer_id = _UNMATCHED_SOURCE_LAYER_ID
|
||
else:
|
||
actual_layer_kind, actual_source_layer_id = cls
|
||
if _upsert_feature(
|
||
db,
|
||
source_layer_id=actual_source_layer_id,
|
||
layer_kind=actual_layer_kind,
|
||
geom_data_id=geom_data_id,
|
||
props=props,
|
||
geom=geom,
|
||
fetched_at=fetched_at,
|
||
):
|
||
n_written += 1
|
||
n_features += 1
|
||
logger.info(
|
||
"gknspecial_harvest: prefix=%r layer_kind=%s fetched=%d written=%d",
|
||
prefix,
|
||
expected_layer_kind,
|
||
len(feats),
|
||
n_written,
|
||
)
|
||
# Durable per-prefix commit: WorkerLostError (OOM / рестарт хоста) в середине
|
||
# прогона не теряет уже записанные префиксы (#1116). Mirror ird_harvest.py.
|
||
db.commit()
|
||
|
||
db.commit() # финальный — no-op если всё уже коммитнуто per-prefix (подстраховка)
|
||
logger.info("gknspecial_harvest: готово, types=%d features=%d", n_types, n_features)
|
||
return {"types": n_types, "features": n_features}
|
||
finally:
|
||
db.close()
|