"""Загрузчик OSM инженерных сетей («инженерные сети») из Overpass API (#1746).

No-B2B open-data путь: тянет ЛЭП / подстанции / трубопроводы (газ/вода) / ЦТП /
вышки связи через Overpass для bbox ЕКБ и UPSERT-ит в
``osm_utility_infrastructure_ekb``. Живой тест дал 3081+ utility-элементов ЕКБ.

Структура зеркалит ``noise_loader.py``: httpx-клиент с явным UA + таймаутом,
геометрия way → LineString / node → Point, per-element SAVEPOINT при UPSERT
(битый элемент не валит батч).

Hardening (#1746 follow-up, prod-инцидент с 429/504): per-query retry-with-backoff
(уважает ``Retry-After``) + фолбэк на зеркала Overpass (``OVERPASS_ENDPOINTS``) +
более мягкая пауза между query-types (``_INTER_QUERY_DELAY_SECONDS``). Раньше
публичный overpass-api.de троттлил поздние тяжёлые query-types → грузилось
~1099/3000 элементов; теперь покрытие полное.

Запускается раз в неделю через Celery beat. Геометрия включает И линии (ЛЭП,
трубопроводы — way), И точки (опоры, подстанции, узлы ВКХ, ЦТП, вышки — node).
"""

import asyncio
import json
import logging
import random

import httpx
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.db import SessionLocal
from app.services.site_finder.noise_loader import (
    EKB_BBOX,
    _way_to_linestring_wkt,
)

logger = logging.getLogger(__name__)

# Primary endpoint первым, далее зеркала. Публичный overpass-api.de агрессивно
# троттлит (429) и отдаёт 504 на тяжёлых query-types (#1746 prod-инцидент:
# power=tower/water_works/gas/heat/man_made=tower/wastewater_plant молча терялись,
# загрузилось 1099/~3000). При исчерпании ретраев на текущем endpoint переходим
# к следующему зеркалу. kumi.systems / private.coffee — общеизвестные mirror'ы.
OVERPASS_ENDPOINTS: list[str] = [
    "https://overpass-api.de/api/interpreter",
    "https://overpass.kumi.systems/api/interpreter",
    "https://overpass.private.coffee/api/interpreter",
]

# Первый endpoint — для обратной совместимости (тесты/импорты на OVERPASS_URL).
OVERPASS_URL = OVERPASS_ENDPOINTS[0]

# Retry-with-backoff на 429 / 5xx (особенно 504). Попыток на ОДИН endpoint;
# при их исчерпании уходим к следующему зеркалу из OVERPASS_ENDPOINTS.
_MAX_RETRIES_PER_ENDPOINT = 3
# Экспоненциальный backoff (сек) между попытками на одном endpoint; индекс =
# номер уже сделанной неуспешной попытки. Если Retry-After в ответе есть — он в
# приоритете над этой таблицей.
_RETRY_BACKOFF_SECONDS: tuple[float, ...] = (10.0, 30.0, 60.0)
# Джиттер (сек) поверх backoff — размазываем повторные попытки, не бьём в ритме.
_RETRY_JITTER_SECONDS = 5.0
# HTTP-статусы, по которым ретраимся (троттлинг + временные сбои шлюза).
_RETRYABLE_STATUS = frozenset({429, 500, 502, 503, 504})
# Пауза между РАЗНЫМИ query-types. 1с был слишком агрессивен для публичного
# endpoint (#1746) — подняли до 3с (gentler на overpass usage policy).
_INTER_QUERY_DELAY_SECONDS = 3.0

# Overpass блокирует дефолтный User-Agent (406) — явный UA с контактом (как noise_loader).
_HEADERS = {
    "User-Agent": "GenDesign-SiteFinder/1.0 (+https://gendsgn.ru)",
    "Accept": "application/json",
}

# Описание запросов: (overpass_key, overpass_value, el_type, infrastructure_kind).
#   el_type: 'way' (out geom) | 'node' (out body) | 'nwr' (out geom, node+way).
#   substance: для pipeline-веток газ/вода дополнительно фильтруем по substance,
#              чтобы не смешивать водопровод с газопроводом — см. _SUBSTANCE_FILTER.
# Виды сети (infrastructure_kind):
#   power | water | gas | heat | communication | sewage
_UTILITY_QUERIES: list[tuple[str, str, str, str]] = [
    # power — магистральные ЛЭП (линии) + опоры (точки) + подстанции (точка/площадка)
    ("power", "line", "way", "power"),
    ("power", "minor_line", "way", "power"),
    ("power", "tower", "node", "power"),
    ("power", "substation", "nwr", "power"),
    # water — водозаборы/узлы ВКХ (точка/площадка) + магистраль pipeline substance=water
    ("man_made", "water_works", "nwr", "water"),
    ("man_made", "water_tower", "nwr", "water"),
    ("pipeline", "water", "way", "water"),
    # gas — магистральные газопроводы (way), pipeline substance=gas
    ("pipeline", "gas", "way", "gas"),
    ("man_made", "pipeline", "way", "gas"),  # substance=gas фильтруется ниже
    # heat — ЦТП / теплопункты (sparse — это нормально)
    ("man_made", "heat_substation", "nwr", "heat"),
    # communication — вышки связи (точка/площадка) tower:type=communication
    ("man_made", "tower", "nwr", "communication"),
    # sewage — магистральная канализация (pipeline substance=sewerage) + очистные
    ("pipeline", "sewerage", "way", "sewage"),
    ("man_made", "wastewater_plant", "nwr", "sewage"),
]

# Для generic man_made=pipeline нужно различать вещество по тегу `substance`
# (или legacy `type`): иначе под газ попадут все трубы. Ключ — (key, value),
# значение — ожидаемый substance. Элементы без совпадения substance пропускаем.
_SUBSTANCE_FILTER: dict[tuple[str, str], frozenset[str]] = {
    ("man_made", "pipeline"): frozenset({"gas", "natural_gas", "cng"}),
}

# tower:type=communication — для man_made=tower оставляем только вышки связи
# (иначе под communication попадут смотровые/водонапорные башни).
_TOWER_TYPE_REQUIRED: dict[tuple[str, str], frozenset[str]] = {
    ("man_made", "tower"): frozenset({"communication", "communications"}),
}


def _build_overpass_query(key: str, value: str, el_type: str) -> str:
    """Запрос Overpass для одного типа инж.-сетевого элемента.

    way / nwr → ``out geom`` (нужны координаты узлов для LineString / way-площадок);
    node → ``out body``. Зеркалит noise_loader._build_overpass_query.
    """
    south, west, north, east = EKB_BBOX
    bbox = f"({south},{west},{north},{east})"
    if el_type == "nwr":
        return f"[out:json][timeout:30];" f'nwr["{key}"="{value}"]{bbox};' f"out geom;"
    if el_type == "way":
        return f"[out:json][timeout:30];" f'way["{key}"="{value}"]{bbox};' f"out geom;"
    return f"[out:json][timeout:30];" f'node["{key}"="{value}"]{bbox};' f"out body;"


def _passes_tag_filters(key: str, value: str, tags: dict) -> bool:
    """Доп-фильтр по тегам элемента (substance / tower:type).

    Для generic ``man_made=pipeline`` оставляем только газовые трубы (substance),
    для ``man_made=tower`` — только вышки связи (tower:type). Остальные ключи
    проходят без фильтра.
    """
    substance_ok = _SUBSTANCE_FILTER.get((key, value))
    if substance_ok is not None:
        substance = (tags.get("substance") or tags.get("type") or "").lower()
        if substance not in substance_ok:
            return False
    tower_ok = _TOWER_TYPE_REQUIRED.get((key, value))
    if tower_ok is not None:
        tower_type = (tags.get("tower:type") or "").lower()
        if tower_type not in tower_ok:
            return False
    return True


def _parse_retry_after(value: str | None) -> float | None:
    """Парсит заголовок ``Retry-After`` (только delta-seconds — самый частый
    формат Overpass). HTTP-date форму игнорируем (Overpass её не отдаёт) — вернём
    None и используем табличный backoff. Отрицательные / мусорные → None.
    """
    if not value:
        return None
    try:
        seconds = float(value.strip())
    except (TypeError, ValueError):
        return None
    return seconds if seconds >= 0 else None


def _backoff_delay(attempt_idx: int, retry_after: float | None) -> float:
    """Задержка перед следующей попыткой.

    ``Retry-After`` (если сервер прислал) приоритетнее таблицы. Иначе —
    экспоненциальный backoff из ``_RETRY_BACKOFF_SECONDS`` по индексу попытки
    (с клампом на последний элемент). Всегда + рандомный джиттер, чтобы повторы
    не били синхронно.
    """
    if retry_after is not None:
        base = retry_after
    else:
        idx = min(attempt_idx, len(_RETRY_BACKOFF_SECONDS) - 1)
        base = _RETRY_BACKOFF_SECONDS[idx]
    # random.uniform — это джиттер планировщика ретраев, не криптография.
    return base + random.uniform(0.0, _RETRY_JITTER_SECONDS)


async def _fetch_one_query(
    client: httpx.AsyncClient, query: str, label: str
) -> list[dict] | None:
    """Один Overpass-запрос с retry-backoff и фолбэком на зеркала (#1746).

    На каждом endpoint из ``OVERPASS_ENDPOINTS`` делаем до
    ``_MAX_RETRIES_PER_ENDPOINT`` попыток, ретраясь по 429/5xx (особенно 504) с
    экспоненциальным backoff + джиттер, уважая ``Retry-After``. Исчерпав попытки
    на endpoint — переходим к следующему зеркалу. Возвращает list элементов при
    успехе или None, если все endpoint'ы исчерпаны (caller логирует + продолжает,
    partial-load, не abort — поведение #1746 сохранено).
    """
    for endpoint in OVERPASS_ENDPOINTS:
        for attempt in range(_MAX_RETRIES_PER_ENDPOINT):
            try:
                r = await client.post(endpoint, data={"data": query})
            except httpx.HTTPError as e:
                # Сетевой сбой / таймаут — это тоже повод ретраить/сменить зеркало.
                last_attempt = attempt == _MAX_RETRIES_PER_ENDPOINT - 1
                logger.warning(
                    "Overpass %s [%s] transport error (attempt %d/%d): %s",
                    label,
                    endpoint,
                    attempt + 1,
                    _MAX_RETRIES_PER_ENDPOINT,
                    e,
                )
                if last_attempt:
                    break  # к следующему зеркалу
                await asyncio.sleep(_backoff_delay(attempt, None))
                continue

            if r.status_code in _RETRYABLE_STATUS:
                retry_after = _parse_retry_after(r.headers.get("Retry-After"))
                last_attempt = attempt == _MAX_RETRIES_PER_ENDPOINT - 1
                logger.warning(
                    "Overpass %s [%s] HTTP %d (attempt %d/%d, Retry-After=%s)",
                    label,
                    endpoint,
                    r.status_code,
                    attempt + 1,
                    _MAX_RETRIES_PER_ENDPOINT,
                    retry_after,
                )
                if last_attempt:
                    break  # к следующему зеркалу
                await asyncio.sleep(_backoff_delay(attempt, retry_after))
                continue

            # Не-ретраибл статус (4xx кроме 429 и т.п.) — поднимем как есть.
            r.raise_for_status()
            return r.json().get("elements", [])

        logger.warning(
            "Overpass %s: endpoint %s exhausted after %d attempts → next mirror",
            label,
            endpoint,
            _MAX_RETRIES_PER_ENDPOINT,
        )

    return None


async def fetch_overpass_utility() -> list[dict]:
    """Запрашивает Overpass API для всех инж.-сетевых типов ЕКБ.

    Возвращает enriched list dict'ов с доп-полями:
    ``_infrastructure_kind``, ``_source_tag`` — для UPSERT.

    Каждый query-type идёт через ``_fetch_one_query`` (retry-backoff + фолбэк на
    зеркала, #1746). Между РАЗНЫМИ query-types — пауза
    ``_INTER_QUERY_DELAY_SECONDS`` (gentler на публичный endpoint, чем прежний 1с).
    Query, провалившийся на всех зеркалах, логируется и пропускается (partial-load,
    не abort) — остальные query-types это не валит.
    """
    all_elements: list[dict] = []
    async with httpx.AsyncClient(timeout=60, headers=_HEADERS) as client:
        for key, value, el_type, kind in _UTILITY_QUERIES:
            query = _build_overpass_query(key, value, el_type)
            label = f"{key}={value} ({kind})"
            try:
                elements = await _fetch_one_query(client, query, label)
            except Exception as e:
                # Не-ретраибл HTTP-ошибка (raise_for_status) или неожиданный сбой —
                # логируем и продолжаем, чтобы один битый query-type не валил весь sync.
                logger.warning("Overpass utility failed for %s=%s: %s", key, value, e)
                elements = None

            if elements is None:
                logger.warning(
                    "Overpass utility: %s skipped — all endpoints failed (retries+mirrors)",
                    label,
                )
            else:
                logger.info(
                    "Overpass utility: %s=%s (%s) → %d elements",
                    key,
                    value,
                    kind,
                    len(elements),
                )
                for el in elements:
                    el["_infrastructure_kind"] = kind
                    el["_source_tag"] = f"{key}={value}"
                    el["_query_key"] = key
                    el["_query_value"] = value
                all_elements.extend(elements)

            await asyncio.sleep(_INTER_QUERY_DELAY_SECONDS)
    logger.info("Overpass utility: total %d elements", len(all_elements))
    return all_elements


def sync_utility_infrastructure_to_db() -> dict[str, int]:
    """UPSERT инж.сетей из Overpass в osm_utility_infrastructure_ekb.

    UPSERT по UNIQUE(osm_type, osm_id). Геометрия:
      - way с nodes → LineString WKT через ST_GeomFromText(..., 4326)
      - node → ST_SetSRID(ST_MakePoint(lon, lat), 4326)
    Per-element SAVEPOINT — битый элемент не валит весь weekly-sync.

    Returns: счётчики fetched/inserted/updated/skipped + per-kind counts.
    """
    elements = asyncio.run(fetch_overpass_utility())
    return _upsert_elements(elements)


def _upsert_elements(elements: list[dict]) -> dict[str, int]:
    """UPSERT уже полученных Overpass-элементов в таблицу (выделено для тестов).

    Принимает enriched-элементы (с ``_infrastructure_kind`` / ``_source_tag``),
    строит геометрию и UPSERT-ит под per-element SAVEPOINT. Возвращает счётчики.
    """
    inserted = 0
    updated = 0
    skipped = 0
    fetched = len(elements)
    by_kind: dict[str, int] = {}

    db = SessionLocal()
    try:
        for el in elements:
            key: str = el.get("_query_key", "")
            value: str = el.get("_query_value", "")
            tags: dict = el.get("tags") or {}

            # substance / tower:type фильтры (generic man_made=pipeline / tower).
            if not _passes_tag_filters(key, value, tags):
                skipped += 1
                continue

            osm_id: int = el["id"]
            osm_type: str = el["type"]  # 'way' | 'node'
            kind: str = el.get("_infrastructure_kind", "power")
            source_tag: str | None = el.get("_source_tag")
            name: str | None = tags.get("name")

            geom_sql: str
            geom_params: dict
            if osm_type == "way":
                nodes: list[dict] = el.get("geometry") or []
                wkt = _way_to_linestring_wkt(nodes)
                if wkt is None:
                    # way-площадка (закрытый контур) тоже даёт LineString из geometry —
                    # если узлов <2, пропускаем (битый/частичный bbox-обрезок).
                    skipped += 1
                    continue
                geom_sql = "ST_GeomFromText(:wkt, 4326)"
                geom_params = {"wkt": wkt}
            elif osm_type == "node":
                lat = el.get("lat")
                lon = el.get("lon")
                if lat is None or lon is None:
                    skipped += 1
                    continue
                geom_sql = "ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)"
                geom_params = {"lat": lat, "lon": lon}
            else:
                # relation — skip (сложная геометрия, не нужна сейчас)
                skipped += 1
                continue

            params: dict = {
                "osm_id": osm_id,
                "osm_type": osm_type,
                "infrastructure_kind": kind,
                "name": name,
                "source_tag": source_tag,
                "tags": json.dumps(tags, ensure_ascii=False),
                **geom_params,
            }

            try:
                with db.begin_nested():  # SAVEPOINT — откат только этой записи
                    result = db.execute(
                        text(f"""
                            INSERT INTO osm_utility_infrastructure_ekb
                                (osm_type, osm_id, infrastructure_kind, name, geom,
                                 source_tag, tags, fetched_at)
                            VALUES (
                                :osm_type, :osm_id, :infrastructure_kind, :name,
                                {geom_sql},
                                :source_tag, CAST(:tags AS jsonb), NOW()
                            )
                            ON CONFLICT (osm_type, osm_id) DO UPDATE
                               SET infrastructure_kind = EXCLUDED.infrastructure_kind,
                                   name                = EXCLUDED.name,
                                   geom                = EXCLUDED.geom,
                                   source_tag          = EXCLUDED.source_tag,
                                   tags                = EXCLUDED.tags,
                                   fetched_at          = NOW()
                            RETURNING (xmax = 0) AS is_insert
                        """),
                        params,
                    ).scalar()
                if result:
                    inserted += 1
                else:
                    updated += 1
                by_kind[kind] = by_kind.get(kind, 0) + 1
            except Exception as e:
                # Дефектный элемент (битая геометрия и т.п.) не должен валить
                # весь weekly-sync. SAVEPOINT откатывает только эту строку.
                logger.warning(
                    "utility_sync upsert failed for %s/%s (kind=%s): %s",
                    osm_type,
                    osm_id,
                    kind,
                    e,
                )
                skipped += 1

        db.commit()
    except Exception as e:
        db.rollback()
        logger.exception("utility_sync: unexpected error, outer tx rolled back: %s", e)
        raise
    finally:
        db.close()

    result_dict: dict[str, int] = {
        "fetched": fetched,
        "inserted": inserted,
        "updated": updated,
        "skipped": skipped,
    }
    # per-kind counts (power/water/gas/heat/communication/sewage) — для отчёта populate.
    for k, c in sorted(by_kind.items()):
        result_dict[f"kind_{k}"] = c
    logger.info("utility_sync done: %s", result_dict)
    return result_dict


async def load_utility_infrastructure(db: Session | None = None) -> dict[str, int]:
    """Async-обёртка для прямого вызова из async-контекста (#1746 spec).

    Тянет Overpass и UPSERT-ит в таблицу. ``db`` принимается для сигнатурной
    совместимости со спекой, но UPSERT идёт через собственную SessionLocal
    (как noise/poi sync) — Overpass-fetch async, запись в БД sync. Возвращает
    те же per-kind counts, что ``sync_utility_infrastructure_to_db``.
    """
    elements = await fetch_overpass_utility()
    return _upsert_elements(elements)


def get_nearby_utility_infrastructure(
    db: Session,
    cad_num: str,
    radius_m: int = 500,
    limit: int = 200,
) -> dict:
    """Инженерные сети из OSM в radius_m от центроида участка (#1746 endpoint).

    Источник: ``osm_utility_infrastructure_ekb`` (наполняется weekly-sync из
    Overpass). Расстояние считается от ЦЕНТРОИДА участка через ST_DWithin /
    ST_Distance в geography. Возвращает feature-list + summary с ближайшими
    расстояниями по видам сети.

    Args:
        db: SQLAlchemy session.
        cad_num: кадастровый номер участка (e.g. '66:41:0204016:10').
        radius_m: радиус поиска, м (caller валидирует диапазон).
        limit: максимум фич в ответе (ближайшие).

    Returns:
        {"features": [...], "summary": {...}}

    Raises:
        ValueError: участок не найден в БД (нет geom).
    """
    # Импорт здесь, чтобы избежать циклического импорта на module-load
    # (quarter_dump_lookup → noise_loader → ... ).
    from app.services.site_finder.quarter_dump_lookup import _get_parcel_wkt

    parcel_wkt = _get_parcel_wkt(db, cad_num)
    if parcel_wkt is None:
        raise ValueError(f"Участок {cad_num!r} не найден в БД")

    rows = (
        db.execute(
            text("""
                SELECT osm_id, osm_type, infrastructure_kind, name, source_tag,
                       ST_Distance(
                           u.geom::geography,
                           ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
                       ) AS distance_m,
                       ST_AsGeoJSON(u.geom) AS geojson
                  FROM osm_utility_infrastructure_ekb u
                 WHERE ST_DWithin(
                       u.geom::geography,
                       ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
                       CAST(:radius_m AS float)
                   )
                 ORDER BY distance_m ASC
                 LIMIT :lim
            """),
            {"wkt": parcel_wkt, "radius_m": radius_m, "lim": limit},
        )
        .mappings()
        .all()
    )

    features: list[dict] = []
    nearest_by_kind: dict[str, float | None] = {}
    nearest_overall: float | None = None
    for r in rows:
        dist = round(float(r["distance_m"]), 1)
        kind = r["infrastructure_kind"]
        # rows отсортированы по distance ASC → первое попадание вида = ближайшее.
        if kind not in nearest_by_kind:
            nearest_by_kind[kind] = dist
        if nearest_overall is None:
            nearest_overall = dist
        features.append(
            {
                "osm_id": int(r["osm_id"]),
                "osm_type": r["osm_type"],
                "infrastructure_kind": kind,
                "name": r["name"],
                "source_tag": r["source_tag"],
                "distance_m": dist,
                "geometry_geojson": json.loads(r["geojson"]) if r["geojson"] else {},
            }
        )

    return {
        "features": features,
        "summary": {
            "total_features": len(features),
            "nearest_distance_m": nearest_overall,
            "nearest_by_kind": nearest_by_kind,
        },
    }