diff --git a/M02_MasterTemplete/M02_Item_Copy.py b/M02_MasterTemplete/M02_Item_Copy.py new file mode 100644 index 00000000..f7e3cd8c --- /dev/null +++ b/M02_MasterTemplete/M02_Item_Copy.py @@ -0,0 +1,262 @@ +"""프로젝트 집계 항목 복사본 읽기 · 쓰기 · 새로 가져오기 · 미리 가져오기 · 계산 (PLAN 25-2b). + +복사본 = `{프로젝트}/templates/items/{폴더}/` (`M02_Item_Snapshot`). 파일마다 판 = 내용 해시 — +쓸 때 판이 다르면 `Stale`(409). 고치면 manifest 에 `고침` 을 남겨 구조물에 안 쓰여도 안 지움. +옛 복사본(새 파일 없음)은 읽을 때 채움(`complete_old`). 계산은 복사본만 씀(마스터 현재값 안 읽음). +""" + +from __future__ import annotations + +import re +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from common_util.common_util_json import atomic_write_json +from common_util.common_util_summary_item import snapshot_folder_name +from M02_MasterTemplete import M02_Item_Prices as prices +from M02_MasterTemplete import M02_Template_Layers as layers +from M02_MasterTemplete.M02_Item_Snapshot import ( + MANIFEST_NAME, + PICKS_FILE, + VALUES_FILE, + build_snapshot, + complete_old, + items_dir, + now_text, + write_snapshot, +) +from M02_MasterTemplete.M02_Item_Source import ( + LABOR_FILE, + LINK_FILE, + PRICE_FILE, + build_source, + read_json, + version_of, +) + +#: 길의 종류 이름 → 복사본 파일 +KINDS = { + "도면": "도면.json", + "면적": "면적.json", + "재료": "재료.json", + "자재조합": "자재조합.json", + "입력연결": LINK_FILE, + "입력값": VALUES_FILE, + "고른자재": PICKS_FILE, +} +#: 뼈대 검사에 쓰는 층 양식 종류(도면 · 면적산출 · 재료산출) +SKELETON = {"도면": "structure", "면적": "area", "재료": "material"} +SEATS = ("면적", "재료", "선택", None) +_FOLDER = re.compile(r"^(system|company|personal)_[0-9A-Za-z-]+_SS\d{6}$") +_MT = re.compile(r"^MT\d{6}$") + + +class Stale(ValueError): + def __init__(self, current: str): + super().__init__("다른 곳에서 먼저 고쳤습니다 — 다시 읽고 고치세요.") + self.current = current + + +def folder_of(project_root: str | Path, folder: str) -> Path: + """복사본 폴더 — 이름 모양이 틀리거나 없으면 LookupError(404).""" + if not _FOLDER.match(folder): + raise LookupError(f"복사본 폴더 이름 모양이 아님 「{folder}」") + target = items_dir(project_root) / folder + if not (target / MANIFEST_NAME).is_file(): + raise LookupError(f"복사본이 없음 「{folder}」") + complete_old(target) + return target + + +def source_changed(manifest: dict[str, Any]) -> bool | None: + """지금 원본(층 · 마스터)을 다시 모은 판이 가져온 판과 다른가 — 원본이 없어졌으면 None.""" + try: + source = build_source(manifest["원본"]) + except (LookupError, ValueError, KeyError): + return None + return version_of(source) != manifest.get("원본판") + + +def list_items(project_root: str | Path) -> list[dict[str, Any]]: + base = items_dir(project_root) + out = [] + for child in sorted(base.iterdir()) if base.is_dir() else []: + if not child.is_dir() or not _FOLDER.match(child.name): + continue + if not (child / MANIFEST_NAME).is_file(): + continue + complete_old(child) + manifest = read_json(child / MANIFEST_NAME) or {} + origin = manifest.get("원본") or {} + changed = source_changed(manifest) + out.append( + { + "폴더": child.name, + "이름": origin.get("name"), + "원본": origin, + "원본판": manifest.get("원본판"), + "가져온날": manifest.get("가져온날"), + "마스터바뀜": bool(changed), + "원본없음": changed is None, + "구조물": manifest.get("구조물") or [], + "고침": bool(manifest.get("고침")), + } + ) + return out + + +def read_file(project_root: str | Path, folder: str, kind: str) -> dict[str, Any]: + target = folder_of(project_root, folder) + content = read_json(target / KINDS[kind]) + return {"종류": kind, "판": version_of(content), "내용": content} + + +# ── 모양 검사 ─────────────────────────────────────────────────────── + + +def _check_values(content: Any) -> None: + if not isinstance(content, dict): + raise ValueError("입력값은 {로직키: {입력이름: {값, 확인}}} 이어야 합니다.") + for logic, inputs in content.items(): + if not isinstance(inputs, dict): + raise ValueError(f"입력값 「{logic}」 은 {{입력이름: {{값, 확인}}}} 이어야 합니다.") + for name, one in inputs.items(): + if not isinstance(one, dict) or set(one) - {"값", "확인"}: + raise ValueError(f"입력값 「{logic}|{name}」 은 {{값, 확인}} 이어야 합니다.") + if not isinstance(one.get("확인", False), bool): + raise ValueError(f"입력값 「{logic}|{name}」 확인은 참 · 거짓이어야 합니다.") + if isinstance(one.get("값"), (dict, list)): + raise ValueError(f"입력값 「{logic}|{name}」 값은 수 · 글 · 빈 값이어야 합니다.") + + +def _check_picks(content: Any) -> None: + if not isinstance(content, dict): + raise ValueError("고른자재는 {로직키: {호표 차례: MT키}} 이어야 합니다.") + for logic, picks in content.items(): + if not isinstance(picks, dict): + raise ValueError(f"고른자재 「{logic}」 은 {{호표 차례: MT키}} 이어야 합니다.") + for at, key in picks.items(): + if not str(at).isdigit() or not isinstance(key, str) or not _MT.match(key): + raise ValueError(f"고른자재 「{logic}」 {at} → 「{key}」 — 차례는 수 · 값은 MT 키") + + +def _check_links(content: Any) -> None: + rows = content.get("줄") if isinstance(content, dict) else None + if not isinstance(rows, list): + raise ValueError("입력연결은 {판, 줄: [...]} 이어야 합니다.") + for row in rows: + if not isinstance(row, dict) or not row.get("로직") or not row.get("입력"): + raise ValueError("입력연결 줄은 {로직, 입력, 자리, 알약?} 이어야 합니다.") + if row.get("자리") not in SEATS: + raise ValueError(f"입력연결 자리 「{row.get('자리')}」 — 면적 · 재료 · 선택 · 빈 값") + + +def check(kind: str, content: Any) -> None: + if kind == "입력값": + _check_values(content) + elif kind == "고른자재": + _check_picks(content) + elif kind == "입력연결": + _check_links(content) + elif kind in SKELETON: + layers.check_skeleton(SKELETON[kind], content) + elif not isinstance(content, dict): + raise ValueError(f"{kind} 은 JSON 객체여야 합니다.") + + +def write_file( + project_root: str | Path, + folder: str, + kind: str, + version: str | None, + content: Any, + recalc: Callable[[dict], dict] | None = None, +) -> dict[str, Any]: + """판이 같을 때만 씀 — 면적 · 재료는 `recalc`(서버 계산값)를 거친 것을 씀.""" + target = folder_of(project_root, folder) + check(kind, content) + current = version_of(read_json(target / KINDS[kind])) + if version != current: + raise Stale(current) + if recalc is not None and kind in ("면적", "재료"): + content = recalc(content) + atomic_write_json(target / KINDS[kind], content) + manifest = read_json(target / MANIFEST_NAME) or {} + atomic_write_json(target / MANIFEST_NAME, {**manifest, "고침": True, "고친날": now_text()}) + return {"종류": kind, "판": version_of(content)} + + +# ── 새로 가져오기 · 미리 가져오기 ───────────────────────────────────── + + +def _kept_values(values: Any, labor: dict[str, Any]) -> dict[str, Any]: + """같은 로직 · 같은 입력 이름만 살림.""" + names = { + str(row.get("키")): {spec["이름"] for spec in row.get("입력", [])} + for row in labor.get("로직", []) + } + out: dict[str, Any] = {} + for logic, inputs in (values or {}).items(): + kept = {k: v for k, v in (inputs or {}).items() if k in names.get(logic, set())} + if kept: + out[logic] = kept + return out + + +def _kept_picks(picks: Any, labor: dict[str, Any]) -> dict[str, Any]: + """같은 로직 · 있는 호표 차례만 살림.""" + lines = {str(row.get("키")): len(row.get("호표", [])) for row in labor.get("로직", [])} + out: dict[str, Any] = {} + for logic, chosen in (picks or {}).items(): + kept = {k: v for k, v in (chosen or {}).items() if int(k) < lines.get(logic, 0)} + if kept: + out[logic] = kept + return out + + +def refresh(project_root: str | Path, folder: str) -> dict[str, Any]: + """층 · 마스터에서 다시 가져옴 — 입력값 · 고른자재는 같은 키면 살림.""" + target = folder_of(project_root, folder) + manifest = read_json(target / MANIFEST_NAME) or {} + values = read_json(target / VALUES_FILE) + picks = read_json(target / PICKS_FILE) + files = build_snapshot(manifest["원본"]) + labor = files.get(LABOR_FILE) or {} + files[VALUES_FILE] = _kept_values(values, labor) + files[PICKS_FILE] = _kept_picks(picks, labor) + for old in target.glob("*.json"): # 원본에서 빠진 파일(도면 없앰 등)은 지움 + if old.name != MANIFEST_NAME and old.name not in files: + old.unlink() + for name, content in files.items(): + atomic_write_json(target / name, content) + source = {k: v for k, v in files.items() if k not in (VALUES_FILE, PICKS_FILE)} + stamp = now_text() + fresh = {**manifest, "판": version_of(source), "원본판": version_of(source), "가져온날": stamp} + atomic_write_json(target / MANIFEST_NAME, fresh) + return {"폴더": folder, "가져온날": stamp, "원본판": fresh["원본판"]} + + +def prefetch(project_root: str | Path, item: dict[str, Any]) -> dict[str, Any]: + """구조물에 아직 안 이은 항목을 미리 가져옴 — 있으면 그대로 · 없으면 새로(쓰는 구조물 없음).""" + folder = snapshot_folder_name(item) + target = items_dir(project_root) / folder + manifest = read_json(target / MANIFEST_NAME) + made = not isinstance(manifest, dict) + if made: + manifest = write_snapshot(target, item, [], {"미리가져옴": now_text()}) + return {"폴더": folder, "만듦": made, "가져온날": manifest.get("가져온날")} + + +def calc(project_root: str | Path, folder: str) -> dict[str, Any]: + """복사본(입력값 · 고른자재 · 단가 · 일위대가)만으로 계산 — 마스터 현재값을 안 읽음.""" + target = folder_of(project_root, folder) + labor = read_json(target / LABOR_FILE) + if not isinstance(labor, dict): + raise LookupError(f"일위대가 조합이 없는 항목 「{folder}」 — 계산할 것이 없음") + return prices.calc( + labor, + read_json(target / PRICE_FILE) or {}, + read_json(target / VALUES_FILE) or {}, + read_json(target / PICKS_FILE) or {}, + ) 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..134c62ce --- /dev/null +++ b/M02_MasterTemplete/M02_Item_Source.py @@ -0,0 +1,138 @@ +"""집계 항목 원본 — 층(시스템 · 회사 · 개인)에서 항목 내용을 모음 (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/`) +#: 열 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 item_name(item: dict[str, Any]) -> str: + """층 SS 줄의 보이는 이름(종류 · 분류1~3) — 없으면 LookupError.""" + row = row_of(read_json(layer_root(item) / layers.COMBO_FILE), str(item.get("key"))) + if row is None: + raise LookupError(f"집계 항목 {item.get('key')} 이 {item.get('layer')} 층에 없음") + parts = (row.get(k) for k in ("종류", "분류1", "분류2", "분류3")) + return " ".join(str(p) for p in parts if p) or str(row["키"]) + + +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`(sub4 `_Inputs.read_file`) · 없으면 빈 연결.""" + from M02_MasterTemplete import M02_MasterTemplete_Inputs as inputs + from M02_MasterTemplete import M02_MasterTemplete_Store as store + + if not column_id: + return {"판": None, "줄": []} + with store.at(root): + found = inputs.read_file(column_id) + 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_MasterTemplete_Inputs.py b/M02_MasterTemplete/M02_MasterTemplete_Inputs.py new file mode 100644 index 00000000..92692c6f --- /dev/null +++ b/M02_MasterTemplete/M02_MasterTemplete_Inputs.py @@ -0,0 +1,229 @@ +"""M02 항목 선택 입력(PLAN 25-1b · 25-2a) — 입력 목록 · 항목 연결 파일 · 알약. + +항목 = 구조물집계표조합(SS) 한 줄 → 그 열의 일위대가(UA) 조합 → 담은 단가산출 로직마다 입력 전부. +로직마다 갈라 보이고 이름이 같아도 합치지 않음(로직마다 값이 다를 수 있음) · 입력 목록은 그 로직 +줄의 `입력` 그대로(하위 로직으로 넘겨지는 입력은 맨 위 로직 입력으로만 · M01 조합 미리 보기와 +같음) · 자동값 = `master_auto.values`(보기용 · 저장 안 함). + +항목 연결 파일 = 층 폴더 `inputs/<열id>.json` = `{판, 줄: [{로직, 입력, 자리, 알약?}]}` — +판은 저장마다 1 씩 늘고, 연결만 적음(값은 프로젝트 몫). 자리 면적 · 재료 = 그 시트의 알약 +(키 `입력|{로직}|{입력}`) · 선택 = 「선택 입력」 카드 · null = 안 씀. +M01 은 읽기만(`M01_MasterData_Store` · `_Store_Combo` · `master_auto`) · 권한은 부르는 쪽 몫. +""" + +from __future__ import annotations + +import json +from decimal import Decimal +from pathlib import Path +from typing import Any + +from common_util.common_util_json import atomic_write_json +from M01_MasterData import M01_MasterData_Store as m01 +from M01_MasterData import M01_MasterData_Store_Combo as combo_store +from M02_MasterTemplete import M02_MasterTemplete_Store as store + +PLACES = ("면적", "재료", "선택", None) +PILL_HEAD = "입력" +_SPEC = ("이름", "설명", "단위", "고르기", "범위") + + +def _num(value: Any) -> Any: + """Decimal → JSON 수(M01 길과 같은 규칙) · 목록 · 묶음 안까지.""" + if isinstance(value, Decimal): + return int(value) if value == value.to_integral_value() else float(value) + if isinstance(value, list): + return [_num(v) for v in value] + if isinstance(value, dict): + return {k: _num(v) for k, v in value.items()} + return value + + +def pill_key(logic: str, name: str) -> str: + return f"{PILL_HEAD}|{logic}|{name}" + + +def _item(key: str) -> dict[str, Any]: + """지금 층(`store.at`)의 항목 줄 — 없으면 404.""" + row = next((r for r in store.read_combos().get("줄", []) if r.get("키") == key), None) + if row is None: + raise store.StoreError(404, f"없는 항목 「{key}」") + return row + + +def _combo_row(ua: str) -> dict[str, Any] | None: + rows = m01.read(combo_store.FILE)[0].get("줄") or [] + return next((r for r in rows if str(r.get("키")) == ua), None) + + +def combo_logics(ua: str | None) -> tuple[dict[str, Any] | None, list[dict], list[str]]: + """(조합 줄, [로직 줄 …], 빠진 로직 키) — 조합이 없거나 못 찾으면 (None, [], []).""" + if not ua: + return None, [], [] + row = _combo_row(str(ua)) + if row is None: + return None, [], [] + whole = m01.loaded()[1] + found, missing = [], [] + for one in combo_store.mcb.items(row): + code = str(one.get("로직") or "") if isinstance(one, dict) else "" + try: + found.append(whole.logic(code)) + except combo_store.cm.mf.FormulaError: + missing.append(code) + return row, found, missing + + +def inputs(key: str) -> dict[str, Any]: + """25-1b 입력 목록 — `{일위대가, 이름, 로직: [{로직, 이름, 결과단위, 입력: [...]}], 빠진로직}`. + 일위대가 없는 항목은 로직 빈 목록.""" + ua = _item(key).get("일위대가") + row, logics, missing = combo_logics(ua) + whole = m01.loaded()[1] + out = [] + for logic in logics: + auto = m01.mau.values(whole, logic) + specs = [] + for spec in logic.get("입력") or []: + one = {k: spec[k] for k in _SPEC if k in spec} + one.setdefault("설명", None) + one.setdefault("단위", None) + specs.append({**one, "자동값": auto.get(spec["이름"])}) + out.append( + { + "로직": logic.get("키"), + "이름": logic.get("이름"), + "결과단위": logic.get("결과단위"), + "입력": specs, + } + ) + return _num( + { + "일위대가": ua or None, + "이름": row.get("이름") if row else None, + "로직": out, + "빠진로직": missing, + } + ) + + +# ── 항목 연결 파일 ───────────────────────────────────── + + +def read_file(column: str) -> dict[str, Any] | None: + """지금 층의 `inputs/<열id>.json` 그대로 — 없으면 None(열 지우기 · 복사 · 스냅샷이 씀).""" + path = store.inputs_path(column) + return json.loads(path.read_bytes()) if path.is_file() else None + + +def write_file(column: str, doc: dict[str, Any]) -> None: + """지금 층에 연결 파일을 그대로 씀(복사용 — 검사 · 판 비교 없음).""" + path = store.inputs_path(column) + path.parent.mkdir(parents=True, exist_ok=True) + atomic_write_json(path, doc) + + +def _known(key: str) -> dict[str, set[str]]: + """{로직 키: {입력 이름}} — 그 항목 조합이 지금 담은 것.""" + _, logics, _ = combo_logics(_item(key).get("일위대가")) + return {str(lg.get("키")): {str(s.get("이름")) for s in lg.get("입력") or []} for lg in logics} + + +def read_links(key: str) -> dict[str, Any]: + """항목 연결 — 없으면 `{판: null, 줄: []}` · 조합이 바뀌어 사라진 로직 · 입력 줄은 + `사라짐: true`.""" + doc = read_file(_item(key)["열"]) or {"판": None, "줄": []} + known = _known(key) + rows = [] + for row in doc.get("줄") or []: + gone = str(row.get("입력")) not in known.get(str(row.get("로직")), set()) + rows.append({**row, "사라짐": True} if gone else row) + return {"판": doc.get("판"), "줄": rows} + + +def _clean_rows(key: str, rows: Any) -> list[dict[str, Any]]: + """없는 로직 · 입력 · 모르는 자리 · 같은 입력 두 번은 422 · 알약 키는 서버가 붙임.""" + if not isinstance(rows, list): + raise store.StoreError(422, "「줄」 이 목록 아님") + known = _known(key) + out, seen, errors = [], set(), [] + for at, row in enumerate(rows, start=1): + if not isinstance(row, dict): + errors.append(f"줄{at} · 묶음 아님") + continue + logic, name, place = str(row.get("로직") or ""), str(row.get("입력") or ""), row.get("자리") + if logic not in known: + errors.append(f"줄{at} · 이 항목 일위대가에 없는 로직 「{logic}」") + elif name not in known[logic]: + errors.append(f"줄{at} · 로직 {logic} 에 없는 입력 「{name}」") + elif place not in PLACES: + errors.append(f"줄{at} · 모르는 자리 「{place}」 — 면적 · 재료 · 선택 · null") + elif (logic, name) in seen: + errors.append(f"줄{at} · 같은 입력 두 번 「{logic}|{name}」") + else: + seen.add((logic, name)) + one: dict[str, Any] = {"로직": logic, "입력": name, "자리": place} + if place in ("면적", "재료"): + one["알약"] = pill_key(logic, name) + out.append(one) + if errors: + raise store.StoreError(422, {"errors": errors}) + return out + + +def write_links(key: str, version: Any, rows: Any) -> dict[str, Any]: + """항목 연결 저장 — 받은 `판` 이 지금 판(없으면 null)과 다르면 409 · 저장마다 판 + 1.""" + column = _item(key)["열"] + clean = _clean_rows(key, rows) + with store._LOCK: + have = (read_file(column) or {}).get("판") + if version != have: + raise store.StoreError(409, {"stale": [f"inputs/{column}"], "판": have}) + doc = {"판": int(have or 0) + 1, "줄": clean} + write_file(column, doc) + return doc + + +# ── 알약 — 면적산출 · 재료산출 시트 ─────────────────────── + + +def pills(key: str, place: str) -> dict[str, Any]: + """그 시트 알약 — 면적이면 조합 물량 알약(M01 `combo_inputs` · 권한 무관 읽기) + 자리를 + 그 시트로 이은 입력 알약 · 재료면 이은 입력 알약만. 입력 알약 = `{키, 이름, 단위, 로직, 입력}` — + 이름은 로직마다 갈라 「입력 · 로직 이름」(같은 이름 입력이 여러 로직에 있어도 안 합침).""" + if place not in ("면적", "재료"): + raise store.StoreError(422, f"모르는 자리 「{place}」 — 면적 · 재료") + ua = _item(key).get("일위대가") + quantity: list[dict[str, Any]] = [] + if place == "면적" and ua and _combo_row(str(ua)): + quantity = combo_store.combo_inputs(str(ua))["물량"] + _, logics, _ = combo_logics(ua) + by_key = {str(lg.get("키")): lg for lg in logics} + linked = [] + for row in read_links(key)["줄"]: + if row.get("자리") != place or row.get("사라짐"): + continue + logic = by_key[str(row["로직"])] + spec = next(s for s in logic.get("입력") or [] if s.get("이름") == row["입력"]) + short = combo_store._TAIL.sub("", str(logic.get("이름") or "")) or row["로직"] + linked.append( + { + "키": pill_key(row["로직"], row["입력"]), + "이름": f"{row['입력']} · {short}", + "단위": spec.get("단위"), + "로직": row["로직"], + "입력": row["입력"], + } + ) + return _num({"물량": quantity, "입력": linked}) + + +def copy_between(source: Path, target: Path, source_column: str, target_column: str) -> bool: + """다른 층(또는 같은 층) 열의 연결 파일을 새 열로 — 판은 1 부터 · 없으면 아무것도 안 함.""" + with store.at(source): + doc = read_file(source_column) + if doc is None: + return False + with store.at(target): + write_file(target_column, {"판": 1, "줄": doc.get("줄") or []}) + return True diff --git a/M02_MasterTemplete/M02_MasterTemplete_Items.py b/M02_MasterTemplete/M02_MasterTemplete_Items.py index ea87928a..f535a722 100644 --- a/M02_MasterTemplete/M02_MasterTemplete_Items.py +++ b/M02_MasterTemplete/M02_MasterTemplete_Items.py @@ -15,6 +15,7 @@ from typing import Any from common_util.common_util_json import atomic_write_json from M01_MasterData import M01_MasterData_Store as m01 from M01_MasterData import M01_MasterData_Store_Combo as combo_store +from M02_MasterTemplete import M02_MasterTemplete_Inputs as inputs_store from M02_MasterTemplete import M02_MasterTemplete_Matset as matset from M02_MasterTemplete import M02_MasterTemplete_Store as store @@ -104,7 +105,7 @@ def _fresh_id(want: str, used: set[str]) -> str: def copy_item(source: Path, target: Path, key: str, origin: dict[str, Any]) -> dict[str, Any]: """다른 층 항목(SS `key`) → `target` 층 새 열 하나 — 머리 · 단위 · 일위대가 · 바인딩 그대로 · - 도면 · 면적산출 · 재료산출 내용(알약 연결 포함) · 자재 조합 줄(새 UM) 복사 · + 도면 · 면적산출 · 재료산출 내용(알약 연결 포함) · 항목 연결 · 자재 조합 줄(새 UM) 복사 · 새 번호 · 새 SS 키 · 열 `본뜬` = `origin`(원래 층 · 주인 · 키). 돌려줌 `{key, 열, 이름}`.""" with store.at(source): @@ -153,4 +154,5 @@ def copy_item(source: Path, target: Path, key: str, origin: dict[str, Any]) -> d with store._LOCK: atomic_write_json(store._path(kind, new_id), body) ss = next(r for r in store.read_combos()["줄"] if r.get("열") == new_id) + inputs_store.copy_between(source, target, col_id, new_id) # 항목 연결도 같이(판 1) return {"key": ss["키"], "열": new_id, "이름": " ".join(h for h in store._heads(made) if h)} diff --git a/M02_MasterTemplete/M02_MasterTemplete_Router_Items.py b/M02_MasterTemplete/M02_MasterTemplete_Router_Items.py index 9cfbe6c8..08f8ad31 100644 --- a/M02_MasterTemplete/M02_MasterTemplete_Router_Items.py +++ b/M02_MasterTemplete/M02_MasterTemplete_Router_Items.py @@ -20,6 +20,7 @@ from pydantic import BaseModel from common_util.common_util_auth import verify_session from M02_MasterTemplete import M02_MasterTemplete_Branch as branch +from M02_MasterTemplete import M02_MasterTemplete_Inputs as inputs_store from M02_MasterTemplete import M02_MasterTemplete_Items as items_store from M02_MasterTemplete import M02_MasterTemplete_Matset as matset from M02_MasterTemplete import M02_MasterTemplete_Store as store @@ -38,6 +39,11 @@ Session = dict[str, Any] NO_COMPANY = "회사부터 등록 — 회사 없이는 회사 · 개인 구조물 집계표를 만들 수 없음" +class InputLinksBody(BaseModel): + 판: int | None = None + 줄: list[Any] = [] + + class CopyItemBody(BaseModel): from_layer: Literal["system", "company"] from_owner: str | None = None @@ -145,6 +151,50 @@ def copy_item(body: CopyItemBody, session: Session = Depends(verify_session)) -> raise HTTPException(status_code=e.status, detail=e.detail) from e +# ── 항목 선택 입력(PLAN 25-1b · 25-2a) — 읽기 = 층 읽기 권한(시스템 층 = 로그인 누구나) ── + + +@router.get("/layers/{layer}/items/{key}/inputs") +def get_item_inputs( + layer: Layer, key: str, owner: str | None = None, session: Session = Depends(verify_session) +) -> dict: + """그 항목 일위대가가 담은 로직마다 입력 전부(합치지 않음) · 자동값 — M01 길은 관리자 전용이라 + 일반 사용자는 이 길로 읽음.""" + return _call(_root(session, layer, owner), inputs_store.inputs, key) + + +@router.get("/layers/{layer}/items/{key}/input-links") +def get_input_links( + layer: Layer, key: str, owner: str | None = None, session: Session = Depends(verify_session) +) -> dict: + return _call(_root(session, layer, owner), inputs_store.read_links, key) + + +@router.put("/layers/{layer}/items/{key}/input-links") +def put_input_links( + layer: Layer, + key: str, + body: InputLinksBody, + owner: str | None = None, + session: Session = Depends(verify_session), +) -> dict: + """본문 `{판, 줄}` — 판 다르면 409 · 없는 로직 · 입력 키 422 · 층 쓰기 권한.""" + folder = _root(session, layer, owner, write=True) + return _call(folder, inputs_store.write_links, key, body.판, body.줄) + + +@router.get("/layers/{layer}/items/{key}/pills") +def get_item_pills( + layer: Layer, + key: str, + place: Literal["면적", "재료"], + owner: str | None = None, + session: Session = Depends(verify_session), +) -> dict: + """면적산출 · 재료산출 시트 알약 — `{물량: [조합 물량], 입력: [그 시트로 이은 입력]}`.""" + return _call(_root(session, layer, owner), inputs_store.pills, key, place) + + # ── 층 길 — 옛 시스템 길과 같은 모양 ───────────────────── diff --git a/M02_MasterTemplete/M02_MasterTemplete_Store.py b/M02_MasterTemplete/M02_MasterTemplete_Store.py index f068b29a..4aa8d3f1 100644 --- a/M02_MasterTemplete/M02_MasterTemplete_Store.py +++ b/M02_MasterTemplete/M02_MasterTemplete_Store.py @@ -101,6 +101,16 @@ def _path(kind: str, name: str) -> Path: return root() / kind / f"{name}.json" +# 항목 연결(PLAN 25-1b) — 열 id 마다 `inputs/<열id>.json` · 종류 폴더(KINDS) 밖 · 읽기 · 쓰기는 +# `M02_MasterTemplete_Inputs` 몫 · 열 지우기 · 분류 더하기 복사는 여기서 같이 +INPUTS = "inputs" + + +def inputs_path(column: str) -> Path: + """그 열의 항목 연결 파일 자리 — 이름 규칙은 종류 파일과 같음.""" + return _path("structure", column).parent.parent / INPUTS / f"{column}.json" + + def _info(kind: str, path: Path) -> dict[str, Any]: raw = path.read_bytes() stamp = datetime.fromtimestamp(path.stat().st_mtime).isoformat(timespec="seconds") @@ -163,6 +173,7 @@ def write(kind: str, name: str, version: str, doc: Any) -> dict[str, Any]: for column, source in origins.items(): # 빈 파일 채우기 앞 — 원본 내용으로 만듦 for part in _KEY: _ensure(part, column, source) + _copy_inputs(source, column) backfill_missing_structures() # 지우기 앞 — 같은 저장의 새 열이 지운 번호를 안 받게 prune_removed_structures() sync_combos() # 번호가 다 선 뒤 — 조합 표 도번 · 면적번 · 재료번 칸 @@ -375,13 +386,26 @@ def copy_from(kind: str, column: str, source: str) -> dict[str, Any]: return _info(kind, _path(kind, column)) +def _copy_inputs(source: str, column: str) -> None: + """분류 더하기로 복사한 새 열 — 원본 열의 항목 연결을 판 1 로(이미 있으면 그대로).""" + have, got = inputs_path(column), inputs_path(source) + if have.is_file() or not got.is_file(): + return + have.parent.mkdir(parents=True, exist_ok=True) + atomic_write_json(have, {"판": 1, "줄": json.loads(got.read_bytes()).get("줄") or []}) + + def prune_removed_structures() -> list[str]: - """집계표에서 지운 열(또는 구조물이 아니게 된 열)의 세 파일을 같이 지움 — + """집계표에서 지운 열(또는 구조물이 아니게 된 열)의 세 파일 · 항목 연결을 같이 지움 — 번호는 비워 둠(다른 열 번호를 안 당김).""" keep = set(_column_ids()) gone: list[str] = [] codes: set[str] = set() with _LOCK: + for path in sorted((root() / INPUTS).glob("[!.]*.json")): + if path.stem not in keep: + path.unlink() + gone.append(f"{INPUTS}/{path.stem}") for kind, key, _, _ in PARTS: for path in sorted((root() / kind).glob("[!.]*.json")): if path.stem not in keep: diff --git a/M02_MasterTemplete/M02_Project_Items_Router.py b/M02_MasterTemplete/M02_Project_Items_Router.py new file mode 100644 index 00000000..4ca0e2f7 --- /dev/null +++ b/M02_MasterTemplete/M02_Project_Items_Router.py @@ -0,0 +1,126 @@ +"""프로젝트 집계 항목 복사본 길 — `/api/projects/{id}/items…` (PLAN 25-1b · 25-2b · 25-2e). + + GET …/items 목록(이름 · 원본 · 가져온날 · 마스터 바뀜) + POST …/items {layer, owner, key} 미리 가져오기(구조물에 안 이은 항목) → {폴더, 만듦, 가져온날} + 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 +from M02_MasterTemplete.M02_Item_Source import item_name + +router = APIRouter(prefix="/api/projects", tags=["M02 Project Items"]) + +Kind = Literal["도면", "면적", "재료", "자재조합", "입력연결", "입력값", "고른자재"] + + +class WriteBody(BaseModel): + 판: str | None = None + 내용: Any + + +class PrefetchBody(BaseModel): + """`{layer, owner, key}`(이름은 층 SS 줄에서) — 옛 모양 `{summary_item}` 도 받음.""" + + layer: str | None = None + owner: str | None = None + key: str | None = None + name: str | None = None + summary_item: dict[str, Any] | None = None + + +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]: + raw = body.summary_item or {"layer": body.layer, "owner": body.owner, "key": body.key} + try: # 모양 먼저(이름은 자리만) — 그다음 층 SS 줄에서 이름 + item = parse_summary_item({**raw, "name": raw.get("name") or body.name or "-"}) + except ValueError as error: + raise HTTPException(status_code=400, detail=str(error)) from error + if not (raw.get("name") or body.name): + try: + item["name"] = item_name(item) + except LookupError as error: + raise HTTPException(status_code=404, 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..1546f92c --- /dev/null +++ b/resources/tester/test_item_copy_complete.py @@ -0,0 +1,342 @@ +"""프로젝트 집계 항목 복사본 완전체 (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["원본판"], + "가져온날": manifest["가져온날"], + "마스터바뀜": False, + "원본없음": 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 + made = client.post(base, json={"layer": "system", "owner": None, "key": "SS000022"}).json() + assert made["폴더"] == "system_none_SS000022" and made["만듦"] is True + names = {i["폴더"]: i["이름"] for i in client.get(base).json()["항목"]} + assert names["system_none_SS000022"] == "돌쌓기 (찰·메마다) H=(높이마다)" # 층 SS 줄 이름 + assert client.post(base, json={"layer": "system", "key": "SS999999"}).status_code == 404 + session["company_id"] = 4 + assert client.get(base).status_code == 403 + assert client.post(f"{base}/{FOLDER}/refresh").status_code == 403 diff --git a/resources/tester/test_m02_item_inputs.py b/resources/tester/test_m02_item_inputs.py new file mode 100644 index 00000000..a72a8557 --- /dev/null +++ b/resources/tester/test_m02_item_inputs.py @@ -0,0 +1,238 @@ +"""M02 항목 선택 입력(PLAN 25-1b · 25-2a) — 입력 목록 · 항목 연결 파일 · 알약 · 권한. + +① 입력 목록은 실제 마스터(시스템 층 · 읽기만)로 — 일위대가 있는 항목 셋에서 로직 수 · 입력 수 · + 자동값이 M01 조합 미리 보기가 쓰는 것(로직 줄 `입력` · `master_auto`)과 같은지 · 로직마다 갈라 + 합치지 않는지 · 일위대가 없는 항목은 빈 목록. +② 연결 파일 쓰기는 임시 storage(회사 · 개인 층) + 견본 시스템 집계표로만 — 판 409 · 없는 키 422 · + 사라짐 표시 · 알약 · 층 권한 · 열 지우기 · 항목 복사 · 분류 더하기 복사가 파일을 같이 다룸. +""" + +from __future__ import annotations + +import json +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 M01_MasterData import M01_MasterData_Store_Combo as combo_store +from M02_MasterTemplete import M02_MasterTemplete_Inputs as inputs_store +from M02_MasterTemplete import M02_MasterTemplete_Router_Items as items_router +from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layers_router +from M02_MasterTemplete import M02_MasterTemplete_Store as store +from M02_MasterTemplete import M02_Template_Layers as layers +from resources.tester.test_m02_structure import write_sample_table + +TABLE = "table/구조물집계표" +UA = "UA000004" # 견본 항목 셋(rv · c4 · c1)이 쓰는 조합 — 담은 로직 여럿 +OTHER_UA = "UA000007" + + +def _app(session: dict[str, Any]) -> TestClient: + app = FastAPI() + app.include_router(layers_router.router) + app.include_router(items_router.router) + app.dependency_overrides[verify_session] = lambda: session + return TestClient(app) + + +def _m01_view(ua: str) -> list[tuple[str, list[str], dict]]: + """M01 조합 미리 보기가 그리는 것 — 담은 로직마다 (키, 입력 이름들, 자동값).""" + row = next(r for r in m01.read(combo_store.FILE)[0]["줄"] if r["키"] == ua) + whole = m01.loaded()[1] + out = [] + for one in row["담은로직"]: + try: + logic = whole.logic(one["로직"]) + except combo_store.cm.mf.FormulaError: + continue # 화면도 못 읽은 로직은 건너뜀 + auto = inputs_store._num(m01.auto(one["로직"])["값"]) + out.append((one["로직"], [s["이름"] for s in logic.get("입력") or []], auto)) + return out + + +# ── ① 입력 목록 — 실제 시스템 층(읽기만) ───────────────────── + + +def test_입력_목록은_M01_조합_미리_보기와_같음(): + client = _app({"user_id": 1, "company_id": 1, "role": "USER", "is_master": False}) + rows = [r for r in json.loads((store.FOLDER / store.COMBO_FILE).read_bytes())["줄"]] + with_ua = [r for r in rows if r.get("일위대가")] + picked, seen = [], set() + for r in with_ua: # 서로 다른 조합 셋 + if r["일위대가"] not in seen: + picked.append(r) + seen.add(r["일위대가"]) + if len(picked) == 3: + break + assert len(picked) == 3 + for r in picked: + res = client.get(f"/api/m02/layers/system/items/{r['키']}/inputs") + assert res.status_code == 200, res.text # 일반 사용자 읽기 + got = res.json() + want = _m01_view(r["일위대가"]) + assert got["일위대가"] == r["일위대가"] and got["이름"] + assert [lg["로직"] for lg in got["로직"]] == [k for k, _, _ in want] + for logic, (_, names, auto) in zip(got["로직"], want): + assert [i["이름"] for i in logic["입력"]] == names # 로직마다 · 합치지 않음 + assert {i["이름"]: i["자동값"] for i in logic["입력"]} == auto + assert all({"이름", "설명", "단위", "자동값"} <= set(i) for i in logic["입력"]) + # 이름이 같은 입력이 두 로직에 있어도 두 번 나옴 + names = [ + i["이름"] + for r in with_ua + for lg in client.get(f"/api/m02/layers/system/items/{r['키']}/inputs").json()["로직"] + for i in lg["입력"] + ] + assert len(names) > len(set(names)) + no_ua = next(r for r in rows if not r.get("일위대가")) + empty = client.get(f"/api/m02/layers/system/items/{no_ua['키']}/inputs").json() + assert empty == {"일위대가": None, "이름": None, "로직": [], "빠진로직": []} + assert client.get("/api/m02/layers/system/items/SS999999/inputs").status_code == 404 + + +# ── ② 연결 파일 — 임시 storage ───────────────────────── + + +@pytest.fixture() +def world(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: + system = tmp_path / "system" + system.mkdir() + write_sample_table(system) + monkeypatch.setattr(store, "FOLDER", system) + monkeypatch.setattr(layers, "SYSTEM_ROOT", system) + store.renumber() + storage = tmp_path / "storage" + storage.mkdir() + monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage)) + session: dict[str, Any] = {"user_id": 42, "company_id": 7, "role": "USER", "is_master": False} + return {"client": _app(session), "session": session, "storage": storage} + + +def _as(world: dict, role: str = "USER", user: int = 42) -> TestClient: + world["session"].update(role=role, user_id=user) + return world["client"] + + +def _put_table(client: TestClient, layer: str, edit) -> None: + url = f"/api/m02/layers/{layer}/templates/{TABLE}" + got = client.get(url).json() + edit(got["문서"]["열"]) + res = client.put(url, json={"판": got["판"], "문서": got["문서"]}) + assert res.status_code == 200, res.text + + +def _col(cid: str, ua: str | None = UA) -> dict[str, Any]: + return { + "id": cid, + "머리": ["sub4 돌쌓기", cid, None, None], + "단위": "㎡", + "꼴": "수", + "일위대가": ua, + } + + +def _key(client: TestClient, layer: str, cid: str) -> str: + items = client.get(f"/api/m02/items?layer={layer}").json() + return next(r["key"] for r in items if r["열"] == cid) + + +def test_연결_저장_다시_읽기_판_422_사라짐_알약(world): + client = _as(world) + _put_table(client, "personal", lambda cols: cols.append(_col("sub4_a"))) + key = _key(client, "personal", "sub4_a") + base = f"/api/m02/layers/personal/items/{key}" + listing = client.get(f"{base}/inputs").json() + first, second = listing["로직"][0], listing["로직"][1] + assert client.get(f"{base}/input-links").json() == {"판": None, "줄": []} + rows = [ + {"로직": first["로직"], "입력": first["입력"][0]["이름"], "자리": "면적"}, + {"로직": second["로직"], "입력": second["입력"][0]["이름"], "자리": "재료"}, + {"로직": first["로직"], "입력": first["입력"][1]["이름"], "자리": "선택"}, + ] + saved = client.put(f"{base}/input-links", json={"판": None, "줄": rows}) + assert saved.status_code == 200, saved.text + assert saved.json()["판"] == 1 + again = client.get(f"{base}/input-links").json() + assert again == saved.json() + assert again["줄"][0]["알약"] == f"입력|{first['로직']}|{first['입력'][0]['이름']}" + assert "알약" not in again["줄"][2] # 선택은 시트 알약이 아님 + file = world["storage"] / "7" / "42" / "templates" / "inputs" / "sub4_a.json" + assert json.loads(file.read_text("utf-8")) == saved.json() + # 판 다르면 409 · 없는 키 · 모르는 자리 · 두 번은 422 + assert client.put(f"{base}/input-links", json={"판": None, "줄": rows}).status_code == 409 + for bad in ( + [{"로직": "GF999999", "입력": "높이", "자리": "면적"}], + [{"로직": first["로직"], "입력": "없는입력", "자리": "면적"}], + [{"로직": first["로직"], "입력": first["입력"][0]["이름"], "자리": "옆"}], + [rows[0], rows[0]], + ): + res = client.put(f"{base}/input-links", json={"판": 1, "줄": bad}) + assert res.status_code == 422, res.text + # 알약 — 면적 = 조합 물량 + 면적으로 이은 입력 · 재료 = 재료로 이은 입력만 + area = client.get(f"{base}/pills?place=면적").json() + assert area["물량"] and [p["키"] for p in area["입력"]] == [again["줄"][0]["알약"]] + mat = client.get(f"{base}/pills?place=재료").json() + assert mat["물량"] == [] and [p["입력"] for p in mat["입력"]] == [second["입력"][0]["이름"]] + assert " · " in mat["입력"][0]["이름"] # 입력 · 로직 이름(합치지 않음) + # 일위대가가 바뀌면 옛 로직 연결은 읽을 때 사라짐 · 알약에서 빠짐 + _put_table(client, "personal", lambda cols: cols[-1].update(일위대가=OTHER_UA)) + gone = client.get(f"{base}/input-links").json()["줄"] + assert all(r.get("사라짐") for r in gone) + assert client.get(f"{base}/pills?place=면적").json()["입력"] == [] + + +def test_권한_일반_사용자_읽기_쓰기(world): + admin = _as(world, "SYSTEM_ADMIN") + _put_table(admin, "system", lambda cols: cols.append(_col("sub4_sys"))) + key = _key(admin, "system", "sub4_sys") + base = f"/api/m02/layers/system/items/{key}" + user = _as(world) + assert user.get(f"{base}/inputs").status_code == 200 + assert user.get(f"{base}/input-links").status_code == 200 + assert user.put(f"{base}/input-links", json={"판": None, "줄": []}).status_code == 403 + assert _as(world, "SYSTEM_ADMIN").put(f"{base}/input-links", json={"줄": []}).status_code == 200 + # 회사 층 — 같은 회사 일반 사용자 읽기 · 쓰기 403 · 회사 관리자 쓰기 + _put_table(_as(world, "ADMIN"), "company", lambda cols: cols.append(_col("sub4_c"))) + ckey = _key(_as(world), "company", "sub4_c") + cbase = f"/api/m02/layers/company/items/{ckey}" + assert _as(world).get(f"{cbase}/inputs").status_code == 200 + assert _as(world).put(f"{cbase}/input-links", json={"줄": []}).status_code == 403 + assert _as(world, "ADMIN").put(f"{cbase}/input-links", json={"줄": []}).status_code == 200 + # 남의 개인 층은 못 읽음 + assert _as(world).get(f"{cbase}/inputs?owner=9").status_code == 403 + + +def test_열_지우기_항목_복사_분류_복사가_연결을_같이(world): + admin = _as(world, "ADMIN") + _put_table(admin, "company", lambda cols: cols.append(_col("sub4_c"))) + key = _key(admin, "company", "sub4_c") + base = f"/api/m02/layers/company/items/{key}" + logic = admin.get(f"{base}/inputs").json()["로직"][0] + rows = [{"로직": logic["로직"], "입력": logic["입력"][0]["이름"], "자리": "면적"}] + assert admin.put(f"{base}/input-links", json={"판": None, "줄": rows}).status_code == 200 + # 항목 개인 복사 — 새 열에 연결이 판 1 로 + user = _as(world) + made = user.post( + "/api/m02/layers/personal/items/copy", json={"from_layer": "company", "key": key} + ).json() + mine = user.get(f"/api/m02/layers/personal/items/{made['key']}/input-links").json() + assert mine["판"] == 1 and [r["입력"] for r in mine["줄"]] == [rows[0]["입력"]] + # 분류 더하기(본뜬열) — 새 열이 원본 열 연결을 받음 + admin = _as(world, "ADMIN") + twin = {**_col("sub4_c2", None), "본뜬열": "sub4_c"} + _put_table(admin, "company", lambda cols: cols.append(twin)) + folder = world["storage"] / "7" / "templates" + with store.at(folder): + assert inputs_store.read_file("sub4_c2")["줄"] == inputs_store.read_file("sub4_c")["줄"] + # 열 지우기 — 연결 파일도 같이 지움 + _put_table( + admin, "company", lambda cols: cols.remove(next(c for c in cols if c["id"] == "sub4_c")) + ) + assert not (folder / "inputs" / "sub4_c.json").is_file() + assert (folder / "inputs" / "sub4_c2.json").is_file()