feat(M02): 집계 항목 복사본 완전체 — 입력연결 · 입력값 · 고른자재 · 단가(닫힌 묶음) · 가져온날 · 원본판 · 복사본 길 · 새로 가져오기 · 미리 가져오기 · 복사본만으로 계산 (PLAN 25-2b · 25-2e)

- 스냅샷 나눔: 원본 모으기(M02_Item_Source) · 맞추기(M02_Item_Snapshot) · 복사본 읽기 · 쓰기(M02_Item_Copy) · 단가 묶음 · 계산(M02_Item_Prices)
- 단가.json = 담은 로직 · 하위 로직 · 요소 · 준용 · 찾기 표 · 재료 고르기 후보 · 지역 · 계약종별 후보(마스터 파일 차례 그대로)
- 길 /api/projects/{id}/items(목록 · 미리 가져오기) · /{폴더}/{종류}(GET · PUT 판 409 · 면적 · 재료 서버 계산값) · /refresh · /calc
- 고친 복사본은 구조물에서 빠져도 안 지움 · 옛 복사본은 읽을 때 채움

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BU9gbNE3aWFUvE61pS2WGt

M02_MasterTemplete/M02_Item_Copy.py

@@ -0,0 +1,259 @@
+"""프로젝트 집계 항목 복사본 읽기 · 쓰기 · 새로 가져오기 · 미리 가져오기 · 계산 (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 {},
+    )

M02_MasterTemplete/M02_Item_Prices.py

@@ -0,0 +1,249 @@
+"""집계 항목 복사본의 단가 묶음 `단가.json` · 복사본만으로 계산 (PLAN 25-1b · 25-2b · 25-2e).
+
+단가 묶음 = 그 항목 일위대가 조합이 담은 로직을 계산할 때 닿을 수 있는 마스터 줄 전부(닫힌 묶음):
+    - 담은 로직 · 로직이 부르는 하위 로직(`로직(GC…)`) 끝까지
+    - 식이 가리키는 요소(인력 · 기계 · 재료 · 유가전력 · 요율 · 환율 …) · 준용 대상
+    - 찾기 표(`찾기(QC…)` · `찾기(QF…)` 소요량 · 계수)
+    - 재료 고르기 조건의 후보 자재품목 줄 · 품셈재료 연결 · 지역 · 계약종별 후보
+원문번호 틀의 `{입력}` 은 그 입력의 고르기 값으로 모두 끼워 모음(고르기가 없으면 그 테이블 전부 ·
+자재품목만은 너무 커서 뺌). 줄은 마스터 파일 차례 그대로 담음 — 고르기의 「첫 줄」 이 같게.
+
+`단가.json` = `{파일 이름: {그룹 · 머리 칸 …, 줄|표: [줄]}}` — `master_formula.Master` 가
+그대로 읽는 꼴.
+계산(`calc`)은 이 묶음과 `입력값.json` · `고른자재.json` 만 씀 — 마스터 현재값을 읽지 않음.
+"""
+
+from __future__ import annotations
+
+import json
+from decimal import Decimal
+from typing import Any
+
+from M01_MasterData import M01_MasterData_Store as m01
+
+mf = m01.cm.mf
+mk = m01.mk
+mcb = m01.mcb
+mm = mf.mp.mm  # 품셈재료 · 자재품목 후보(`master_material`)
+
+#: 틀을 못 풀었을 때 통째로 담지 않는 테이블 — 자재품목(3만 줄)
+_TOO_BIG = ("MT",)
+
+
+def master() -> tuple[dict[str, dict], Any]:
+    """(마스터 파일 묶음, 색인) — 읽기 전용(시험은 `m01.FOLDER` 를 사본으로 바꿈)."""
+    return m01.loaded()
+
+
+def _texts(row: dict) -> list[str]:
+    out = [step["식"] for step in row.get("중간", [])]
+    if "결과" in row:
+        out.append(row["결과"]["식"])
+    for item in row.get("호표", []):
+        out.append(item["수량"])
+        if not isinstance(item.get("요소"), dict):
+            out.append(item["요소"])
+    out += [extra["식"] for extra in row.get("덧줄", [])]
+    return out
+
+
+class _Collector:
+    def __init__(self, whole: Any):
+        self.whole = whole
+        self.keys: set[str] = set()
+        self.logics: set[str] = set()
+        self.missing: list[str] = []
+
+    def refs(self, row: dict, template: str) -> list[str]:
+        """원문번호 틀 → 닿을 수 있는 요소 참조 — 고르기로 끼우고, 못 풀면 그 테이블 전부."""
+        if "{" not in template:
+            return [template]
+        expanded = mf._expand(row, template, self.whole)
+        if expanded is not None:
+            return expanded
+        tid = mk.id_of(template)
+        if not tid or tid in _TOO_BIG:
+            self.missing.append(f"{row.get('키')} 「{template}」 틀을 못 풂")
+            return []
+        return list(self.whole.index.get(tid, {}))
+
+    def add(self, ref: str) -> None:
+        try:
+            row = self.whole.get(ref)
+        except mf.FormulaError:
+            return  # 없는 요소 — 계산이 같은 까닭으로 멈춤(마스터와 같게)
+        key = str(row.get("키"))
+        if key in self.keys:
+            return
+        self.keys.add(key)
+        if row.get("준용") not in (None, ""):
+            self.add(str(row["준용"]))
+        link = row.get("연결") if key[:2] == mf.LINKED else None
+        if isinstance(link, str):
+            self.add(link)
+        elif isinstance(link, dict):
+            self.widely(link, row.get("단위"))
+
+    def widely(self, cond: dict, unit: Any) -> None:
+        """지역 · 계약종별 「입력」 조건 — 입력값과 상관없이 닿을 수 있는 후보 전부."""
+        want = mm.plain(cond.get("이름"))
+        if cond.get("계약종별") == "입력":
+            for key, one in self.whole.index.get(mm.PRICED, {}).items():
+                if one.get("구분") == "전력" and mm.plain(one.get("이름")).startswith(want):
+                    self.keys.add(key)
+            return
+        spec, unit_p = mm.plain_spec(cond.get("규격")), mm.plain_unit(unit)
+        for root, _, s, u, key in mm._market_index(self.whole):
+            if root.endswith(want) and u == unit_p and (not spec or s == spec):
+                self.keys.add(key)
+
+    def pick(self, row: dict, cond: dict, unit: Any) -> None:
+        """재료 고르기 조건 — 대표 줄 · 후보 줄(`{입력}` 은 고르기 값으로 끼움)."""
+        if cond.get("지역") == "입력" or cond.get("계약종별") == "입력":
+            self.widely(cond, unit)
+            return
+        holes = sorted(
+            {h for v in cond.values() if isinstance(v, str) for h in mf._HOLE.findall(v)}
+        )
+        variants = [cond]
+        if holes:
+            env_sets = mf._expand(row, "|".join(f"{{{h}}}" for h in holes), self.whole)
+            if env_sets is None:
+                variants = [mf.static(cond)]
+            else:
+                variants = []
+                for filled in env_sets:
+                    env = dict(zip(holes, filled.split("|")))
+                    variants.append(
+                        {k: mf.fill(v, env) if isinstance(v, str) else v for k, v in cond.items()}
+                    )
+        for one in variants:
+            if one.get("대표"):
+                self.add(str(one["대표"]))
+            if not one.get("구분") and not one.get("검색어"):
+                continue
+            for key in mf.candidates(self.whole, one, None, None):
+                self.add(key)
+
+    def logic(self, key: str) -> None:
+        if key in self.logics:
+            return
+        try:
+            row = self.whole.logic(key)
+        except mf.FormulaError as error:
+            self.missing.append(f"{key} · {error}")
+            return
+        self.logics.add(key)
+        self.keys.add(key)
+        for text in _texts(row):
+            try:
+                node = mf.parse(text)
+            except mf.FormulaError:
+                continue  # 못 읽는 식 — 계산도 같은 까닭으로 멈춤
+            for part in mf.walk(node):
+                kind = part[0]
+                if kind == "ref" or kind == "find":
+                    for ref in self.refs(row, part[1]):
+                        self.add(ref)
+                elif kind == "logic":
+                    for ref in self.refs(row, part[1]):
+                        self.logic(ref)
+        for item in row.get("호표", []):
+            if isinstance(item.get("요소"), dict):
+                self.pick(row, item["요소"], item.get("단위"))
+
+
+def collect(logic_keys: list[str]) -> dict[str, Any]:
+    """담은 로직의 닫힌 단가 묶음 — `{파일 이름: 파일(줄은 모은 것만 · 차례 그대로)}`."""
+    files, whole = master()
+    collector = _Collector(whole)
+    for key in logic_keys:
+        collector.logic(key)
+    out: dict[str, Any] = {}
+    for name, data in files.items():
+        slot = m01.items_key(data)
+        rows = [row for row in data.get(slot) or [] if str(row.get("키")) in collector.keys]
+        if rows:
+            out[name] = {**{k: v for k, v in data.items() if k != slot}, slot: rows}
+    if collector.missing:
+        out["_빠진"] = sorted(set(collector.missing))
+    return out
+
+
+def to_json(value: Any) -> Any:
+    """Decimal 을 JSON 수로 — 정수는 int · 아니면 원래 자릿수 그대로(float 로 흔들지 않음)."""
+    return json.loads(json.dumps(value, ensure_ascii=False, default=_number))
+
+
+def _number(value: Any) -> Any:
+    if isinstance(value, Decimal):
+        text = str(value)  # 「14480.0」 은 소수 그대로 — 끝수 · 표기가 마스터와 같게
+        return int(value) if "." not in text and "E" not in text.upper() else float(value)
+    raise TypeError(type(value).__name__)
+
+
+def load_prices(prices: dict[str, Any]) -> Any:
+    """`단가.json` → 계산 색인(수는 Decimal · 마스터 적재와 같게)."""
+    raw = json.dumps({k: v for k, v in prices.items() if not k.startswith("_")}, ensure_ascii=False)
+    files = json.loads(raw, parse_float=Decimal, parse_int=Decimal)
+    return mf.Master(files)
+
+
+# ── 계산(25-2e) ─────────────────────────────────────────────────────
+
+
+def missing_inputs(snapshot: Any, labor: dict[str, Any], values: dict[str, Any]) -> list[dict]:
+    """담은 로직마다 값이 없는(null · 안 적음) 입력 — `{로직, 입력}`."""
+    out = []
+    for key in mcb.logic_keys(labor.get("줄") or {}):
+        try:
+            row = snapshot.logic(key)
+        except mf.FormulaError:
+            continue
+        mine = values.get(key) or {}
+        for spec in row.get("입력", []):
+            got = mine.get(spec["이름"])
+            if not isinstance(got, dict) or got.get("값") in (None, ""):
+                out.append({"로직": key, "입력": spec["이름"]})
+    return out
+
+
+def calc(
+    labor: dict[str, Any], prices: dict[str, Any], values: dict[str, Any], picks: dict[str, Any]
+) -> dict[str, Any]:
+    """복사본만으로 계산 — 입력이 빠진 로직은 안 돌리고 `멈춤` · 세 비목 · 계는 돈 로직 합."""
+    snapshot = load_prices(prices)
+    gaps = missing_inputs(snapshot, labor, values)
+    blocked = {gap["로직"] for gap in gaps}
+    sums = dict.fromkeys(mf.COST_ITEMS, Decimal(0))
+    lines, stuck = [], []
+    for at, key in enumerate(mcb.logic_keys(labor.get("줄") or {}), start=1):
+        head: dict[str, Any] = {"차례": at, "로직": key}
+        if key in blocked:
+            names = [gap["입력"] for gap in gaps if gap["로직"] == key]
+            lines.append({**head, "ok": False, "까닭": f"입력 없음 — {', '.join(names)}"})
+            stuck.append(key)
+            continue
+        given = {name: one.get("값") for name, one in (values.get(key) or {}).items()}
+        try:
+            with mf.mp.choosing((picks or {}).get(key)):
+                got = mf.run(snapshot, key, given)
+        except (mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as error:
+            lines.append({**head, "ok": False, "까닭": str(error) or type(error).__name__})
+            stuck.append(key)
+            continue
+        if "결과" in got:
+            lines.append({**head, "ok": True, "갈래": "결과", "결과": got["결과"]})
+            continue
+        for name in mf.COST_ITEMS:
+            sums[name] += got[name]
+        lines.append(
+            {
+                **head,
+                "ok": True,
+                "갈래": "돈",
+                **{k: got[k] for k in mf.COST_ITEMS},
+                "계": got["계"],
+            }
+        )
+    return to_json({"줄": lines, **sums, "계": sum(sums.values()), "빠진입력": gaps, "멈춤": stuck})

M02_MasterTemplete/M02_Item_Snapshot.py

@@ -1,19 +1,19 @@
-"""구조물 집계표 항목 스냅샷 — 프로젝트 `templates/items/{폴더}/` (PLAN 21-1 · 21-3).
+"""구조물 집계표 항목 스냅샷 — 프로젝트 `templates/items/{폴더}/` (PLAN 21-1 · 21-3 · 25-2b).
 
 [저장](`PUT /route/structures` · `PUT /drainage/pipe-points`) 뒤 서버가 맞춤:
     - 구조물 · 관 지점이 가리키는 항목(`summary_item`)마다 원본 층에서 내용을 복사
-      (`집계.json` SS 줄 + 열 정의 · `도면.json` SD · `면적.json` SA · `재료.json` SM ·
-      `자재조합.json` UM 줄 · `일위대가.json` UA 줄 + 담은 로직 줄 · `manifest.json`)
+      (`M02_Item_Source.build_source` — 집계 · 도면 · 면적 · 재료 · 자재조합 · 일위대가 · 입력연결 ·
+      단가) + 프로젝트 값 `입력값.json`(처음엔 빈 값 · 자동값은 안 적음) · `고른자재.json` ·
+      `manifest.json`
     - 이미 있는 폴더는 내용을 다시 쓰지 않음 — 원본(마스터 · 회사 · 개인)이 바뀌어도 프로젝트
-      값은 지정한 때 그대로 · `manifest` 의 쓰는 구조물 목록만 새로
-    - 아무 구조물도 안 쓰는 폴더는 지움(지정 바뀜 · 해제 · 구조물 삭제)
-원본 층 자리 = `M02_Template_Layers` 경로(시스템 층은 지금 파일) · 일위대가는 M01 시스템 것.
+      값은 지정한 때 그대로 · `manifest` 의 쓰는 구조물 목록만 새로([새로 가져오기]는 사용자가)
+    - 아무 구조물도 안 쓰는 폴더는 지움(지정 바뀜 · 해제 · 구조물 삭제) — 단 프로젝트에서 고친
+      흔적(`manifest.고침`)이 있으면 남김
+원본 층 자리 = `M02_Template_Layers` 경로(시스템 층은 지금 파일) · 일위대가 · 단가는 M01 시스템 것.
 """
 
 from __future__ import annotations
 
-import hashlib
-import json
 import logging
 import shutil
 from datetime import datetime
@@ -25,97 +25,54 @@ from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pip
 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_Template_Layers as layers
+from M02_MasterTemplete.M02_Item_Source import (
+    LINK_FILE,
+    PRICE_FILE,
+    build_source,
+    read_json,
+    version_of,
+)
 
 logger = logging.getLogger(__name__)
 
 ITEMS_DIRNAME = "items"
 MANIFEST_NAME = "manifest.json"
-TABLE_NAME = "구조물집계표"
-UM_FILE = "자재조합.json"
-UA_FILE = "일위대가조합.json"  # M01 시스템(`resources/master_data/`)
-#: 열 id → 스냅샷 파일(구조물 도면 · 면적산출 · 재료산출)
-PART_FILES = (("structure", "도면.json"), ("area", "면적.json"), ("material", "재료.json"))
+VALUES_FILE = "입력값.json"  # {로직키: {입력이름: {값, 확인}}}
+PICKS_FILE = "고른자재.json"  # {로직키: {호표 차례: MT키}}
 
 
 def items_dir(project_root: str | Path) -> Path:
     return layers.project_dir(project_root) / ITEMS_DIRNAME
 
 
-def layer_root(item: dict[str, Any]) -> Path:
-    """항목 원본 층 뿌리 — 시스템 · 회사 · 개인(21-2 층 폴더 함수가 오면 그리로)."""
-    if item["layer"] == "system":
-        return layers.system_dir()
-    if item["layer"] == "company":
-        return layers.company_dir(item["owner"])
-    company, user = str(item["owner"]).split("/")
-    return layers.personal_dir(company, user)
-
-
-def _read(path: Path) -> Any:
-    try:
-        return json.loads(path.read_text(encoding="utf-8"))
-    except (OSError, ValueError):
-        return None
-
-
-def _row(document: Any, key: str) -> dict[str, Any] | None:
-    rows = document.get("줄") if isinstance(document, dict) else None
-    for row in rows or []:
-        if isinstance(row, dict) and row.get("키") == key:
-            return row
-    return None
-
-
-def _labor_rows(ua_key: str) -> dict[str, Any] | None:
-    """UA 줄 + 담은 로직 줄(M01 시스템 · 없는 로직은 빼고 키만 `빠진로직`)."""
-    from M01_MasterData import M01_MasterData_Store as m01
-
-    ua = _row(_read(Path(m01.FOLDER) / UA_FILE), ua_key)
-    if ua is None:
-        return None
-    wanted = [str(entry.get("로직")) for entry in ua.get("담은로직") or [] if entry.get("로직")]
-    found: dict[str, dict[str, Any]] = {}
-    raw: dict[str, Any] = {}  # M01 적재본은 수가 Decimal — 줄은 원본 파일 JSON 그대로 담음
-    for key in wanted:
-        try:
-            file = m01.logic(key)["file"]
-        except m01.StoreError:
-            continue
-        if file not in raw:
-            raw[file] = _read(Path(m01.FOLDER) / file)
-        row = _row(raw[file], key)
-        if row is not None:
-            found[key] = row
-    return {
-        "줄": ua,
-        "로직": [found[key] for key in wanted if key in found],
-        "빠진로직": [key for key in wanted if key not in found],
-    }
+def now_text() -> str:
+    return datetime.now().isoformat(timespec="seconds")
 
 
 def build_snapshot(item: dict[str, Any]) -> dict[str, Any]:
-    """원본 층에서 항목 내용을 모음 — `{파일 이름: 내용}` · SS 줄이 없으면 LookupError."""
-    root = layer_root(item)
-    row = _row(_read(root / layers.COMBO_FILE), item["key"])
-    if row is None:
-        raise LookupError(f"집계 항목 {item['key']} 이 {item['layer']} 층에 없음")
-    column_id = str(row.get("열") or "")
-    table = _read(root / "table" / f"{TABLE_NAME}.json") or {}
-    column = next((c for c in table.get("열") or [] if c.get("id") == column_id), None)
-    files: dict[str, Any] = {"집계.json": {"줄": row, "열": column}}
-    for kind, name in PART_FILES:
-        part = _read(root / kind / f"{column_id}.json") if column_id else None
-        if part is not None:
-            files[name] = part
-    if row.get("자재조합"):
-        um = _row(_read(root / UM_FILE), str(row["자재조합"]))
-        if um is not None:
-            files["자재조합.json"] = um
-    if row.get("일위대가"):
-        labor = _labor_rows(str(row["일위대가"]))
-        if labor is not None:
-            files["일위대가.json"] = labor
-    return files
+    """원본 내용 + 빈 프로젝트 값 — `{파일 이름: 내용}` · SS 줄이 없으면 LookupError."""
+    return {**build_source(item), VALUES_FILE: {}, PICKS_FILE: {}}
+
+
+def write_snapshot(target: Path, item: dict[str, Any], users: list[str], extra: dict) -> dict:
+    """폴더에 새로 씀 — manifest 는 마지막(있으면 다 쓴 폴더 · 끊기면 다음 저장이 다시 씀)."""
+    files = build_snapshot(item)
+    target.mkdir(parents=True, exist_ok=True)
+    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()
+    manifest = {
+        "원본": item,
+        "판": version_of(source),
+        "원본판": version_of(source),
+        "적용일": stamp,
+        "가져온날": stamp,
+        "구조물": users,
+        **extra,
+    }
+    atomic_write_json(target / MANIFEST_NAME, manifest)
+    return manifest
 
 
 def assignments(project_root: str | Path) -> dict[str, tuple[dict[str, Any], list[str]]]:
@@ -134,11 +91,6 @@ def assignments(project_root: str | Path) -> dict[str, tuple[dict[str, Any], lis
     return found
 
 
-def _version(files: dict[str, Any]) -> str:
-    raw = json.dumps(files, ensure_ascii=False, sort_keys=True).encode("utf-8")
-    return hashlib.sha256(raw).hexdigest()[:16]
-
-
 def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
     """지정 항목 스냅샷을 맞춤 — `{만듦, 지움, 못만듦}` 폴더 이름."""
     base = items_dir(project_root)
@@ -147,37 +99,30 @@ def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
     failed: list[str] = []
     for folder, (item, users) in sorted(wanted.items()):
         target = base / folder
-        manifest = _read(target / MANIFEST_NAME)
+        manifest = read_json(target / MANIFEST_NAME)
         if isinstance(manifest, dict):
             if manifest.get("구조물") != users:
                 atomic_write_json(target / MANIFEST_NAME, {**manifest, "구조물": users})
             continue
         try:
-            files = build_snapshot(item)
+            write_snapshot(target, item, users, {})
         except (LookupError, ValueError) as error:
             logger.warning("집계 항목 스냅샷을 못 만듦 (%s): %s", folder, error)
             failed.append(folder)
             continue
-        target.mkdir(parents=True, exist_ok=True)
-        for name, content in files.items():
-            atomic_write_json(target / name, content)
-        # manifest 는 마지막 — 있으면 다 쓴 폴더(도중에 끊기면 다음 저장이 다시 씀)
-        atomic_write_json(
-            target / MANIFEST_NAME,
-            {
-                "원본": item,
-                "판": _version(files),
-                "적용일": datetime.now().isoformat(timespec="seconds"),
-                "구조물": users,
-            },
-        )
         made.append(folder)
     removed: list[str] = []
     if base.is_dir():
         for child in sorted(base.iterdir()):
-            if child.is_dir() and child.name not in wanted:
-                shutil.rmtree(child)
-                removed.append(child.name)
+            if not child.is_dir() or child.name in wanted:
+                continue
+            manifest = read_json(child / MANIFEST_NAME)
+            if isinstance(manifest, dict) and manifest.get("고침"):
+                if manifest.get("구조물"):  # 고친 복사본은 남기되 쓰는 구조물은 비움
+                    atomic_write_json(child / MANIFEST_NAME, {**manifest, "구조물": []})
+                continue
+            shutil.rmtree(child)
+            removed.append(child.name)
     return {"만듦": made, "지움": removed, "못만듦": failed}
 
 
@@ -191,7 +136,44 @@ def safe_sync(project_root: str | Path) -> None:
         logger.exception("집계 항목 스냅샷 맞춤 실패 (저장은 유지): %s", project_root)
 
 
+def complete_old(folder: Path) -> bool:
+    """옛 스냅샷(21-3 · 새 파일 없음)을 채움 — 입력연결 · 단가는 지금 원본에서 · 값은 빈 것.
+
+    옛 내용 파일(집계 · 도면 …)은 그대로 둠. 채웠으면 True(manifest 에 `옮겨받음` 날).
+    """
+    manifest = read_json(folder / MANIFEST_NAME)
+    if not isinstance(manifest, dict):
+        return False
+    lacking = [
+        name
+        for name in (LINK_FILE, PRICE_FILE, VALUES_FILE, PICKS_FILE)
+        if not (folder / name).is_file()
+    ]
+    if not lacking and "원본판" in manifest:
+        return False
+    source = (
+        build_source(manifest["원본"]) if (LINK_FILE in lacking or PRICE_FILE in lacking) else {}
+    )
+    for name in lacking:
+        atomic_write_json(folder / name, source.get(name, {}))
+    kept = {
+        name: read_json(folder / name)
+        for name in sorted(p.name for p in folder.glob("*.json"))
+        if name not in (MANIFEST_NAME, VALUES_FILE, PICKS_FILE)
+    }
+    atomic_write_json(
+        folder / MANIFEST_NAME,
+        {
+            **manifest,
+            "가져온날": manifest.get("가져온날") or manifest.get("적용일"),
+            "원본판": manifest.get("원본판") or version_of(kept),
+            "옮겨받음": now_text(),
+        },
+    )
+    return True
+
+
 def read_snapshot(project_root: str | Path, item: dict[str, Any]) -> dict[str, Any] | None:
     """프로젝트 스냅샷의 `집계.json`(`{줄, 열}`) — 없으면 None."""
-    data = _read(items_dir(project_root) / snapshot_folder_name(item) / "집계.json")
+    data = read_json(items_dir(project_root) / snapshot_folder_name(item) / "집계.json")
     return data if isinstance(data, dict) else None

M02_MasterTemplete/M02_Item_Source.py

@@ -0,0 +1,124 @@
+"""집계 항목 원본 — 층(시스템 · 회사 · 개인)에서 항목 내용을 모음 (PLAN 21-1 · 25-1b · 25-2b).
+
+모으는 것 = `{파일 이름: 내용}`:
+    `집계.json` SS 줄 + 열 정의 · `도면.json` SD · `면적.json` SA · `재료.json` SM ·
+    `자재조합.json` UM 줄 · `일위대가.json` UA 줄 + 담은 로직 줄(M01 시스템) ·
+    `입력연결.json` 층의 `inputs/<열id>.json`(없으면 빈 연결) ·
+    `단가.json` 계산에 닿는 마스터 줄 전부
+`원본판` = 모은 내용의 해시 — 지금 층 · 마스터에서 다시 모은 판과 대어 「마스터가 바뀜」 을 앎.
+프로젝트가 고치는 값(`입력값.json` · `고른자재.json`)은 원본이 아니라 여기서 안 모음.
+원본 층 자리 = `M02_Template_Layers` 경로 · 일위대가 · 단가는 M01 시스템 것.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+from M02_MasterTemplete import M02_Template_Layers as layers
+
+TABLE_NAME = "구조물집계표"
+UM_FILE = "자재조합.json"
+UA_FILE = "일위대가조합.json"  # M01 시스템(`resources/master_data/`)
+INPUTS_DIRNAME = "inputs"  # 층의 항목 연결 파일 자리(25-1b · sub4 층 자료 틀)
+#: 열 id → 스냅샷 파일(구조물 도면 · 면적산출 · 재료산출)
+PART_FILES = (("structure", "도면.json"), ("area", "면적.json"), ("material", "재료.json"))
+LINK_FILE = "입력연결.json"
+PRICE_FILE = "단가.json"
+LABOR_FILE = "일위대가.json"
+
+
+def layer_root(item: dict[str, Any]) -> Path:
+    """항목 원본 층 뿌리 — 시스템 · 회사 · 개인."""
+    if item["layer"] == "system":
+        return layers.system_dir()
+    if item["layer"] == "company":
+        return layers.company_dir(item["owner"])
+    company, user = str(item["owner"]).split("/")
+    return layers.personal_dir(company, user)
+
+
+def read_json(path: Path) -> Any:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+
+
+def row_of(document: Any, key: str) -> dict[str, Any] | None:
+    rows = document.get("줄") if isinstance(document, dict) else None
+    for row in rows or []:
+        if isinstance(row, dict) and row.get("키") == key:
+            return row
+    return None
+
+
+def version_of(content: Any) -> str:
+    raw = json.dumps(content, ensure_ascii=False, sort_keys=True).encode("utf-8")
+    return hashlib.sha256(raw).hexdigest()[:16]
+
+
+def labor_rows(ua_key: str) -> dict[str, Any] | None:
+    """UA 줄 + 담은 로직 줄(M01 시스템 · 없는 로직은 빼고 키만 `빠진로직`)."""
+    from M01_MasterData import M01_MasterData_Store as m01
+
+    ua = row_of(read_json(Path(m01.FOLDER) / UA_FILE), ua_key)
+    if ua is None:
+        return None
+    wanted = [str(entry.get("로직")) for entry in ua.get("담은로직") or [] if entry.get("로직")]
+    found: dict[str, dict[str, Any]] = {}
+    raw: dict[str, Any] = {}  # M01 적재본은 수가 Decimal — 줄은 원본 파일 JSON 그대로 담음
+    for key in wanted:
+        try:
+            file = m01.logic(key)["file"]
+        except m01.StoreError:
+            continue
+        if file not in raw:
+            raw[file] = read_json(Path(m01.FOLDER) / file)
+        row = row_of(raw[file], key)
+        if row is not None:
+            found[key] = row
+    return {
+        "줄": ua,
+        "로직": [found[key] for key in wanted if key in found],
+        "빠진로직": [key for key in wanted if key not in found],
+    }
+
+
+def input_links(root: Path, column_id: str) -> dict[str, Any]:
+    """층의 항목 연결 `inputs/<열id>.json` — 없으면 빈 연결(같은 모양)."""
+    found = read_json(root / INPUTS_DIRNAME / f"{column_id}.json") if column_id else None
+    if isinstance(found, dict) and isinstance(found.get("줄"), list):
+        return found
+    return {"판": None, "줄": []}
+
+
+def build_source(item: dict[str, Any]) -> dict[str, Any]:
+    """원본 층 · 마스터에서 항목 내용을 모음 — SS 줄이 없으면 LookupError."""
+    from M02_MasterTemplete import M02_Item_Prices as prices
+
+    root = layer_root(item)
+    row = row_of(read_json(root / layers.COMBO_FILE), item["key"])
+    if row is None:
+        raise LookupError(f"집계 항목 {item['key']} 이 {item['layer']} 층에 없음")
+    column_id = str(row.get("열") or "")
+    table = read_json(root / "table" / f"{TABLE_NAME}.json") or {}
+    column = next((c for c in table.get("열") or [] if c.get("id") == column_id), None)
+    files: dict[str, Any] = {"집계.json": {"줄": row, "열": column}}
+    for kind, name in PART_FILES:
+        part = read_json(root / kind / f"{column_id}.json") if column_id else None
+        if part is not None:
+            files[name] = part
+    if row.get("자재조합"):
+        um = row_of(read_json(root / UM_FILE), str(row["자재조합"]))
+        if um is not None:
+            files[UM_FILE] = um
+    labor = labor_rows(str(row["일위대가"])) if row.get("일위대가") else None
+    if labor is not None:
+        files[LABOR_FILE] = labor
+    files[LINK_FILE] = input_links(root, column_id)
+    logic_keys = [str(one.get("키")) for one in (labor or {}).get("로직", [])]
+    files[PRICE_FILE] = prices.to_json(prices.collect(logic_keys)) if logic_keys else {}
+    return files

M02_MasterTemplete/M02_Project_Items_Router.py

@@ -0,0 +1,113 @@
+"""프로젝트 집계 항목 복사본 길 — `/api/projects/{id}/items…` (PLAN 25-1b · 25-2b · 25-2e).
+
+    GET  …/items                      목록(이름 · 원본 · 가져온날 · 마스터 바뀜)
+    POST …/items {summary_item}       미리 가져오기(구조물에 안 이은 항목)
+    GET  …/items/{폴더}/{종류}         {종류, 판, 내용}
+    PUT  …/items/{폴더}/{종류}         {판, 내용} — 판이 다르면 409 · 면적 · 재료는 서버 계산값
+    POST …/items/{폴더}/refresh        새로 가져오기(입력값 · 고른자재는 같은 키면 살림)
+    POST …/items/{폴더}/calc           복사본만으로 계산(세 비목 · 계 · 빠진 입력)
+종류 = 도면 · 면적 · 재료 · 자재조합 · 입력연결 · 입력값 · 고른자재.
+⚠ 등록은 `main.py`(로그인만) — 프로젝트 권한(같은 회사)은 여기서 `_project` 로 봄(403 · 404).
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Literal
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from common_util import common_util_spreadsheet as spreadsheet
+from common_util.common_util_auth import verify_session
+from common_util.common_util_summary_item import parse_summary_item
+from M02_MasterTemplete import M02_Item_Copy as copies
+from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layer_routes
+
+router = APIRouter(prefix="/api/projects", tags=["M02 Project Items"])
+
+Kind = Literal["도면", "면적", "재료", "자재조합", "입력연결", "입력값", "고른자재"]
+
+
+class WriteBody(BaseModel):
+    판: str | None = None
+    내용: Any
+
+
+class PrefetchBody(BaseModel):
+    summary_item: dict[str, Any]
+
+
+async def _root(session: dict[str, Any], project_id: str):
+    return (await layer_routes._project(session, project_id))["root"]
+
+
+async def _run(fn, *args):
+    try:
+        return await asyncio.to_thread(fn, *args)
+    except copies.Stale as error:
+        raise HTTPException(
+            status_code=409, detail={"message": str(error), "판": error.current}
+        ) from error
+    except LookupError as error:
+        raise HTTPException(status_code=404, detail=str(error)) from error
+    except ValueError as error:
+        raise HTTPException(status_code=400, detail=str(error)) from error
+
+
+@router.get("/{project_id}/items")
+async def list_items(
+    project_id: str, session: dict[str, Any] = Depends(verify_session)
+) -> dict[str, Any]:
+    return {"항목": await _run(copies.list_items, await _root(session, project_id))}
+
+
+@router.post("/{project_id}/items")
+async def prefetch_item(
+    project_id: str, body: PrefetchBody, session: dict[str, Any] = Depends(verify_session)
+) -> dict[str, Any]:
+    try:
+        item = parse_summary_item(body.summary_item)
+    except ValueError as error:
+        raise HTTPException(status_code=400, detail=str(error)) from error
+    return await _run(copies.prefetch, await _root(session, project_id), item)
+
+
+@router.get("/{project_id}/items/{folder}/{kind}")
+async def read_item_file(
+    project_id: str, folder: str, kind: Kind, session: dict[str, Any] = Depends(verify_session)
+) -> dict[str, Any]:
+    return await _run(copies.read_file, await _root(session, project_id), folder, kind)
+
+
+def _recalc(document: dict) -> dict:
+    try:
+        return spreadsheet.with_server_values(document)
+    except spreadsheet.RecalcError as error:
+        raise HTTPException(status_code=503, detail=str(error)) from error
+
+
+@router.put("/{project_id}/items/{folder}/{kind}")
+async def write_item_file(
+    project_id: str,
+    folder: str,
+    kind: Kind,
+    body: WriteBody,
+    session: dict[str, Any] = Depends(verify_session),
+) -> dict[str, Any]:
+    root = await _root(session, project_id)
+    return await _run(copies.write_file, root, folder, kind, body.판, body.내용, _recalc)
+
+
+@router.post("/{project_id}/items/{folder}/refresh")
+async def refresh_item(
+    project_id: str, folder: str, session: dict[str, Any] = Depends(verify_session)
+) -> dict[str, Any]:
+    return await _run(copies.refresh, await _root(session, project_id), folder)
+
+
+@router.post("/{project_id}/items/{folder}/calc")
+async def calc_item(
+    project_id: str, folder: str, session: dict[str, Any] = Depends(verify_session)
+) -> dict[str, Any]:
+    return await _run(copies.calc, await _root(session, project_id), folder)

main.py

@@ -65,6 +65,7 @@ from M02_MasterTemplete.M02_MasterTemplete_Router import router as m02_master_te
 from M02_MasterTemplete.M02_MasterTemplete_Router_Drawing import router as m02_drawing_router
 from M02_MasterTemplete.M02_MasterTemplete_Router_Items import router as m02_items_router
 from M02_MasterTemplete.M02_MasterTemplete_Router_Layers import router as m02_layers_router
+from M02_MasterTemplete.M02_Project_Items_Router import router as m02_project_items_router
 from common_util.common_util_audit import note_api_call, record_call_burst
 from common_util.common_util_auth import (
     require_company,
@@ -641,6 +642,8 @@ app.include_router(m02_master_template_router, dependencies=system_admin_only)
 app.include_router(m02_layers_router, dependencies=protected)
 app.include_router(m02_items_router, dependencies=protected)
 app.include_router(m02_drawing_router, dependencies=protected)
+# 프로젝트 집계 항목 복사본(PLAN 25-2b) — 로그인만 · 프로젝트 권한은 길 안에서.
+app.include_router(m02_project_items_router, dependencies=protected)
 # 개발 전용 잠금 해제 — 다른 라우터와 **같은 보호**를 받는다(로그인·회사·프로젝트 접근).
 # 그 위에 서버가 환경까지 한 번 더 본다.
 app.include_router(dev_unlock_router, dependencies=protected_with_company)

resources/tester/test_item_copy_complete.py

@@ -0,0 +1,335 @@
+"""프로젝트 집계 항목 복사본 완전체 (PLAN 25-2b · 25-2e · 임시 저장소 · 임시 마스터 사본).
+
+- 지정 → [저장] → 폴더에 새 파일(입력연결 · 입력값 · 고른자재 · 단가) · manifest 가져온날 · 원본판
+- 단가 묶음이 닫혀 있음 — 조합 37개 · 로직 170개를 마스터와 복사본으로 돌려 같은 값
+- 마스터 단가 한 줄을 바꿔도 프로젝트 계산 그대로 · 목록에 「마스터 바뀜」 · 새로 가져오기 뒤 바뀜
+  (입력값 · 고른자재는 같은 키면 살림)
+- 쓰기 뒤 다시 읽기 같음 · 판 다르면 409 · 모양 틀리면 400 · 고친 복사본은 안 지움
+- 옛 복사본은 읽을 때 채움 · 미리 가져오기 · 남의 프로젝트 403
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+from pathlib import Path
+from typing import Any
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from common_util.common_util_auth import verify_session
+from config import config_system
+from M01_MasterData import M01_MasterData_Store as m01
+from M02_MasterTemplete import M02_Item_Copy as copies
+from M02_MasterTemplete import M02_Item_Prices as prices
+from M02_MasterTemplete import M02_Item_Snapshot as snap
+from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layer_routes
+from M02_MasterTemplete import M02_Project_Items_Router as items_router
+from M02_MasterTemplete import M02_Template_Layers as layers
+
+REAL = config_system.PROJECT_ROOT / "resources/master_template"
+MASTER = config_system.PROJECT_ROOT / "resources/master_data"
+PIPE = {"layer": "system", "owner": None, "key": "SS000011", "name": "관공 관매설"}
+FOLDER = "system_none_SS000011"
+LABOR_PRICE = "LB000002"  # 보통인부 — 되메우기 · 파형강관 부설 호표가 씀
+
+
+@pytest.fixture(scope="module")
+def master_copy(tmp_path_factory) -> Path:
+    target = tmp_path_factory.mktemp("master_data")
+    for path in MASTER.glob("*.json"):
+        shutil.copy2(path, target / path.name)
+    return target
+
+
+@pytest.fixture
+def world(tmp_path: Path, monkeypatch, master_copy: Path) -> dict[str, Any]:
+    system = tmp_path / "master"
+    system.mkdir()
+    for name in (layers.COMBO_FILE, "자재조합.json"):
+        shutil.copy2(REAL / name, system / name)
+    for kind in ("table", "structure", "area", "material"):
+        shutil.copytree(REAL / kind, system / kind)
+    monkeypatch.setattr(layers, "SYSTEM_ROOT", system)
+    monkeypatch.setattr(m01, "FOLDER", master_copy)
+    storage = tmp_path / "storage"
+    monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage))
+    root = storage / "3" / "12" / "proj"
+    (root / "B05_Profile/route").mkdir(parents=True)
+    (root / "B04_PreProcess/drainage/edits").mkdir(parents=True)
+    return {"system": system, "root": root, "master": master_copy}
+
+
+def _save(root: Path, pipes: list[dict]) -> dict[str, list[str]]:
+    (root / "B05_Profile/route/structures.json").write_text(
+        json.dumps({"revision": 1, "structures": []}), encoding="utf-8"
+    )
+    (root / "B04_PreProcess/drainage/edits/pipe_points.json").write_text(
+        json.dumps({"points": pipes}, ensure_ascii=False), encoding="utf-8"
+    )
+    return snap.sync_snapshots(root)
+
+
+def _pipe(summary: dict | None = PIPE) -> dict[str, Any]:
+    return {"chainage_m": 40.0, "source": "user", "summary_item": summary}
+
+
+def _read(path: Path) -> Any:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _auto_values(root: Path) -> dict[str, Any]:
+    """담은 로직마다 자동값을 확인한 값으로 — 사용자가 다 채운 것과 같게."""
+    labor = _read(root / "templates/items" / FOLDER / "일위대가.json")
+    _, whole = m01.loaded()
+    out: dict[str, Any] = {}
+    for row in labor["로직"]:
+        values = m01.mau.values(whole, whole.logic(row["키"]))
+        out[row["키"]] = {k: {"값": prices.to_json(v), "확인": True} for k, v in values.items()}
+    return out
+
+
+def _fill_values(root: Path) -> dict[str, Any]:
+    current = copies.read_file(root, FOLDER, "입력값")
+    values = _auto_values(root)
+    copies.write_file(root, FOLDER, "입력값", current["판"], values)
+    return values
+
+
+# ── 새 파일 ──────────────────────────────────────────
+
+
+def test_지정하면_새_파일까지_복사(world):
+    root = world["root"]
+    assert _save(root, [_pipe()])["만듦"] == [FOLDER]
+    folder = root / "templates/items" / FOLDER
+    assert _read(folder / "입력연결.json") == {"판": None, "줄": []}  # 층 연결이 아직 없음
+    assert _read(folder / "입력값.json") == {} and _read(folder / "고른자재.json") == {}
+    price = _read(folder / "단가.json")
+    rows = {k: len(v.get("줄") or v.get("표")) for k, v in price.items()}
+    assert rows["인력.json"] >= 2 and rows["재료_자재품목.json"] > 0 and "_빠진" not in price
+    keys = {row["키"] for data in price.values() for row in data.get("줄") or data.get("표")}
+    assert {"GF000176", "GC000088", "GF000113", "GC000267", LABOR_PRICE} <= keys
+    manifest = _read(folder / "manifest.json")
+    assert manifest["가져온날"] and len(manifest["원본판"]) == 16
+    assert copies.list_items(root) == [
+        {
+            "폴더": FOLDER,
+            "이름": "관공 관매설",
+            "원본": PIPE,
+            "가져온날": manifest["가져온날"],
+            "마스터바뀜": False,
+            "구조물": ["pipe@40.0"],
+            "고침": False,
+        }
+    ]
+
+
+def test_층_입력연결을_복사(world):
+    root, system = world["root"], world["system"]
+    (system / "inputs").mkdir()
+    link = {"판": "a1", "줄": [{"로직": "GF000176", "입력": "관경", "자리": "면적", "알약": "x"}]}
+    column = next(
+        r["열"] for r in _read(system / layers.COMBO_FILE)["줄"] if r["키"] == PIPE["key"]
+    )
+    (system / "inputs" / f"{column}.json").write_text(json.dumps(link, ensure_ascii=False), "utf-8")
+    _save(root, [_pipe()])
+    assert copies.read_file(root, FOLDER, "입력연결")["내용"] == link
+
+
+def test_단가_묶음은_닫혀_있음(world):
+    """조합마다 복사본 단가 묶음만으로 돌린 값 = 마스터로 돌린 값(자동값 · 고르기 끝값)."""
+    files, whole = prices.master()
+    combos = [r for d in files.values() if d.get("그룹") == "일위대가조합" for r in d["줄"]]
+    runs = 0
+    for ua in combos:
+        keys = prices.mcb.logic_keys(ua)
+        snapshot = prices.load_prices(prices.to_json(prices.collect(keys)))
+        for key in keys:
+            try:
+                row = whole.logic(key)
+            except prices.mf.FormulaError:
+                continue
+            for pick in (0, -1):
+                values = m01.mau.values(whole, row)
+                for spec in row.get("입력", []):
+                    if spec.get("고르기"):
+                        values[spec["이름"]] = spec["고르기"][pick]
+                assert _run(whole, key, values) == _run(snapshot, key, values), (ua["키"], key)
+                runs += 1
+    assert len(combos) >= 30 and runs >= 300
+
+
+def _run(master: Any, key: str, values: dict) -> str:
+    try:
+        got = prices.mf.run(master, key, dict(values))
+    except Exception as error:  # 까닭도 같아야 함
+        return f"멈춤 {error}"
+    return str(got.get("계", got.get("결과")))
+
+
+# ── 계산 · 마스터 바뀜 · 새로 가져오기 ───────────────────
+
+
+def test_입력이_없으면_빠진_입력_목록(world):
+    root = world["root"]
+    _save(root, [_pipe()])
+    got = copies.calc(root, FOLDER)
+    assert got["계"] == 0 and len(got["멈춤"]) == 3
+    assert {"로직": "GF000176", "입력": "관경"} in got["빠진입력"]
+
+
+def test_마스터_단가를_바꿔도_계산_그대로_새로_가져오면_바뀜(world):
+    root, master_dir = world["root"], world["master"]
+    _save(root, [_pipe()])
+    values = _fill_values(root)
+    before = copies.calc(root, FOLDER)
+    assert before["빠진입력"] == [] and before["멈춤"] == [] and before["계"] > 0
+    files, whole = m01.loaded()
+    ua = next(
+        r
+        for d in files.values()
+        if d.get("그룹") == "일위대가조합"
+        for r in d["줄"]
+        if r["키"] == "UA000036"
+    )
+    given = {k: {n: v["값"] for n, v in one.items()} for k, one in values.items()}
+    expected = prices.to_json(prices.mcb.run(whole, ua, given))
+    assert before["계"] == pytest.approx(expected["계"]) and before["노무비"] > 0
+    labor_file = master_dir / "인력.json"
+    original = labor_file.read_bytes()
+    try:
+        data = _read(labor_file)
+        row = next(r for r in data["줄"] if r["키"] == LABOR_PRICE)
+        row["값"] = row["값"] * 2
+        labor_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+        assert copies.calc(root, FOLDER) == before  # 복사본만 씀
+        assert copies.list_items(root)[0]["마스터바뀜"] is True
+        copies.refresh(root, FOLDER)
+        after = copies.calc(root, FOLDER)
+        assert after["노무비"] > before["노무비"] and after["재료비"] == before["재료비"]
+        assert copies.read_file(root, FOLDER, "입력값")["내용"] == values  # 같은 키 살림
+        assert copies.list_items(root)[0]["마스터바뀜"] is False
+    finally:
+        labor_file.write_bytes(original)
+
+
+def test_새로_가져오면_없는_키는_버림(world):
+    root = world["root"]
+    _save(root, [_pipe()])
+    folder = root / "templates/items" / FOLDER
+    (folder / "입력값.json").write_text(
+        json.dumps(
+            {
+                "GF000176": {"관경": {"값": 600, "확인": True}, "옛입력": {"값": 1}},
+                "GX999999": {"a": {"값": 1}},
+            },
+            ensure_ascii=False,
+        ),
+        "utf-8",
+    )
+    (folder / "고른자재.json").write_text(
+        json.dumps({"GF000176": {"0": "MT000001", "99": "MT000002"}}), "utf-8"
+    )
+    copies.refresh(root, FOLDER)
+    assert _read(folder / "입력값.json") == {"GF000176": {"관경": {"값": 600, "확인": True}}}
+    assert _read(folder / "고른자재.json") == {"GF000176": {"0": "MT000001"}}
+
+
+# ── 쓰기 · 판 · 모양 · 지우기 규칙 ─────────────────────────
+
+
+def test_쓰기_뒤_다시_읽기_같음_판_다르면_409(world):
+    root = world["root"]
+    _save(root, [_pipe()])
+    first = copies.read_file(root, FOLDER, "고른자재")
+    wrote = copies.write_file(
+        root, FOLDER, "고른자재", first["판"], {"GF000176": {"0": "MT000010"}}
+    )
+    again = copies.read_file(root, FOLDER, "고른자재")
+    assert again["내용"] == {"GF000176": {"0": "MT000010"}} and again["판"] == wrote["판"]
+    with pytest.raises(copies.Stale):
+        copies.write_file(root, FOLDER, "고른자재", first["판"], {})
+    for kind, bad in (
+        ("입력값", {"GF000176": {"관경": 600}}),
+        ("입력값", {"GF000176": {"관경": {"값": 1, "확인": "네"}}}),
+        ("고른자재", {"GF000176": {"첫": "MT000001"}}),
+        ("입력연결", {"줄": [{"로직": "GF000176", "입력": "관경", "자리": "어딘가"}]}),
+        ("도면", {}),
+    ):
+        with pytest.raises(ValueError):
+            copies.write_file(root, FOLDER, kind, again["판"], bad)
+    with pytest.raises(LookupError):
+        copies.read_file(root, "../../etc", "입력값")
+
+
+def test_고친_복사본은_구조물에서_빠져도_안_지움(world):
+    root = world["root"]
+    _save(root, [_pipe()])
+    assert _save(root, [_pipe(None)])["지움"] == [FOLDER]  # 안 고쳤으면 지움(지금 규칙)
+    _save(root, [_pipe()])
+    current = copies.read_file(root, FOLDER, "입력값")
+    copies.write_file(root, FOLDER, "입력값", current["판"], {"GF000176": {"관경": {"값": 600}}})
+    assert _save(root, [_pipe(None)])["지움"] == []
+    assert copies.list_items(root)[0]["구조물"] == [] and copies.list_items(root)[0]["고침"]
+
+
+def test_미리_가져오기(world):
+    root = world["root"]
+    _save(root, [])
+    made = copies.prefetch(root, PIPE)
+    assert made["폴더"] == FOLDER and made["만듦"] is True
+    assert copies.prefetch(root, PIPE)["만듦"] is False  # 있으면 그대로
+    assert copies.list_items(root)[0]["구조물"] == []
+    assert _save(root, [])["지움"] == [FOLDER]  # 안 고쳤으면 [저장] 때 지움
+
+
+def test_옛_복사본은_읽을_때_채움(world):
+    root = world["root"]
+    _save(root, [_pipe()])
+    folder = root / "templates/items" / FOLDER
+    for name in ("입력연결.json", "입력값.json", "고른자재.json", "단가.json"):
+        (folder / name).unlink()
+    manifest = _read(folder / "manifest.json")
+    for key in ("원본판", "가져온날"):
+        manifest.pop(key)
+    (folder / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False), "utf-8")
+    assert copies.read_file(root, FOLDER, "입력값")["내용"] == {}
+    filled = _read(folder / "manifest.json")
+    assert filled["가져온날"] == filled["적용일"] and filled["옮겨받음"]
+    assert (folder / "단가.json").is_file() and copies.calc(root, FOLDER)["멈춤"]
+    assert copies.list_items(root)[0]["마스터바뀜"] is False  # 원본이 안 바뀐 옛 복사본
+
+
+# ── 길 · 권한 ────────────────────────────────────────
+
+
+def test_길_권한_남의_프로젝트는_403(world, monkeypatch):
+    root = world["root"]
+    _save(root, [_pipe()])
+
+    async def fake_row(project_id: str) -> dict[str, Any]:
+        return {"company_id": 3, "storage_path": str(root)}
+
+    monkeypatch.setattr(layer_routes, "_project_row", fake_row)
+    monkeypatch.setattr(layer_routes, "resolve_stored_project_path", lambda path: path)
+    app = FastAPI()
+    app.include_router(items_router.router)
+    session = {"role": "USER", "company_id": 3}
+    app.dependency_overrides[verify_session] = lambda: session
+    client = TestClient(app)
+    base = "/api/projects/p1/items"
+    assert client.get(base).json()["항목"][0]["폴더"] == FOLDER
+    got = client.get(f"{base}/{FOLDER}/입력값").json()
+    put = client.put(f"{base}/{FOLDER}/입력값", json={"판": got["판"], "내용": {}})
+    assert put.status_code == 200
+    stale = client.put(f"{base}/{FOLDER}/입력값", json={"판": got["판"] + "x", "내용": {}})
+    assert stale.status_code == 409 and stale.json()["detail"]["판"] == put.json()["판"]
+    assert client.get(f"{base}/{FOLDER}/모름").status_code == 422
+    assert client.post(f"{base}/{FOLDER}/calc").json()["멈춤"]
+    assert client.post(base, json={"summary_item": {"layer": "x"}}).status_code == 400
+    session["company_id"] = 4
+    assert client.get(base).status_code == 403
+    assert client.post(f"{base}/{FOLDER}/refresh").status_code == 403