Aislo/B06_Section/B06_Section_Router.py

"""B06 종횡단 생성 FastAPI 라우터."""

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

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

from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions, resolve_grade_options
from B05_Profile.B05_Profile_Engine_Grade_Profile import rebuild_alignment_profile
from B05_Profile.B05_Profile_Engine_Sections import (
    prune_stale_cross_files,
    run_section_generation,
)
from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions
from B06_Section.B06_Section_Engine_Culvert import attach_culvert_sets
from B06_Section.B06_Section_Engine_Design import compute_cross_design, curve_widening_args
from B06_Section.B06_Section_Engine_Structures_Wall import attach_wall_structures
from B06_Section.B06_Section_Repository import (
    count_cross_sections,
    create_longitudinal_section,
    delete_sections_for_route,
    get_cross_section_designs,
    get_latest_section_options,
    get_longitudinal_section,
    get_project_standard_cross_section,
    get_route_generation_source,
    get_workflow_route_context,
    insert_cross_sections,
    list_recent_company_projects,
    update_cross_section_design,
)
from B06_Section.B06_Section_Router_Design import (
    PREVIEW_DESIGN_FIELDS as _PREVIEW_DESIGN_FIELDS,
)
from B06_Section.B06_Section_Router_Design import (
    USER_TOUCHED_KEYS,
    ford_drop_at,
    ford_surface_drops,
    stored_berm,
    stored_cut_slope,
)
from B06_Section.B06_Section_Router_Design import (
    attach_default_designs as _attach_default_designs,
)
from B06_Section.B06_Section_Router_Design import (
    read_cross_design_inputs as _read_cross_design_inputs,
)
from B06_Section.B06_Section_Router_Design import (
    recompute_designs_for_alignment as _recompute_designs_for_alignment,
)
from B06_Section.B06_Section_Router_Design import (
    stored_standard_cross_section as _stored_standard_cross_section,
)
from B06_Section.B06_Section_Schema import (
    CompanyStandardListResponse,
    CompanyStandardProject,
    CompanyStandardResponse,
    CrossDesignPreviewRequest,
    CrossDesignPreviewResponse,
    CrossDesignRequest,
    CrossDesignResponse,
    HaulEquipmentLimit,
    SectionContextResponse,
    SectionDetailResponse,
    SectionOptionDefaults,
    SectionRegenerateRequest,
    SectionSummaryResponse,
)
from B06_Section.B06_Section_Server_Calc_Prebuild import conversion_factors_for
from common_util.common_util_auth import verify_session
from common_util.common_util_storage import resolve_stored_project_path
from common_util.common_util_surface_confirmation import get_surface_confirmation_params
from common_util.common_util_workflow_state import get_workflow_state
from config.config_db import get_db_pool, run_with_connection
from config.config_system import (
    EARTHWORK_HAUL_EQUIPMENT_LIMITS_M,
    FOREST_ROAD_MIN_WIDTH_M,
    NATURAL_SPOIL_MIN_GROUND_SLOPE,
    SECTION_VERTICAL_EXAGGERATION,
    STANDARD_CROSS_SECTION,
    STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
    STANDARD_ROCK_BOUNDARY_STEP_M,
)

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["B06 Profile Cross"])


@router.get("/{project_id}/sections/context", response_model=SectionContextResponse)
async def get_section_context(project_id: UUID) -> SectionContextResponse | JSONResponse:
    """최신 확정 경로와 B04 확정값, config 기반 생성 기본값을 반환한다."""
    try:
        # 임도 종류(projects.road_type) — B05가 계획선 법정 기준을 정하는 데 쓴다.
        async def _road_type(connection: aiomysql.Connection) -> str | None:
            async with connection.cursor() as cursor:
                await cursor.execute(
                    "SELECT road_type FROM projects WHERE id = %s", (str(project_id),)
                )
                row = await cursor.fetchone()
            return row[0] if row else None

        # 화면이 보는 경로 = 최신 경로(확정 여부 무관) — B05와 같은 규칙이어야
        # 두 화면이 같은 노선의 같은 값을 본다(2026-09-03 일원화).
        # 셋은 서로 기다릴 이유가 없다 — 원격 DB 왕복(약 12ms)이 더해지지 않게 같이 보낸다.
        route_context, surface_params, road_type = await asyncio.gather(
            run_with_connection(get_workflow_route_context, project_id),
            run_with_connection(get_surface_confirmation_params, str(project_id)),
            run_with_connection(_road_type),
        )

        # 저장된 표준 횡단면 — 브라우저 계산이 서버와 같은 값을 쓰게 함께 내려보낸다
        # (2026-09-07). 노선이 없으면 저장분도 없다.
        stored_standard: dict[str, Any] | None = None
        if route_context and route_context.get("route_id") is not None:
            longitudinal_row = await run_with_connection(
                get_longitudinal_section, project_id, int(route_context["route_id"])
            )
            stored_standard = _stored_standard_cross_section(longitudinal_row)

        defaults = SectionGenerationOptions()
        return SectionContextResponse(
            project_id=str(project_id),
            route_id=route_context["route_id"] if route_context else None,
            filter_key=surface_params["source_filter"] if route_context else None,
            method=surface_params["method"] if route_context else None,
            smooth=bool(surface_params["smooth"]) if route_context else None,
            crs_epsg=route_context["crs_epsg"] if route_context else None,
            road_type=road_type,
            defaults=SectionOptionDefaults(
                station_interval_m=defaults.station_interval_m,
                cross_half_width_m=defaults.cross_half_width_m,
                cross_sample_interval_m=defaults.cross_sample_interval_m,
                long_sample_interval_m=defaults.long_sample_interval_m,
                vertical_exaggeration=SECTION_VERTICAL_EXAGGERATION,
            ),
            standard_cross_section=STANDARD_CROSS_SECTION,
            stored_standard_cross_section=stored_standard,
            rock_boundary_default_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
            rock_boundary_step_m=STANDARD_ROCK_BOUNDARY_STEP_M,
            # ⚠ 상수를 직접 들지 않는다 — 프로젝트가 고른 계수가 있으면 화면도 그 값으로
            #   그려야 서버가 뒤에 다시 셈한 값과 갈리지 않는다(CLAUDE.md 5장).
            earthwork_conversion=await conversion_factors_for(project_id),
            haul_equipment_limits=[
                HaulEquipmentLimit(key=key, max_distance_m=limit)
                for key, limit in EARTHWORK_HAUL_EQUIPMENT_LIMITS_M
            ],
            natural_spoil_min_ground_slope=NATURAL_SPOIL_MIN_GROUND_SLOPE,
        )
    except Exception:
        logger.exception("B06 종횡단 컨텍스트 조회 실패: project_id=%s", project_id)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "종횡단 컨텍스트 조회 중 오류가 발생했습니다."},
        )


@router.get("/{project_id}/sections/road-widths")
async def get_forest_road_min_widths(project_id: UUID) -> dict[str, dict[str, float]]:
    """B05 측점 가로선에 적용할 임도 등급별 법정 최소너비를 반환한다."""
    return {"forest_road_min_width_m": FOREST_ROAD_MIN_WIDTH_M}


# 아래 두 엔드포인트는 `/sections/{route_id}`(int)보다 먼저 선언해 라우팅 충돌을 막는다.
@router.get("/{project_id}/sections/company-standards", response_model=CompanyStandardListResponse)
async def list_company_standards(
    project_id: UUID, session: dict[str, Any] = Depends(verify_session)
) -> CompanyStandardListResponse:
    """같은 회사의 최근 프로젝트 5개를 반환한다(설계값 보유 여부 무관)."""
    company_id = session.get("company_id")
    if company_id is None:
        return CompanyStandardListResponse(projects=[])
    pool = get_db_pool()
    async with pool.acquire() as connection:
        rows = await list_recent_company_projects(connection, company_id, project_id)
    return CompanyStandardListResponse(
        projects=[
            CompanyStandardProject(project_id=str(row["project_id"]), name=row["name"])
            for row in rows
        ]
    )


@router.get(
    "/{project_id}/sections/company-standards/{source_project_id}",
    response_model=CompanyStandardResponse,
)
async def get_company_standard(
    project_id: UUID,
    source_project_id: UUID,
    session: dict[str, Any] = Depends(verify_session),
) -> CompanyStandardResponse | JSONResponse:
    """특정 프로젝트의 표준횡단 설정값을 미리보기용으로 반환한다(회사 스코프 강제)."""
    company_id = session.get("company_id")
    if company_id is None:
        return JSONResponse(
            status_code=404, content={"status": "error", "message": "설계값을 찾을 수 없습니다."}
        )
    pool = get_db_pool()
    async with pool.acquire() as connection:
        standard = await get_project_standard_cross_section(
            connection, company_id, source_project_id
        )
    if standard is None:
        return JSONResponse(
            status_code=404, content={"status": "error", "message": "설계값을 찾을 수 없습니다."}
        )
    return CompanyStandardResponse(
        project_id=str(source_project_id), standard_cross_section=standard
    )


@router.get("/{project_id}/sections/{route_id}", response_model=SectionSummaryResponse)
async def get_sections(project_id: UUID, route_id: int) -> SectionSummaryResponse | JSONResponse:
    """경로의 종단면 요약을 조회한다."""
    pool = get_db_pool()
    try:
        async with pool.acquire() as connection:
            longitudinal = await get_longitudinal_section(connection, project_id, route_id)
            cross_section_count = await count_cross_sections(connection, route_id)
        data = longitudinal.get("data") if longitudinal else None
        length_m = data.get("length_m") if isinstance(data, dict) else None
        return SectionSummaryResponse(
            project_id=str(project_id),
            route_id=route_id,
            longitudinal=longitudinal,
            length_m=length_m,
            cross_section_count=cross_section_count,
        )
    except Exception:
        logger.exception("B06 종횡단 조회 실패: project_id=%s", project_id)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "종횡단 조회 중 오류가 발생했습니다."},
        )


def _read_section_detail(project_root: Path, longitudinal_file_path: str) -> dict:
    """검증된 프로젝트 루트 안의 종단 및 횡단 JSON을 읽는다."""
    root = project_root.resolve()
    longitudinal_path = (root / longitudinal_file_path).resolve()
    if root not in longitudinal_path.parents:
        raise ValueError("종단면 파일 경로가 프로젝트 저장소를 벗어났습니다.")
    if not longitudinal_path.is_file():
        raise FileNotFoundError("종단면 상세 파일을 찾을 수 없습니다.")

    stage_root = longitudinal_path.parent.parent
    cross_dir = stage_root / "cross_sections"
    if not cross_dir.is_dir():
        raise FileNotFoundError("횡단면 상세 파일을 찾을 수 없습니다.")

    longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8"))
    stations = longitudinal.get("stations") if isinstance(longitudinal, dict) else None
    valid_names = prune_stale_cross_files(cross_dir, stations if isinstance(stations, list) else [])
    cross_sections = [
        json.loads(path.read_text(encoding="utf-8"))
        for path in sorted(cross_dir.glob("cross_*.json"))
        if not valid_names or path.name in valid_names
    ]
    if not isinstance(longitudinal, dict) or not all(
        isinstance(section, dict) for section in cross_sections
    ):
        raise ValueError("종횡단 상세 파일 형식이 올바르지 않습니다.")
    # 배수관 측점에 세트(배관·기슭막이·보호공) 제원을 얹는다 — 횡단 카드가 그림을 그린다.
    # 독립 기슭막이도 2026-08-28 이관으로 **관 숨김 세트**로 여기 함께 얹힌다(pipe_points).
    attach_culvert_sets(root, cross_sections)
    # 좌측 「구조물 배치」로 넣은 C군 벽(옹벽·돌쌓기 등)도 같은 제원 자리에 얹는다 —
    # 그래야 횡단도·설계선 트림·폐회로 면적이 그 벽을 본다(2026-09-06 사용자 확정).
    attach_wall_structures(root, cross_sections)
    return {"longitudinal": longitudinal, "cross_sections": cross_sections}


def _read_balloon_offsets(row: dict[str, Any]) -> dict[str, list[float]] | None:
    """확정 시 저장해 둔 유토곡선 balloon 위치를 꺼낸다. 없거나 형태가 깨졌으면 None."""
    data = row.get("data")
    if isinstance(data, (str, bytes)):
        try:
            data = json.loads(data)
        except (ValueError, TypeError):
            return None
    if not isinstance(data, dict):
        return None
    offsets = (data.get("mass_haul") or {}).get("balloon_offsets")
    if not isinstance(offsets, dict):
        return None
    cleaned: dict[str, list[float]] = {}
    for key, value in offsets.items():
        if isinstance(value, (list, tuple)) and len(value) == 2:
            try:
                cleaned[str(key)] = [float(value[0]), float(value[1])]
            except (TypeError, ValueError):
                continue
    return cleaned or None


@router.get("/{project_id}/sections/{route_id}/detail", response_model=SectionDetailResponse)
async def get_section_detail(
    project_id: UUID, route_id: int
) -> SectionDetailResponse | JSONResponse:
    """경로의 SVG 렌더링용 종단·횡단 원시 샘플을 반환한다."""
    try:
        # 서로 기다릴 이유가 없는 읽기 셋 — 원격 DB 라 순차로 내면 왕복이 그대로 더해진다
        # (질의 하나 약 12ms, 2026-09-06 실측). 같이 보내 가장 느린 하나의 시간만 쓴다.
        longitudinal, stored_path, designs = await asyncio.gather(
            run_with_connection(get_longitudinal_section, project_id, route_id),
            run_with_connection(get_project_storage_relative_path, project_id),
            run_with_connection(get_cross_section_designs, route_id),
        )
        if not longitudinal:
            return JSONResponse(
                status_code=404,
                content={"status": "error", "message": "종횡단 상세 결과가 없습니다."},
            )
        project_root = Path(resolve_stored_project_path(stored_path))
        detail = await asyncio.to_thread(
            _read_section_detail,
            project_root,
            str(longitudinal["longitudinal_file_path"]),
        )
        # DB에만 있는 잠정 설계 지정을 chainage 근사로 각 횡단에 얹어 화면 복원을 돕는다.
        for record in designs:
            for section in detail["cross_sections"]:
                if abs(float(section.get("chainage_m", 0.0)) - record["chainage_m"]) < 0.01:
                    section["design"] = record["design"]
                    break
        # 포장 구간·세월교 노면 하강 보정은 **저장 때**로 옮겼다(2026-09-06 사용자 확정:
        # 「읽을 때는 영구저장소에서 가져오기만」). 조회는 저장분을 그대로 싣는다 —
        # 보정은 `B06_Section_Server_Calc_Prebuild.recompute_server_side` 가 [저장]·[확정]과
        # 자동설계 체인에서 돌려 정본에 남긴다.
        standard = _stored_standard_cross_section(longitudinal)
        # 지정값이 없는 측점은 기본값(토사/좌절토)으로 즉석 계산해 프리뷰로 채운다.
        # 저장분이 있으면 **아무것도 하지 않는다**(측점마다 design 유무만 본다) — 아직
        # 저장된 적 없는 비정규 측점만 이 폴백을 탄다.
        await asyncio.to_thread(
            _attach_default_designs,
            detail["longitudinal"],
            detail["cross_sections"],
            project_root,
            standard,
        )
        return SectionDetailResponse(**detail, balloon_offsets=_read_balloon_offsets(longitudinal))
    except FileNotFoundError as exc:
        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        logger.warning(
            "B06 종횡단 상세 파일 조회 실패: project_id=%s route_id=%s error=%s",
            project_id,
            route_id,
            exc,
        )
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "종횡단 상세 파일을 읽지 못했습니다."},
        )
    except Exception:
        logger.exception(
            "B06 종횡단 상세 조회 실패: project_id=%s route_id=%s", project_id, route_id
        )
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "종횡단 상세 조회 중 오류가 발생했습니다."},
        )


def _regeneration_grade_options(stage_params: dict[str, Any] | None) -> GradeDesignOptions | None:
    """재생성 시 종단 계획선 기준을 확정 당시 저장 파라미터(stage 2)로 재구성한다.

    grade_options 없이 재생성하면 종단 계획선(profile_alignment)이 통째로 소실돼 계획
    횡단도선·유토곡선·테이블이 사라진다(2026-08-06 사용자 보고 — 옛 재계산 버튼 사고의
    원인). B04 파일입력→B06 자동 계산 파이프라인이 쓰는 같은 엔진 경로를 재활용한다.
    """
    if not stage_params:
        return None
    options = stage_params.get("options") or {}
    grade_class = options.get("grade_class")
    if not grade_class:
        return None
    stored = {
        key: stage_params.get(key)
        for key in (
            "max_grade_pct",
            "min_vertical_radius_m",
            "min_tangent_length_m",
            "balance_segment_length_m",
            "start_elevation_offset_m",
            "end_elevation_offset_m",
            # 횡단배수 최소고 강제 스위치도 확정 당시 값을 따라야 한다 — 빠뜨리면 재생성이
            # 기본(해제)으로 돌아가 켜 둔 계획선이 내려앉는다(2026-09-01).
            "enforce_pipe_clearance",
        )
        if stage_params.get(key) is not None
    }
    # 설계속도는 B05가 저장해 둔 값을 그대로 따른다 — 없으면 종류별 기본(20).
    speed = options.get("design_speed_kph")
    return resolve_grade_options(
        str(grade_class),
        terrain_type=str(options.get("terrain_type") or "normal"),
        paved=bool(options.get("paved", False)),
        design_speed_kph=int(speed) if speed is not None else None,
        stored=stored,
    )


def _regeneration_options(
    stored_options: dict[str, Any] | None,
    stage_params: dict[str, Any] | None,
    cross_half_width_m: float,
) -> SectionGenerationOptions:
    """DB 저장 옵션(단일 소스) → stage 2 params → config 순으로 유지하고 반폭만 교체한다."""
    defaults = SectionGenerationOptions()
    stored = stored_options or {}
    params = stage_params or {}

    def pick(key: str, default: float) -> float:
        return stored.get(key) or params.get(key) or default

    return SectionGenerationOptions(
        station_interval_m=pick("station_interval_m", defaults.station_interval_m),
        cross_half_width_m=cross_half_width_m,
        cross_sample_interval_m=pick("cross_sample_interval_m", defaults.cross_sample_interval_m),
        long_sample_interval_m=pick("long_sample_interval_m", defaults.long_sample_interval_m),
        include_endpoint=defaults.include_endpoint,
    )


@router.post("/{project_id}/sections/{route_id}/regenerate", response_model=SectionDetailResponse)
async def regenerate_sections(
    project_id: UUID, route_id: int, request: SectionRegenerateRequest
) -> SectionDetailResponse | JSONResponse:
    """표시 옵션의 횡단 반폭으로 종횡단을 재생성해 저장하고 상세를 반환한다."""
    pool = get_db_pool()
    try:
        async with pool.acquire() as connection:
            source = await get_route_generation_source(connection, project_id, route_id)
            if not source:
                return JSONResponse(
                    status_code=404,
                    content={"status": "error", "message": "재생성할 경로가 없습니다."},
                )
            surface_params = await get_surface_confirmation_params(connection, str(project_id))
            stored_path = await get_project_storage_relative_path(connection, project_id)
            stored_options = await get_latest_section_options(connection, project_id)
            async with connection.cursor(aiomysql.DictCursor) as cursor:
                workflow = await get_workflow_state(cursor, str(project_id))
            route_stage = next(
                (stage for stage in workflow["stages"] if stage["stage_no"] == 2),
                None,
            )
            project_root = Path(resolve_stored_project_path(stored_path))
            crs_epsg = source["crs_epsg"]
            stage_params = route_stage.get("params") if route_stage else None
            sections = await asyncio.to_thread(
                run_section_generation,
                project_root,
                source["route_data_path"],
                surface_params["source_filter"],
                surface_params["method"],
                bool(surface_params["smooth"]),
                # 비정규 측점은 run_section_generation이 정본 파일에서 해석한다(단일 공급자).
                options=_regeneration_options(
                    stored_options,
                    stage_params,
                    request.cross_half_width_m,
                ),
                # 계획선까지 함께 재계산·저장 — 없으면 계획 횡단도선·유토곡선이 사라진다.
                grade_options=_regeneration_grade_options(stage_params),
                crs=f"EPSG:{crs_epsg}" if crs_epsg is not None else None,
            )
            await connection.begin()
            try:
                await delete_sections_for_route(connection, route_id)
                await create_longitudinal_section(
                    connection,
                    project_id=project_id,
                    route_id=route_id,
                    data=sections["longitudinal"]["data"],
                    longitudinal_file_path=sections["longitudinal"]["file_path"],
                )
                await insert_cross_sections(
                    connection,
                    project_id=project_id,
                    route_id=route_id,
                    sections=sections["cross_sections"],
                )
                await connection.commit()
            except Exception:
                await connection.rollback()
                raise
        result = sections["result"]
        # 재생성 응답도 상세 조회와 같은 배수관 세트·구조물 벽 정보를 실어야 화면이 어긋나지 않는다.
        attach_culvert_sets(project_root, result["cross_sections"])
        attach_wall_structures(project_root, result["cross_sections"])
        return SectionDetailResponse(
            longitudinal=result["longitudinal"], cross_sections=result["cross_sections"]
        )
    except FileNotFoundError as exc:
        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
    except ValueError as exc:
        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
    except Exception:
        logger.exception("B06 종횡단 재생성 실패: project_id=%s route_id=%s", project_id, route_id)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "종횡단 재생성 처리 중 오류가 발생했습니다."},
        )


@router.post(
    "/{project_id}/sections/{route_id}/cross-design/preview",
    response_model=CrossDesignPreviewResponse,
)
async def preview_cross_designs(
    project_id: UUID, route_id: int, request: CrossDesignPreviewRequest
) -> CrossDesignPreviewResponse | JSONResponse:
    """계획선 편집 델타로 계획선과 전 측점 횡단 설계를 다시 계산해 돌려준다(저장 없음).

    ⚠ **평상시 조작 경로가 아니다**(2026-09-03 이후). 계획선을 만지는 동안의 재계산은
    브라우저가 직접 한다(`common_util/common_util_cross_design.ts`) — 조작 중 계산이
    서버로 나가면 왕복이 조작 속도를 지배하기 때문이다(사용자 확정). 이 엔드포인트는
    선형 저장분이 없어 브라우저가 계획고를 풀 수 없는 **옛 데이터 폴백**으로만 남는다
    (`B06_Section_Cross_Refresh.refreshCrossDesigns`).

    영속화는 각 페이지의 임시저장·확정이 맡는다.
    """
    pool = get_db_pool()
    try:
        async with pool.acquire() as connection:
            longitudinal_row = await get_longitudinal_section(connection, project_id, route_id)
            if not longitudinal_row:
                return JSONResponse(
                    status_code=404,
                    content={"status": "error", "message": "종횡단 상세 결과가 없습니다."},
                )
            stored_path = await get_project_storage_relative_path(connection, project_id)
            designs = await get_cross_section_designs(connection, route_id)
        project_root = Path(resolve_stored_project_path(stored_path))
        detail = await asyncio.to_thread(
            _read_section_detail,
            project_root,
            str(longitudinal_row["longitudinal_file_path"]),
        )

        def rebuild() -> None:
            alignment, profile = rebuild_alignment_profile(detail["longitudinal"], request.edits())
            detail["longitudinal"]["profile_alignment"] = alignment
            detail["longitudinal"]["design_profiles"] = [profile]
            _recompute_designs_for_alignment(
                detail["longitudinal"],
                detail["cross_sections"],
                designs,
                request.standard_cross_section,
                request.rock_boundary_offsets,
                project_root,
                request.berms,
                request.cut_slope_ratios,
            )

        await asyncio.to_thread(rebuild)
        # 계획고를 끄는 동안 오가는 값이라 기본은 **유토곡선이 실제로 읽는 필드만** 싣는다.
        # 지반선 샘플·설계선 좌표(design_line)·선형 구조를 다 담으면 1MB가 넘어 편집이 굼떠진다
        # (2026-08-03 사용자 지적). 계획선 자체는 화면이 이미 같은 규칙으로 계산해 들고 있다.
        # `full_designs`(B06 진입 시 stale 일괄 재계산)일 때만 설계 전체를 1회 싣는다 —
        # 측점별 순차 호출 N번보다 왕복 1번이 훨씬 싸다(2026-08-04 사용자 확인).
        return CrossDesignPreviewResponse(
            designs=[
                {
                    "chainage_m": section.get("chainage_m"),
                    "design": section["design"]
                    if request.full_designs
                    else {
                        key: section["design"].get(key)
                        for key in _PREVIEW_DESIGN_FIELDS
                        if key in section["design"]
                    },
                }
                for section in detail["cross_sections"]
                if section.get("design")
            ],
        )
    except FileNotFoundError as exc:
        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
    except Exception:
        logger.exception(
            "B06 횡단 설계 프리뷰 실패: project_id=%s route_id=%s", project_id, route_id
        )
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "횡단 설계 프리뷰 계산 중 오류가 발생했습니다."},
        )


@router.post("/{project_id}/sections/{route_id}/cross-design", response_model=CrossDesignResponse)
async def compute_cross_section_design(
    project_id: UUID, route_id: int, request: CrossDesignRequest
) -> CrossDesignResponse | JSONResponse:
    """측점 표준횡단 설계(지반유형·단면유형)를 즉시 계산해 잠정치로 저장한다."""
    pool = get_db_pool()
    try:
        async with pool.acquire() as connection:
            longitudinal = await get_longitudinal_section(connection, project_id, route_id)
            if not longitudinal:
                return JSONResponse(
                    status_code=404,
                    content={"status": "error", "message": "종횡단 상세 결과가 없습니다."},
                )
            stored_path = await get_project_storage_relative_path(connection, project_id)
            # 저장분을 **계산 전에** 읽는다 — 소단은 사용자 조작값이면서 **기하 입력**이라,
            # 나중에 키만 베껴 붙이면 설계선·면적은 계단 없이 나오고 `berm` 값만 남아
            # 서로 어긋난다(2026-09-07). 값을 나르는 다른 사용자 키와 다른 점이다.
            stored_designs = await get_cross_section_designs(connection, route_id)
        stored_design: dict[str, Any] | None = None
        for record in stored_designs:
            if abs(float(record["chainage_m"]) - request.chainage_m) < 0.01:
                candidate = record.get("design")
                stored_design = candidate if isinstance(candidate, dict) else None
                break
        project_root = Path(resolve_stored_project_path(stored_path))
        samples, design_elevation, pavement_suggested, cross_record = await asyncio.to_thread(
            _read_cross_design_inputs,
            project_root,
            str(longitudinal["longitudinal_file_path"]),
            request.chainage_m,
        )
        design = compute_cross_design(
            samples,
            design_elevation,
            ground_type=request.ground_type,
            section_mode=request.section_mode,
            ditch_side=request.ditch_side,
            ditch_type=request.ditch_type,
            paved=request.paved,
            standard=request.standard_cross_section,
            rock_boundary_offset_m=request.rock_boundary_offset_m,
            two_stage_slope=request.two_stage_slope,
            # 측점별 암 절토 경사 — 요청값이 없으면 저장분에서 잇는다(2026-09-07).
            cut_slope_ratio=request.cut_slope_ratio or stored_cut_slope(stored_design or {}),
            ditch_enabled=request.ditch_enabled,
            ditch_choice=request.ditch_choice,
            surface_drop_m=ford_drop_at(request.chainage_m, ford_surface_drops(project_root)),
            berm=stored_berm(stored_design or {}),
            **curve_widening_args(cross_record),
        )
        # B06 지정 시점은 잠정치. B07 도면 확정 시 동일 엔진으로 재계산해 confirmed로 승격한다.
        design["status"] = "provisional"
        # 법정 근거 문구 표기용 — 사용자가 포장을 바꿔도 제안 여부는 그대로 남긴다.
        design["pavement_suggested"] = pavement_suggested
        # 표시 설정(측점 개별 반폭)은 계산 입력이 아니다 — 저장분에서 이월해 재계산이
        # 지우지 않게 한다(2026-08-06). 목록을 여기 다시 적지 않는다 — 서버의 한 벌은
        # `B06_Section_Router_Design.USER_TOUCHED_KEYS` 다(2026-09-07). 예전에는 여기에
        # 따로 적어 두어 `extra_spans` 가 빠져 있었다.
        if stored_design is not None:
            for key in USER_TOUCHED_KEYS:
                if stored_design.get(key) is not None:
                    design[key] = stored_design[key]
        async with pool.acquire() as connection:
            await connection.begin()
            try:
                updated = await update_cross_section_design(
                    connection,
                    route_id=route_id,
                    chainage_m=request.chainage_m,
                    design=design,
                    project_id=project_id,
                )
                await connection.commit()
            except Exception:
                await connection.rollback()
                raise
        if not updated:
            return JSONResponse(
                status_code=404,
                content={"status": "error", "message": "해당 측점 횡단 레코드가 없습니다."},
            )
        return CrossDesignResponse(chainage_m=request.chainage_m, design=design)
    except FileNotFoundError as exc:
        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
    except ValueError as exc:
        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
    except Exception:
        logger.exception(
            "B06 측점 횡단 설계 계산 실패: project_id=%s route_id=%s", project_id, route_id
        )
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "횡단 설계 계산 중 오류가 발생했습니다."},
        )