- 모달 = 담은 로직 이름별 입력(library-inputs) · 거리 칸 거리에서 · 손 값(own) 이김
- 사본 폴더 = 종류@층@주인@키~값해시8 · 같은 값 = 같은 사본 · 다른 거리 = 따로 · 옛 이름 읽힘
- 셈 때 서버가 거리로 사본 다시 고름(resolve_hauls · fix_keys) · [저장] = 바뀐 연결 저장 값에도 · prune
- 연결 {tab dg, key, lib, given, own} 저장 꼴 · 원본 읽기 전 모양 검사
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JywHxQuWPk6UYXTebyrBoU
362 lines
16 KiB
Python
362 lines
16 KiB
Python
"""B09 설계내역서 연결 — 서고 구조물조합 · 일위대가 사전조합 → 프로젝트 사본(PLAN 204-2).
|
|
|
|
자리 = 프로젝트 `B09_Estimation/v1/library/{폴더}/` · 폴더 = `{종류}@{층}@{주인}@{키}`
|
|
(주인 `/` → `-` · 마스터 `-`) — 같은 것을 다시 고르면 새로 안 들여오고 있는 사본(같은 호표):
|
|
combo 구조물조합 = B08 구조물 사본과 한 꼴(`M02_Item_Snapshot.build_snapshot` — 집계 ·
|
|
도면 · 면적 · 일위대가 · 단가 · 입력값 · 고른자재 첫 값)
|
|
precombo 사전조합 = `일위대가.json` {줄: {담은로직: [{참조, 수량: {손: 1}}]}, 참조사본} ·
|
|
`단가.json`(`M02_Item_Source.price_bundle` — 사본 단가 묶음 한 벌)
|
|
사본 = 원본이 바뀌거나 지워져도 그대로(굳힘) ·
|
|
대가 블록 = `Labor_Source.daega_inputs`(`libs` {폴더: 블록 id}) ·
|
|
단가 수요 = `Linked.linked_resource_keys` · 줄 연결 = {tab: dg, key: 블록 id, lib: 폴더} ·
|
|
[저장] 때 줄이 안 쓰는 사본은 지움(`prune`) · 층 읽기 권한 = 부르는 쪽(`M02_Layer_Access.grant`).
|
|
|
|
운반 값 사본(263-2) — 운반공 줄 연결
|
|
`{tab: dg, key, lib, given{줄 id: {입력: 값}}, own[줄 id|입력]}` ·
|
|
담은 로직 줄(id = `{참조 id}#{차례}` · 맨 위 줄 = `#{차례}`)의 거리 칸
|
|
(운반거리 · Cm · 사이클시간 · t2) =
|
|
거리이정표 거리로 셈(짝 `B08_Quantity_Haul_Cycle`) · 그 밖 = 손 값 → 값 묶음 `값` ·
|
|
폴더 = `{종류}@{층}@{주인}@{키}~{값 해시 8}`(값 없으면 해시 없음 · 옛 이름 그대로 읽힘) ·
|
|
같은 항목 같은 값 = 같은 사본 ·
|
|
manifest `원값` = 덮기 전 값(다른 값 사본을 이 사본에서 다시 만듦) ·
|
|
단가 묶음 = 로직 · 자재 키로만 모음 → 값이 바뀌어도 그대로 ·
|
|
셈 때(`resolve_hauls`) 서버가 거리로 값을 다시 셈 → 다른 해시면 새 사본 ·
|
|
줄 연결 lib · key 바꿔 셈.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from B08_Quantity.B08_Quantity_Haul_Cycle import FILL, cycle_inputs, cycle_time
|
|
from B09_Estimation.B09_Estimation_Store import folder as store_folder
|
|
from B09_Estimation.B09_Estimation_Store import ref_rows
|
|
from common_util.common_util_json import atomic_write_json
|
|
from M01_MasterData import M01_MasterData_Store as m01
|
|
from M02_MasterTemplete import M02_Labor_Ref as refs
|
|
from M02_MasterTemplete.M02_Item_Snapshot import (
|
|
MANIFEST_NAME,
|
|
VALUES_FILE,
|
|
build_snapshot,
|
|
now_text,
|
|
origin_of,
|
|
seed_values,
|
|
)
|
|
from M02_MasterTemplete.M02_Item_Source import (
|
|
LABOR_FILE,
|
|
PRICE_FILE,
|
|
item_name,
|
|
labor_doc_rows,
|
|
price_bundle,
|
|
read_json,
|
|
)
|
|
|
|
LIBRARY = "library"
|
|
KINDS = ("combo", "precombo")
|
|
#: 폴더 이름 — 종류 · 층 · 주인 · 키(영문 · 숫자 · `-` `_`) · 값 해시(263-2 · 없을 수 있음)
|
|
NAME = re.compile(
|
|
r"^(combo|precombo)@(system|company|personal)@[0-9-]+@[A-Za-z0-9_-]{1,40}(~[0-9a-f]{8})?$"
|
|
)
|
|
|
|
|
|
def library_dir(project_root: str | Path) -> Path:
|
|
return store_folder(project_root) / LIBRARY
|
|
|
|
|
|
def tag(values: dict[str, Any]) -> str:
|
|
"""값 묶음 해시 8 — 같은 값 = 같은 표지."""
|
|
text = json.dumps(values, sort_keys=True, ensure_ascii=False)
|
|
return hashlib.sha1(text.encode("utf-8")).hexdigest()[:8]
|
|
|
|
|
|
def folder_name(
|
|
kind: str, layer: str, owner: str | None, key: str, values: dict[str, Any] | None = None
|
|
) -> str:
|
|
"""고른 서고 것 → 사본 폴더 이름(값 있으면 `~해시`) — 규칙 밖이면 ValueError."""
|
|
name = f"{kind}@{layer}@{(owner or '-').replace('/', '-')}@{key}"
|
|
name += f"~{tag(values)}" if values else ""
|
|
if not NAME.match(name):
|
|
raise ValueError(f"서고 항목 모양이 틀림 「{name}」")
|
|
return name
|
|
|
|
|
|
def _precombo_files(layer: str, owner: str | None, key: str) -> dict[str, Any]:
|
|
ref = {"층": layer, "주인": owner, "키": key}
|
|
doc = refs.read_ref(ref)
|
|
if doc is None or doc.get("종류") != refs.TITLE["precombo"]:
|
|
raise LookupError(f"사전조합 {key} 이 {layer} 층에 없음")
|
|
line = {"참조": ref, "수량": {"손": 1}}
|
|
labor = labor_doc_rows({"이름": doc.get("이름"), "단위": doc.get("단위"), "담은로직": [line]})
|
|
return {
|
|
LABOR_FILE: labor,
|
|
PRICE_FILE: price_bundle(labor, refs.folder_of(layer, owner)),
|
|
"_원본": {"참조": ref, "name": doc.get("이름") or key, "단위": doc.get("단위") or ""},
|
|
}
|
|
|
|
|
|
def _build(kind: str, layer: str, owner: str | None, key: str) -> tuple[dict, dict]:
|
|
"""서고 원본 → (사본 파일들, 원본 표지)."""
|
|
if kind not in KINDS:
|
|
raise ValueError(f"서고 종류는 {', '.join(KINDS)} 중 하나")
|
|
folder_name(kind, layer, owner, key) # 모양 먼저(원본 읽기 전)
|
|
if kind == "combo":
|
|
item = {"layer": layer, "owner": owner, "key": key}
|
|
item["name"] = item_name(item)
|
|
return build_snapshot(item), origin_of(item)
|
|
files = _precombo_files(layer, owner, key)
|
|
return files, files.pop("_원본")
|
|
|
|
|
|
# ── 운반 값 사본(263-2) ────────────────────────────────────────────────
|
|
def _logic(code: str) -> dict[str, Any]:
|
|
try:
|
|
return dict(m01.loaded()[1].logic(code) or {})
|
|
except Exception: # noqa: BLE001 — 없는 로직 = 입력 없음
|
|
return {}
|
|
|
|
|
|
def _line_at(labor: Any, lid: str) -> dict[str, Any] | None:
|
|
rid, _, at = lid.rpartition("#")
|
|
doc = (labor or {}).get("줄") if not rid else ((labor or {}).get("참조사본") or {}).get(rid)
|
|
lines = doc.get("담은로직") if isinstance(doc, dict) else None
|
|
if not at.isdigit() or not isinstance(lines, list) or int(at) >= len(lines):
|
|
return None
|
|
line = lines[int(at)]
|
|
return line if isinstance(line, dict) and line.get("로직") else None
|
|
|
|
|
|
def lines_of(files: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]:
|
|
"""사본 안 로직 줄 `(줄 id, 줄)` — 맨 위 줄 → 굳힌 참조 문서 줄."""
|
|
labor = files.get(LABOR_FILE) or {}
|
|
docs = [("", labor.get("줄")), *((labor.get("참조사본") or {}).items())]
|
|
return [
|
|
(f"{rid}#{i}", one)
|
|
for rid, doc in docs
|
|
for i, one in enumerate((doc or {}).get("담은로직") or [])
|
|
if isinstance(one, dict) and one.get("로직")
|
|
]
|
|
|
|
|
|
def haul_values(
|
|
lines: list[tuple[str, dict[str, Any]]], given: Any, own: Any, dist: Any
|
|
) -> dict[str, dict[str, Any]]:
|
|
"""값 묶음 `{줄 id: {입력: 값}}` — 거리 칸 = 거리로 셈(`own` 이면 손 값) · 그 밖 = 손 값."""
|
|
given = given if isinstance(given, dict) else {}
|
|
own = set(own if isinstance(own, list) else [])
|
|
got = cycle_time(cycle_inputs(None, dist)) if isinstance(dist, dict) else None
|
|
out: dict[str, dict[str, Any]] = {}
|
|
for lid, line in lines:
|
|
hand = given.get(lid) if isinstance(given.get(lid), dict) else {}
|
|
vals = {}
|
|
for name in (str(i.get("이름")) for i in _logic(str(line["로직"])).get("입력") or []):
|
|
if got and name in FILL and f"{lid}|{name}" not in own:
|
|
vals[name] = got[FILL[name]]
|
|
elif isinstance(hand.get(name), (str, int, float)):
|
|
vals[name] = hand[name]
|
|
if vals:
|
|
out[lid] = vals
|
|
return out
|
|
|
|
|
|
def _apply(files: dict[str, Any], kind: str, values: dict[str, Any]) -> dict[str, Any]:
|
|
"""값 덮기 → 덮기 전 값(`원값` · 없던 칸 = None)."""
|
|
labor = files.get(LABOR_FILE)
|
|
old: dict[str, Any] = {}
|
|
for lid, vals in values.items():
|
|
if (line := _line_at(labor, lid)) is None:
|
|
continue
|
|
given = dict(line.get("입력값") or {})
|
|
old[lid] = {k: given.get(k) for k in vals}
|
|
line["입력값"] = {**given, **vals}
|
|
if kind == "combo" and old:
|
|
files[VALUES_FILE] = seed_values(labor)
|
|
return old
|
|
|
|
|
|
def _restore(files: dict[str, Any], kind: str, old: Any) -> None:
|
|
"""`원값` 되돌림 — 덮기 전 사본 내용."""
|
|
labor = files.get(LABOR_FILE)
|
|
for lid, vals in (old if isinstance(old, dict) else {}).items():
|
|
if (line := _line_at(labor, lid)) is None:
|
|
continue
|
|
given = dict(line.get("입력값") or {})
|
|
for k, v in vals.items():
|
|
if v is None:
|
|
given.pop(k, None)
|
|
else:
|
|
given[k] = v
|
|
line["입력값"] = given
|
|
if kind == "combo" and old:
|
|
files[VALUES_FILE] = seed_values(labor)
|
|
|
|
|
|
def _write(project_root: str | Path, name: str, files: dict, manifest: dict, values: dict) -> None:
|
|
target = library_dir(project_root) / name
|
|
old = _apply(files, str(manifest.get("종류")), values)
|
|
if target.is_dir(): # 끊긴 쓰기(manifest 없음) 자리
|
|
shutil.rmtree(target)
|
|
target.mkdir(parents=True)
|
|
for file, content in files.items():
|
|
if isinstance(content, bytes):
|
|
(target / file).write_bytes(content)
|
|
else:
|
|
atomic_write_json(target / file, content)
|
|
extra = {"값": values, "원값": old} if values else {}
|
|
atomic_write_json(target / MANIFEST_NAME, {**manifest, **extra}) # 마지막 — 있으면 다 쓴 폴더
|
|
|
|
|
|
def _read_copy(folder: Path) -> tuple[dict[str, Any], dict[str, Any]] | None:
|
|
"""다 쓴 사본 → (원값 되돌린 파일들, manifest(값 · 원값 뺌)) — 없으면 None."""
|
|
manifest = read_json(folder / MANIFEST_NAME)
|
|
if not NAME.match(folder.name) or not isinstance(manifest, dict):
|
|
return None
|
|
files: dict[str, Any] = {}
|
|
for path in sorted(folder.iterdir()):
|
|
if path.is_file() and path.name != MANIFEST_NAME:
|
|
files[path.name] = read_json(path) if path.suffix == ".json" else path.read_bytes()
|
|
_restore(files, str(manifest.get("종류")), manifest.get("원값"))
|
|
return files, {k: v for k, v in manifest.items() if k not in ("값", "원값")}
|
|
|
|
|
|
def import_copy(
|
|
project_root: str | Path,
|
|
kind: str,
|
|
layer: str,
|
|
owner: str | None,
|
|
key: str,
|
|
who: dict[str, Any] | None = None,
|
|
haul: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""서고 것 하나를 프로젝트 사본으로 — `{폴더, 만듦}` · 이미 있으면 그대로(만듦 거짓) ·
|
|
haul = 운반 값(263-2 · `{given, own, dist}`) ·
|
|
원본이 없으면 LookupError · 모양이 틀리면 ValueError."""
|
|
haul = haul or {}
|
|
lib = haul.get("lib") # 있는 사본에서(굳힌 내용 · 같은 항목만)
|
|
if isinstance(lib, str) and lib.split("~")[0] == folder_name(kind, layer, owner, key):
|
|
if name := resolve(project_root, {**haul, "lib": lib}, haul.get("dist")):
|
|
return {"폴더": name, "만듦": False}
|
|
files, origin = _build(kind, layer, owner, key)
|
|
values = haul_values(lines_of(files), haul.get("given"), haul.get("own"), haul.get("dist"))
|
|
name = folder_name(kind, layer, owner, key, values)
|
|
if isinstance(read_json(library_dir(project_root) / name / MANIFEST_NAME), dict):
|
|
return {"폴더": name, "만듦": False}
|
|
manifest = {"종류": kind, "원본": origin, "가져온날": now_text(), "가져온이": who}
|
|
_write(project_root, name, files, manifest, values)
|
|
return {"폴더": name, "만듦": True}
|
|
|
|
|
|
def library_inputs(
|
|
project_root: str | Path, kind: str, layer: str, owner: str | None, key: str, lib: Any = None
|
|
) -> list[dict[str, Any]]:
|
|
"""담은 로직 입력 · 값(263-2 · 읽기만) — 있는 사본(`lib`)이면 그 덮기 전 값 ·
|
|
없으면 서고 원본."""
|
|
got = None
|
|
if isinstance(lib, str) and NAME.match(lib):
|
|
got = _read_copy(library_dir(project_root) / lib)
|
|
files = got[0] if got else _build(kind, layer, owner, key)[0]
|
|
out = []
|
|
for lid, line in lines_of(files):
|
|
logic = _logic(str(line["로직"]))
|
|
out.append(
|
|
{
|
|
"id": lid,
|
|
"로직": line["로직"],
|
|
"이름": logic.get("이름") or line["로직"],
|
|
"입력": logic.get("입력") or [],
|
|
"값": line.get("입력값") or {},
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def resolve(project_root: str | Path, ref: dict[str, Any], dist: Any) -> str | None:
|
|
"""운반 연결 사본 → 지금 거리 · 손 값의 사본 폴더(없으면 그 사본에서 새로 만듦) ·
|
|
그 사본이 지워졌으면(prune 뒤 옛 연결) 같은 바탕 사본 → 서고 원본에서 · 못 하면 None."""
|
|
lib = str(ref.get("lib") or "")
|
|
if not NAME.match(lib):
|
|
return None
|
|
base = lib.split("~")[0]
|
|
got = _read_copy(library_dir(project_root) / lib)
|
|
if got is None: # 같은 바탕 사본
|
|
same = [f for f, _m in copies(project_root) if f.name.split("~")[0] == base]
|
|
got = _read_copy(same[0]) if same else None
|
|
if got is None:
|
|
kind, layer, owner, key = base.split("@")
|
|
owner = None if owner == "-" else owner.replace("-", "/")
|
|
try:
|
|
got = _build(kind, layer, owner, key)
|
|
except (LookupError, ValueError):
|
|
return None
|
|
got = (got[0], {"종류": kind, "원본": got[1], "가져온날": now_text(), "가져온이": None})
|
|
files, manifest = got
|
|
values = haul_values(lines_of(files), ref.get("given"), ref.get("own"), dist)
|
|
name = base + (f"~{tag(values)}" if values else "")
|
|
target = library_dir(project_root) / name / MANIFEST_NAME
|
|
if not isinstance(read_json(target), dict):
|
|
_write(project_root, name, files, manifest, values)
|
|
return name
|
|
|
|
|
|
def resolve_hauls(project_root: str | Path, user: dict[str, Any]) -> dict[str, Any]:
|
|
"""셈 사용자 값 — 운반공 줄 서고 사본 연결(덮개 줄 · 거리이정표 연결)을 지금 거리의 사본으로."""
|
|
dist = user.get("haul_dist") or {}
|
|
|
|
def fix(ref: Any, key: Any) -> Any:
|
|
if key not in dist or not (isinstance(ref, dict) and ref.get("tab") == "dg"):
|
|
return ref
|
|
name = resolve(project_root, ref, dist[key]) if ref.get("lib") else None
|
|
return {**ref, "lib": name} if name and name != ref["lib"] else ref
|
|
|
|
bill = user.get("bill") if isinstance(user.get("bill"), dict) else {}
|
|
rows = [
|
|
{**r, "ref": fix(r["ref"], r.get("key"))} if isinstance(r, dict) and "ref" in r else r
|
|
for r in bill.get("rows") or []
|
|
]
|
|
hauls = {k: fix(ref, k) for k, ref in (user.get("haul_refs") or {}).items()}
|
|
return {**user, "bill": {**bill, "rows": rows}, "haul_refs": hauls}
|
|
|
|
|
|
def fix_keys(user: dict[str, Any], libs: dict[str, str]) -> None:
|
|
"""사본 폴더 → 대가 블록 id — 운반 연결 key 를 바뀐 사본 블록으로(제자리)."""
|
|
for row in ref_rows(user):
|
|
ref = row.get("ref")
|
|
if isinstance(ref, dict) and ref.get("tab") == "dg" and ref.get("lib") in libs:
|
|
ref["key"] = libs[ref["lib"]]
|
|
|
|
|
|
def copies(project_root: str | Path) -> list[tuple[Path, dict[str, Any]]]:
|
|
"""다 쓴 사본 `(폴더, manifest)` — 폴더 이름 차례."""
|
|
base = library_dir(project_root)
|
|
out = []
|
|
for child in sorted(base.iterdir()) if base.is_dir() else []:
|
|
manifest = read_json(child / MANIFEST_NAME)
|
|
if child.is_dir() and NAME.match(child.name) and isinstance(manifest, dict):
|
|
out.append((child, manifest))
|
|
return out
|
|
|
|
|
|
def used_folders(user: dict[str, Any]) -> set[str]:
|
|
"""설계내역서 덮개 줄 · 거리이정표 운반 연결(251-5)이 잇는 서고 사본 폴더."""
|
|
return {
|
|
str(ref["lib"])
|
|
for row in ref_rows(user)
|
|
if isinstance(ref := row.get("ref"), dict) and ref.get("tab") == "dg" and ref.get("lib")
|
|
}
|
|
|
|
|
|
def prune(project_root: str | Path, user: dict[str, Any]) -> list[str]:
|
|
"""[저장] 뒤 — 줄이 안 잇는 사본 폴더를 지움(다시 고르면 새로 들여옴) · 지운 이름."""
|
|
keep = used_folders(user)
|
|
base = library_dir(project_root)
|
|
gone = []
|
|
for child in sorted(base.iterdir()) if base.is_dir() else []:
|
|
if child.is_dir() and NAME.match(child.name) and child.name not in keep:
|
|
shutil.rmtree(child)
|
|
gone.append(child.name)
|
|
return gone
|