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,
) -> int:
    """구조물 목록을 정본에 덮어쓰고 새 판번호를 돌려준다."""
    items = list(structures)
    _validate_types(items)

    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:
    """레지스트리에 없는 타입, 다른 정본이 관리하는 타입(배관)을 걸러낸다."""
    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} 정본이 관리합니다 — "
                "구조물 목록에 저장할 수 없습니다."
            )