Aislo/B05_Profile/B05_Profile_Structures_Repository.py

"""B05 구조물 정본(`B05_Profile/route/structures.json`) 읽기·쓰기.

정본은 이 파일 **하나뿐**이다. DB에는 참조 메타(개수·revision)만 남긴다 — 같은 값을 두 곳에
두면 한쪽 저장이 실패했을 때 어느 쪽이 진짜인지 알 수 없다(관 매설 지점의 "복원 유령" 전례).

저장은 임시 파일에 쓰고 교체하는 방식이라, 쓰는 도중 죽어도 반쪽짜리 파일이 남지 않는다.
동시에 두 화면이 저장하면 `base_revision`이 어긋나 뒤엣것이 거절된다(앞의 편집을 덮지 않게).
"""

import json
import os
import uuid
from typing import Iterable

from B05_Profile.B05_Profile_Structures_Schema import StructureInstance, structure_type_map
from common_util.common_util_json import atomic_write_json

STRUCTURES_FILE_NAME = "structures.json"
_STAGE_DIR = os.path.join("B05_Profile", "route")


class StructureRevisionConflict(RuntimeError):
    """다른 저장이 먼저 반영되어 판번호가 어긋났다."""

    def __init__(self, expected: int, actual: int) -> None:
        super().__init__(
            f"구조물 정본이 이미 갱신되었습니다 (요청 판번호 {expected}, 현재 {actual}). "
            "최신 내용을 다시 불러온 뒤 저장해 주세요."
        )
        self.expected = expected
        self.actual = actual


def structures_file_path(project_root: str) -> str:
    """프로젝트 저장소 안의 구조물 정본 경로."""
    return os.path.join(project_root, _STAGE_DIR, STRUCTURES_FILE_NAME)


def load_structures(project_root: str) -> tuple[int, list[StructureInstance]]:
    """정본을 읽어 (판번호, 구조물 목록)을 돌려준다.

    파일이 없거나 깨졌으면 빈 정본(판번호 0)으로 본다 — 화면이 못 열리는 것보다 낫고,
    다음 저장이 온전한 파일로 덮어쓴다.
    """
    path = structures_file_path(project_root)
    if not os.path.exists(path):
        return 0, []
    try:
        with open(path, encoding="utf-8") as handle:
            payload = json.load(handle)
        revision = int(payload.get("revision", 0))
        structures = [
            StructureInstance.model_validate(item) for item in payload.get("structures", [])
        ]
    except (OSError, ValueError, TypeError):
        return 0, []
    return revision, structures


def save_structures(
    project_root: str,
    structures: Iterable[StructureInstance],
    *,
    base_revision: int,
    max_chainage_m: float | None = None,
) -> int:
    """구조물 목록을 정본에 덮어쓰고 새 판번호를 돌려준다.

    `max_chainage_m`는 노선 총연장(m) — 주어지면 범위 밖 배치를 거절한다.
    """
    items = list(structures)
    _validate_types(items)
    _validate_unique_ids(items)
    _validate_range(items, max_chainage_m)

    current_revision, _ = load_structures(project_root)
    if current_revision != base_revision:
        raise StructureRevisionConflict(base_revision, current_revision)

    for item in items:
        if not item.structure_id:
            item.structure_id = uuid.uuid4().hex

    revision = current_revision + 1
    payload = {
        "revision": revision,
        "structures": [item.model_dump(mode="json") for item in items],
    }
    atomic_write_json(structures_file_path(project_root), payload)
    return revision


def requires_downstream_invalidation(
    previous: Iterable[StructureInstance], current: Iterable[StructureInstance]
) -> bool:
    """구조물 변경이 B06 이후 결과를 못 쓰게 만드는지 판정한다.

    메모나 표시용 값만 바뀐 경우까지 후속 단계를 깨면, 사용자가 메모 한 줄 고칠 때마다
    횡단·수량을 다시 돌려야 한다. 그래서 설계에 실제로 영향을 주는 값
    (타입·위치·범위·측·오프셋·제원)만 비교한다. 목록 순서는 설계와 무관하므로 무시한다.
    """
    return _design_fingerprint(previous) != _design_fingerprint(current)


def _design_fingerprint(items: Iterable[StructureInstance]) -> set[str]:
    return {
        json.dumps(
            [
                item.structure_id,
                item.type_id,
                item.placement,
                item.chainage_m,
                item.start_m,
                item.end_m,
                item.side,
                item.offset_m,
                item.options,
                item.geometry,
            ],
            ensure_ascii=False,
            sort_keys=True,
        )
        for item in items
    }


def _validate_types(items: list[StructureInstance]) -> None:
    """레지스트리 대조 검증 — 타입 존재·관리 주체·배치형태·옵션까지 서버가 지킨다.

    화면만 믿으면 조작된 요청(placement 불일치·음수 제원·미정의 옵션)이 정본에
    들어간다(2026-08-16 크로스체크 지적 2). 정본에 닿는 마지막 관문은 여기다.
    """
    types = structure_type_map()
    for item in items:
        definition = types.get(item.type_id)
        if definition is None:
            raise ValueError(f"등록되지 않은 구조물 타입입니다: {item.type_id}")
        if definition.managed_by:
            raise ValueError(
                f"{definition.name}은(는) {definition.managed_by} 정본이 관리합니다 — "
                "구조물 목록에 저장할 수 없습니다."
            )
        if item.placement != definition.placement:
            raise ValueError(
                f"{definition.name}의 배치형태는 {definition.placement}인데 "
                f"{item.placement}로 보냈습니다."
            )
        _validate_options(item, definition)


def _validate_options(item: StructureInstance, definition) -> None:
    allowed = {option.key: option for option in definition.options}
    for key, value in item.options.items():
        option = allowed.get(key)
        if option is None:
            raise ValueError(f"{definition.name}에 정의되지 않은 옵션입니다: {key}")
        if option.input == "number":
            if isinstance(value, bool) or not isinstance(value, (int, float)):
                raise ValueError(f"{definition.name}의 {option.label}은(는) 숫자여야 합니다.")
            if not (value == value and abs(value) != float("inf")) or value < 0:
                raise ValueError(f"{definition.name}의 {option.label}은(는) 0 이상이어야 합니다.")
        elif option.input == "select" and option.choices and str(value) not in option.choices:
            raise ValueError(f"{definition.name}의 {option.label} 값이 선택지에 없습니다: {value}")
    for option in definition.options:
        if option.required and item.options.get(option.key) in (None, ""):
            raise ValueError(
                f"{definition.name}의 {option.label}은(는) 필수 입력입니다 — "
                "미확정 항목이라 기본값이 없습니다."
            )


def _validate_unique_ids(items: list[StructureInstance]) -> None:
    seen: set[str] = set()
    for item in items:
        if not item.structure_id:
            continue
        if item.structure_id in seen:
            raise ValueError(f"구조물 식별자가 중복되었습니다: {item.structure_id}")
        seen.add(item.structure_id)


def _validate_range(items: list[StructureInstance], max_chainage_m: float | None) -> None:
    """노선 연장을 알 때만 범위를 지킨다 — 연장 밖 배치는 도면·수량 어디에도 못 실린다."""
    if max_chainage_m is None:
        return
    limit = max_chainage_m + 1e-6
    for item in items:
        positions = [item.chainage_m, item.start_m, item.end_m]
        if any(value is not None and value > limit for value in positions):
            raise ValueError(
                f"구조물 위치가 노선 연장({max_chainage_m:.1f} m)을 벗어났습니다: {item.type_id}"
            )