Aislo/B06_Section/B06_Section_Server_Calc_Prebuild.py

"""브라우저에서만 돌던 횡단 계산을 **서버가** 돌려 정본에 남긴다(2026-09-06).

대상 둘 —
  ① 구조물이 선 측점의 절·성토 면적: 기슭막이·세월교·BOX암거가 서면 성토 사면이 벽에서
     끊겨 지반선과 설계선이 이루는 폐회로가 달라진다.
  ② 그 면적을 쌓아 만드는 유토곡선.

왜 — 둘 다 화면에서만 돌아, 사용자가 B06 을 한 번도 안 열면 값이 없고 저장·확정 뒤에는
구조물을 모르는 표준값으로 되돌아갔다(CLAUDE.md 5장 「계산 자리」 — 금지 항목이던 자리).

**계산을 다시 짜지 않는다.** 화면이 쓰는 TS 를 Node 진입점(`B06_Section_Server_Calc_Node.ts`)
으로 감싸 그대로 돌린다. 파이썬으로 포팅하면 같은 기하가 두 벌이 되어 「그림은 이런데
수량은 저렇다」가 생긴다.

실패는 비치명적이다 — 보정 전(표준) 값이 그대로 남고 화면은 예전처럼 스스로 고친다.
"""

from __future__ import annotations

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

from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B06_Section.B06_Section_Repository import (
    get_longitudinal_section,
    merge_cross_section_design_patch,
    merge_longitudinal_section_data,
    update_cross_section_design,
)
from common_util.common_util_node_bundle import run_bundle_json
from common_util.common_util_storage import resolve_stored_project_path
from config.config_db import get_db_pool
from config.config_system import (
    EARTHWORK_CONVERSION_FACTORS,
    EARTHWORK_HAUL_EQUIPMENT_LIMITS_M,
    NATURAL_SPOIL_MIN_GROUND_SLOPE,
)

logger = logging.getLogger(__name__)

ROOT = Path(__file__).resolve().parents[1]
BUNDLE = ROOT / "config" / "server_calc_node" / "B06_Section_Server_Calc_Node.js"
_NPM_SCRIPT = "build:server-calc"
# 정본에 얹는 값만 받는다 — Node 가 다른 키를 내도 설계 데이터에 흘리지 않는다.
_AREA_KEYS = ("cut_area_m2", "fill_area_m2", "cut_soil_area_m2", "cut_rock_area_m2")


def _mass_haul_context() -> dict[str, Any]:
    """유토곡선 계산에 필요한 값 — 화면이 `sections/context`로 받는 것과 같은 상수다."""
    return {
        "earthwork_conversion": EARTHWORK_CONVERSION_FACTORS,
        "natural_spoil_min_ground_slope": NATURAL_SPOIL_MIN_GROUND_SLOPE,
        "haul_equipment_limits": [
            {"key": key, "max_distance_m": limit}
            for key, limit in EARTHWORK_HAUL_EQUIPMENT_LIMITS_M
        ],
    }


def _enforce_stored_designs(
    longitudinal: dict[str, Any],
    sections: list[dict[str, Any]],
    project_root: Path,
    standard: dict[str, Any] | None,
) -> None:
    """저장분 설계를 **쓰는 시점에** 바로잡는다 — 포장 구간·세월교 노면 하강.

    예전에는 상세를 **읽을 때마다** 돌려 화면이 볼 때만 맞았다(저장분은 낡은 채로).
    2026-09-06 사용자 확정대로 「읽기는 영구저장소에서 가져오기만」이므로 이쪽으로 옮겼다.
    """
    from B06_Section.B06_Section_Router_Design import (
        enforce_ford_surface_drops,
        enforce_pavement_ranges,
    )

    enforce_pavement_ranges(longitudinal, sections, project_root, standard)
    enforce_ford_surface_drops(longitudinal, sections, project_root, standard)


async def recompute_server_side(project_id: UUID | str, route_id: int) -> int:
    """구조물 면적·유토곡선을 다시 계산해 정본에 얹는다. 고친 측점 수를 돌려준다."""
    from B06_Section.B06_Section_Router import get_section_detail

    project_uuid = UUID(str(project_id))
    response = await get_section_detail(project_uuid, route_id)
    payload = getattr(response, "model_dump", None)
    if payload is None:  # JSONResponse = 실패
        logger.warning("서버 재계산: 종횡단 상세를 못 받음 (route_id=%s)", route_id)
        return 0
    detail = payload(mode="json")
    sections = detail.get("cross_sections") or []

    pool = get_db_pool()
    async with pool.acquire() as connection:
        stored_path = await get_project_storage_relative_path(connection, project_uuid)
        longitudinal_row = await get_longitudinal_section(connection, project_uuid, route_id)
    project_root = Path(resolve_stored_project_path(stored_path))
    from B06_Section.B06_Section_Router_Design import stored_standard_cross_section

    standard = stored_standard_cross_section(longitudinal_row)
    # 포장 구간·세월교 보정 — 고쳐진 설계 위에서 면적·유토곡선이 나와야 한다.
    before = [json.dumps(item.get("design"), sort_keys=True, default=str) for item in sections]
    await asyncio.to_thread(
        _enforce_stored_designs, detail.get("longitudinal") or {}, sections, project_root, standard
    )
    fixed = [
        item
        for index, item in enumerate(sections)
        if json.dumps(item.get("design"), sort_keys=True, default=str) != before[index]
    ]

    output = await asyncio.to_thread(
        run_bundle_json,
        BUNDLE,
        _NPM_SCRIPT,
        {"detail": detail, "context": _mass_haul_context()},
    )
    if not isinstance(output, dict):
        return 0
    rows = output.get("areas")
    mass_haul = output.get("mass_haul")

    updated = 0
    async with pool.acquire() as connection:
        # balloon 위치는 **사용자가 끌어 옮긴 화면값**이다 — 서버가 만들지 않으므로
        # 저장분에서 떼어 새 유토곡선에 도로 붙인다(2026-09-06).
        if isinstance(mass_haul, dict):
            existing = await get_longitudinal_section(connection, project_uuid, route_id)
            stored = (existing or {}).get("data") or {}
            offsets = (stored.get("mass_haul") or {}).get("balloon_offsets")
            if offsets is not None:
                mass_haul["balloon_offsets"] = offsets

        await connection.begin()
        try:
            for item in fixed:
                await update_cross_section_design(
                    connection,
                    route_id=route_id,
                    chainage_m=float(item.get("chainage_m") or 0.0),
                    design=item["design"],
                    project_id=project_uuid,
                )
            for row in rows if isinstance(rows, list) else []:
                patch: dict[str, Any] = {
                    key: float(row[key])
                    for key in _AREA_KEYS
                    if isinstance(row.get(key), (int, float))
                }
                if not patch:
                    continue
                if await merge_cross_section_design_patch(
                    connection,
                    route_id=route_id,
                    chainage_m=float(row["chainage_m"]),
                    patch=patch,
                ):
                    updated += 1
            if isinstance(mass_haul, dict):
                await merge_longitudinal_section_data(
                    connection, route_id=route_id, data_patch={"mass_haul": mass_haul}
                )
            await connection.commit()
        except Exception:
            await connection.rollback()
            raise
    logger.info(
        "서버 재계산: route_id=%s 설계 보정 %s곳, 구조물 면적 %s곳, 유토곡선 %s",
        route_id,
        len(fixed),
        updated,
        "갱신" if isinstance(mass_haul, dict) else "없음",
    )
    return updated