Aislo/M02_MasterTemplete/M02_MasterTemplete_Store.py

"""M02 마스터 템플릿 — 시스템 층 저장소 (`resources/master_template/{종류}/<이름>.json`).

M01 `Store` 방식 — 판(파일 sha256 앞 16자) · 판이 다르면 409 · 원자 쓰기.
권한은 등록하는 쪽(`main.py`)이 붙임.
"""

from __future__ import annotations

import hashlib
import json
import re
import threading
from datetime import datetime
from pathlib import Path
from typing import Any

from common_util import common_util_spreadsheet as spreadsheet
from common_util.common_util_json import atomic_write_json
from M02_MasterTemplete import M02_Template_Layers as layers

FOLDER: Path = Path(__file__).resolve().parent.parent / "resources" / "master_template"
KINDS = layers.KINDS  # 시험은 FOLDER 를 사본으로 바꿈
# 구조물 도면 = 구조물집계표 열 id 마다 하나(파일 이름 = 열 id) · 도각 없는 전용 화면(그린 만큼이 설계 영역)
# 산출근거 = 같은 열 id 로 따로 한 파일(`basis/<열 id>.json` · 스프레드시트 통합문서)
STRUCTURE_TABLE = "구조물집계표"
_BASIS_HEAD = ("공종", "규격", "산출 내역", "단위", "수량(m당)")
_BASIS_HEAD_STYLE = {
    "굵게": True,
    "가로": "center",
    "세로": "center",
    "테두리": {side: {"선": "thin"} for side in ("위", "아래", "왼", "오른")},
}
_NUMBER = re.compile(r"^구-(\d+)$")
_BAD_NAME = re.compile(r'[\\/:*?"<>|\x00-\x1f]|\.\.')
# 도각 없는 빈 CAD 문서의 기본 도면층 — 없으면 B07 CAD 가 못 엶("Cannot read properties
# of undefined (reading 'length')") · CAD 쪽도 없으면 새로 채우지만(2026-09-27 sub1 고침)
# 뿌리도 처음부터 채움
_BLANK_LAYER = {"id": "기본", "name": "기본", "isVisible": True, "isLocked": False}
# ponytail: 저장은 한 번에 하나(프로세스 안 잠금) · 서버를 여럿 띄우면 파일 잠금으로
_LOCK = threading.Lock()


class StoreError(Exception):
    def __init__(self, status: int, detail):
        super().__init__(detail)
        self.status, self.detail = status, detail


def version_of(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()[:16]


def _path(kind: str, name: str) -> Path:
    if kind not in KINDS:
        raise StoreError(404, f"없는 종류 「{kind}」")
    if not name or name != name.strip() or name.startswith(".") or _BAD_NAME.search(name):
        raise StoreError(400, f"쓸 수 없는 글자가 있는 이름 「{name}」")
    return FOLDER / kind / f"{name}.json"


def _info(kind: str, path: Path) -> dict[str, Any]:
    raw = path.read_bytes()
    stamp = datetime.fromtimestamp(path.stat().st_mtime).isoformat(timespec="seconds")
    return {"종류": kind, "이름": path.stem, "판": version_of(raw), "수정일": stamp}


def list_all() -> list[dict[str, Any]]:
    rows = [
        _info(kind, p)
        for kind in KINDS
        for p in sorted((FOLDER / kind).glob("*.json"))
        if not p.name.startswith(".")
    ]
    return rows


def read(kind: str, name: str) -> dict[str, Any]:
    path = _path(kind, name)
    if not path.is_file():
        raise StoreError(404, f"없는 양식 「{name}」")
    raw = path.read_bytes()
    return {"종류": kind, "이름": name, "판": version_of(raw), "문서": json.loads(raw)}


def _check_skeleton(kind: str, doc: Any) -> None:
    """빈 문서·뼈대 없는 문서는 거절 — 층 저장과 같은 규칙(`layers.check_skeleton`)."""
    try:
        layers.check_skeleton(kind, doc)
    except ValueError as e:
        raise StoreError(400, str(e)) from e


def write(kind: str, name: str, version: str, doc: Any) -> dict[str, Any]:
    """`version` 이 빈 글이면 새로 만듦(이미 있으면 409) · 아니면 그 판일 때만 덮어씀.
    구조물집계표를 저장하면 — 새 열의 빈 구조물 도면 · 산출근거도 같이 만듦(잠금 밖에서 ·
    `create_structure` 가 다시 잠금을 잡으므로 안에서 부르면 죽음).
    산출근거(basis)는 서버가 같은 TS 엔진을 Node 로 돌려 `계산값` 을 새로 적음
    (브라우저 값 버림 · 5장 ②)."""
    path = _path(kind, name)
    _check_skeleton(kind, doc)
    if kind == "basis":
        doc = _server_values(doc)
    with _LOCK:
        have = version_of(path.read_bytes()) if path.is_file() else ""
        if have != (version or ""):
            raise StoreError(409, {"stale": [name], "판": have})
        atomic_write_json(path, doc)
        info = _info(kind, path)
    if kind == "table" and name == STRUCTURE_TABLE:
        backfill_missing_structures()
    return info


def delete(kind: str, name: str, version: str | None = None) -> None:
    path = _path(kind, name)
    with _LOCK:
        if not path.is_file():
            raise StoreError(404, f"없는 양식 「{name}」")
        if version is not None and version_of(path.read_bytes()) != version:
            raise StoreError(409, {"stale": [name], "판": version_of(path.read_bytes())})
        path.unlink()


# ── 구조물 도면 ───────────────────────────────────────


def _structures() -> list[tuple[str, Any]]:
    folder = FOLDER / "structure"
    return [(p.stem, json.loads(p.read_bytes())) for p in sorted(folder.glob("[!.]*.json"))]


def structure_numbers() -> dict[str, str | None]:
    """`{열 id: 도번}` — 집계표 도면 줄이 한 번에 받음."""
    return {
        name: (doc.get("도번") if isinstance(doc, dict) else None) for name, doc in _structures()
    }


def _number(doc: Any) -> int:
    found = _NUMBER.match(str(doc.get("도번") or "")) if isinstance(doc, dict) else None
    return int(found.group(1)) if found else 0


def _column_ids() -> list[str]:
    table = read("table", STRUCTURE_TABLE)["문서"]
    return [col["id"] for col in table.get("열", []) if isinstance(col, dict) and col.get("id")]


def create_structure(column: str) -> dict[str, Any]:
    """열 id 하나의 구조물 도면 새로 — 도번 = 있는 것 중 가장 큰 번호 + 1 · 이미 있으면 409.
    산출근거 파일이 없으면 빈 통합문서도 같이 만듦."""
    path = _path("structure", column)
    if column not in _column_ids():
        raise StoreError(404, f"{STRUCTURE_TABLE}에 없는 열 「{column}」")
    with _LOCK:
        if path.is_file():
            raise StoreError(409, f"이미 있는 구조물 도면 「{column}」")
        number = max((_number(doc) for _name, doc in _structures()), default=0) + 1
        doc = {
            "양식": "구조물도면",
            "종류": "structure",
            "판": 1,
            "열": column,
            "도번": f"구-{number:02d}",
            "도면": {"format": 6, "entities": [], "layers": [dict(_BLANK_LAYER)]},
        }
        atomic_write_json(path, doc)
    ensure_basis(column)
    return read("structure", column)


# ── 산출근거 ──────────────────────────────────────────


def _col_letters(index: int) -> str:
    letters = ""
    index += 1
    while index:
        index, rest = divmod(index - 1, 26)
        letters = chr(65 + rest) + letters
    return letters


def basis_doc(column: str, old: Any = None) -> dict[str, Any]:
    """빈 산출근거 통합문서 — 1 행 머리(공종 · 규격 · 산출 내역 · 단위 · 수량(m당)).
    `old` = 옛 구조물 문서 `산출근거` 표(열 · 줄) — 머리 글 · 칸 값 · 열 폭을 옮김
    (옛 열 식은 없어 안 옮김)."""
    columns = old.get("열") if isinstance(old, dict) and isinstance(old.get("열"), list) else None
    if columns:
        heads = [
            " ".join(str(h) for h in (c.get("머리") or []) if h) or c.get("id", "") for c in columns
        ]
    else:
        heads, columns = list(_BASIS_HEAD), []
    cells: dict[str, Any] = {
        f"{_col_letters(i)}1": {"값": head, "서식": 1} for i, head in enumerate(heads) if head
    }
    for r, row in enumerate(old.get("줄", []) if columns else [], start=2):
        values = row.get("값", {}) if isinstance(row, dict) else {}
        for i, col in enumerate(columns):
            value = values.get(col.get("id"))
            if value not in (None, ""):
                cells[f"{_col_letters(i)}{r}"] = {"값": value}
    widths = ((old or {}).get("보기") or {}).get("열너비") or {} if columns else {}
    # 옛 폭은 px · 새 폭은 엑셀 글자 단위(숫자 폭 7px 어림)
    info = {
        _col_letters(i): {"폭": round(widths[c["id"]] / 7, 1)}
        for i, c in enumerate(columns)
        if isinstance(widths.get(c.get("id")), (int, float))
    }
    sheet: dict[str, Any] = {"id": "s1", "이름": "산출근거", "칸": cells}
    if info:
        sheet["열"] = info
    return {
        "종류": "통합문서",
        "판": 1,
        "열": column,
        "서식": [{}, _BASIS_HEAD_STYLE],
        "시트": [sheet],
        "활성": "s1",
    }


def ensure_basis(column: str, old: Any = None) -> bool:
    """산출근거 파일이 없으면 만듦(식 없는 새 문서라 `계산값` 없음 · 첫 [저장] 때 서버가 적음)."""
    path = _path("basis", column)
    with _LOCK:
        if path.is_file():
            return False
        atomic_write_json(path, basis_doc(column, old))
    return True


def _server_values(doc: Any) -> dict[str, Any]:
    try:
        return spreadsheet.with_server_values(doc)
    except spreadsheet.RecalcError as e:
        raise StoreError(503, str(e)) from e


def migrate_basis() -> dict[str, list[str]]:
    """한 번 — 옛 구조물 문서 안 `산출근거` 표를 `basis/<열 id>.json` 으로 옮기고
    구조물 문서에서 뺌 · 그 뒤 산출근거가 없는 집계표 열은 빈 통합문서.
    두 번 돌려도 같음(있는 파일은 안 덮음)."""
    moved: list[str] = []
    for name, doc in _structures():
        if not isinstance(doc, dict) or "산출근거" not in doc:
            continue
        ensure_basis(name, doc["산출근거"])
        rest = {key: value for key, value in doc.items() if key != "산출근거"}
        with _LOCK:
            atomic_write_json(_path("structure", name), rest)
        moved.append(name)
    made = [column for column in _column_ids() if ensure_basis(column)]
    return {"옮김": moved, "새로": made}


def backfill_missing_structures() -> list[str]:
    """구조물집계표 열마다 빈 구조물 도면 · 산출근거가 있게 — 없는 열만 열 차례로 새로
    만듦(도번 이어서).
    그 사이 남이 만들었거나(409) 열 규칙에 안 맞으면(404) 그 열만 건너뜀."""
    have = {name for name, _ in _structures()}
    made: list[str] = []
    for cid in _column_ids():
        if cid in have:
            ensure_basis(cid)
            continue
        try:
            create_structure(cid)
            made.append(cid)
        except StoreError:
            pass
    return made