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

"""Celery task: ingest ОКН-объектов ЕКБ/Свердл.обл из АИС ЕГРКН (#1066).

Обходит 6 комбинаций (address, category_type):
  - «Екатеринбург», 1/2/4
  - «Свердловская», 1/2/4

Один объект может встречаться в нескольких проходах (например «Екатеринбург»-категория-2
и «Свердловская»-категория-2). Дедупликация по source_id: первый проход, выдавший объект,
устанавливает category (самая «высокая» категория первой встретившейся).

Порядок проходов (от федерального к местному, Екатеринбург раньше области) обеспечивает,
что при конфликте source_id category берётся из наиболее значимой категории.

UPSERT: ON CONFLICT (source_id) DO UPDATE обновляет geom, fetched_at (category не меняется
при повторном импорте — первый 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_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",
}

# Порядок проходов: (address, category_type).
# Федеральные идут первыми — при source_id-конфликте category='federal' выигрывает.
_PASSES: list[tuple[str, int]] = [
    ("Екатеринбург", 1),
    ("Екатеринбург", 2),
    ("Екатеринбург", 4),
    ("Свердловская", 1),
    ("Свердловская", 2),
    ("Свердловская", 4),
]

_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


@celery_app.task(name="tasks.okn_objects_sync.sync_okn_objects", queue="celery")
def sync_okn_objects() -> dict[str, int]:
    """Ingest ОКН-объектов из ЕГРКН в okn_objects (#1066).

    Проходит 6 комбинаций (address, category_type), дедуплицирует по source_id
    (первый проход устанавливает category), UPSERT в БД с SAVEPOINT per-row.

    Returns:
        {"okn": N} — число успешно обработанных фич (upsert + skip дублей).
    """
    # Дедуп по source_id в памяти: source_id → category (first-wins)
    seen: dict[str, str] = {}
    # Очередь для UPSERT: (source_id, category, lon, lat, raw_props_json)
    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))

    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 errors=%d",
        inserted,
        errors,
    )
    return {"okn": inserted}


__all__ = ["sync_okn_objects"]