Aislo/B06_Section/B06_Section_Router_DesignSave_Route.py

"""노선 바뀐 [저장] 한 본문 몫(PLAN 472-6 6-B) — `B06_Section_Router_DesignSave` 가 부름.

노선 초안이 오면 차례 맨 앞:

    노선 저장(`/route/replan` 체인 · 관 · 구조물 · 추가 측점 좌표 따라감 · 종횡단 · 계획고 ·
    횡단 이월)
    → 관 · 구조물 초안 다시 얹기(Node `facility.rebase` · 서버 따라감 위에)
    → 6-A 차례(새 노선) → 횡단 = 옛 노선 세션 칸을 새 측점에 넘긴 patch(Node `stations.carry`)

브라우저가 하던 다시 얹기(`rebasePipeDraft` · `carryAfterRouteSave`)와 같은 TS 를 서버가 부름 —
브라우저는 노선 초안 + 사용자 칸만 보냄. 손 상단측은 안 넘김(요구 274 · 새 지형이 정함).
"""

from __future__ import annotations

from typing import Any
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field

from common_util.common_util_calc_runner import run_calc


class RouteCarryRequest(BaseModel):
    """옛 노선 세션 칸 — 브라우저 `routeCarryInputs` 꼴 그대로(넘김 규칙은 TS 한 벌)."""

    model_config = ConfigDict(extra="forbid")

    #: 칸 이름 → 세션 원본(측점 열쇠 칸 다섯 · 누가거리 칸 · `kindreset`)
    states: dict[str, Any] = Field(default_factory=dict)
    #: 옛 노선 측점 열쇠 목록 [{key, chainageM}] · 브라우저가 못 세웠으면 null
    from_keys: list[dict[str, Any]] | None = None
    #: 측점을 덮는 관 · 구조물 [{id, kind, at, role}]
    owners: list[dict[str, Any]] = Field(default_factory=list)
    #: 관 세션 장부 {next, ids: [{id, at}]}
    pipe_ids: dict[str, Any] | None = None


class StepFailed(Exception):
    def __init__(self, step: str, message: str) -> None:
        super().__init__(message)
        self.step = step


def _node(ids: list[str], inputs: dict[str, Any], step: str) -> dict[str, Any]:
    out = run_calc(ids, inputs)
    if not isinstance(out, dict) or ids[0] not in out:
        raise StepFailed(step, "다시 얹기 계산(Node)을 못 돌렸습니다")
    return out[ids[0]]


def _pipes(listing: Any) -> list[dict[str, Any]]:
    """관 GET 응답 → 관 목록(지도용 lonlat 뺌 · 브라우저 `fetchDetailPipePoints` 짝)."""
    points = listing.get("pipe_points") if isinstance(listing, dict) else None
    return [{k: v for k, v in point.items() if k != "lonlat"} for point in points or []]


async def saved_facilities(project_id: UUID) -> dict[str, Any]:
    """노선 저장 직전 저장 관 · 구조물(다시 얹기 · 넘김의 `before`)."""
    from B04_PreProcess.B04_PreProcess_Router_Basins import get_pipe_points
    from B05_Profile.B05_Profile_Structures_Router import read_structures

    pipes = await get_pipe_points(project_id)
    structures = await read_structures(project_id)
    return {
        "pipes": _pipes(pipes),
        "structures": [
            item.model_dump(mode="json") for item in getattr(structures, "structures", None) or []
        ],
    }


async def rebase_drafts(project_id: UUID, request: Any, before: dict[str, Any]) -> None:
    """관 · 구조물 초안을 서버가 따라 옮긴 정본 위에(Node `facility.rebase`) — 본문을 제자리에서
    고침.

    구조물 판번호 = 노선 저장 뒤 지금 판(따라감이 판을 올림 · 옛 길 = 저장 뒤 받은 판).
    """
    from B04_PreProcess.B04_PreProcess_Router_Basins import get_pipe_points
    from B05_Profile.B05_Profile_Structures_Router import read_structures

    rebase: dict[str, Any] = {}
    points = (request.pipes or {}).get("points") if isinstance(request.pipes, dict) else None
    if points is not None:
        after = _pipes(await get_pipe_points(project_id))
        rebase["pipes"] = {"draft": points, "before": before["pipes"], "after": after}
    stored = None
    if request.structures is not None:
        stored = await read_structures(project_id)
        draft = [item.model_dump(mode="json") for item in request.structures.structures]
        after = [item.model_dump(mode="json") for item in stored.structures]
        rebase["structures"] = {"draft": draft, "before": before["structures"], "after": after}
    if not rebase:
        return
    out = _node(["facility.rebase"], {"rebase": rebase}, "route")
    if out.get("pipes") is not None:
        request.pipes = {**request.pipes, "points": out["pipes"]}
    if out.get("structures") is not None and stored is not None:
        request.structures = type(request.structures).model_validate(
            {"base_revision": stored.revision, "structures": out["structures"]}
        )


async def carry_cross(
    project_id: UUID,
    route_id: int,
    carry: RouteCarryRequest | None,
    before: dict[str, Any],
) -> dict[str, Any]:
    """옛 노선 세션 칸 → 새 노선 횡단 patch(Node `stations.carry`) → {patches, outside}."""
    from B04_PreProcess.B04_PreProcess_Router_Basins import get_pipe_points
    from B06_Section.B06_Section_Router import get_section_detail

    detail = await get_section_detail(project_id, route_id)
    if not hasattr(detail, "model_dump"):
        raise StepFailed("cross", "새 노선 횡단 상세를 못 읽었습니다")
    inputs = {
        "detail": detail.model_dump(mode="json"),
        "pipes_before": before["pipes"],
        "pipes_after": _pipes(await get_pipe_points(project_id)),
        **(carry or RouteCarryRequest()).model_dump(mode="json"),
    }
    return _node(["stations.carry"], {"carry": inputs}, "cross")