343 lines
16 KiB
Python
343 lines
16 KiB
Python
"""Celery task: ingest ОКН-объектов ЕКБ/Свердл.обл из АИС ЕГРКН (#1066, #1159).
|
||
|
||
Обходит расширенный список ``(address, category_type)``-проходов: помимо базовых
|
||
«Екатеринбург»/«Свердловская» (3 категории каждая) добавлены варианты подстрок
|
||
адреса — районы ЕКБ, городские округа, обиходные склонения области. Запрос
|
||
``Maps/searchMap`` фильтрует ОКН по подстроке в поле ``cob_address``: объекты,
|
||
у которых адрес начинается с улицы без слова «Екатеринбург», ловятся только
|
||
запросом по району/округу. Это закрывает coverage gap (428 vs ~798 ЕГРОКН).
|
||
|
||
Один объект может встречаться в нескольких проходах. Дедупликация по
|
||
``source_id``: первый проход устанавливает category (самая «высокая» категория
|
||
первой встретившейся). Порядок проходов: federal → regional → local;
|
||
«Екатеринбург» раньше областных подстрок → при конфликте source_id выбирается
|
||
наиболее значимая категория.
|
||
|
||
После сбора уникальных фич воркер догружает per-object detail-карточку
|
||
``Show/Show?id=…``: searchMap отдаёт ``properties={id}``, а название/адрес/датировка/
|
||
категория охраны лежат отдельно. ``raw_props`` обогащается полями ``name``,
|
||
``address``, ``protection_category``, ``object_kind``, ``dating``,
|
||
``protection_doc``, ``regnumber`` (#1159 acceptance 2). Detail-фейлы не валят
|
||
прогон — объект остаётся с базовыми полями (минимум {id}).
|
||
|
||
UPSERT: ON CONFLICT (source_id) DO UPDATE обновляет geom, raw_props, fetched_at
|
||
(category не меняется при повторном импорте — first wins). SAVEPOINT per-row через
|
||
``db.begin_nested()``.
|
||
|
||
Beat: еженедельно в воскресенье в 04:30 МСК — до ird-harvest (понедельник 05:00).
|
||
|
||
WAF-hygiene (#2445 item D3): пэйсинг и 429/5xx backoff между HTTP-запросами живут
|
||
в ``okn_egrkn_client`` (``_wait_rate_limit`` — общий для 33 list-проходов и до
|
||
``_DETAIL_FETCH_LIMIT`` detail-запросов, один WAF-хост). Здесь, в
|
||
``_enrich_with_detail``, — circuit breaker: ``_DETAIL_FAIL_BREAKER`` детейл-фейлов
|
||
ПОДРЯД → прерываем detail-loop (не долбим WAF до конца лимита при устойчивом бане),
|
||
аналогично ``gisogd66.py`` (``_CARD_FAIL_BREAKER``).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.scrapers.okn_egrkn_client import fetch_okn_detail, fetch_okn_points
|
||
from app.workers.celery_app import celery_app
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Маппинг cob_category_type → строковое название категории
|
||
_CATEGORY_MAP: dict[int, str] = {
|
||
1: "federal",
|
||
2: "regional",
|
||
4: "local",
|
||
}
|
||
|
||
# Категории по приоритету (federal → regional → local): порядок важен,
|
||
# при конфликте source_id у одного и того же объекта побеждает более высокая.
|
||
_CATEGORY_ORDER: list[int] = [1, 2, 4]
|
||
|
||
# Подстроки для cob_address. ЕГРКН не возвращает все объекты «Свердловской обл.»
|
||
# по одному запросу — обходим расширенным списком:
|
||
# - город целиком («Екатеринбург»),
|
||
# - области («Свердловская», «Свердловской области»),
|
||
# - районы ЕКБ — для записей без слова «Екатеринбург», только улица + район.
|
||
# Порядок: «Екатеринбург» раньше других → при дедупе district-вариант не
|
||
# перетирает category «городского» прохода.
|
||
_ADDRESS_PASSES: list[str] = [
|
||
"Екатеринбург",
|
||
"г. Екатеринбург",
|
||
"Кировский",
|
||
"Ленинский",
|
||
"Октябрьский",
|
||
"Орджоникидзевский",
|
||
"Чкаловский",
|
||
"Верх-Исетский",
|
||
"Железнодорожный",
|
||
"Свердловская",
|
||
"Свердловской области",
|
||
]
|
||
|
||
# Декартово произведение (address × category) — генерируется единожды при импорте.
|
||
# Категория федеральная идёт первой по каждому address: при конфликте source_id
|
||
# (тот же объект встречается в нескольких категориях) federal/regional перебивают
|
||
# local в той же address-strata.
|
||
_PASSES: list[tuple[str, int]] = [
|
||
(addr, cat) for addr in _ADDRESS_PASSES for cat in _CATEGORY_ORDER
|
||
]
|
||
|
||
# Лимит per-run на detail-fetch. ~700 объектов * ~0.3 s сетевой задержки ≈ 4 min,
|
||
# плюс retries — приемлемо для weekly cron. Жёсткий cap защищает от runaway
|
||
# при сетевом шторме.
|
||
_DETAIL_FETCH_LIMIT = 1500
|
||
|
||
# Логгировать прогресс detail-fetch каждые N объектов.
|
||
_DETAIL_PROGRESS_EVERY = 100
|
||
|
||
# Circuit breaker (#2445 item D3, аналогично gisogd66._CARD_FAIL_BREAKER): столько
|
||
# detail-фейлов ПОДРЯД (raise ИЛИ пустой detail считается "не подряд", см. ниже)
|
||
# → прерываем detail-loop. WAF-бан обычно даёт устойчивую серию ошибок; долбить
|
||
# оставшиеся до ~1500 detail-запросов бессмысленно и рискованно для IP-репутации.
|
||
_DETAIL_FAIL_BREAKER = 20
|
||
|
||
_UPSERT_SQL = text("""
|
||
INSERT INTO okn_objects (source_id, category, geom, raw_props, fetched_at)
|
||
VALUES (
|
||
CAST(:source_id AS text),
|
||
CAST(:category AS text),
|
||
ST_SetSRID(
|
||
ST_MakePoint(
|
||
CAST(:lon AS double precision),
|
||
CAST(:lat AS double precision)
|
||
),
|
||
4326
|
||
),
|
||
CAST(:raw_props AS jsonb),
|
||
now()
|
||
)
|
||
ON CONFLICT (source_id) DO UPDATE SET
|
||
geom = EXCLUDED.geom,
|
||
raw_props = EXCLUDED.raw_props,
|
||
fetched_at = EXCLUDED.fetched_at
|
||
""")
|
||
|
||
|
||
def _extract_coords(feature: dict[str, Any]) -> tuple[float, float] | None:
|
||
"""Извлечь (lon, lat) из фичи ЕГРКН.
|
||
|
||
ЕГРКН возвращает coordinates в порядке [lat, lon] (Яндекс-конвенция).
|
||
Для PostGIS берём: lon=coordinates[1], lat=coordinates[0].
|
||
|
||
Валидный bbox ЕКБ + ближайшего окружения (55–58°N, 59–62°E).
|
||
"""
|
||
try:
|
||
geom = feature.get("geometry") or {}
|
||
coords = geom.get("coordinates")
|
||
if not coords or len(coords) < 2:
|
||
return None
|
||
lat = float(coords[0])
|
||
lon = float(coords[1])
|
||
# Базовая санитарная проверка — координаты должны попасть в Урал
|
||
if not (55.0 <= lat <= 58.5 and 58.0 <= lon <= 63.0):
|
||
logger.warning(
|
||
"okn_objects_sync: подозрительные координаты lat=%.4f lon=%.4f, пропускаем",
|
||
lat,
|
||
lon,
|
||
)
|
||
return None
|
||
return lon, lat
|
||
except (TypeError, ValueError, KeyError) as exc:
|
||
logger.warning("okn_objects_sync: ошибка разбора координат фичи: %s", exc)
|
||
return None
|
||
|
||
|
||
def _enrich_with_detail(
|
||
to_upsert: list[dict[str, Any]],
|
||
*,
|
||
detail_fetcher: Any = fetch_okn_detail,
|
||
limit: int = _DETAIL_FETCH_LIMIT,
|
||
fail_breaker: int = _DETAIL_FAIL_BREAKER,
|
||
) -> int:
|
||
"""Обогатить raw_props detail-полями (name, address, …).
|
||
|
||
Принимает список row-dict'ов, уже подготовленных к UPSERT. Для каждой
|
||
строки вызывает ``detail_fetcher(source_id)`` и сливает результат в JSON
|
||
``raw_props`` (detail-поля имеют приоритет над сохранёнными ранее).
|
||
Записи без detail-полей сохраняются с тем же ``raw_props={id}`` — детали
|
||
могут быть недоступны (объект удалён из реестра / detail-endpoint лежит).
|
||
|
||
Пэйсинг и 429/5xx backoff — на уровне HTTP-клиента (``okn_egrkn_client``),
|
||
общий rate-limit между списком и detail-запросами. Circuit breaker здесь
|
||
(#2445 item D3): ``fail_breaker`` вызовов ``detail_fetcher`` ПОДРЯД,
|
||
завершившихся исключением (сетевая ошибка / устойчивый WAF-бан после
|
||
исчерпания ретраев клиента) → прерываем цикл, не долбим оставшийся лимит.
|
||
Успешный вызов (даже с пустым detail — объект без карточки) сбрасывает
|
||
счётчик подряд-фейлов.
|
||
|
||
Args:
|
||
to_upsert: список row-dict для UPSERT (мутируется in-place).
|
||
detail_fetcher: callable source_id → dict (для DI в тестах).
|
||
limit: максимум detail-вызовов за прогон.
|
||
fail_breaker: детейл-фейлов ПОДРЯД до прерывания цикла.
|
||
|
||
Returns:
|
||
Число фич, для которых detail-поля были успешно подмёрджены.
|
||
"""
|
||
enriched = 0
|
||
consecutive_fail = 0
|
||
total = min(len(to_upsert), limit)
|
||
for idx, row in enumerate(to_upsert[:limit], 1):
|
||
try:
|
||
detail = detail_fetcher(row["source_id"])
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"okn_objects_sync: detail-fetch source_id=%s raised: %s",
|
||
row["source_id"],
|
||
exc,
|
||
)
|
||
detail = {}
|
||
consecutive_fail += 1
|
||
if consecutive_fail >= fail_breaker:
|
||
logger.error(
|
||
"okn_objects_sync: %d detail-фейлов подряд — прерываю "
|
||
"detail-loop (устойчивый бан/сбой WAF), обработано %d/%d",
|
||
fail_breaker,
|
||
idx,
|
||
total,
|
||
)
|
||
break
|
||
continue
|
||
else:
|
||
consecutive_fail = 0
|
||
|
||
if not detail:
|
||
continue
|
||
|
||
try:
|
||
base_props = json.loads(row["raw_props"]) if row["raw_props"] else {}
|
||
except (TypeError, ValueError):
|
||
base_props = {}
|
||
if not isinstance(base_props, dict):
|
||
base_props = {}
|
||
|
||
merged = {**base_props, **detail}
|
||
row["raw_props"] = json.dumps(merged, ensure_ascii=False)
|
||
enriched += 1
|
||
|
||
if idx % _DETAIL_PROGRESS_EVERY == 0:
|
||
logger.info(
|
||
"okn_objects_sync: detail-fetch progress %d/%d (enriched=%d)",
|
||
idx,
|
||
total,
|
||
enriched,
|
||
)
|
||
|
||
logger.info(
|
||
"okn_objects_sync: detail-fetch done enriched=%d of %d (limit=%d)",
|
||
enriched,
|
||
total,
|
||
limit,
|
||
)
|
||
return enriched
|
||
|
||
|
||
@celery_app.task(name="tasks.okn_objects_sync.sync_okn_objects", queue="celery")
|
||
def sync_okn_objects() -> dict[str, int]:
|
||
"""Ingest ОКН-объектов из ЕГРКН в okn_objects (#1066, #1159).
|
||
|
||
Проходит ``_PASSES`` — декартово (адресная подстрока × category_type),
|
||
дедуплицирует по source_id (первый проход устанавливает category),
|
||
догружает detail-карточку для каждого уникального source_id и
|
||
UPSERT'ит в БД с SAVEPOINT per-row.
|
||
|
||
Returns:
|
||
{"okn": N, "enriched": M} — N успешно обработанных фич (upsert),
|
||
M фич, для которых detail-карточка была подмёрджена в raw_props.
|
||
"""
|
||
# Дедуп по source_id в памяти: source_id → category (first-wins)
|
||
seen: dict[str, str] = {}
|
||
# Очередь для UPSERT: row-dict для исполнения _UPSERT_SQL.
|
||
to_upsert: list[dict[str, Any]] = []
|
||
|
||
# Пэйсинг между 33 list-запросами (#2445 item D3): fetch_okn_points → _do_request
|
||
# → okn_egrkn_client._wait_rate_limit перед КАЖДЫМ POST (общий rate-limit с
|
||
# detail-запросами, один WAF-хост okn-mk.mkrf.ru). Ничего дополнительно спать
|
||
# здесь не нужно — пэйсинг/backoff инкапсулированы в клиенте.
|
||
for address, category_type in _PASSES:
|
||
category = _CATEGORY_MAP[category_type]
|
||
logger.info(
|
||
"okn_objects_sync: запрос address=%r category_type=%d (%s)",
|
||
address,
|
||
category_type,
|
||
category,
|
||
)
|
||
features = fetch_okn_points(address, category_type)
|
||
logger.info(
|
||
"okn_objects_sync: получено %d фич для address=%r category=%s",
|
||
len(features),
|
||
address,
|
||
category,
|
||
)
|
||
|
||
for feature in features:
|
||
source_id = str(feature.get("id") or "").strip()
|
||
if not source_id:
|
||
logger.warning("okn_objects_sync: фича без id, пропускаем: %s", feature)
|
||
continue
|
||
|
||
coords = _extract_coords(feature)
|
||
if coords is None:
|
||
continue
|
||
lon, lat = coords
|
||
|
||
if source_id in seen:
|
||
# Объект уже встречался в более приоритетном проходе — пропускаем
|
||
continue
|
||
seen[source_id] = category
|
||
|
||
props = feature.get("properties") or {}
|
||
to_upsert.append(
|
||
{
|
||
"source_id": source_id,
|
||
"category": category,
|
||
"lon": lon,
|
||
"lat": lat,
|
||
"raw_props": json.dumps(props, ensure_ascii=False),
|
||
}
|
||
)
|
||
|
||
logger.info("okn_objects_sync: всего уникальных фич для UPSERT: %d", len(to_upsert))
|
||
|
||
# Догрузить detail (#1159 acceptance 2). Не валит прогон при сетевых ошибках.
|
||
enriched = _enrich_with_detail(to_upsert)
|
||
|
||
inserted = 0
|
||
errors = 0
|
||
|
||
with SessionLocal() as db:
|
||
for row in to_upsert:
|
||
try:
|
||
with db.begin_nested():
|
||
db.execute(_UPSERT_SQL, row)
|
||
inserted += 1
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"okn_objects_sync: UPSERT source_id=%s failed: %s",
|
||
row["source_id"],
|
||
exc,
|
||
)
|
||
errors += 1
|
||
db.commit()
|
||
|
||
logger.info(
|
||
"okn_objects_sync: done inserted/updated=%d enriched=%d errors=%d",
|
||
inserted,
|
||
enriched,
|
||
errors,
|
||
)
|
||
return {"okn": inserted, "enriched": enriched}
|
||
|
||
|
||
__all__ = ["sync_okn_objects"]
|