"""배수유역 분석 API 라우터 (B04 — 관리자 확인용).

계획 노선(B03 업로드 CSV)과 도엽 등고선·세류선으로 배수유역을 끝까지 분석하고, 결과를
`storage/{프로젝트}/B04_PreProcess/drainage/`에 남긴다. 30초 안팎이 걸리므로 여기서 한 번만
돌리고, 일반 사용자가 쓰는 B05는 저장분을 읽어 쓴다(2026-07-31 사용자 지시).

좌표는 사업지 CRS(m)에서 계산하고 응답만 WGS84로 바꿔 내보낸다.
"""

import asyncio
import base64
import json
import logging
import math
from pathlib import Path
from typing import Any
from uuid import UUID

import numpy as np
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from pyproj import Transformer
from shapely.geometry import Point, Polygon, box

from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import preview_stages
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import (
    drainage_dir,
    write_grid_arrays,
    write_stage,
)
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import (
    AZIMUTH_INVALID,
    AZIMUTH_SINK,
    AZIMUTH_STEPS,
    mask_row_spans,
)
from B05_Profile.B05_Profile_Repository import get_surface_crs_epsg
from common_util.common_util_route_geometry import (
    StructureCandidate,
    find_planned_route_file,
    read_planned_route_csv,
)
from common_util.common_util_storage import resolve_stored_project_path
from common_util.common_util_wamis_rainfall import (
    build_rainfall_table,
    ensure_contour_cache,
)
from common_util.common_util_wamis_station import (
    build_station_rainfall_table,
    is_jeju,
)
from config.config_db import get_db_pool
from config.config_system import (
    DRAINAGE_ARROW_SPACING_M,
    DRAINAGE_DESIGN_RETURN_PERIOD_YR,
    DRAINAGE_RAINFALL_FILENAME,
    DRAINAGE_RAINFALL_IDF_METHOD,
    DRAINAGE_RAINFALL_STATION_DIRNAME,
    DRAINAGE_RESPONSE_FILENAME,
    WAMIS_CONTOUR_CACHE_DIR,
    WAMIS_STATION_RADIUS_M,
)

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["B04 Surface Watershed"])

# 도엽 레이어 파일명 (B04 전처리 산출물과 같은 위치)
# B07 유역도가 같은 배경을 그리므로 파일명은 공개 상수로 둔다(사본 금지).
CONTOUR_FILE = "도엽_등고선.geojson"
STREAM_FILE = "도엽_하천중심선.geojson"

# 응답 형식 판(版). 응답에 항목을 더하거나 값의 의미를 바꾸면 이 값을 올린다 —
# 저장해 둔 옛 응답을 그대로 돌려주면 화면이 조용히 어긋나기 때문이다.
#   2 = 강도 곡선 간격 1m(구간 합) + 유입 집중점(inflow_hotspots) 추가 (2026-08-01)
#   3 = 강도 곡선 구간 기준을 round→floor로 바로잡고(2m 주기 빗살 제거),
#       유입 집중점이 기본 관 옆 최소간격 안쪽을 피하도록 수정 (2026-08-01)
RESPONSE_SCHEMA_VERSION = 3


def _sheet_dir(stored_path: str) -> Path:
    return Path(resolve_stored_project_path(stored_path)) / "B04_PreProcess" / "processed"


def _route_input_dir(stored_path: str) -> Path:
    """B03 업로드 폴더 — 계획 노선 파일이 여기 들어온다."""
    return Path(resolve_stored_project_path(stored_path)) / "B03_FileInput" / "input"


def _load_features(directory: Path, filename: str) -> list[dict[str, Any]]:
    """도엽 GeoJSON을 읽어 피처 목록만 돌려준다. 없으면 빈 목록."""
    path = directory / filename
    if not path.exists():
        return []
    try:
        with path.open("r", encoding="utf-8") as file:
            data = json.load(file)
    except (OSError, json.JSONDecodeError):
        logger.warning("도엽 GeoJSON을 읽지 못했습니다: %s", path)
        return []
    features = data.get("features")
    return features if isinstance(features, list) else []


def _reproject_features(
    features: list[dict[str, Any]],
    transformer: Transformer | None,
) -> list[dict[str, Any]]:
    """WGS84 도엽 좌표를 사업지 CRS(m)로 바꾼다. 거리·면적을 미터로 계산하기 위함."""
    if transformer is None:
        return features
    converted: list[dict[str, Any]] = []
    for feature in features:
        geometry = feature.get("geometry")
        if not geometry:
            continue
        coordinates = _map_coordinates(geometry.get("coordinates"), transformer)
        if coordinates is None:
            continue
        converted.append(
            {
                "type": "Feature",
                "properties": feature.get("properties") or {},
                "geometry": {"type": geometry.get("type"), "coordinates": coordinates},
            }
        )
    return converted


def _map_coordinates(coordinates: Any, transformer: Transformer) -> Any:
    """중첩 좌표 배열을 재귀적으로 변환한다."""
    if not isinstance(coordinates, list) or not coordinates:
        return None
    first = coordinates[0]
    if isinstance(first, (int, float)):
        x, y = transformer.transform(float(coordinates[0]), float(coordinates[1]))
        return [x, y]
    mapped = [_map_coordinates(item, transformer) for item in coordinates]
    return [item for item in mapped if item is not None]


def _candidate_payload(
    candidate: StructureCandidate,
    to_lonlat: Any,
) -> dict[str, Any]:
    lon, lat = to_lonlat(candidate.x, candidate.y)
    return {
        "chainage_m": round(candidate.chainage_m, 2),
        "x": candidate.x,
        "y": candidate.y,
        "lon": lon,
        "lat": lat,
        "reason": candidate.reason,
        "stream_name": candidate.stream_name,
    }


async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse:
    """계획 노선 파일과 도엽 피처, 좌표 변환기를 준비한다.

    노선은 **B03에 업로드된 계획 노선 파일**에서 읽는다 — B05의 확정 경로가 아니다.
    배수유역 분석은 노선 설계보다 먼저 끝나 있어야 하기 때문이다(2026-07-31 사용자 지시).
    """
    pool = get_db_pool()
    async with pool.acquire() as connection:
        stored_path = await get_project_storage_relative_path(connection, project_id)
        epsg = await get_surface_crs_epsg(connection, project_id, 0)

    route_file = find_planned_route_file(_route_input_dir(stored_path))
    if route_file is None:
        return JSONResponse(
            status_code=404,
            content={"status": "error", "message": "계획 노선 파일이 업로드되지 않았습니다."},
        )
    planned = read_planned_route_csv(route_file)
    if planned is None or len(planned.vertices) < 2:
        return JSONResponse(
            status_code=400,
            content={
                "status": "error",
                "message": f"계획 노선 파일을 읽지 못했습니다: {route_file.name}",
            },
        )

    # 노선 파일이 CRS를 명시하면 그 값을 따른다. 도엽 재투영도 같은 좌표계로 맞춘다.
    source_crs = f"EPSG:{planned.epsg or epsg or 5186}"
    to_lonlat_transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True)
    to_metric_transformer = Transformer.from_crs("EPSG:4326", source_crs, always_xy=True)

    directory = _sheet_dir(stored_path)
    streams = _reproject_features(_load_features(directory, STREAM_FILE), to_metric_transformer)
    contour_features = _reproject_features(
        _load_features(directory, CONTOUR_FILE), to_metric_transformer
    )
    return {
        "route_source": route_file.name,
        "vertices": planned.vertices,
        "route_line": planned.line,
        "streams": streams,
        "contours": contour_features,
        "stored_path": stored_path,
        "to_lonlat": lambda x, y: to_lonlat_transformer.transform(x, y),
    }


def _response_path(stored_path: str) -> Path:
    """분석 응답 캐시 경로. 재산정하지 않는 한 이 파일을 그대로 돌려준다."""
    return drainage_dir(stored_path) / DRAINAGE_RESPONSE_FILENAME


def _load_saved_response(stored_path: str) -> dict[str, Any] | None:
    path = _response_path(stored_path)
    if not path.exists():
        return None
    try:
        with path.open("r", encoding="utf-8") as file:
            saved = json.load(file)
    except (OSError, json.JSONDecodeError):
        logger.warning("배수유역: 저장된 분석 응답을 읽지 못했습니다 (%s).", path)
        return None
    # 응답 형식이 바뀌면 옛 저장분을 그대로 주면 안 된다 — 화면이 없는 항목을 그리려다
    # 조용히 어긋난다(예: 강도 곡선 간격 5m→1m, 유입 집중점 신설). 다시 계산하게 둔다.
    if saved.get("schema_version") != RESPONSE_SCHEMA_VERSION:
        logger.info("배수유역: 저장분이 옛 형식이라 다시 분석합니다 (%s).", path)
        return None
    return saved


def _save_response(stored_path: str, payload: dict[str, Any]) -> None:
    path = _response_path(stored_path)
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("w", encoding="utf-8") as file:
            json.dump(payload, file, ensure_ascii=False)
    except OSError:
        logger.warning("배수유역: 분석 응답을 저장하지 못했습니다 (%s).", path)


def _rainfall_path(stored_path: str) -> Path:
    return drainage_dir(stored_path) / DRAINAGE_RAINFALL_FILENAME


def _route_center_lonlat(stored_path: str, fallback_epsg: int | None) -> tuple[float, float] | None:
    """계획 노선 중간점(WGS84). 강우량 내삽 기준 좌표 — 유역 규모 대비 등우선 간격이
    훨씬 넓어 노선 대표점 하나로 고정한다(설계기준.md 4절, 2026-08-05 협의)."""
    route_file = find_planned_route_file(_route_input_dir(stored_path))
    if route_file is None:
        return None
    planned = read_planned_route_csv(route_file)
    if planned is None or not planned.vertices:
        return None
    middle = planned.vertices[len(planned.vertices) // 2]
    source_crs = f"EPSG:{planned.epsg or fallback_epsg or 5186}"
    lon, lat = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True).transform(
        middle.x, middle.y
    )
    return lat, lon


def _build_rainfall_sync(lat: float, lon: float, stored_path: str) -> dict[str, Any]:
    """계획노선 좌표로 지역을 분별해 강우량표를 만든다 (PLAN.md A, 2026-08-13).

    제주: 등우선 내삽(서버 원본이 제주만 커버) / 본토: 관측소 방식(최근접+10km
    최대값). 본토에서 관측소 수급이 실패하면 그대로 예외를 올린다 — 제주 등우선으로
    폴백하면 "조용히 틀린 값"이 재발하므로 금지.
    """
    if is_jeju(lat, lon):
        cached, failures = ensure_contour_cache(WAMIS_CONTOUR_CACHE_DIR)
        table = build_rainfall_table(
            lat,
            lon,
            WAMIS_CONTOUR_CACHE_DIR,
            design_return_period=DRAINAGE_DESIGN_RETURN_PERIOD_YR,
        )
        table["contour_cache_files"] = cached
        if failures:
            table["failures"] = (table.get("failures") or []) + failures
        table["region_mode"] = "jeju_contour"
        return table
    table = build_station_rainfall_table(
        lat,
        lon,
        drainage_dir(stored_path) / DRAINAGE_RAINFALL_STATION_DIRNAME,
        design_return_period=DRAINAGE_DESIGN_RETURN_PERIOD_YR,
        radius_m=WAMIS_STATION_RADIUS_M,
        idf_method=DRAINAGE_RAINFALL_IDF_METHOD,
    )
    table["region_mode"] = "mainland_station"
    return table


async def _ensure_rainfall_table(stored_path: str, fallback_epsg: int | None) -> None:
    """rainfall_table.json이 없으면 백그라운드로 만든다. 실패는 비치명(로그만)."""
    path = _rainfall_path(stored_path)
    if path.exists():
        return
    center = _route_center_lonlat(stored_path, fallback_epsg)
    if center is None:
        return
    try:
        table = await asyncio.to_thread(_build_rainfall_sync, *center, stored_path)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(table, ensure_ascii=False, indent=1), encoding="utf-8")
        logger.info("확률강우량표 저장: %s (값 %d개)", path, len(table.get("values") or []))
    except Exception:  # noqa: BLE001 — 외부망 차단 등. 배수유역 해석은 계속되어야 한다.
        logger.warning("확률강우량표 생성 실패 (%s)", stored_path, exc_info=True)


@router.get("/{project_id}/drainage/rainfall", response_model=None)
async def get_drainage_rainfall(
    project_id: UUID, refresh: bool = False
) -> dict[str, Any] | JSONResponse:
    """프로젝트 지점의 확률강우량표(설계빈도 IDF 적합계수 포함)를 돌려준다.

    저장분이 있으면 그대로 주고, 없으면 즉석 생성한다(최초 1회는 등우선 96개
    다운로드로 수십 초 걸릴 수 있다). B05 세션 캐시가 이 응답을 물고 다닌다.
    """
    pool = get_db_pool()
    async with pool.acquire() as connection:
        stored_path = await get_project_storage_relative_path(connection, project_id)
        epsg = await get_surface_crs_epsg(connection, project_id, 0)
    path = _rainfall_path(stored_path)
    if not refresh and path.exists():
        try:
            return {**json.loads(path.read_text(encoding="utf-8")), "from_cache": True}
        except (OSError, json.JSONDecodeError):
            logger.warning("강우량표 저장분을 읽지 못해 다시 만듭니다 (%s).", path)
    center = _route_center_lonlat(stored_path, epsg)
    if center is None:
        return JSONResponse(
            status_code=404,
            content={"status": "error", "message": "계획 노선 파일이 업로드되지 않았습니다."},
        )
    try:
        table = await asyncio.to_thread(_build_rainfall_sync, *center, stored_path)
    except Exception as exc:  # noqa: BLE001
        return JSONResponse(
            status_code=502,
            content={
                "status": "error",
                "message": f"확률강우량 자료를 받지 못했습니다: {exc}",
            },
        )
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(table, ensure_ascii=False, indent=1), encoding="utf-8")
    return {**table, "from_cache": False}


@router.get("/{project_id}/drainage/primary-region", response_model=None)
async def get_primary_region(
    project_id: UUID, refresh: bool = False
) -> dict[str, Any] | JSONResponse:
    """배수유역 분석 결과를 돌려준다.

    기본은 **영구저장소에 남은 결과를 그대로** 준다 — 분석이 30초 걸리므로 화면을 열
    때마다 다시 돌릴 이유가 없다. `refresh=true`면 처음부터 다시 계산하고 덮어쓴다
    (2026-07-31 사용자 지시).
    """
    pool = get_db_pool()
    async with pool.acquire() as connection:
        stored_path = await get_project_storage_relative_path(connection, project_id)
        epsg = await get_surface_crs_epsg(connection, project_id, 0)
    # 확률강우량표가 없으면 백그라운드로 만들어 둔다 — 유효직경 계산(B05)이 이 파일을 쓴다.
    asyncio.create_task(_ensure_rainfall_table(stored_path, epsg))
    if not refresh:
        saved = _load_saved_response(stored_path)
        if saved is not None:
            logger.info("배수유역: 저장된 분석 결과를 그대로 돌려줍니다 (%s).", stored_path)
            return {**saved, "from_cache": True}

    prepared = await _prepare(project_id)
    if isinstance(prepared, JSONResponse):
        return prepared
    preview = await asyncio.to_thread(
        preview_stages,
        prepared["vertices"],
        prepared["contours"],
        prepared["streams"],
    )
    if preview is None:
        return JSONResponse(
            status_code=400,
            content={"status": "error", "message": "1차 배수유역을 정할 등고선이 없습니다."},
        )
    region = preview.region
    to_lonlat = prepared["to_lonlat"]
    # 확장을 거치면 격자·해석 영역이 1차 영역보다 커진다. 최종본을 써야 화면과 어긋나지 않는다.
    spec = preview.spec or region.spec
    domain = preview.domain if preview.domain is not None else region.cell_mask
    payload = {
        "status": "success",
        "schema_version": RESPONSE_SCHEMA_VERSION,
        "project_id": str(project_id),
        "route_source": prepared["route_source"],
        "radius_m": region.radius_m,
        # 채택된 상류 세류망 = 1차 영역의 기준선.
        "upstream_lines": [_line_lonlat(line, to_lonlat) for line in region.split.upstream],
        # 도로 아래로 이어진 하류망 — 판정이 맞는지 눈으로 대조하기 위해 함께 준다.
        "downstream_lines": [_line_lonlat(line, to_lonlat) for line in region.split.downstream],
        "no_contact_count": region.split.no_contact,
        # 노선이 1차 영역 밖으로 나간 길이(m). 크면 반경을 올려야 한다는 신호.
        "road_outside_m": round(region.road_outside_m, 1),
        # 1차 영역(버퍼 합집합) 외곽 링 목록.
        "region_rings": _polygon_rings(region.area, to_lonlat),
        "grid": {
            "cell_m": spec.cell_m,
            "rows": spec.n_rows,
            "cols": spec.n_cols,
            # bbox 전체 셀 수와, 해석 영역에 실제로 생성된 셀 수(확장 반영).
            "bbox_cells": spec.size,
            "cells": int(domain.sum()) if domain is not None else 0,
            "width_m": round(spec.n_cols * spec.cell_m, 1),
            "height_m": round(spec.n_rows * spec.cell_m, 1),
            # 격자 bbox 링. 프론트는 이 사각형을 rows×cols로 나눠 행·열 좌표를 얻는다.
            "bbox_lonlat": _grid_bbox_lonlat(spec, to_lonlat),
            # 실제 생성된 셀을 행별 연속 구간 [행, 시작열, 끝열]으로 압축해 보낸다.
            # 셀을 낱개로 보내면 수십만 건이라 응답이 감당되지 않는다.
            "row_spans": [list(span) for span in mask_row_spans(domain)]
            if domain is not None
            else [],
        },
        # 최외곽 적색 셀 주변 확장 결과.
        "expansion": {
            "rounds": preview.expand_rounds,
            "closed": preview.expand_closed,
            "added_cells": preview.expand_added_cells,
            "initial_cells": region.active_cells,
        },
        # 셀별 흐름 방향과 도로 도달 여부. row_spans 순서(행 → 구간 → 열 오름차순)로 1바이트씩.
        "flow": _flow_payload(preview, domain),
        # ⑦ 2차 전체 배수유역 외곽선(= 분수령)과 면적.
        "basin_polygon_lonlat": [list(to_lonlat(x, y)) for x, y in preview.basin_boundary_xy],
        "basin_area_m2": round(preview.basin_area_m2, 1),
        # ⑥ 도로 위 흐름 강도 — [누가거리 m, 그 구간으로 모이는 상류 면적 ㎡].
        "strength_profile": [
            [round(chainage, 1), round(area, 1)] for chainage, area in preview.strength_profile
        ],
        # ⑥-1 유입 집중점 — 기본 관으로 나눈 구역마다 물이 많이 모이는 자리.
        # [누가거리 m, 유입면적 ㎡, 구역 번호, 구역 내 순위]. 화면 마커·하이라이트 대상이다.
        "inflow_hotspots": [
            [round(chainage, 1), round(area, 1), zone, rank]
            for chainage, area, zone, rank in preview.inflow_hotspots
        ],
        # ⑧ 기본 관 매설 위치 — 도로 × 세류선 교차점.
        "pipes": [_candidate_payload(pipe, to_lonlat) for pipe in preview.pipes],
        # B05용 평균 흐름 화살표 — [lon, lat, 방위(도), 도로도달, 셀 수].
        # 세류·도로 셀을 뺀 블록 평균이라 사면 경향만 남는다.
        "flow_arrows": [
            [*to_lonlat(x, y), round(math.degrees(angle), 1), reaches, cells]
            for x, y, angle, reaches, cells in preview.flow_arrows
        ],
        # 화살표 간격(m). 화면이 화살표 크기를 정할 때 쓴다 — 서로 닿지 않게 이 값보다 짧게 그린다.
        "arrow_spacing_m": DRAINAGE_ARROW_SPACING_M,
        "from_cache": False,
    }
    # 단계 산출물을 영구저장소에 남긴다 — 기능을 붙일 때마다 여기에 단계가 하나씩 늘어난다.
    payload["saved_to"] = write_stage(
        prepared["stored_path"],
        "primary_region",
        {
            "primary_region": _as_polygons(region.area),
            "upstream": region.split.upstream,
            "downstream": region.split.downstream,
            "route": [prepared["route_line"]],
            "grid_bbox": [_grid_bbox_polygon(spec)],
            # ⑦ 2차 전체 배수유역 외곽선, ⑧ 기본 관 위치(누가거리·근거 포함).
            "basin_boundary": _boundary_geometry(preview.basin_boundary_xy),
            "pipe": [
                (
                    Point(pipe.x, pipe.y),
                    {
                        "chainage_m": round(pipe.chainage_m, 2),
                        "reason": pipe.reason,
                        "stream_name": pipe.stream_name,
                    },
                )
                for pipe in preview.pipes
            ],
        },
        {
            "radius_m": region.radius_m,
            "road_outside_m": payload["road_outside_m"],
            "no_contact_count": region.split.no_contact,
            "basin_area_m2": payload["basin_area_m2"],
            "pipe_count": len(preview.pipes),
            # 기하(bbox·셀 구간)는 파일 본문에 있으므로 요약 수치만 남긴다.
            "grid": {
                key: value
                for key, value in payload["grid"].items()
                if key not in {"bbox_lonlat", "row_spans"}
            },
        },
        to_lonlat,
    )
    _write_stage_arrays(prepared["stored_path"], preview, domain, spec)
    _write_road_routing(prepared["stored_path"], preview, spec, prepared["route_line"], to_lonlat)
    # 응답 자체를 캐시로 남긴다 — 다음 조회는 배열을 재조립하지 않고 이 파일을 그대로 준다.
    _save_response(prepared["stored_path"], payload)
    return payload


def _write_road_routing(
    stored_path: str, preview: Any, spec: Any, route_line: Any, to_lonlat: Any
) -> None:
    """B05가 세부유역을 나눌 때 쓸 최소 산출물을 남긴다.

    B05는 일반 사용자용이라 가벼워야 한다. 화살표(방향 코드)·밴드 표고 같은 확인용 배열은
    빼고, **셀 → 도로 셀 귀속**과 도로 셀 제원만 담는다. 여기에 표고를 함께 넣는 이유는
    유역 낙차를 내려면 셀 표고가 필요해서다(2026-07-31 사용자 지시).
    """
    routing = preview.routing
    road = preview.road
    if routing is None or road is None or road.count == 0:
        return
    write_grid_arrays(
        stored_path,
        "road_routing",
        spec,
        {
            "road_slot": routing.road_slot,
            "path_length": routing.path_length,
            "strength": routing.strength,
            "road_cell_index": road.cell_index,
            "road_chainage": road.chainage,
            "elevation": preview.terrain.elevation.reshape(-1),
        },
        {
            "road_cells": road.count,
            "reached_cells": int((routing.road_slot >= 0).sum()),
            "basin_area_m2": round(preview.basin_area_m2, 1),
            "pipe_count": len(preview.pipes),
        },
    )
    # B05가 그대로 그릴 기하 — 계획도로선 · 기본 배관 · 2차 전체 배수유역, 이 셋뿐이다.
    write_stage(
        stored_path,
        "road_routing",
        {
            "route": [route_line],
            "basin_boundary": _boundary_geometry(preview.basin_boundary_xy),
            "pipe": [
                (
                    Point(pipe.x, pipe.y),
                    {"chainage_m": round(pipe.chainage_m, 2), "reason": pipe.reason},
                )
                for pipe in preview.pipes
            ],
            # 평균 흐름 화살표 — B05도 같은 그림을 그려야 하므로 여기 함께 남긴다.
            "flow_arrow": [
                (
                    Point(x, y),
                    {
                        # B05 화면은 사업지 CRS(m)로 그리므로 미터 좌표도 함께 남긴다.
                        "x": round(x, 2),
                        "y": round(y, 2),
                        "azimuth_deg": round(math.degrees(angle), 1),
                        "reaches_road": reaches,
                        "cells": cells,
                    },
                )
                for x, y, angle, reaches, cells in preview.flow_arrows
            ],
        },
        {
            "basin_area_m2": round(preview.basin_area_m2, 1),
            "pipe_count": len(preview.pipes),
            "route_length_m": round(route_line.length, 1),
            "arrow_count": len(preview.flow_arrows),
            "arrow_spacing_m": DRAINAGE_ARROW_SPACING_M,
        },
        to_lonlat,
    )


def _write_stage_arrays(stored_path: str, preview: Any, domain: Any, spec: Any) -> None:
    """격자 규모 배열(셀 마스크·흐름 방향·도달 여부)을 단계별 `.npz`로 남긴다."""
    if domain is not None:
        write_grid_arrays(
            stored_path,
            "primary_region",
            spec,
            {"mask": domain},
            {
                "cells": int(domain.sum()),
                "bbox_cells": spec.size,
                "expand_rounds": preview.expand_rounds,
                "expand_closed": preview.expand_closed,
            },
        )
    flow = preview.flow
    if flow is None:
        return
    arrays = {
        "direction": flow.direction.reshape(spec.n_rows, spec.n_cols),
        "reaches_road": flow.reaches_road.reshape(spec.n_rows, spec.n_cols),
        "analyzed": flow.analyzed.reshape(spec.n_rows, spec.n_cols),
        # 사후 진단용 — 수신 셀이 있어야 사슬을 다시 따라가 볼 수 있다.
        "receiver": preview.terrain.receiver.reshape(spec.n_rows, spec.n_cols),
    }
    if flow.burned is not None:
        arrays["burned"] = flow.burned.reshape(spec.n_rows, spec.n_cols)
    if preview.descent is not None:
        arrays["band_elevation"] = preview.descent.band_elevation
    # ⑥ 흐름 강도 곡선 — 기하가 아니라 수치 곡선이라 GeoJSON이 아닌 여기에 함께 담는다.
    if preview.strength_profile:
        curve = np.asarray(preview.strength_profile, dtype=np.float64)
        arrays["strength_chainage_m"] = curve[:, 0]
        arrays["strength_area_m2"] = curve[:, 1]
    write_grid_arrays(
        stored_path,
        "flow_direction",
        spec,
        arrays,
        {
            "azimuth_steps": AZIMUTH_STEPS,
            "sink_code": AZIMUTH_SINK,
            "invalid_code": AZIMUTH_INVALID,
            "analyzed": int(flow.analyzed.sum()),
            "reaches_road": int((flow.reaches_road & flow.analyzed).sum()),
            "no_road": int((~flow.reaches_road & flow.analyzed).sum()),
            "burned": 0 if flow.burned is None else int(flow.burned.sum()),
            "outer_seeds": flow.outer_seeds,
            "interior_seeds": flow.interior_seeds,
            "strength_points": len(preview.strength_profile),
            "strength_total_m2": round(sum(area for _, area in preview.strength_profile), 1),
        },
    )


def _flow_payload(preview: Any, domain: Any) -> dict[str, Any] | None:
    """셀별 흐름 방향·도로 도달 여부를 바이트 배열로 압축한다.

    셀이 수십만 개라 JSON 객체로는 못 보낸다. 셀 하나당 1바이트로 줄이고 base64로 싣는다:
      하위 6비트(0x3F) = 32방위 코드(0~31, 0=화면 오른쪽·시계방향), 32=제자리, 33=표고 없음
      최상위 비트(0x80) = 도로 도달(적색). 꺼져 있으면 미도달(백색 화살표).
    바이트 순서는 `grid.row_spans`를 행 → 구간 → 열 오름차순으로 훑은 순서와 같다.
    """
    flow = preview.flow
    if flow is None or domain is None:
        return None
    order = np.flatnonzero(domain.reshape(-1))
    analyzed = flow.analyzed[order]
    reaches = flow.reaches_road[order]
    packed = np.clip(flow.direction[order], 0, AZIMUTH_INVALID).astype(np.uint8)
    packed |= np.where(reaches, 0x80, 0).astype(np.uint8)
    burned = flow.burned
    return {
        "encoding": "base64-uint8",
        "azimuth_steps": AZIMUTH_STEPS,
        "sink_code": AZIMUTH_SINK,
        "invalid_code": AZIMUTH_INVALID,
        "cells": int(order.size),
        "reaches_road": int((reaches & analyzed).sum()),
        "no_road": int((~reaches & analyzed).sum()),
        # 격자에는 있으나 등고선 TIN 밖이라 표고가 없어 판정 못한 셀.
        "unanalyzed": int((~analyzed).sum()),
        # 확정된 상류 세류망을 따라 흐름을 강제로 새긴 셀 수.
        "burned": 0 if burned is None else int(burned[order].sum()),
        "outer_seeds": flow.outer_seeds,
        "interior_seeds": flow.interior_seeds,
        "data": base64.b64encode(packed.tobytes()).decode("ascii"),
    }


def _boundary_geometry(ring: list[tuple[float, float]]) -> list[Any]:
    """2차 유역 외곽 링을 저장용 폴리곤으로 만든다(정점 3개 미만이면 비운다)."""
    return [Polygon(ring)] if len(ring) >= 4 else []


def _as_polygons(geometry: Any) -> list[Any]:
    if geometry is None or geometry.is_empty:
        return []
    return list(geometry.geoms) if geometry.geom_type == "MultiPolygon" else [geometry]


def _grid_bbox_polygon(spec: Any) -> Polygon:
    x_max = spec.x_min + spec.n_cols * spec.cell_m
    y_min = spec.y_max - spec.n_rows * spec.cell_m
    return box(spec.x_min, y_min, x_max, spec.y_max)


def _line_lonlat(line: Any, to_lonlat: Any) -> list[list[float]]:
    return [list(to_lonlat(x, y)) for x, y in line.coords]


def _polygon_rings(geometry: Any, to_lonlat: Any) -> list[list[list[float]]]:
    """폴리곤/멀티폴리곤의 외곽 링만 뽑아 lonlat으로 바꾼다."""
    if geometry is None or geometry.is_empty:
        return []
    parts = geometry.geoms if geometry.geom_type == "MultiPolygon" else [geometry]
    return [[list(to_lonlat(x, y)) for x, y in part.exterior.coords] for part in parts]


def _grid_bbox_lonlat(spec: Any, to_lonlat: Any) -> list[list[float]]:
    x_min = spec.x_min
    x_max = spec.x_min + spec.n_cols * spec.cell_m
    y_max = spec.y_max
    y_min = spec.y_max - spec.n_rows * spec.cell_m
    corners = ((x_min, y_min), (x_min, y_max), (x_max, y_max), (x_max, y_min), (x_min, y_min))
    return [list(to_lonlat(x, y)) for x, y in corners]