Merge remote-tracking branch 'origin/sub_laptop_6' into sub_laptop_5
This commit is contained in:
@@ -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 {},
|
||||
)
|
||||
@@ -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})
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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)}
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
# ── 층 길 — 옛 시스템 길과 같은 모양 ─────────────────────
|
||||
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user