diff --git a/M02_MasterTemplete/M02_Item_Copy.py b/M02_MasterTemplete/M02_Item_Copy.py new file mode 100644 index 00000000..4117b2c5 --- /dev/null +++ b/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 {}, + ) diff --git a/M02_MasterTemplete/M02_Item_Prices.py b/M02_MasterTemplete/M02_Item_Prices.py new file mode 100644 index 00000000..9db65771 --- /dev/null +++ b/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}) diff --git a/M02_MasterTemplete/M02_Item_Snapshot.py b/M02_MasterTemplete/M02_Item_Snapshot.py index 1c35e542..cd583500 100644 --- a/M02_MasterTemplete/M02_Item_Snapshot.py +++ b/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 diff --git a/M02_MasterTemplete/M02_Item_Source.py b/M02_MasterTemplete/M02_Item_Source.py new file mode 100644 index 00000000..e5c45c27 --- /dev/null +++ b/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 diff --git a/M02_MasterTemplete/M02_Project_Items_Router.py b/M02_MasterTemplete/M02_Project_Items_Router.py new file mode 100644 index 00000000..0b183cae --- /dev/null +++ b/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) diff --git a/main.py b/main.py index 2513076b..bee37211 100644 --- a/main.py +++ b/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) diff --git a/resources/tester/test_item_copy_complete.py b/resources/tester/test_item_copy_complete.py new file mode 100644 index 00000000..3367b7b8 --- /dev/null +++ b/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