gendesign/backend/app/workers/tasks/okn_objects_sync.py

"""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).
"""

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

_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,
) -> int:
    """Обогатить raw_props detail-полями (name, address, …).

    Принимает список row-dict'ов, уже подготовленных к UPSERT. Для каждой
    строки вызывает ``detail_fetcher(source_id)`` и сливает результат в JSON
    ``raw_props`` (detail-поля имеют приоритет над сохранёнными ранее).
    Записи без detail-полей сохраняются с тем же ``raw_props={id}`` — детали
    могут быть недоступны (объект удалён из реестра / detail-endpoint лежит).

    Args:
        to_upsert: список row-dict для UPSERT (мутируется in-place).
        detail_fetcher: callable source_id → dict (для DI в тестах).
        limit: максимум detail-вызовов за прогон.

    Returns:
        Число фич, для которых detail-поля были успешно подмёрджены.
    """
    enriched = 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 = {}

        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]] = []

    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"]