"""상세 배수유역 API 라우터 (B04 — 관리자 검토용).

격자 해석은 하지 않는다. 배수유역 분석이 저장해 둔 `03_road_routing` 산출물을 읽어

  · 기본 관(도로 × 상류 세류선 교차점) + 최대 간격 자동 보충으로 관 목록을 만들고
  · 계획노선 종단 Z로 노면 물이 어느 관으로 가는지 정해 세부유역을 나눈다

계산 알고리즘은 B05 사용자 화면과 공용이다(`common_util_drainage_detail`) — 같은 이름의
버튼은 같은 결과를 내야 하기 때문이다(2026-08-01 사용자 지시).

편집분(관 지점)은 "이 모델 확정" 시점에만 파일로 남는다. 그 전까지는 화면 메모리에만 있고,
확정 없이 나가면 저장된 값으로 되돌아온다.
"""

from __future__ import annotations

import asyncio
import copy
import hashlib
import json
import logging
import math
import os
import pickle
from typing import Any
from uuid import UUID

from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse

from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import STAGES, drainage_dir
from B05_Profile.B05_Profile_Structures_Schema import DEFINITION_PATH, structure_type_map
from B06_Section.B06_Section_Engine_Culvert import pipe_wall_min_height_m
from common_util.common_util_auth import get_optional_session
from common_util.common_util_drainage_context import DrainageContext, load_drainage_context
from common_util.common_util_drainage_detail import (
    DrainageDetail,
    build_detail,
    pipes_from_chainages,
)
from common_util.common_util_drainage_pipes import (
    PIPE_SOURCE_USER,
    PipePoint,
    analysis_route_signature,
    carry_facility_attributes,
    carry_indices,
    clear_pipe_points,
    load_pipe_points,
    parse_pipe_points,
    pipe_signature,
    route_signature,
    save_detail_basins,
    save_pipe_points,
)
from common_util.common_util_json import LONLAT_DIGITS, round_floats
from common_util.common_util_route_geometry import StructureCandidate
from config.config_system import (
    DRAINAGE_PIPE_MAX_SPACING_M,
    DRAINAGE_PIPE_MIN_SPACING_M,
    DRAINAGE_RAINFALL_FILENAME,
    FORD_MANNING_N,
    FORD_SLOPE,
)

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

_NO_ANALYSIS = "배수유역 분석 결과가 없습니다. [유역 분석]을 먼저 실행하세요."
# 노선 지문이 분석 때와 다름(옛 저장분은 지문 없음) — 옛 격자로 나누면 틀린 유역이 나온다(342-3).
STALE_MESSAGE = (
    "배수유역 낡음 — 노선이 분석 때와 다릅니다. 유역을 다시 분석해야 세부 유역을 나눕니다."
)


def _build(
    stored_path: str,
    context: DrainageContext,
    points: list[PipePoint] | None,
    routing: Any = None,
    upstream_lonlat: list | None = None,
) -> tuple[DrainageDetail | None, list[PipePoint]]:
    """세부유역을 계산하고, 그 결과에 쓰인 관 목록을 함께 돌려준다.

    `points`가 None 이면 B04 기본 관에 최대 간격 규칙으로 자동 보충한 목록이 만들어진다.
    있으면 그 자리를 그대로 쓰되, 관이 왜 거기 있는지(기본/자동/수동)는 넘겨받은 목록의
    표시를 유지한다 — 계산기는 확정 좌표만 보므로 그대로 두면 전부 "수동"이 된다.
    """
    directory = drainage_dir(stored_path)
    chainages = [point.chainage_m for point in points] if points is not None else None
    detail = build_detail(directory, context.vertices, chainages, routing, upstream_lonlat)
    if detail is None:
        return None, []
    if points is not None:
        _retag(detail.pipes, points)
    rebuilt = _rebuilt(detail)
    # 원본에 없던 자리 = 이번에 새로 놓인 관(자동 배치 · 간격 보충) — 등록부 기본값을 받을 관
    fresh = [
        pick is None
        for pick in carry_indices(
            [point.chainage_m for point in rebuilt], [point.chainage_m for point in points or []]
        )
    ]
    # 계산기는 chainage만 다뤄 시설 종류·구간·옵션이 사라진다 — 원본에서 되붙인다
    # (2026-08-17 계곡 통과 시설: 배관/BOX암거/물넘이/세월교).
    carried = carry_facility_attributes(rebuilt, points or [])
    # 관경 · 시설 자동 지정 없음 — 10-16 사용자 지시로 끔 · 제안(recommended_*)만 응답에 실음.
    # 새 관은 아래 등록부 기본값(구조물정의 배관 관경)을 받는다.
    flows = {round(basin.chainage_m, 2): basin.design_flow_m3s for basin in detail.basins}
    _apply_new_defaults([point for point, new in zip(carried, fresh) if new], flows)
    return detail, carried


def _rebuilt(detail: DrainageDetail) -> list[PipePoint]:
    return [PipePoint(chainage_m=pipe.chainage_m, source=pipe.reason) for pipe in detail.pipes]


def _build_stale(context: DrainageContext, points: list[PipePoint]) -> DrainageDetail:
    """낡은 분석 — 세부 유역 없이 관 자리만(지금 노선 위) · 시설 속성은 넘겨받은 그대로."""
    to_lonlat = context.to_lonlat
    detail = DrainageDetail(
        route_lonlat=[list(to_lonlat(vertex.x, vertex.y)) for vertex in context.vertices],
        pipes=pipes_from_chainages(context.vertices, [point.chainage_m for point in points]),
    )
    _retag(detail.pipes, points)
    return detail


# 세부 유역 저장본(342-3 · 부록 4 R15) — 열쇠 = 관 목록 · 노선(평면 · 종단 Z) · 분석 산출물 ·
# 강우표 · 구조물 정의 자국. 같으면 다시 나누지 않는다 — 메모리 최근 몇 벌 + [저장] 때 파일 한 벌.
_DETAIL_CACHE: dict[str, tuple[DrainageDetail, list[PipePoint]]] = {}
_DETAIL_CACHE_SIZE = 4
DETAIL_CACHE_FILENAME = "04_detail_cache.pkl"


def _detail_key(context: DrainageContext, points: list[PipePoint] | None) -> str:
    directory = drainage_dir(context.stored_path)
    digest = hashlib.sha1(usedforsecurity=False)
    listed = None if points is None else [point.as_dict() for point in points]
    digest.update(json.dumps(listed, sort_keys=True, ensure_ascii=False).encode())
    for vertex in context.vertices:
        digest.update(f"{vertex.x:.3f},{vertex.y:.3f},{vertex.z:.3f};".encode())
    routing = STAGES["road_routing"]
    for path in (
        directory / f"{routing}_road_routing.npz",
        directory / f"{routing}_road_routing.geojson",
        directory / f"{STAGES['primary_region']}_primary_region.geojson",
        directory / DRAINAGE_RAINFALL_FILENAME,
        DEFINITION_PATH,
    ):
        try:
            info = os.stat(path)
            digest.update(f"{info.st_mtime_ns}:{info.st_size};".encode())
        except OSError:
            digest.update(b"-;")
    return digest.hexdigest()


def _cache_path(stored_path: str):
    return drainage_dir(stored_path) / DETAIL_CACHE_FILENAME


def _remember(key: str, value: tuple[DrainageDetail, list[PipePoint]]) -> None:
    _DETAIL_CACHE.pop(key, None)
    _DETAIL_CACHE[key] = value
    while len(_DETAIL_CACHE) > _DETAIL_CACHE_SIZE:
        _DETAIL_CACHE.pop(next(iter(_DETAIL_CACHE)))


def _build_cached(
    context: DrainageContext, points: list[PipePoint] | None
) -> tuple[DrainageDetail | None, list[PipePoint]]:
    """`_build` + 저장본 — 관 목록은 사본으로 돌려준다(저장이 좌표를 채워도 저장본은 그대로)."""
    key = _detail_key(context, points)
    hit = _DETAIL_CACHE.get(key)
    if hit is None:
        try:
            with _cache_path(context.stored_path).open("rb") as file:
                saved = pickle.load(file)
            hit = saved["value"] if saved.get("key") == key else None
        except (OSError, pickle.PickleError, EOFError, AttributeError, KeyError, TypeError):
            hit = None
    if hit is None:
        detail, carried = _build(context.stored_path, context, points)
        if detail is None:
            return None, []
        hit = (detail, carried)
    _remember(key, hit)
    return hit[0], copy.deepcopy(hit[1])


def _save_detail_cache(
    context: DrainageContext, points: list[PipePoint] | None, detail: DrainageDetail, carried
) -> None:
    """[저장] 뒤 — 다음 GET 이 읽을 관 목록(저장분) 열쇠로 파일 한 벌을 남긴다."""
    key = _detail_key(context, points)
    value = (detail, copy.deepcopy(carried))
    _remember(key, value)
    try:
        with _cache_path(context.stored_path).open("wb") as file:
            pickle.dump({"key": key, "value": value}, file)
    except OSError:
        logger.warning("배수유역: 세부 유역 저장본을 남기지 못했습니다(%s).", context.stored_path)


# 관 보호공 · 유입 집수정 높이 칸 — 새 관 기본 = 관경 + 0.5(139-1) · 유입 흙막이만 + 0.7(211-1)
PIPE_HEIGHT_KEYS = ("inlet_revet_height_m", "outlet_revet_height_m", "inlet_basin_height_m")


def _float(value: Any) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


def _apply_new_defaults(
    points: list[PipePoint], flows: dict[float, float | None] | None = None
) -> None:
    """새로 놓인 관에만 등록부 기본값을 **빈 키에** 채움(PLAN 46-7).

    시설이 생길 때 기본값을 칸에 실제 값으로 넣어 두고 사용자가 고침 — 이미 있던 관의 빈 칸은 그대로
    (옛 프로젝트 빈칸 = 미확정 유지 · 사용자 결정 2026-10-05).
    세월교 빈 월류 높이 = 담당 유역 설계유량(`flows` · 측점 → ㎥/s)을 월류 폭으로 넘기는
    필요 수심(65-4).
    """
    types = structure_type_map()
    for point in points:
        definition = types.get(point.facility)
        if definition is None:
            continue
        options = dict(point.options or {})
        for field in definition.options:
            if field.key not in options and field.default not in (None, ""):
                options[field.key] = field.default
        # 집수정 모양도 실제 값으로(54-39 · 54-36 기본 L형 — 화면 짝 `DEFAULT_BASIN_SHAPE`)
        if options.get("inlet_type") == "집수정":
            options.setdefault("inlet_structure", "L")
        # 관 보호공 · 유입 집수정 높이 = 관경 + 0.5(139-1 · 화면 짝 `PIPE_HEIGHT_KEYS`)
        keys = {field.key for field in definition.options}
        diameter_m = _float(options.get("pipe_diameter_mm")) / 1000
        if diameter_m > 0:
            for key in PIPE_HEIGHT_KEYS:
                if key in keys:
                    side = (
                        "inlet" if key == "inlet_revet_height_m" else ""
                    )  # 유입 흙막이 + 0.7(211-1)
                    options.setdefault(key, pipe_wall_min_height_m(diameter_m, side))
        if point.facility == "ford_bridge" and "ford_height_m" not in options:
            flow = (flows or {}).get(round(point.chainage_m, 2))
            height = ford_height_m(flow, float(options.get("ford_width_m") or 0))
            if height is not None:
                options["ford_height_m"] = height
        point.options = options or None


def ford_height_m(flow_m3s: float | None, width_m: float) -> float | None:
    """월류 폭(m) → 필요 월류 높이(m) · 0.01 m 올림 — 유량 · 폭이 없으면 None.

    짝: `A00_Common/design/design_structure_ford_calc.fordHeightM`(+ `fordSection` 같은 풀이 ·
    Manning n · S = `FORD_MANNING_N` · `FORD_SLOPE`) · 거울 시험 test_65_ford_defaults.
    """
    if not flow_m3s or flow_m3s <= 0 or not math.isfinite(width_m) or width_m <= 0:
        return None
    root = math.sqrt(FORD_SLOPE)

    def velocity(depth: float) -> float:
        return (1 / FORD_MANNING_N) * ((width_m * depth) / (width_m + 2 * depth)) ** (2 / 3) * root

    depth = ((flow_m3s * FORD_MANNING_N) / (width_m * root)) ** (3 / 5)
    for _ in range(20):
        nxt = flow_m3s / (width_m * velocity(depth))
        if abs(nxt - depth) < 1e-6:
            depth = nxt
            break
        depth = nxt
    return math.ceil(depth * 100 - 1e-9) / 100


def _retag(pipes: list[StructureCandidate], points: list[PipePoint]) -> None:
    """확정 좌표로 되돌아온 관에 원래의 생성 사유를 다시 붙인다."""
    by_chainage = {round(point.chainage_m, 2): point.source for point in points}
    for pipe in pipes:
        source = by_chainage.get(round(pipe.chainage_m, 2))
        if source is None and points:
            nearest = min(points, key=lambda item: abs(item.chainage_m - pipe.chainage_m))
            source = nearest.source
        pipe.reason = source or PIPE_SOURCE_USER


def _payload(
    project_id: UUID,
    context: DrainageContext,
    detail: DrainageDetail,
    points: list[PipePoint],
    saved: bool,
    stale: bool = False,
) -> dict[str, Any]:
    """관 목록과 세부유역을 화면 좌표(WGS84)로 정리한다.

    좌표는 위경도 7자리(약 1.1cm)로 맞춰 내보낸다 — 저장된 값에는 뜻 없는 자리가 붙어 있어
    응답이 151KB → 100KB(압축까지 얹으면 22KB)가 된다(2026-09-06 실측). 화면이 그리는 값이고
    다시 계산에 넣지 않으므로 줄여도 결과가 안 갈린다.
    """
    payload = _raw_payload(project_id, context, detail, points, saved)
    # 지문(342-3) — 화면이 지금 노선 · 관 자리와 견줌 · 낡으면 까닭 글(세부 유역 없음)
    payload["route_signature"] = route_signature(context.vertices)
    payload["pipe_signature"] = pipe_signature([pipe.chainage_m for pipe in detail.pipes])
    payload["stale"] = STALE_MESSAGE if stale else None
    return round_floats(payload, LONLAT_DIGITS)


def _raw_payload(
    project_id: UUID,
    context: DrainageContext,
    detail: DrainageDetail,
    points: list[PipePoint],
    saved: bool,
) -> dict[str, Any]:
    """자릿수를 줄이기 전의 응답 본문."""
    to_lonlat = context.to_lonlat
    return {
        "status": "success",
        "project_id": str(project_id),
        # 종단 Z를 어디서 가져왔는지 — 관 담당 구간이 갈리는 근거라 화면에서 확인 가능해야 한다.
        "z_source": context.z_source,
        "route_length_m": round(context.vertices[-1].chainage_m, 2),
        "max_spacing_m": DRAINAGE_PIPE_MAX_SPACING_M,
        "min_spacing_m": DRAINAGE_PIPE_MIN_SPACING_M,
        "saved": saved,
        # 도로 1m 구간별 유입 면적 — 계획선을 색으로 칠하는 데 쓴다(B05와 같은 값).
        "strength_profile": detail.strength_profile,
        # 유입 집중점 — 관을 어디에 둘지 판단하는 근거. 화면에서 토글로 켠다.
        "inflow_hotspots": detail.inflow_hotspots,
        # ── 아래는 B05 배수유역도가 그리는 데 필요한 값. B04 지도는 자체 오버레이가 있어
        #    쓰지 않지만, 두 화면이 같은 응답을 받아야 결과가 갈리지 않는다(2026-08-01 일원화).
        "route_lonlat": detail.route_lonlat,
        # 2차 전체 유역 외곽선 = 분수령. 해석 결과 그대로(손으로 고치는 기능 없음).
        "main_polygon_lonlat": detail.basin_lonlat,
        "grid_cell_m": detail.grid_cell_m,
        "flow_arrows": detail.flow_arrows,
        "arrow_spacing_m": detail.arrow_spacing_m,
        "upstream_lonlat": detail.upstream_lonlat,
        # 시설 종류·구간·옵션까지 함께 싣는다(as_dict — 기본 배관·폭 0은 필드 생략).
        "pipe_points": [
            {
                **point.as_dict(),
                "chainage_m": round(pipe.chainage_m, 2),
                "lonlat": list(to_lonlat(pipe.x, pipe.y)),
                "source": pipe.reason,
            }
            for pipe, point in zip(detail.pipes, points)
        ],
        "basins": [
            {
                "index": basin.index,
                "chainage_m": round(basin.chainage_m, 2),
                "outlet_lonlat": list(to_lonlat(basin.outlet_x, basin.outlet_y)),
                # 옛 소비처를 위한 외곽 링 하나. 그리기는 아래 링 목록을 쓴다.
                "polygon_lonlat": [list(to_lonlat(x, y)) for x, y in basin.boundary_xy],
                # 조각·구멍을 모두 편 링 목록 — 도넛 유역과 떨어진 조각을 그대로 그린다.
                "polygon_rings_lonlat": [
                    [list(to_lonlat(x, y)) for x, y in ring] for ring in basin.boundary_rings
                ],
                "area_m2": round(basin.area_m2, 1),
                "relief_m": round(basin.relief_m, 2),
                "flow_length_m": round(basin.flow_length_m, 1),
                # 배수 유효직경(합리식 산출). 강우량표가 아직 없으면 None → "미정" 표기.
                "pipe_diameter_mm": basin.pipe_diameter_mm,
                # 산출 근거 — 도달시간(분)·설계강우강도(mm/hr)·설계유량(m³/s, 2.0배 반영).
                "tc_minutes": basin.tc_minutes,
                "intensity_mm_hr": basin.intensity_mm_hr,
                "design_flow_m3s": basin.design_flow_m3s,
                # 유효직경이 관 최대 규격 초과 — 세월교·물넘이·교량 검토 대상(임도설치규정 제12조).
                "bridge_required": basin.bridge_required,
                # 필요 통수단면적(㎡) — 물넘이·세월교 개략 단면의 출발값.
                "required_area_m2": basin.required_area_m2,
                # 유량 근거 추천(2026-08-17 사용자 확정) — 배관이면 규격 스냅 관경이 붙는다.
                "recommended_facility": basin.recommended_facility,
                "recommended_diameter_mm": basin.recommended_diameter_mm,
            }
            for basin in detail.basins
        ],
        "pipe_count": len(points),
        # 관 자리 후보(347-3) — 제안만 · [적용] 때 화면이 관 초안에 더함.
        "pipe_candidates": [
            {"chainage_m": round(pipe.chainage_m, 2), "lonlat": list(to_lonlat(pipe.x, pipe.y))}
            for pipe in detail.pipe_candidates
        ],
    }


def _basin_features(
    context: DrainageContext, detail: DrainageDetail, points: list[PipePoint]
) -> list[dict[str, Any]]:
    """세부유역을 저장용 GeoJSON 피처로 바꾼다(관 지점도 같은 파일에 함께 남긴다)."""
    to_lonlat = context.to_lonlat
    features: list[dict[str, Any]] = []
    for basin in detail.basins:
        # GeoJSON 규격 그대로 — 조각마다 [외곽, 구멍...], 조각이 여럿이면 MultiPolygon.
        parts = [
            [[list(to_lonlat(x, y)) for x, y in ring] for ring in part if len(ring) >= 4]
            for part in basin.boundary_parts
        ]
        parts = [part for part in parts if part]
        if not parts:
            continue
        geometry = (
            {"type": "Polygon", "coordinates": parts[0]}
            if len(parts) == 1
            else {"type": "MultiPolygon", "coordinates": parts}
        )
        features.append(
            {
                "type": "Feature",
                "properties": {
                    "kind": "detail_basin",
                    "index": basin.index,
                    "chainage_m": round(basin.chainage_m, 2),
                    "area_m2": round(basin.area_m2, 1),
                    "relief_m": round(basin.relief_m, 2),
                    "flow_length_m": round(basin.flow_length_m, 1),
                    "pipe_diameter_mm": basin.pipe_diameter_mm,
                    # 규격관(도면 배수규격 표기용) — 소요 관경과 구분해 함께 남긴다.
                    "recommended_diameter_mm": basin.recommended_diameter_mm,
                    "tc_minutes": basin.tc_minutes,
                    "intensity_mm_hr": basin.intensity_mm_hr,
                    "design_flow_m3s": basin.design_flow_m3s,
                    "bridge_required": basin.bridge_required,
                },
                "geometry": geometry,
            }
        )
    for pipe, point in zip(detail.pipes, points):
        features.append(
            {
                "type": "Feature",
                # 시설 종류·구간·옵션도 함께 남긴다(as_dict — 기본값 생략) — GeoJSON만
                # 보고도 세월교인지 배관인지 알 수 있어야 한다.
                "properties": {"kind": "pipe", **point.as_dict()},
                "geometry": {"type": "Point", "coordinates": list(to_lonlat(pipe.x, pipe.y))},
            }
        )
    return features


async def _resolve(
    project_id: UUID, payload: dict[str, Any] | None, *, use_stored: bool, fresh: bool = False
) -> tuple[DrainageContext, DrainageDetail, list[PipePoint], bool, bool] | JSONResponse:
    """요청 본문 → 저장분 → 자동 생성 순으로 관 목록을 정하고 세부유역까지 계산한다.

    분석의 노선 지문이 지금 노선과 다르면(옛 저장분 = 지문 없음) 나누지 않고 관 자리만 낡음으로
    돌려준다 — 관 목록도 없으면(자동 생성 자리) 409(342-3). 마지막 값 = 낡음.

    빈 목록(`[]`)은 「관 0개」로 그대로 쓴다 — 자동 생성은 요청에 칸이 없고 저장분도 없을(None) 때만(312).
    """
    context, reason = await load_drainage_context(project_id)
    if context is None:
        return JSONResponse(status_code=404, content={"status": "error", "message": reason})

    try:
        raw = (payload or {}).get("points")
        requested = None if raw is None else parse_pipe_points(raw)
    except ValueError as error:  # 집계 항목(summary_item) 모양이 틀림
        return JSONResponse(status_code=400, content={"status": "error", "message": str(error)})
    signature = route_signature(context.vertices)
    stored = (
        load_pipe_points(context.stored_path, signature, context.vertices) if use_stored else None
    )
    points = requested if requested is not None else stored
    saved = requested is None and stored is not None

    if analysis_route_signature(context.stored_path) != signature:
        if points is None:
            return JSONResponse(
                status_code=409, content={"status": "stale", "message": STALE_MESSAGE}
            )
        detail = _build_stale(context, points)
        return context, detail, carry_facility_attributes(_rebuilt(detail), points), saved, True

    # [저장]은 정본으로 다시 셈 — 저장본은 GET · POST 만
    if fresh:
        result = await asyncio.to_thread(_build, context.stored_path, context, points)
    else:
        result = await asyncio.to_thread(_build_cached, context, points)
    detail, resolved = result
    if detail is None:
        return JSONResponse(status_code=404, content={"status": "error", "message": _NO_ANALYSIS})
    return context, detail, resolved, saved, False


async def _analysis_stale(project_id: UUID) -> bool:
    context, _ = await load_drainage_context(project_id)
    if context is None:
        return False
    return analysis_route_signature(context.stored_path) != route_signature(context.vertices)


@router.get("/{project_id}/drainage/pipe-points", response_model=None)
async def get_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
    """저장된 관 지점과 그 세부유역을 돌려준다. 저장분이 없으면 자동 생성한 목록."""
    resolved = await _resolve(project_id, None, use_stored=True)
    if isinstance(resolved, JSONResponse):
        return resolved
    context, detail, points, saved, stale = resolved
    return _payload(project_id, context, detail, points, saved, stale)


@router.post("/{project_id}/drainage/detail-basins", response_model=None)
async def post_detail_basins(
    project_id: UUID, payload: dict[str, Any] | None = None
) -> dict[str, Any] | JSONResponse:
    """화면에서 편집 중인 관 목록으로 세부유역을 다시 나눈다(저장하지 않는다).

    `points` 칸 없이 부르면 저장분을 무시하고 기본 관 + 자동 보충으로 되돌린다 —
    노선이 바뀌어 전체 분석을 다시 돌린 직후의 경로다. 빈 목록은 관 0개(312).
    """
    resolved = await _resolve(project_id, payload, use_stored=False)
    if isinstance(resolved, JSONResponse):
        return resolved
    context, detail, points, _, stale = resolved
    return _payload(project_id, context, detail, points, False, stale)


@router.delete("/{project_id}/drainage/pipe-points", response_model=None)
async def delete_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
    """저장된 관 지점을 버리고 기본 관 + 자동 보충 배치로 되돌린다("초기화").

    화면만 되돌리면 다시 들어왔을 때 옛 관이 살아난다 — 저장분과 파생 산출물까지 지운다
    (2026-08-02 사용자 보고).
    """
    context, reason = await load_drainage_context(project_id)
    if context is None:
        return JSONResponse(status_code=404, content={"status": "error", "message": reason})
    await asyncio.to_thread(clear_pipe_points, context.stored_path)

    resolved = await _resolve(project_id, None, use_stored=False)
    if isinstance(resolved, JSONResponse):
        return resolved
    context, detail, points, _, stale = resolved
    return _payload(project_id, context, detail, points, False, stale)


@router.put("/{project_id}/drainage/pipe-points", response_model=None)
async def put_pipe_points(
    project_id: UUID,
    payload: dict[str, Any] | None = None,
    session: dict[str, Any] | None = Depends(get_optional_session),  # 가져온이(142-2)
    recompute: bool = False,
) -> dict[str, Any] | JSONResponse:
    """관 지점을 정본으로 확정하고 세부유역 산출물까지 함께 남긴다(모델 확정 시점)."""
    # [확정](recompute) = 낡았으면 지금 노선으로 배수유역부터 다시 셈(정본 · 작업본에 씀 · 347-2) ·
    # [저장] = 낡음 표시 달고 관만 저장.
    if recompute and await _analysis_stale(project_id):
        from B04_PreProcess.B04_PreProcess_Router_Watershed import get_primary_region

        region = await get_primary_region(project_id, refresh=True)
        if isinstance(region, JSONResponse):
            return region
    resolved = await _resolve(project_id, payload, use_stored=True, fresh=True)
    if isinstance(resolved, JSONResponse):
        return resolved
    context, detail, points, _, stale = resolved

    signature = route_signature(context.vertices)
    # 좌표를 같이 남긴다 — 다른 선(B05 최적 경로)으로 읽어도 그 자리에 되놓는다.
    saved = await asyncio.to_thread(
        save_pipe_points, context.stored_path, signature, points, context.vertices
    )
    if not stale:  # 낡은 분석 — 관 자리 · 속성만 정본으로 · 세부 유역 산출물은 안 씀
        await asyncio.to_thread(
            save_detail_basins,
            context.stored_path,
            _basin_features(context, detail, points),
            context.crs,
        )
        stored = load_pipe_points(context.stored_path, signature, context.vertices)
        await asyncio.to_thread(_save_detail_cache, context, stored, detail, points)
    # 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지)
    from common_util.common_util_storage import resolve_stored_project_path
    from M02_MasterTemplete.M02_Item_Snapshot import importer, safe_sync

    failed = await asyncio.to_thread(
        safe_sync, resolve_stored_project_path(context.stored_path), importer(session)
    )
    result = _payload(project_id, context, detail, points, True, stale)
    result["saved_count"] = saved
    result["snapshot_failed"] = failed  # 못 만든 구조물 조합 복사본(142-2) — [저장] 알림
    return result