Aislo/B06_wf3_ProfileCross/B06_wf3_ProfileCross_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, Body, Depends
from fastapi.responses import JSONResponse

from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B05_wf2_Route.B05_wf2_Route_Engine_Sections import (
    cross_filename,
    prune_stale_cross_files,
    run_section_generation,
)
from B05_wf2_Route.B05_wf2_Route_Engine_Sections_Core import SectionGenerationOptions
from B05_wf2_Route.B05_wf2_Route_Router_Confirm import _merge_uphill_overrides_into_longitudinal
from B06_wf3_ProfileCross.B06_wf3_ProfileCross_Engine_Design import compute_cross_design
from B06_wf3_ProfileCross.B06_wf3_ProfileCross_Repository import (
    confirm_sections_for_route,
    count_cross_sections,
    create_longitudinal_section,
    delete_sections_for_route,
    get_confirmed_route_context,
    get_cross_section_designs,
    get_cross_sections_missing_design_chainages,
    get_latest_section_options,
    get_longitudinal_section,
    get_project_standard_cross_section,
    get_route_generation_source,
    insert_cross_sections,
    list_recent_company_projects,
    merge_cross_section_design_patch,
    merge_longitudinal_section_data,
    merge_longitudinal_section_options,
    update_cross_section_design,
)
from B06_wf3_ProfileCross.B06_wf3_ProfileCross_Schema import (
    CompanyStandardListResponse,
    CompanyStandardProject,
    CompanyStandardResponse,
    CrossDesignRequest,
    CrossDesignResponse,
    SectionConfirmRequest,
    SectionConfirmResponse,
    SectionContextResponse,
    SectionDetailResponse,
    SectionOptionDefaults,
    SectionRegenerateRequest,
    SectionSummaryResponse,
)
from common_util.common_util_auth import verify_session
from common_util.common_util_route_profile import design_elevation_from_longitudinal
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 complete_stage, get_workflow_state
from config.config_db import get_db_pool
from config.config_system import (
    EARTHWORK_CONVERSION_FACTORS,
    FOREST_ROAD_MIN_WIDTH_M,
    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 기반 생성 기본값을 반환한다."""
    pool = get_db_pool()
    try:
        async with pool.acquire() as connection:
            route_context = await get_confirmed_route_context(connection, project_id)
            surface_params = await get_surface_confirmation_params(connection, str(project_id))

        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,
            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,
            rock_boundary_default_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
            rock_boundary_step_m=STANDARD_ROCK_BOUNDARY_STEP_M,
            earthwork_conversion=EARTHWORK_CONVERSION_FACTORS,
        )
    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("종횡단 상세 파일 형식이 올바르지 않습니다.")
    return {"longitudinal": longitudinal, "cross_sections": cross_sections}


@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 렌더링용 종단·횡단 원시 샘플을 반환한다."""
    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)
            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["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
        # 지정값이 없는 측점은 기본값(토사/좌절토)으로 즉석 계산해 프리뷰로 채운다.
        # (미저장 프리뷰: 실제 저장은 사용자가 카드를 조작하거나 확정할 때 이뤄진다.)
        await asyncio.to_thread(
            _attach_default_designs, detail["longitudinal"], detail["cross_sections"]
        )
        return SectionDetailResponse(**detail)
    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_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"]
            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"]),
                options=_regeneration_options(
                    stored_options,
                    route_stage.get("params") if route_stage else None,
                    request.cross_half_width_m,
                ),
                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"]
        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": "종횡단 재생성 처리 중 오류가 발생했습니다."},
        )


def _read_cross_design_inputs(
    project_root: Path, longitudinal_file_path: str, chainage_m: float
) -> tuple[list[dict], float | None, bool]:
    """측점 하나의 지반 샘플·계획고·포장 제안 여부를 파일에서 읽는다 (경로 이탈 검증 포함)."""
    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("종단면 상세 파일을 찾을 수 없습니다.")
    longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8"))
    design_elevation = design_elevation_from_longitudinal(longitudinal, chainage_m)
    pavement_suggested = _pavement_suggestions(longitudinal).get(round(chainage_m, 3), False)

    cross_dir = longitudinal_path.parent.parent / "cross_sections"
    cross_path = (cross_dir / cross_filename(chainage_m)).resolve()
    if cross_dir.resolve() not in cross_path.parents or not cross_path.is_file():
        raise FileNotFoundError("해당 측점의 횡단 상세 파일을 찾을 수 없습니다.")
    cross = json.loads(cross_path.read_text(encoding="utf-8"))
    samples = cross.get("samples") if isinstance(cross, dict) else None
    if not isinstance(samples, list):
        raise ValueError("횡단 상세 파일 형식이 올바르지 않습니다.")
    return samples, design_elevation, pavement_suggested


def _pavement_suggestions(longitudinal: dict[str, Any]) -> dict[float, bool]:
    """측점별 포장 제안(B05 solve가 법정 경사 기준으로 판정) 매핑을 만든다."""
    stations = longitudinal.get("stations")
    mapping: dict[float, bool] = {}
    if isinstance(stations, list):
        for station in stations:
            suggested = station.get("pavement_suggested")
            if isinstance(suggested, bool):
                mapping[round(float(station.get("chainage_m", 0.0)), 3)] = suggested
    return mapping


def _default_section_modes(longitudinal: dict[str, Any]) -> dict[float, str]:
    """측점별 기본 단면유형 매핑: 상단측(uphill_side)이 절토측이 되는 편절편성.

    B05 solve가 자동 판정하고 사용자가 3D 램프로 바꾼 값(확정 시 정본 병합)을 그대로
    소비한다. 판정 불가 측점은 매핑에서 빠지고 호출부가 좌절토로 폴백한다.
    """
    stations = longitudinal.get("stations")
    mapping: dict[float, str] = {}
    if isinstance(stations, list):
        for station in stations:
            side = station.get("uphill_side")
            if side in ("left", "right"):
                mapping[round(float(station.get("chainage_m", 0.0)), 3)] = f"{side}_cut"
    return mapping


def _attach_default_designs(
    longitudinal: dict[str, Any], cross_sections: list[dict[str, Any]]
) -> None:
    """지정 설계가 없는 횡단에 기본값(토사 + 상단측 절토) 프리뷰 설계를 즉석 계산해 얹는다.

    detail 조회가 이미 읽어온 samples와 종단 계획선을 그대로 써서 추가 파일 I/O 없이
    전 측점 프리뷰를 만든다(미저장). 계산 불가 측점은 건너뛴다.
    """
    default_modes = _default_section_modes(longitudinal)
    pavement = _pavement_suggestions(longitudinal)
    for section in cross_sections:
        if section.get("design"):
            continue
        try:
            chainage_m = float(section.get("chainage_m", 0.0))
            suggested = pavement.get(round(chainage_m, 3), False)
            design = compute_cross_design(
                section.get("samples", []),
                design_elevation_from_longitudinal(longitudinal, chainage_m),
                ground_type="soil",
                section_mode=default_modes.get(round(chainage_m, 3), "left_cut"),
                paved=suggested,
            )
            design["status"] = "provisional"
            design["pavement_suggested"] = suggested
            section["design"] = design
        except (ValueError, KeyError):
            continue


def _compute_default_designs(
    project_root: Path,
    longitudinal_file_path: str,
    chainages: list[float],
    standard: dict[str, Any] | None = None,
) -> list[tuple[float, dict[str, Any]]]:
    """미지정 측점들을 기본값(토사/좌절토)으로 계산한 (chainage, design) 목록을 만든다.

    standard가 오면(확정 요청의 패널 편집값) 그 값으로 표준단면 기하를 계산한다.
    계획고 부재 등으로 계산 불가한 측점은 조용히 건너뛴다(확정을 막지 않기 위함).
    """
    root = project_root.resolve()
    longitudinal_path = (root / longitudinal_file_path).resolve()
    if root not in longitudinal_path.parents or not longitudinal_path.is_file():
        return []
    longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8"))
    cross_dir = longitudinal_path.parent.parent / "cross_sections"
    default_modes = _default_section_modes(longitudinal)
    pavement = _pavement_suggestions(longitudinal)
    results: list[tuple[float, dict[str, Any]]] = []
    for chainage_m in chainages:
        cross_path = cross_dir / cross_filename(chainage_m)
        if not cross_path.is_file():
            continue
        try:
            cross = json.loads(cross_path.read_text(encoding="utf-8"))
            samples = cross.get("samples")
            if not isinstance(samples, list):
                continue
            suggested = pavement.get(round(chainage_m, 3), False)
            design = compute_cross_design(
                samples,
                design_elevation_from_longitudinal(longitudinal, chainage_m),
                ground_type="soil",
                section_mode=default_modes.get(round(chainage_m, 3), "left_cut"),
                paved=suggested,
                standard=standard,
            )
            design["status"] = "provisional"
            design["pavement_suggested"] = suggested
            results.append((chainage_m, design))
        except (ValueError, KeyError, OSError, json.JSONDecodeError):
            continue
    return results


@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)
        project_root = Path(resolve_stored_project_path(stored_path))
        samples, design_elevation, pavement_suggested = 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,
            ditch_enabled=request.ditch_enabled,
        )
        # B06 지정 시점은 잠정치. B07 도면 확정 시 동일 엔진으로 재계산해 confirmed로 승격한다.
        design["status"] = "provisional"
        # 법정 근거 문구 표기용 — 사용자가 포장을 바꿔도 제안 여부는 그대로 남긴다.
        design["pavement_suggested"] = pavement_suggested
        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": "횡단 설계 계산 중 오류가 발생했습니다."},
        )


@router.post("/{project_id}/sections/{route_id}/confirm", response_model=SectionConfirmResponse)
async def confirm_sections(
    project_id: UUID,
    route_id: int,
    request: SectionConfirmRequest | None = Body(default=None),
) -> SectionConfirmResponse | JSONResponse:
    """경로의 종횡단면을 확정(CONFIRMED)한다.

    지반유형을 지정하지 않은 측점은 기본값(토사/좌절토)으로 자동 채운 뒤 확정한다.
    표준 횡단면 설정값이 함께 오면 longitudinal_sections.data.options에 저장한다.
    """
    pool = get_db_pool()
    try:
        async with pool.acquire() as connection:
            existing = await get_longitudinal_section(connection, project_id, route_id)
            if not existing:
                return JSONResponse(
                    status_code=404,
                    content={"status": "error", "message": "확정할 종횡단이 없습니다."},
                )
            stored_path = await get_project_storage_relative_path(connection, project_id)
            missing = await get_cross_sections_missing_design_chainages(connection, route_id)

        # 미지정 측점을 기본값으로 계산해 채운다 (계산 불가 측점은 조용히 건너뜀).
        default_designs: list[tuple[float, dict[str, Any]]] = []
        if missing:
            project_root = Path(resolve_stored_project_path(stored_path))
            default_designs = await asyncio.to_thread(
                _compute_default_designs,
                project_root,
                str(existing["longitudinal_file_path"]),
                missing,
                request.standard_cross_section if request else None,
            )

        async with pool.acquire() as connection:
            await connection.begin()
            try:
                for chainage_m, design in default_designs:
                    await update_cross_section_design(
                        connection,
                        route_id=route_id,
                        chainage_m=chainage_m,
                        design=design,
                        project_id=project_id,
                    )
                if request and request.standard_cross_section:
                    await merge_longitudinal_section_options(
                        connection,
                        route_id=route_id,
                        options_patch={"standard_cross_section": request.standard_cross_section},
                    )
                # 유토곡선 결과는 생성 옵션이 아니므로 data.options가 아니라 최상위 키에 둔다.
                if request and request.mass_haul:
                    await merge_longitudinal_section_data(
                        connection,
                        route_id=route_id,
                        data_patch={"mass_haul": request.mass_haul},
                    )
                # 프론트 세션 보관값(암 경계선 오프셋 등)을 측점별 design에 병합.
                if request and request.cross_patches:
                    for patch_item in request.cross_patches:
                        patch: dict[str, Any] = {}
                        if patch_item.rock_boundary_offset_m is not None:
                            patch["rock_boundary_offset_m"] = patch_item.rock_boundary_offset_m
                        if patch:
                            await merge_cross_section_design_patch(
                                connection,
                                route_id=route_id,
                                chainage_m=patch_item.chainage_m,
                                patch=patch,
                            )
                await confirm_sections_for_route(connection, route_id)
                async with connection.cursor() as cursor:
                    await complete_stage(cursor, str(project_id), 3)
                await connection.commit()
            except Exception:
                await connection.rollback()
                raise

        # 측구 방향(design.ditch_side) 변경을 B05 종단 정본 stations.uphill_side에 역반영한다(E-7).
        # 파일 기반·비치명적: 실패해도 확정은 유지한다.
        try:
            async with pool.acquire() as connection:
                designs = await get_cross_section_designs(connection, route_id)
            overrides = [
                {"chainage_m": record["chainage_m"], "side": record["design"]["ditch_side"]}
                for record in designs
                if isinstance(record.get("design"), dict)
                and record["design"].get("ditch_side") in ("left", "right")
            ]
            if overrides:
                await asyncio.to_thread(
                    _merge_uphill_overrides_into_longitudinal,
                    Path(resolve_stored_project_path(stored_path)),
                    str(existing["longitudinal_file_path"]),
                    overrides,
                )
        except Exception:
            logger.exception(
                "B06 측구 방향 B05 역반영 실패 (확정은 유지): project_id=%s route_id=%s",
                project_id,
                route_id,
            )
        return SectionConfirmResponse(project_id=str(project_id), route_id=route_id)
    except Exception:
        logger.exception("B06 종횡단 확정 실패: project_id=%s", project_id)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": "종횡단 확정 처리 중 오류가 발생했습니다."},
        )