"""M02 일위대가 조합 서고 — 단가승격 · 사전조합 · 참조(PLAN 50-3 · 조사 갈래 A). 자리 = 층 양식 길 새 종류 둘(`M02_Template_Layers.LIBRARY_KINDS`) — 파일 이름 = 키: promote `{층}/promote/SP-######.json` 단가승격 `{종류:"단가승격", 키, 이름, 단위, 배, 본뜬, 담은로직:[{로직, 입력값}]}` — 줄 하나 · 입력값은 승격 때 고정 · 값 = 로직 기준 ÷ 배(1 단위 값 · `배` 는 저장 때) precombo `{층}/precombo/SC-######.json` 사전조합 `{종류:"사전조합", 키, 이름, 단위, 본뜬, 담은로직:[{참조, 수량:{손}, 메모} | {로직, 입력값, 메모, 수량:{손}}]}` — 승격 · 사전조합 · 단가산출 로직(51-11) · 수량은 손(기본 1) · 값 = Σ 안 줄 1 단위 값 × 수량(로직 줄 = 기준 ÷ 배) 참조 = `{층, 주인, 키}`(마스터 주인 null · 회사 `C` · 개인 `C/U`) — 종류는 키 머리(SP · SC) · 위 층만 가리킴(개인 → 그 개인 · 그 회사 · 마스터 / 회사 → 그 회사 · 마스터 / 마스터 → 마스터) · 같은 참조 두 번 · 순환(422)은 저장 때 막음 · 계산 때도 깊이 상한(`DEPTH`)에서 멈추고 까닭. 지우기 = 쓰는 곳(모든 층 일위대가 · 사전조합)이 있으면 409 · 층 사이 복사 = 대상 층이 못 읽는 참조는 그 문서도 같이 복사(새 키 · `본뜬`) · 프로젝트 = 참조사본으로 굳힘(`frozen`). """ from __future__ import annotations import json import re from collections.abc import Callable from decimal import Decimal from pathlib import Path from typing import Any from M02_MasterTemplete import M02_MasterTemplete_Store as store from M02_MasterTemplete import M02_Template_Layers as layers PREFIX = {"promote": "SP", "precombo": "SC"} TITLE = {"promote": "단가승격", "precombo": "사전조합"} DEPTH = 8 # 참조 깊이 상한(저장 · 계산 둘 다) REF_HEAD = "참조|" _KEY = re.compile(r"^(SP|SC)-(\d{6})$") _OWNER = {"company": re.compile(r"^\d+$"), "personal": re.compile(r"^\d+/\d+$")} Ref = dict[str, Any] Here = tuple[str, str | None] def _bad(text: str, status: int = 400) -> store.StoreError: return store.StoreError(status, text) def kind_of(key: str) -> str: return "promote" if str(key).startswith("SP") else "precombo" def ref_id(ref: Ref) -> str: """참조 한 줄의 이름표 — 알약 키 · 줄 가름(짝 `Labor_Calc.ts` `lineId`).""" return f"{REF_HEAD}{ref.get('층')}|{ref.get('주인') or ''}|{ref.get('키')}" # ── 자리 ────────────────────────────────────────────────────────────── def where(folder: str | Path) -> Here: """층 폴더 → (층, 주인) — 저장소 밖 · 프로젝트는 ("project", None).""" path = Path(folder).resolve() if path in (Path(store.FOLDER).resolve(), layers.system_dir().resolve()): return "system", None try: parts = path.relative_to(layers.storage_root()).parts except ValueError: return "project", None if len(parts) == 2 and parts[1] == layers.TEMPLATES_DIRNAME: return "company", parts[0] if len(parts) == 3 and parts[2] == layers.TEMPLATES_DIRNAME: return "personal", f"{parts[0]}/{parts[1]}" return "project", None def here() -> Here: return where(store.root()) def folder_of(layer: str, owner: str | None) -> Path: if layer == "system": return Path(store.FOLDER) if layer == "company": return layers.company_dir(owner or "") company, user = str(owner).split("/", 1) return layers.personal_dir(company, user) def ref_path(ref: Ref) -> Path: return folder_of(ref["층"], ref.get("주인")) / kind_of(ref["키"]) / f"{ref['키']}.json" def read_ref(ref: Ref) -> dict[str, Any] | None: try: got = json.loads(ref_path(ref).read_bytes()) except (OSError, ValueError): return None return got if isinstance(got, dict) else None def readable(at: Here, ref: Ref) -> bool: """`at` 층이 `ref` 를 가리켜도 되나 — 위 층만(프로젝트는 아무것도 · 굳힌 사본만 씀).""" layer, owner = at if layer == "project": return False if ref["층"] == "system": return True if ref["층"] == "company": return layer in ("company", "personal") and str(owner).split("/")[0] == ref["주인"] return layer == "personal" and owner == ref["주인"] def check_ref(raw: Any, at: str, here_: Here) -> Ref: """참조 꼴 · 위 층만 · 있는 문서 — 아니면 400.""" if not isinstance(raw, dict): raise _bad(f"{at} · 참조는 {{층, 주인, 키}}") layer, owner, key = raw.get("층"), raw.get("주인"), str(raw.get("키") or "") if layer == "system": owner = None elif layer not in _OWNER or not _OWNER[layer].match(str(owner or "")): raise _bad(f"{at} · 참조 층 · 주인 꼴이 틀림") if not _KEY.match(key): raise _bad(f"{at} · 참조 키는 SP-###### · SC-######") ref = {"층": layer, "주인": None if owner is None else str(owner), "키": key} if not readable(here_, ref): raise _bad(f"{at} · 위 층만 가리킴 — 「{key}」({layer} {owner or ''})") if read_ref(ref) is None: raise _bad(f"{at} · 없는 참조 「{key}」") return ref # ── 서고 문서 정리 ───────────────────────────────────────────────────── def check_key(kind: str, name: str) -> None: found = _KEY.match(name) if not found or found.group(1) != PREFIX[kind]: raise _bad(f"{TITLE[kind]} 키는 {PREFIX[kind]}-######") def clean_doc(kind: str, doc: Any) -> dict[str, Any]: """서고 문서를 서버가 받는 꼴로 — `doc["키"]` = 파일 이름(부르는 쪽이 넣음).""" from M01_MasterData import M01_MasterData_Store_Combo as combo_store from M02_MasterTemplete import M02_Labor_Doc as labor if not isinstance(doc, dict): raise _bad("서고 문서가 묶음 아님") key = str(doc.get("키") or "") check_key(kind, key) name, unit = str(doc.get("이름") or "").strip(), str(doc.get("단위") or "").strip() out: dict[str, Any] = {"종류": TITLE[kind], "키": key} if kind == "promote": lines = doc.get("담은로직") or [] if len(lines) != 1 or not isinstance(lines[0], dict) or not lines[0].get("로직"): raise _bad("단가승격은 단가산출 로직 한 줄") line = labor.clean({"담은로직": lines})["담은로직"][0] pill = (combo_store.combo_pills({"담은로직": [line]})["물량"] or [{}])[0] out.update( 이름=name or pill.get("이름") or line["로직"], 단위=unit or pill.get("단위") or None, 배=pill.get("배") or 1, # ponytail: 저장 때 적음 · 마스터 결과단위가 바뀌면 다시 저장 담은로직=[{"로직": line["로직"], "입력값": line["입력값"]}], ) else: if not name: raise _bad("사전조합 이름이 빔") out.update(이름=name, 단위=unit or None, 담은로직=_combo_lines(doc, key)) out["본뜬"] = doc["본뜬"] if isinstance(doc.get("본뜬"), dict) else None return out def _combo_lines(doc: dict, key: str) -> list[dict[str, Any]]: """참조 줄 `{참조, 수량, 메모}` · 로직 줄 `{로직, 입력값, 메모, 수량}`(51-11) — 검사는 일위대가와 같은 길(`labor.clean` · 같은 로직 · 같은 참조 두 번 · 입력값) · 수량 기본 {손: 1}.""" from M02_MasterTemplete import M02_Labor_Doc as labor lines = [] for one in labor.clean({"담은로직": doc.get("담은로직") or []})["담은로직"]: head = {"참조": one["참조"]} if "참조" in one else {"로직": one["로직"]} if "로직" in one: head["입력값"] = one["입력값"] lines.append({**head, "수량": one["수량"] or {"손": 1}, "메모": one["메모"]}) at_ = here() me = {"층": at_[0], "주인": at_[1], "키": key} for one in lines: if "참조" not in one: continue if ref_id(one["참조"]) == ref_id(me): raise _bad(f"순환 — 자기 자신을 담음 「{key}」", 422) _no_loop(one["참조"], [me, one["참조"]]) return lines def _no_loop(ref: Ref, trail: list[Ref]) -> None: """저장 때 순환 · 깊이 — 아래로 따라가 처음 문서를 다시 만나면 422 「A → B → A」.""" if len(trail) > DEPTH: raise _bad(f"참조가 {DEPTH}단을 넘음 — " + " → ".join(r["키"] for r in trail), 422) for one in (read_ref(ref) or {}).get("담은로직") or []: child = one.get("참조") if isinstance(one, dict) else None if not isinstance(child, dict) or not child.get("키"): continue if ref_id(child) == ref_id(trail[0]): raise _bad("순환 — " + " → ".join(r["키"] for r in [*trail, child]), 422) _no_loop(child, [*trail, child]) def next_key(kind: str) -> str: """지금 층 그 종류의 다음 키(가장 큰 번호 + 1).""" folder = store.root() / kind have = [int(m.group(2)) for p in folder.glob("*.json") if (m := _KEY.match(p.stem))] return f"{PREFIX[kind]}-{max(have, default=0) + 1:06d}" def create(kind: str, doc: dict[str, Any]) -> dict[str, Any]: """지금 층에 새 키로 씀 → `store.write` 정보(+ 키) — 그새 남이 같은 키를 잡으면 다음 키.""" for _ in range(5): key = next_key(kind) try: return store.write(kind, key, "", {**doc, "키": key}) except store.StoreError as e: if e.status != 409 or not isinstance(e.detail, dict): raise raise _bad("새 키를 못 잡음 — 다시 누름", 409) def summary(folder: Path, kind: str, key: str, expand: Expand | None = None) -> dict[str, Any]: """층 목록 한 줄에 붙일 것 — `{제목, 단위}` · `expand` 면 1 단위 기준 계(`기준금액` · 51-4).""" try: doc = json.loads((Path(folder) / kind / f"{key}.json").read_bytes()) except (OSError, ValueError): return {} out = {"제목": doc.get("이름"), "단위": doc.get("단위")} if expand is not None: layer, owner = where(folder) base = expand({"층": layer, "주인": owner, "키": key}).get("기준") or {} out["기준금액"] = float(base["계"]) if "계" in base else None return out def summaries(folder: Path, rows: list[dict[str, Any]]) -> None: """층 목록의 서고 줄마다 `summary`(값 포함) — 한 번 셈 캐시를 함께 씀.""" from M01_MasterData import M01_MasterData_Store as m01 expand = Expand(m01.loaded()[1]) for row in rows: if row["종류"] in layers.LIBRARY_KINDS: row.update(summary(folder, row["종류"], row["이름"], expand)) # ── 쓰는 곳 · 지우기 ────────────────────────────────────────────────── def users(ref: Ref) -> list[str]: """`ref` 를 담은 일위대가 · 사전조합 — 「어디 이름」 목록(모든 층 훑음 · 프로젝트 뺌).""" rid, out = ref_id(ref), [] for kind, what in (("labor", "일위대가"), ("precombo", "사전조합")): for at, path in layers.layer_files(kind): try: doc = json.loads(path.read_bytes()) except (OSError, ValueError): continue lines = doc.get("담은로직") if isinstance(doc, dict) else None if any( isinstance(one, dict) and isinstance(one.get("참조"), dict) and ref_id(one["참조"]) == rid for one in lines or [] ): name = doc.get("일위대가번") or doc.get("이름") or path.stem out.append(f"{what} {at} 「{name}」") return out def refuse_used(folder: Path, kind: str, key: str) -> None: """쓰는 곳이 있으면 409 + 이름(서고 종류만).""" if kind not in PREFIX: return layer, owner = where(folder) found = users({"층": layer, "주인": owner, "키": key}) if found: raise _bad(f"쓰는 곳이 있어 못 지움 「{key}」 — " + " · ".join(found), 409) # ── 층 사이 복사 ────────────────────────────────────────────────────── def carry(doc: dict[str, Any], source: Here) -> dict[str, Any]: """복사(원본 층 `source` → 지금 층) — 지금 층이 못 읽는 참조는 그 문서를 지금 층에 새 키로 같이 복사(`본뜬` = 원래 참조) · 참조를 새 키로 바꿈. 원본 층 읽기 권한은 부르는 쪽 몫.""" target, done = here(), {} def move(ref: Ref, depth: int) -> Ref: if readable(target, ref): return ref if ref_id(ref) in done: return done[ref_id(ref)] if depth > DEPTH or not readable(source, ref): raise _bad(f"복사할 수 없는 참조 「{ref.get('키')}」") got = read_ref(ref) if got is None: raise _bad(f"없는 참조 「{ref['키']}」") body = {**got, "담은로직": _moved(got.get("담은로직"), move, depth + 1), "본뜬": ref} made = create(kind_of(ref["키"]), body) done[ref_id(ref)] = {"층": target[0], "주인": target[1], "키": made["이름"]} return done[ref_id(ref)] return {**doc, "담은로직": _moved(doc.get("담은로직"), move, 1)} def _moved(lines: Any, move: Callable[[Ref, int], Ref], depth: int) -> list[Any]: return [ {**one, "참조": move(one["참조"], depth)} if isinstance(one, dict) and isinstance(one.get("참조"), dict) else one for one in lines or [] ] # ── 펼침(계산) ──────────────────────────────────────────────────────── class Expand: """참조 → `{이름, 단위, 기준(1 단위), 까닭, 안?}` — 한 번 셈 안에서 키별 캐시 · 깊이 상한 · 순환은 까닭(통째로 멈추지 않음). `read` = 참조 → 문서(프로젝트는 굳힌 사본).""" def __init__(self, whole: Any, read: Callable[[Ref], dict | None] = read_ref) -> None: self.whole, self.read, self.memo = whole, read, {} def __call__(self, ref: Ref, trail: tuple[str, ...] = ()) -> dict[str, Any]: rid = ref_id(ref) if rid in trail: keys = [t.rsplit("|", 1)[-1] for t in (*trail, rid)] return self._stop(ref, "순환 — " + " → ".join(keys)) if len(trail) >= DEPTH: return self._stop(ref, f"참조가 {DEPTH}단을 넘음") if rid not in self.memo: doc = self.read(ref) if not isinstance(doc, dict): return self._stop(ref, f"없는 참조 「{ref['키']}」") if kind_of(ref["키"]) == "promote": self.memo[rid] = self._promote(doc) else: self.memo[rid] = self._combo(doc, (*trail, rid)) return self.memo[rid] @staticmethod def _stop(ref: Ref, why: str) -> dict[str, Any]: return {"이름": ref["키"], "단위": None, "기준": None, "까닭": why} def _promote(self, doc: dict) -> dict[str, Any]: from M02_MasterTemplete import M02_Labor_Doc as labor line = (doc.get("담은로직") or [{}])[0] got = labor.run_line(self.whole, str(line.get("로직") or ""), line.get("입력값")) times = labor._number(doc.get("배")) or Decimal(1) base = got.get("기준") why = got.get("까닭") or (None if base else "돈 아닌 로직") return { "이름": doc.get("이름"), "단위": doc.get("단위"), "로직": line.get("로직"), "기준": {k: Decimal(str(v)) / times for k, v in base.items()} if base else None, "까닭": why, # 원 로직 줄(줄 누름 하위 행 · 51-2) — 1 단위로 나눔(하위 합 = 기준) "줄": [ { **x, "수량": Decimal(str(x["수량"])) / times, "금액": Decimal(str(x["금액"])) / times, "비목": {k: Decimal(str(v)) / times for k, v in (x.get("비목") or {}).items()}, } for x in got.get("줄") or [] ], } def _logic(self, one: dict, pill: dict | None) -> dict[str, Any]: """사전조합 안 로직 줄 — 일위대가 로직 줄과 같은 `run_line` · 배는 물량 알약 것.""" from M02_MasterTemplete import M02_Labor_Doc as labor pill = pill or {} got = labor.run_line(self.whole, one["로직"], one.get("입력값")) times = labor._number(pill.get("배")) or Decimal(1) base = got.get("기준") return { "이름": pill.get("이름") or one["로직"], "단위": pill.get("단위"), "기준": {k: Decimal(str(v)) / times for k, v in base.items()} if base else None, "까닭": got.get("까닭") or (None if base else "돈 아닌 로직"), } def _combo(self, doc: dict, trail: tuple[str, ...]) -> dict[str, Any]: from M01_MasterData import M01_MasterData_Store_Combo as combo_store from M02_MasterTemplete import M02_Labor_Doc as labor lines = [one for one in doc.get("담은로직") or [] if isinstance(one, dict)] by_logic = { code: pill for pill in combo_store.combo_pills({"담은로직": lines})["물량"] for code in pill["로직"] } inner, rows, why = [], [], None for one in lines: ref = one.get("참조") if isinstance(ref, dict): head, sub = {"참조": ref}, self(ref, trail) elif one.get("로직"): # 로직 줄(51-11) = 1 단위 값(기준 ÷ 배 · 승격과 같음) head, sub = {"로직": one["로직"]}, self._logic(one, by_logic.get(one["로직"])) else: continue qty = labor._number((one.get("수량") or {}).get("손")) qty = Decimal(1) if qty is None else qty keep = ("이름", "단위", "기준", "까닭") inner.append({**head, "수량": qty, **{k: sub.get(k) for k in keep}}) if sub.get("기준") is None: why = why or f"「{sub.get('이름')}」 {sub.get('까닭')}" continue rows.append({k: v * qty for k, v in sub["기준"].items()}) return { "이름": doc.get("이름"), "단위": doc.get("단위"), "기준": None if why else labor._sum(rows), "까닭": why, "안": inner, } def pill(ref: Ref, read: Callable[[Ref], dict | None] = read_ref) -> dict[str, Any]: """참조 줄 면적산출 알약 — 키 = `ref_id` · 이름 · 단위 = 가리킨 문서 것.""" doc = read(ref) or {} return { "키": ref_id(ref), "이름": doc.get("이름") or ref["키"], "단위": doc.get("단위") or "", "로직": [], "참조": ref, } # ── 프로젝트 굳힘 ───────────────────────────────────────────────────── def frozen(doc: dict[str, Any]) -> dict[str, dict[str, Any]]: """일위대가 문서가 닿는 참조 문서 전부 `{ref_id: 문서}` — 가져올 때 값으로 굳힘 (원본이 바뀌어도 프로젝트는 그대로).""" out: dict[str, dict[str, Any]] = {} def walk(lines: Any, depth: int) -> None: for one in lines or []: ref = one.get("참조") if isinstance(one, dict) else None if not isinstance(ref, dict) or ref_id(ref) in out or depth > DEPTH: continue got = read_ref(ref) if got is not None: out[ref_id(ref)] = got walk(got.get("담은로직"), depth + 1) walk(doc.get("담은로직"), 1) return out def frozen_logics(copies: dict[str, Any]) -> list[str]: """굳힌 단가승격 · 사전조합이 담은 로직 키 — 프로젝트 단가 묶음에 넣음.""" return [ str(line["로직"]) for doc in copies.values() for line in doc.get("담은로직") or [] if isinstance(line, dict) and line.get("로직") ]