Aislo/B06_Section/B06_Section_Server_Calc_Prebuild.py

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

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

왜 — 사용자가 B06 을 한 번도 안 열어도 **초기값**에는 이 값이 있어야 한다.

**부르는 자리 둘** — 파일입력 자동설계 체인(초기값)과 [저장]·[확정](정본).
사용자가 화면을 만지는 **동안**은 브라우저 몫이지만(왕복 없이 즉시 따라와야 한다),
저장 시점은 서버가 다시 낸다(2026-09-06 저녁 사용자 확정). 이유는 속도가 아니라
**보안**이다 — 저장 경로가 브라우저 계산이면 유토 배분·운반거리 코드가 번들에 남아야
해서 화면에서 안 그려도 뺄 수 없다. 대가는 저장 대기 850ms → 약 980ms(Node 131ms).

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

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

from __future__ import annotations

import asyncio
import json
import logging
import time
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 B03_FileInput.B03_FileInput_Service_Chain import _log_steps
from B06_Section.B06_Section_Repository import (
    get_longitudinal_section,
    merge_longitudinal_section_data,
)
from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs
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, run_with_connection
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 가 다른 키를 내도 설계 데이터에 흘리지 않는다.
# ⚠ **TS 쪽 `STRUCTURE_ROW_KEYS` 와 짝이다.** 한쪽만 늘리면 Node 가 값을 내도 여기서 조용히
# 버려진다(2026-09-08 관 길이를 더하며 실제로 걸린 자리). 시험이 두 목록을 대조한다.
_AREA_KEYS = (
    "cut_area_m2",
    "fill_area_m2",
    "cut_soil_area_m2",
    "cut_rock_area_m2",
    "pipe_length_m",
)


async def haul_inputs_for(project_id: Any) -> dict[str, Any]:
    """B08 이 낸 **구조물 몫**(채집석 공제·구조물 잔토)을 받아 온다.

    ⚠ **여기서 다시 세지 않는다** — 두 값 다 B08 전개에서 나오는 것이라 이쪽이 세면
    같은 계산이 두 벌이 된다(CLAUDE.md 5장). 못 읽으면 빈 값으로 두고 **0 으로 눅이지 않는다**.
    """
    try:
        from B08_Quantity.B08_Quantity_Router_Material import project_haul_inputs

        data = await project_haul_inputs(project_id)
        return data if isinstance(data, dict) else {}
    except Exception:
        logger.exception("구조물 몫(공제·잔토) 조회 실패 — 값 없이 진행: project_id=%s", project_id)
        return {}


def _mass_haul_context(haul_inputs: dict[str, Any] | None = None) -> dict[str, Any]:
    """유토곡선 계산에 필요한 값 — 화면이 `sections/context`로 받는 것과 같은 상수다.

    ⚠ 채집석 공제(`collected_stone_deduction_m3`)만 상수가 아니라 **B08 이 내는 값**이다.
    `None` 은 「아직 안 옴」이고 `0` 은 「공제 없음」이라 **서로 다르다** — 값이 안 온 것을
    공제 0 으로 읽으면 조용히 넘어간다(2026-09-09 네 창 합의).

        채집석 공제는 사토에서 한 번만 뺀다.
        B08 은 소요량(collected_stone_deduction_m3, ㎥ 양수)을 내기만 하고 공제하지 않으며,
        빼는 자리는 유토곡선의 사토뿐이다 —
        실어 내는 몫(spoil_m3 − natural_spoil_m3)에서 먼저 빼고 모자라면 자연방토에서 뺀다.
    """
    inputs = haul_inputs or {}
    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
        ],
        # ⚠ B08 이 아직 이 값을 내지 않는다(2026-09-09) — 그때까지 `None`(아직 안 옴)이다.
        # 값을 내기 시작하면 여기에 실어 주기만 하면 통로가 이어진다.
        "collected_stone_deduction_m3": inputs.get("collected_stone_deduction_m3"),
        # 갈래별 채집석(2026-09-09 세 창 확정) — **벽 입적**이라 자연 축으로 보고 곡선 쪽에서
        # ×C 해 다짐 축에 맞춰 뺀다. 갈래를 못 가른 몫은 계수가 없어 **환산하지 않는다**.
        "collected_stone_by_ground_m3": inputs.get("collected_stone_by_ground_m3") or {},
        "collected_stone_ground_unknown_m3": inputs.get("collected_stone_ground_unknown_m3"),
        # 구조물 터파기 잔토(㎥, 양수) — **사토에 더한다**(공제는 빼고 이것은 더한다).
        #     구조물 잔토는 사토에 한 번만 더한다.
        #     B08 은 소요량(structure_spoil_m3, ㎥ 양수)을 내기만 하고,
        #     더하는 자리는 유토곡선의 사토뿐이다.
        "structure_spoil_m3": inputs.get("structure_spoil_m3"),
        # 측점별 잔토 — 오면 이쪽이 이긴다(구조물이 선 자리 잔량에 얹어 운반거리를 맞춘다).
        "structure_spoil_points": inputs.get("structure_spoil_points"),
    }


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_Engine_SpoilFill import enforce_spoil_fills
    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)
    # ⚠ 사토장은 **맨 뒤**다 — 앞의 두 보정이 설계를 다시 계산하면서 사토장 칸을 지운다.
    #   맨 뒤에 두면 그 결과 위에 사토장 단면이 얹힌다(2026-09-09).
    enforce_spoil_fills(longitudinal, sections, project_root, standard)


async def recompute_server_side(project_id: UUID | str, route_id: int) -> int:
    """포장 구간·세월교 노면 하강 보정에 더해 구조물 면적·유토곡선까지 Node 로 만들어 저장한다.

    부르는 자리 둘 — 파일입력 자동설계 체인(초기값)과 [저장]·[확정](정본). 저장 때도
    서버가 내는 것으로 2026-09-06 저녁 되돌렸다(그 사이 잠깐 브라우저 계산이었다):
    이유는 속도가 아니라 **배분·운반거리 코드를 브라우저 번들에서 빼기 위해서**다.
    """
    return await _recompute(project_id, route_id)


async def _recompute(project_id: UUID | str, route_id: int) -> int:
    from B06_Section.B06_Section_Router import get_section_detail

    project_uuid = UUID(str(project_id))
    marks = [("시작", time.perf_counter())]
    # 상세 만들기(파일 읽기 위주)와 DB 두 건은 서로 기다릴 이유가 없다 — 같이 보낸다.
    # 원격 DB 라 순차로 내면 왕복이 그대로 더해진다(질의 하나 약 12ms, 2026-09-06 실측).
    pool = get_db_pool()
    # 구조물 몫(채집석 공제·구조물 잔토)도 함께 받아 온다 — **B08 이 낸 값**이고, 안 넘기면
    # 통로만 있고 값이 안 흐른다(2026-09-09 실측: 공제가 늘 `None` 이라 사토가 안 줄었다).
    response, stored_path, longitudinal_row, haul_inputs = await asyncio.gather(
        get_section_detail(project_uuid, route_id),
        run_with_connection(get_project_storage_relative_path, project_uuid),
        run_with_connection(get_longitudinal_section, project_uuid, route_id),
        haul_inputs_for(project_uuid),
    )
    marks.append(("종횡단 상세+DB 조회(병렬)", time.perf_counter()))
    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 []
    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]
    ]

    marks.append(("포장·세월교 보정", time.perf_counter()))

    output = await asyncio.to_thread(
        run_bundle_json,
        BUNDLE,
        _NPM_SCRIPT,
        {
            "detail": detail,
            "context": _mass_haul_context(haul_inputs),
        },
    )
    marks.append(("Node 번들(면적·유토곡선)", time.perf_counter()))
    if not isinstance(output, dict):
        output = {}
    rows = output.get("areas")
    mass_haul = output.get("mass_haul")
    if not fixed and not rows and not mass_haul:
        return 0

    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:
            # 행마다 쓰면 원격 DB 왕복이 행 수만큼 난다(측정: 22행 670ms) — 한 문장으로 묶는다.
            await merge_cross_section_designs(
                connection,
                route_id=route_id,
                entries=[(float(item.get("chainage_m") or 0.0), item["design"]) for item in fixed],
                replace=True,
                project_id=project_uuid,
            )
            area_entries: list[tuple[float, dict[str, Any]]] = []
            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 patch:
                    area_entries.append((float(row["chainage_m"]), patch))
            updated = await merge_cross_section_designs(
                connection, route_id=route_id, entries=area_entries, replace=False
            )
            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
    marks.append(("정본 저장", time.perf_counter()))
    _log_steps("서버 재계산 내부", marks)
    logger.info(
        "서버 재계산: route_id=%s 설계 보정 %s곳, 구조물 면적 %s곳, 유토곡선 %s",
        route_id,
        len(fixed),
        updated,
        "갱신" if isinstance(mass_haul, dict) else "없음",
    )
    return updated