260725_0

2026-07-25 08:16:43 +09:00
parent 7b453d7a46
commit 42de855ab3
4 changed files with 138 additions and 111 deletions
@@ -223,8 +223,9 @@ async def send_analysis_error_email(
            _summary_row("오류 내용", html.escape(error_message)),
        ]
    )
+    safe_name = html.escape(project_name)
    body = f"""
-      <p>프로젝트 <strong>{html.escape(project_name)}</strong>의 지표면 분석 중 오류가 발생했습니다.</p>
+      <p>프로젝트 <strong>{safe_name}</strong>의 지표면 분석 중 오류가 발생했습니다.</p>
      <div class="box">
        {rows}
      </div>
@@ -16,10 +16,7 @@ from B05_wf2_Route.B05_wf2_Route_Debug import log_b05_debug
 from B05_wf2_Route.B05_wf2_Route_Engine import run_route_design
 from B05_wf2_Route.B05_wf2_Route_Engine_Grade import GradeDesignOptions, resolve_grade_options
 from B05_wf2_Route.B05_wf2_Route_Engine_Grade_Profile import rebuild_alignment_profile
-from B05_wf2_Route.B05_wf2_Route_Engine_Sections import (
-    generate_irregular_sections,
-    run_section_generation,
-)
+from B05_wf2_Route.B05_wf2_Route_Engine_Sections import run_section_generation
 from B05_wf2_Route.B05_wf2_Route_Engine_Sections_Core import SectionGenerationOptions
 from B05_wf2_Route.B05_wf2_Route_Repository import (
    confirm_route,
@@ -31,6 +28,10 @@ from B05_wf2_Route.B05_wf2_Route_Repository import (
    insert_route_points,
    update_longitudinal_grade_summary,
 )
+from B05_wf2_Route.B05_wf2_Route_Router_Confirm import (
+    _append_irregular_cross_sections,
+    _merge_uphill_overrides_into_longitudinal,
+)
 from B05_wf2_Route.B05_wf2_Route_Schema import (
    GRADE_PERCENT_FIELDS,
    ContourIntervalUpdateRequest,
@@ -113,111 +114,6 @@ def _normalized_route_params(params: dict[str, Any] | None) -> dict[str, Any] |
    }


-def _section_options_from_stored(stored: dict[str, Any] | None) -> SectionGenerationOptions:
-    """저장된 종횡단 옵션(단일 소스)만으로 옵션을 재구성한다(확정 시 비정규 측점 샘플링용)."""
-    defaults = SectionGenerationOptions()
-    stored = stored or {}
-    return SectionGenerationOptions(
-        station_interval_m=stored.get("station_interval_m") or defaults.station_interval_m,
-        cross_half_width_m=stored.get("cross_half_width_m") or defaults.cross_half_width_m,
-        cross_sample_interval_m=stored.get("cross_sample_interval_m")
-        or defaults.cross_sample_interval_m,
-        long_sample_interval_m=stored.get("long_sample_interval_m")
-        or defaults.long_sample_interval_m,
-        include_endpoint=defaults.include_endpoint,
-    )
-
-
-def _merge_uphill_overrides_into_longitudinal(
-    project_root: Path, longitudinal_file_path: str, overrides: list[dict[str, Any]]
-) -> None:
-    """종단 정본 파일 stations의 uphill_side를 사용자 변경값으로 덮어쓴다.
-
-    solve가 자동 판정한 상단측(=측구 방향)을 3D 램프 클릭으로 바꾼 경우, 확정 시점에
-    정본에 반영해 B06이 값만 읽으면 되게 한다. 사용자 지정임을 소스 필드로 남긴다.
-    """
-    if not overrides:
-        return
-    root = project_root.resolve()
-    path = (root / longitudinal_file_path).resolve()
-    if root not in path.parents or not path.is_file():
-        return
-    data = json.loads(path.read_text(encoding="utf-8"))
-    stations = data.get("stations")
-    if not isinstance(stations, list):
-        return
-    by_chainage = {round(float(item["chainage_m"]), 3): str(item["side"]) for item in overrides}
-    changed = False
-    for station in stations:
-        side = by_chainage.get(round(float(station.get("chainage_m", -1.0)), 3))
-        if side is None:
-            continue
-        station["uphill_side"] = side
-        station["uphill_side_source"] = "user"
-        changed = True
-    if changed:
-        data["stations"] = stations
-        atomic_write_json(path, data)
-
-
-def _merge_irregular_into_longitudinal(
-    project_root: Path, longitudinal_file_path: str, irregular_stations: list[dict[str, Any]]
-) -> None:
-    """종단 정본 파일의 stations에 비정규 측점을 병합한다(기존 비정규는 교체·정렬).
-
-    B06 상세는 종단 파일의 stations를 읽고 그에 맞는 cross 파일만 노출하므로, 이 병합과
-    (엔진이 이미 쓴) cross 파일만으로 다음 페이지에 횡단이 나타난다. 계획선·규칙 측점·표고
-    샘플은 손대지 않는다. 파일 기반이라 재확정해도 중복이 생기지 않는다.
-    """
-    root = project_root.resolve()
-    path = (root / longitudinal_file_path).resolve()
-    if root not in path.parents or not path.is_file():
-        return
-    data = json.loads(path.read_text(encoding="utf-8"))
-    stations = data.get("stations")
-    if not isinstance(stations, list):
-        return
-    regular = [station for station in stations if station.get("kind") != "irregular"]
-    merged = regular + list(irregular_stations)
-    merged.sort(key=lambda station: float(station.get("chainage_m", 0.0)))
-    data["stations"] = merged
-    atomic_write_json(path, data)
-
-
-async def _append_irregular_cross_sections(
-    connection: aiomysql.Connection,
-    project_id: UUID,
-    route: dict[str, Any],
-    request: RouteConfirmRequest,
-) -> None:
-    """확정 시 비정규 측점의 횡단을 생성해 종단 파일에 병합한다(파일 기반, 비치명적 호출용)."""
-    stored_path = await get_project_storage_relative_path(connection, project_id)
-    project_root = Path(resolve_stored_project_path(stored_path))
-    stored_options = await get_latest_section_options(connection, project_id)
-    crs_epsg = await get_surface_crs_epsg(connection, project_id, request.surface_model_id)
-    irregular_stations = await asyncio.to_thread(
-        generate_irregular_sections,
-        project_root,
-        str(route["route_data_path"]),
-        request.filter_key,
-        request.method,
-        request.smooth,
-        extra_stations=request.extra_stations(),
-        options=_section_options_from_stored(stored_options),
-        crs=f"EPSG:{crs_epsg}" if crs_epsg is not None else None,
-    )
-    if not irregular_stations:
-        return
-    longitudinal = await get_longitudinal_section(connection, project_id, route["id"])
-    if longitudinal:
-        await asyncio.to_thread(
-            _merge_irregular_into_longitudinal,
-            project_root,
-            str(longitudinal["longitudinal_file_path"]),
-            irregular_stations,
-        )
-
-
 def _grade_options(
    request: RouteSolveRequest, stored_grade_options: dict[str, Any] | None
 ) -> GradeDesignOptions:
@@ -0,0 +1,130 @@
+"""B05 경로 확정 시 종단 정본 파일에 비정규 측점·상단측 변경을 병합하는 헬퍼.
+
+라우터(`B05_wf2_Route_Router.py`)의 confirm 엔드포인트가 호출한다. 파일 기반 병합이라
+재확정해도 중복이 생기지 않으며, 모든 호출은 비치명적(실패해도 경로 확정은 진행)이다.
+"""
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+import aiomysql
+
+from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
+from B05_wf2_Route.B05_wf2_Route_Engine_Sections import generate_irregular_sections
+from B05_wf2_Route.B05_wf2_Route_Engine_Sections_Core import SectionGenerationOptions
+from B05_wf2_Route.B05_wf2_Route_Repository import get_surface_crs_epsg
+from B05_wf2_Route.B05_wf2_Route_Schema import RouteConfirmRequest
+from B06_wf3_ProfileCross.B06_wf3_ProfileCross_Repository import (
+    get_latest_section_options,
+    get_longitudinal_section,
+)
+from common_util.common_util_json import atomic_write_json
+from common_util.common_util_storage import resolve_stored_project_path
+
+
+def _section_options_from_stored(stored: dict[str, Any] | None) -> SectionGenerationOptions:
+    """저장된 종횡단 옵션(단일 소스)만으로 옵션을 재구성한다(확정 시 비정규 측점 샘플링용)."""
+    defaults = SectionGenerationOptions()
+    stored = stored or {}
+    return SectionGenerationOptions(
+        station_interval_m=stored.get("station_interval_m") or defaults.station_interval_m,
+        cross_half_width_m=stored.get("cross_half_width_m") or defaults.cross_half_width_m,
+        cross_sample_interval_m=stored.get("cross_sample_interval_m")
+        or defaults.cross_sample_interval_m,
+        long_sample_interval_m=stored.get("long_sample_interval_m")
+        or defaults.long_sample_interval_m,
+        include_endpoint=defaults.include_endpoint,
+    )
+
+
+def _merge_uphill_overrides_into_longitudinal(
+    project_root: Path, longitudinal_file_path: str, overrides: list[dict[str, Any]]
+) -> None:
+    """종단 정본 파일 stations의 uphill_side를 사용자 변경값으로 덮어쓴다.
+
+    solve가 자동 판정한 상단측(=측구 방향)을 3D 램프 클릭으로 바꾼 경우, 확정 시점에
+    정본에 반영해 B06이 값만 읽으면 되게 한다. 사용자 지정임을 소스 필드로 남긴다.
+    """
+    if not overrides:
+        return
+    root = project_root.resolve()
+    path = (root / longitudinal_file_path).resolve()
+    if root not in path.parents or not path.is_file():
+        return
+    data = json.loads(path.read_text(encoding="utf-8"))
+    stations = data.get("stations")
+    if not isinstance(stations, list):
+        return
+    by_chainage = {round(float(item["chainage_m"]), 3): str(item["side"]) for item in overrides}
+    changed = False
+    for station in stations:
+        side = by_chainage.get(round(float(station.get("chainage_m", -1.0)), 3))
+        if side is None:
+            continue
+        station["uphill_side"] = side
+        station["uphill_side_source"] = "user"
+        changed = True
+    if changed:
+        data["stations"] = stations
+        atomic_write_json(path, data)
+
+
+def _merge_irregular_into_longitudinal(
+    project_root: Path, longitudinal_file_path: str, irregular_stations: list[dict[str, Any]]
+) -> None:
+    """종단 정본 파일의 stations에 비정규 측점을 병합한다(기존 비정규는 교체·정렬).
+
+    B06 상세는 종단 파일의 stations를 읽고 그에 맞는 cross 파일만 노출하므로, 이 병합과
+    (엔진이 이미 쓴) cross 파일만으로 다음 페이지에 횡단이 나타난다. 계획선·규칙 측점·표고
+    샘플은 손대지 않는다. 파일 기반이라 재확정해도 중복이 생기지 않는다.
+    """
+    root = project_root.resolve()
+    path = (root / longitudinal_file_path).resolve()
+    if root not in path.parents or not path.is_file():
+        return
+    data = json.loads(path.read_text(encoding="utf-8"))
+    stations = data.get("stations")
+    if not isinstance(stations, list):
+        return
+    regular = [station for station in stations if station.get("kind") != "irregular"]
+    merged = regular + list(irregular_stations)
+    merged.sort(key=lambda station: float(station.get("chainage_m", 0.0)))
+    data["stations"] = merged
+    atomic_write_json(path, data)
+
+
+async def _append_irregular_cross_sections(
+    connection: aiomysql.Connection,
+    project_id: UUID,
+    route: dict[str, Any],
+    request: RouteConfirmRequest,
+) -> None:
+    """확정 시 비정규 측점의 횡단을 생성해 종단 파일에 병합한다(파일 기반, 비치명적 호출용)."""
+    stored_path = await get_project_storage_relative_path(connection, project_id)
+    project_root = Path(resolve_stored_project_path(stored_path))
+    stored_options = await get_latest_section_options(connection, project_id)
+    crs_epsg = await get_surface_crs_epsg(connection, project_id, request.surface_model_id)
+    irregular_stations = await asyncio.to_thread(
+        generate_irregular_sections,
+        project_root,
+        str(route["route_data_path"]),
+        request.filter_key,
+        request.method,
+        request.smooth,
+        extra_stations=request.extra_stations(),
+        options=_section_options_from_stored(stored_options),
+        crs=f"EPSG:{crs_epsg}" if crs_epsg is not None else None,
+    )
+    if not irregular_stations:
+        return
+    longitudinal = await get_longitudinal_section(connection, project_id, route["id"])
+    if longitudinal:
+        await asyncio.to_thread(
+            _merge_irregular_into_longitudinal,
+            project_root,
+            str(longitudinal["longitudinal_file_path"]),
+            irregular_stations,
+        )
@@ -509,7 +509,7 @@ export async function renderB07DesignDetail(root: HTMLElement): Promise<void> {
  // TODO: [임시 테스트용] B08 강제 이동 버튼 (테스트 완료 후 제거 예정)
  const tempB08Btn = createButton({
    label: "[임시] B08 이동",
-    variant: "outlined",
+    variant: "ghost",
    onClick: () => {
      if (projectId) goToWorkflowStage(projectId, WORKFLOW_STEP_ROUTES[5]);
    },