Aislo/M02_MasterTemplete/M02_Item_Copy.py

"""프로젝트 집계 항목 복사본 읽기 · 쓰기 · 새로 가져오기 · 미리 가져오기 · 계산 (PLAN 25-2b).

복사본 = `{프로젝트}/templates/items/{폴더}/` (`M02_Item_Snapshot`). 파일마다 판 = 내용 해시 —
쓸 때 판이 다르면 `Stale`(409). 고치면 manifest 에 `고침` 을 남겨 구조물에 안 쓰여도 안 지움.
옛 복사본(새 파일 없음)은 읽을 때 채움(`complete_old`). 계산은 복사본만 씀(마스터 현재값 안 읽음).
"""

from __future__ import annotations

import re
from collections.abc import Callable
from pathlib import Path
from typing import Any

from common_util.common_util_json import atomic_write_json
from common_util.common_util_summary_item import snapshot_folder_name
from M02_MasterTemplete import M02_Item_Prices as prices
from M02_MasterTemplete import M02_Template_Layers as layers
from M02_MasterTemplete.M02_Item_Snapshot import (
    MANIFEST_NAME,
    PICKS_FILE,
    VALUES_FILE,
    build_snapshot,
    complete_old,
    items_dir,
    now_text,
    write_snapshot,
)
from M02_MasterTemplete.M02_Item_Source import (
    LABOR_FILE,
    LINK_FILE,
    PRICE_FILE,
    build_source,
    read_json,
    version_of,
)

#: 길의 종류 이름 → 복사본 파일
KINDS = {
    "도면": "도면.json",
    "면적": "면적.json",
    "재료": "재료.json",
    "자재조합": "자재조합.json",
    "입력연결": LINK_FILE,
    "입력값": VALUES_FILE,
    "고른자재": PICKS_FILE,
}
#: 뼈대 검사에 쓰는 층 양식 종류(도면 · 면적산출 · 재료산출)
SKELETON = {"도면": "structure", "면적": "area", "재료": "material"}
SEATS = ("면적", "재료", "선택", None)
_FOLDER = re.compile(r"^(system|company|personal)_[0-9A-Za-z-]+_SS\d{6}$")
_MT = re.compile(r"^MT\d{6}$")


class Stale(ValueError):
    def __init__(self, current: str):
        super().__init__("다른 곳에서 먼저 고쳤습니다 — 다시 읽고 고치세요.")
        self.current = current


def folder_of(project_root: str | Path, folder: str) -> Path:
    """복사본 폴더 — 이름 모양이 틀리거나 없으면 LookupError(404)."""
    if not _FOLDER.match(folder):
        raise LookupError(f"복사본 폴더 이름 모양이 아님 「{folder}」")
    target = items_dir(project_root) / folder
    if not (target / MANIFEST_NAME).is_file():
        raise LookupError(f"복사본이 없음 「{folder}」")
    complete_old(target)
    return target


def source_changed(manifest: dict[str, Any]) -> bool | None:
    """지금 원본(층 · 마스터)을 다시 모은 판이 가져온 판과 다른가 — 원본이 없어졌으면 None."""
    try:
        source = build_source(manifest["원본"])
    except (LookupError, ValueError, KeyError):
        return None
    return version_of(source) != manifest.get("원본판")


def list_items(project_root: str | Path) -> list[dict[str, Any]]:
    base = items_dir(project_root)
    out = []
    for child in sorted(base.iterdir()) if base.is_dir() else []:
        if not child.is_dir() or not _FOLDER.match(child.name):
            continue
        if not (child / MANIFEST_NAME).is_file():
            continue
        complete_old(child)
        manifest = read_json(child / MANIFEST_NAME) or {}
        origin = manifest.get("원본") or {}
        changed = source_changed(manifest)
        out.append(
            {
                "폴더": child.name,
                "이름": origin.get("name"),
                "원본": origin,
                "원본판": manifest.get("원본판"),
                "가져온날": manifest.get("가져온날"),
                "마스터바뀜": bool(changed),
                "원본없음": changed is None,
                "구조물": manifest.get("구조물") or [],
                "고침": bool(manifest.get("고침")),
            }
        )
    return out


def read_file(project_root: str | Path, folder: str, kind: str) -> dict[str, Any]:
    target = folder_of(project_root, folder)
    content = read_json(target / KINDS[kind])
    return {"종류": kind, "판": version_of(content), "내용": content}


# ── 모양 검사 ───────────────────────────────────────────────────────


def _check_values(content: Any) -> None:
    if not isinstance(content, dict):
        raise ValueError("입력값은 {로직키: {입력이름: {값, 확인}}} 이어야 합니다.")
    for logic, inputs in content.items():
        if not isinstance(inputs, dict):
            raise ValueError(f"입력값 「{logic}」 은 {{입력이름: {{값, 확인}}}} 이어야 합니다.")
        for name, one in inputs.items():
            if not isinstance(one, dict) or set(one) - {"값", "확인"}:
                raise ValueError(f"입력값 「{logic}|{name}」 은 {{값, 확인}} 이어야 합니다.")
            if not isinstance(one.get("확인", False), bool):
                raise ValueError(f"입력값 「{logic}|{name}」 확인은 참 · 거짓이어야 합니다.")
            if isinstance(one.get("값"), (dict, list)):
                raise ValueError(f"입력값 「{logic}|{name}」 값은 수 · 글 · 빈 값이어야 합니다.")


def _check_picks(content: Any) -> None:
    if not isinstance(content, dict):
        raise ValueError("고른자재는 {로직키: {호표 차례: MT키}} 이어야 합니다.")
    for logic, picks in content.items():
        if not isinstance(picks, dict):
            raise ValueError(f"고른자재 「{logic}」 은 {{호표 차례: MT키}} 이어야 합니다.")
        for at, key in picks.items():
            if not str(at).isdigit() or not isinstance(key, str) or not _MT.match(key):
                raise ValueError(f"고른자재 「{logic}」 {at} → 「{key}」 — 차례는 수 · 값은 MT 키")


def _check_links(content: Any) -> None:
    rows = content.get("줄") if isinstance(content, dict) else None
    if not isinstance(rows, list):
        raise ValueError("입력연결은 {판, 줄: [...]} 이어야 합니다.")
    for row in rows:
        if not isinstance(row, dict) or not row.get("로직") or not row.get("입력"):
            raise ValueError("입력연결 줄은 {로직, 입력, 자리, 알약?} 이어야 합니다.")
        if row.get("자리") not in SEATS:
            raise ValueError(f"입력연결 자리 「{row.get('자리')}」 — 면적 · 재료 · 선택 · 빈 값")


def check(kind: str, content: Any) -> None:
    if kind == "입력값":
        _check_values(content)
    elif kind == "고른자재":
        _check_picks(content)
    elif kind == "입력연결":
        _check_links(content)
    elif kind in SKELETON:
        layers.check_skeleton(SKELETON[kind], content)
    elif not isinstance(content, dict):
        raise ValueError(f"{kind} 은 JSON 객체여야 합니다.")


def write_file(
    project_root: str | Path,
    folder: str,
    kind: str,
    version: str | None,
    content: Any,
    recalc: Callable[[dict], dict] | None = None,
) -> dict[str, Any]:
    """판이 같을 때만 씀 — 면적 · 재료는 `recalc`(서버 계산값)를 거친 것을 씀."""
    target = folder_of(project_root, folder)
    check(kind, content)
    current = version_of(read_json(target / KINDS[kind]))
    if version != current:
        raise Stale(current)
    if recalc is not None and kind in ("면적", "재료"):
        content = recalc(content)
    atomic_write_json(target / KINDS[kind], content)
    manifest = read_json(target / MANIFEST_NAME) or {}
    atomic_write_json(target / MANIFEST_NAME, {**manifest, "고침": True, "고친날": now_text()})
    return {"종류": kind, "판": version_of(content)}


# ── 새로 가져오기 · 미리 가져오기 ─────────────────────────────────────


def _kept_values(values: Any, labor: dict[str, Any]) -> dict[str, Any]:
    """같은 로직 · 같은 입력 이름만 살림."""
    names = {
        str(row.get("키")): {spec["이름"] for spec in row.get("입력", [])}
        for row in labor.get("로직", [])
    }
    out: dict[str, Any] = {}
    for logic, inputs in (values or {}).items():
        kept = {k: v for k, v in (inputs or {}).items() if k in names.get(logic, set())}
        if kept:
            out[logic] = kept
    return out


def _kept_picks(picks: Any, labor: dict[str, Any]) -> dict[str, Any]:
    """같은 로직 · 있는 호표 차례만 살림."""
    lines = {str(row.get("키")): len(row.get("호표", [])) for row in labor.get("로직", [])}
    out: dict[str, Any] = {}
    for logic, chosen in (picks or {}).items():
        kept = {k: v for k, v in (chosen or {}).items() if int(k) < lines.get(logic, 0)}
        if kept:
            out[logic] = kept
    return out


def refresh(project_root: str | Path, folder: str) -> dict[str, Any]:
    """층 · 마스터에서 다시 가져옴 — 입력값 · 고른자재는 같은 키면 살림."""
    target = folder_of(project_root, folder)
    manifest = read_json(target / MANIFEST_NAME) or {}
    values = read_json(target / VALUES_FILE)
    picks = read_json(target / PICKS_FILE)
    files = build_snapshot(manifest["원본"])
    labor = files.get(LABOR_FILE) or {}
    files[VALUES_FILE] = _kept_values(values, labor)
    files[PICKS_FILE] = _kept_picks(picks, labor)
    for old in target.glob("*.json"):  # 원본에서 빠진 파일(도면 없앰 등)은 지움
        if old.name != MANIFEST_NAME and old.name not in files:
            old.unlink()
    for name, content in files.items():
        atomic_write_json(target / name, content)
    source = {k: v for k, v in files.items() if k not in (VALUES_FILE, PICKS_FILE)}
    stamp = now_text()
    fresh = {**manifest, "판": version_of(source), "원본판": version_of(source), "가져온날": stamp}
    atomic_write_json(target / MANIFEST_NAME, fresh)
    return {"폴더": folder, "가져온날": stamp, "원본판": fresh["원본판"]}


def prefetch(project_root: str | Path, item: dict[str, Any]) -> dict[str, Any]:
    """구조물에 아직 안 이은 항목을 미리 가져옴 — 있으면 그대로 · 없으면 새로(쓰는 구조물 없음)."""
    folder = snapshot_folder_name(item)
    target = items_dir(project_root) / folder
    manifest = read_json(target / MANIFEST_NAME)
    made = not isinstance(manifest, dict)
    if made:
        manifest = write_snapshot(target, item, [], {"미리가져옴": now_text()})
    return {"폴더": folder, "만듦": made, "가져온날": manifest.get("가져온날")}


def calc(project_root: str | Path, folder: str) -> dict[str, Any]:
    """복사본(입력값 · 고른자재 · 단가 · 일위대가)만으로 계산 — 마스터 현재값을 안 읽음."""
    target = folder_of(project_root, folder)
    labor = read_json(target / LABOR_FILE)
    if not isinstance(labor, dict):
        raise LookupError(f"일위대가 조합이 없는 항목 「{folder}」 — 계산할 것이 없음")
    return prices.calc(
        labor,
        read_json(target / PRICE_FILE) or {},
        read_json(target / VALUES_FILE) or {},
        read_json(target / PICKS_FILE) or {},
    )