"""프로젝트 집계 항목 복사본 읽기 · 쓰기 · 새로 가져오기 · 미리 가져오기 · 계산 (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 {} out.append( { "폴더": child.name, "이름": origin.get("name"), "원본": origin, "가져온날": manifest.get("가져온날"), "마스터바뀜": source_changed(manifest), "구조물": 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 {}, )