Merge remote-tracking branch 'origin/sub_laptop_6' into sub_laptop_5

This commit is contained in:
2026-09-30 07:47:31 +09:00
8 changed files with 683 additions and 151 deletions
+92 -22
View File
@@ -1,8 +1,11 @@
"""프로젝트 집계 항목 복사본 읽기 · 쓰기 · 새로 가져오기 · 미리 가져오기 · 계산 (PLAN 25-2b).
"""구조물 집계 항목 복사본 읽기 · 쓰기 · 재적용 · 미리 가져오기 · 개인에 저장 · 새 항목 · 계산
(PLAN 25-2b · 26-2b).
복사본 = `{프로젝트}/templates/items/{폴더}/` (`M02_Item_Snapshot`). 파일마다 판 = 내용 해시 —
쓸 때 판이 다르면 `Stale`(409). 고치면 manifest 에 `고침` 을 남겨 구조물에 안 쓰여도 안 지움.
옛 복사본(새 파일 없음)은 읽을 때 채움(`complete_old`). 계산은 복사본만 씀(마스터 현재값 안 읽음).
복사본 = `{프로젝트}/templates/structures/{구조물 id}/`
(`M02_Item_Snapshot`) · 길의 폴더 = 구조물 id.
파일마다 판 = 내용 해시 — 쓸 때 판이 다르면 `Stale`(409). 고치면 manifest 에 `고침` 을 남겨 지울 때
`_지난/` 에 옮겨 둠. 옛 복사본(새 파일 없음) · 옛 항목 폴더는 읽을 때 채우고 옮겨 받음.
계산은 복사본만 씀(마스터 현재값 안 읽음) — 원본 항목이 바뀌거나 지워져도 그대로.
"""
from __future__ import annotations
@@ -13,17 +16,25 @@ 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_Personal as personal
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 (
FOLDER_PATTERN,
LOCAL_FILES,
MANIFEST_NAME,
PICKS_FILE,
VALUES_FILE,
build_snapshot,
complete_old,
folder_name,
items_dir,
migrate_items,
now_text,
retire,
same_item,
source_files,
structures_dir,
write_snapshot,
)
from M02_MasterTemplete.M02_Item_Source import (
@@ -48,7 +59,6 @@ KINDS = {
#: 뼈대 검사에 쓰는 층 양식 종류(도면 · 면적산출 · 재료산출)
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}$")
@@ -58,11 +68,18 @@ class Stale(ValueError):
self.current = current
def _migrated(project_root: str | Path) -> Path:
"""구조물 복사본 뿌리 — 옛 항목 폴더가 남았으면 먼저 옮겨 받음."""
if items_dir(project_root).is_dir():
migrate_items(project_root)
return structures_dir(project_root)
def folder_of(project_root: str | Path, folder: str) -> Path:
"""복사본 폴더 — 이름 모양이 틀리거나 없으면 LookupError(404)."""
if not _FOLDER.match(folder):
"""구조물 복사본 폴더 — 이름 모양이 틀리거나 없으면 LookupError(404)."""
if not FOLDER_PATTERN.match(folder):
raise LookupError(f"복사본 폴더 이름 모양이 아님 「{folder}」")
target = items_dir(project_root) / folder
target = _migrated(project_root) / folder
if not (target / MANIFEST_NAME).is_file():
raise LookupError(f"복사본이 없음 「{folder}」")
complete_old(target)
@@ -79,10 +96,10 @@ def source_changed(manifest: dict[str, Any]) -> bool | None:
def list_items(project_root: str | Path) -> list[dict[str, Any]]:
base = items_dir(project_root)
base = _migrated(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):
if not child.is_dir() or not FOLDER_PATTERN.match(child.name):
continue
if not (child / MANIFEST_NAME).is_file():
continue
@@ -215,12 +232,15 @@ def _kept_picks(picks: Any, labor: dict[str, Any]) -> dict[str, Any]:
return out
def refresh(project_root: str | Path, folder: str) -> dict[str, Any]:
"""층 · 마스터에서 다시 가져옴 — 입력값 · 고른자재는 같은 키면 살림."""
def reapply(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)
before = {name: read_json(target / file) for name, file in KINDS.items()}
files = build_snapshot(manifest["원본"])
labor = files.get(LABOR_FILE) or {}
files[VALUES_FILE] = _kept_values(values, labor)
@@ -230,24 +250,74 @@ def refresh(project_root: str | Path, folder: str) -> dict[str, Any]:
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)}
covered = [
name
for name, file in KINDS.items()
if file not in LOCAL_FILES and before[name] != files.get(file)
]
source = version_of(source_files(files))
stamp = now_text()
fresh = {**manifest, "판": version_of(source), "원본판": version_of(source), "가져온날": stamp}
fresh = {**manifest, "판": source, "원본판": source, "가져온날": stamp}
atomic_write_json(target / MANIFEST_NAME, fresh)
return {"폴더": folder, "가져온날": stamp, "원본판": fresh["원본판"]}
return {"폴더": folder, "가져온날": stamp, "원본판": source, "덮임": covered}
def prefetch(project_root: str | Path, item: dict[str, Any]) -> dict[str, Any]:
"""구조물에 아직 안 이은 항목을 미리 가져옴 — 있으면 그대로 · 없으면 새로(쓰는 구조물 없음)."""
folder = snapshot_folder_name(item)
target = items_dir(project_root) / folder
refresh = reapply # 25-1b 옛 이름(새로 가져오기)
def prefetch(project_root: str | Path, owner_id: str, item: dict[str, Any]) -> dict[str, Any]:
"""구조물 복사본을 미리 가져옴(아직 [저장] 전) — 같은 항목이면 그대로 · 다르면 새로.
[저장] 뒤에도 그 구조물이 이 항목을 안 쓰면 맞추기가 지움(고친 흔적이 있으면 `_지난/`)."""
folder = folder_name(owner_id)
if not FOLDER_PATTERN.match(folder):
raise ValueError(f"구조물 id 모양이 아님 「{owner_id}」")
target = _migrated(project_root) / folder
manifest = read_json(target / MANIFEST_NAME)
made = not isinstance(manifest, dict)
made = not (isinstance(manifest, dict) and same_item(manifest, item))
if made:
manifest = write_snapshot(target, item, [], {"미리가져옴": now_text()})
if target.is_dir():
retire(target)
manifest = write_snapshot(target, item, owner_id, {"미리가져옴": now_text()})
return {"폴더": folder, "만듦": made, "가져온날": manifest.get("가져온날")}
def save_as(
project_root: str | Path, folder: str, company: int, user: int, path: list[str], name: str
) -> dict[str, Any]:
"""[개인에 저장] — 복사본 내용으로 개인 층에 새 항목(단가 · 입력값 · 고른자재는 안 가져감)."""
target = folder_of(project_root, folder)
manifest = read_json(target / MANIFEST_NAME) or {}
made = personal.create_item(
personal.personal_folder(company, user),
path,
name,
None,
copy=target,
origin=manifest.get("원본"),
)
return {**made, "summary_item": personal.personal_item(company, user, made)}
def new_item(
project_root: str | Path,
owner_id: str,
company: int,
user: int,
path: list[str],
name: str,
unit: str | None,
) -> dict[str, Any]:
"""프로젝트에서 새 항목 — 개인 층에 빈 항목을 만들고 그 구조물 복사본으로.
돌려준 `summary_item` 을 화면이 그 구조물에 지정해 [저장] 해야 복사본이 남음(안 하면 맞추기가
다른 항목 · 지정 없음으로 보고 지움)."""
made = personal.create_item(personal.personal_folder(company, user), path, name, unit)
item = personal.personal_item(company, user, made)
got = prefetch(project_root, owner_id, item)
return {**made, "폴더": got["폴더"], "summary_item": item}
def calc(project_root: str | Path, folder: str) -> dict[str, Any]:
"""복사본(입력값 · 고른자재 · 단가 · 일위대가)만으로 계산 — 마스터 현재값을 안 읽음."""
target = folder_of(project_root, folder)
+120
View File
@@ -0,0 +1,120 @@
"""프로젝트에서 개인 층으로 — [개인에 저장](save-as) · 프로젝트에서 새 항목(new) (PLAN 26-2b).
개인 층 새 항목 = 구조물집계표 새 열 하나 + 빈 도면 · 면적산출 · 재료산출(번호 · SS 키는 저장소가
줌). [개인에 저장]은 구조물 복사본의 열 정의 · 도면 · 면적 · 재료 · 자재조합 · 입력연결을 담음 —
단가 묶음 · 입력값 · 고른자재는 프로젝트 몫이라 안 가져감. 쓰기는 지금 층 저장소 함수(`store.at` ·
`store.write` · `matset.create` · 항목 연결 `write_file`) — sub4 나무 길
(`POST /layers/{layer}/tree`)이 오면 그리로.
권한(본인 개인 층 · 회사 없음 409)은 부르는 쪽 몫.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from common_util.common_util_json import atomic_write_json
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
from M02_MasterTemplete import M02_Template_Layers as layers
from M02_MasterTemplete.M02_Item_Source import LINK_FILE, UM_FILE, read_json
#: 복사본 파일 → 층 양식 종류
PART_KINDS = (("도면.json", "structure"), ("면적.json", "area"), ("재료.json", "material"))
LEVELS = 4 # 지금 집계표 제목행 층 수 — 26-2a 경로 모델이 오면 깊이 자유
def heads_of(path: list[str], name: str | None) -> list[str]:
"""경로 + 이름 → 제목행(`머리`) — 빈 칸은 뺌 · 넷을 넘으면 넷째 칸에 이어 붙임.
ponytail: 제목행 4층 한계 — 26-2a(sub4) 경로 모델이 오면 경로를 그대로 적음."""
full = [str(p).strip() for p in [*path, name or ""] if str(p or "").strip()]
if not full:
raise ValueError("새 항목의 경로 · 이름이 비어 있습니다.")
if len(full) > LEVELS:
full = [*full[: LEVELS - 1], " ".join(full[LEVELS - 1 :])]
return [*full, *[""] * (LEVELS - len(full))]
def _new_matset(row: dict[str, Any], fallback: str) -> str:
"""자재 조합 줄을 이 층에 새 키로 — 이름이 겹치면 「이름 (2)」."""
taken = {str(r.get("이름") or "") for r in matset.listing()["줄"]}
name = base = str(row.get("이름") or fallback)
n = 2
while name in taken:
name, n = f"{base} ({n})", n + 1
got = {k: row.get(k) for k in ("담은자재", "소유", "비고")}
return matset.create({**got, "이름": name, "본뜬키": None})["키"]
def create_item(
folder: Path,
path: list[str],
name: str | None,
unit: str | None,
copy: Path | None = None,
origin: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""개인 층 `folder` 에 새 항목 — `copy` = 구조물 복사본 폴더(있으면 그 내용을 담음).
돌려줌 `{key, 열, 이름}`."""
heads = heads_of(path, name)
total = read_json(copy / "집계.json") if copy else None
base_col = dict((total or {}).get("열") or {})
base_row = (total or {}).get("줄") or {}
items_store.ensure(folder)
with store.at(folder):
table = store.read("table", store.STRUCTURE_TABLE)
doc = table["문서"]
used = {c.get("id") for c in doc.get("열", []) if isinstance(c, dict)}
new_id = items_store._fresh_id(str(base_col.get("id") or "item").split("@")[0], used)
um_row = read_json(copy / UM_FILE) if copy else None
um_key = _new_matset(um_row, heads[0]) if isinstance(um_row, dict) else None
column = {
**json.loads(json.dumps(base_col)),
"id": new_id,
"머리": heads,
"꼴": "수",
"단위": unit or base_col.get("단위") or base_row.get("단위"),
"일위대가": base_col.get("일위대가") or base_row.get("일위대가"),
"자재조합": um_key,
}
column.pop("본뜬열", None)
if origin:
column["본뜬"] = origin
doc["열"] = [*doc.get("열", []), column]
store.write("table", store.STRUCTURE_TABLE, table["판"], doc) # 빈 세 파일 · 번호 · SS
for file, kind in PART_KINDS if copy else ():
got = read_json(copy / file)
if not isinstance(got, dict):
continue
number = store._KEY[kind]
have = store.read(kind, new_id)["문서"]
body = {k: v for k, v in got.items() if k != "계산값"}
body = {**body, "열": new_id, number: have.get(number)}
if kind in store.SHEETS:
body = store._server_values(body)
with store._LOCK:
atomic_write_json(store._path(kind, new_id), body)
links = read_json(copy / LINK_FILE) if copy else None
if isinstance(links, dict) and links.get("줄"):
inputs_store.write_file(new_id, {"판": 1, "줄": links["줄"]})
ss = next(r for r in store.read_combos()["줄"] if r.get("열") == new_id)
return {"key": ss["키"], "열": new_id, "이름": " ".join(h for h in heads if h)}
def personal_item(company: int, user: int, made: dict[str, Any]) -> dict[str, Any]:
"""새로 만든 개인 층 항목의 `summary_item`."""
return {
"layer": "personal",
"owner": f"{company}/{user}",
"key": made["key"],
"name": made["이름"],
}
def personal_folder(company: int, user: int) -> Path:
return layers.personal_dir(company, user)
+152 -63
View File
@@ -1,20 +1,24 @@
"""구조물 집계표 항목 스냅샷 — 프로젝트 `templates/items/{폴더}/` (PLAN 21-1 · 21-3 · 25-2b).
"""구조물 집계 항목 복사본 — 프로젝트 `templates/structures/{구조물 id}/` (PLAN 25-2b · 26-2b).
복사본은 **구조물마다 하나** — 같은 항목을 고른 구조물 둘도 각자 도면 · 면적 · 재료 · 자재조합 ·
일위대가 · 입력연결 · 입력값 · 고른자재 · 단가를 가짐(길이 · 소재 · 높이 · 옵션이 구조물마다 다름).
폴더 이름 = 구조물 id(`structure_id`) · 관 지점은 `pipe@측점`.
[저장](`PUT /route/structures` · `PUT /drainage/pipe-points`) 뒤 서버가 맞춤:
- 구조물 · 관 지점이 가리키는 항목(`summary_item`)마다 원본 층에서 내용을 복사
(`M02_Item_Source.build_source` — 집계 · 도면 · 면적 · 재료 · 자재조합 · 일위대가 · 입력연결 ·
단가) + 프로젝트 값 `입력값.json`(처음엔 빈 값 · 자동값은 안 적음) · `고른자재.json` ·
`manifest.json`
- 이미 있는 폴더는 내용을 다시 쓰지 않음 — 원본(마스터 · 회사 · 개인)이 바뀌어도 프로젝트
값은 지정한 때 그대로 · `manifest` 의 쓰는 구조물 목록만 새로([새로 가져오기]는 사용자가)
- 아무 구조물도 안 쓰는 폴더는 지움(지정 바뀜 · 해제 · 구조물 삭제) — 단 프로젝트에서 고친
흔적(`manifest.고침`)이 있으면 남김
원본 층 자리 = `M02_Template_Layers` 경로(시스템 층은 지금 파일) · 일위대가 · 단가는 M01 시스템 것.
- 항목을 지정한 구조물마다 원본 층에서 내용을 복사(`M02_Item_Source.build_source`) + 프로젝트 값
`입력값.json`(처음엔 빈 값 · 자동값은 안 적음) · `고른자재.json` · `manifest.json`
- 이미 있는 복사본은 다시 쓰지 않음 — 원본(마스터 · 회사 · 개인)이 바뀌거나 지워져도 그대로
([재적용]은 사용자가)
- 지정을 다른 항목으로 바꾸면 새로 가져옴 · 구조물을 지우거나 지정을 풀면 복사본을 지움 —
프로젝트에서 고친 흔적(`manifest.고침`)이 있으면 지우지 않고 `_지난/` 에 옮겨 둠
- 옛 항목 폴더(`templates/items/{층}_{주인}_{키}/` · 25-2b)는 그 항목을 쓰는 구조물마다 복사해
옮겨 받음(고친 값 유지)
"""
from __future__ import annotations
import logging
import re
import shutil
from datetime import datetime
from pathlib import Path
@@ -29,118 +33,197 @@ from M02_MasterTemplete.M02_Item_Source import (
LINK_FILE,
PRICE_FILE,
build_source,
item_path,
read_json,
version_of,
)
logger = logging.getLogger(__name__)
ITEMS_DIRNAME = "items"
STRUCTURES_DIRNAME = "structures"
OLD_ITEMS_DIRNAME = "items" # 25-2b 항목 폴더 — 옮겨 받기만
PAST_DIRNAME = "_지난" # 고친 흔적이 있는 복사본을 지우지 않고 옮겨 두는 자리
MANIFEST_NAME = "manifest.json"
VALUES_FILE = "입력값.json" # {로직키: {입력이름: {값, 확인}}}
PICKS_FILE = "고른자재.json" # {로직키: {호표 차례: MT키}}
LOCAL_FILES = (VALUES_FILE, PICKS_FILE)
#: 복사본 폴더 이름 — 구조물 id(영문 · 숫자 · `_` `-`) 또는 `pipe@측점`
FOLDER_PATTERN = re.compile(r"^(?:pipe@\d+(?:\.\d+)?|[0-9A-Za-z][0-9A-Za-z_-]*)$")
def structures_dir(project_root: str | Path) -> Path:
return layers.project_dir(project_root) / STRUCTURES_DIRNAME
def items_dir(project_root: str | Path) -> Path:
return layers.project_dir(project_root) / ITEMS_DIRNAME
"""옛 항목 폴더 뿌리(25-2b) — 옮겨 받기에만 씀."""
return layers.project_dir(project_root) / OLD_ITEMS_DIRNAME
def folder_name(owner_id: str) -> str:
"""구조물 id → 복사본 폴더 이름 — id 없는 옛 구조물(`#차례`)은 `n차례`."""
return owner_id.replace("#", "n")
def now_text() -> str:
return datetime.now().isoformat(timespec="seconds")
def same_item(manifest: Any, item: dict[str, Any]) -> bool:
origin = manifest.get("원본") if isinstance(manifest, dict) else None
if not isinstance(origin, dict):
return False
return all(origin.get(k) == item.get(k) for k in ("layer", "owner", "key"))
def origin_of(item: dict[str, Any]) -> dict[str, Any]:
"""manifest `원본` — 층 · 주인 · 키 · 이름 + 경로(26-1 · 가져온 때 층 SS 줄)."""
try:
path = item_path(item)
except LookupError:
path = []
return {**item, "경로": path}
def build_snapshot(item: dict[str, Any]) -> dict[str, Any]:
"""원본 내용 + 빈 프로젝트 값 — `{파일 이름: 내용}` · 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:
def source_files(files: dict[str, Any]) -> dict[str, Any]:
return {k: v for k, v in files.items() if k not in LOCAL_FILES}
def write_snapshot(target: Path, item: dict[str, Any], owner_id: 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()
source = version_of(source_files(files))
manifest = {
"원본": item,
"판": version_of(source),
"원본판": version_of(source),
"원본": origin_of(item),
"판": source,
"원본판": source,
"적용일": stamp,
"가져온날": stamp,
"구조물": users,
"구조물": [owner_id],
**extra,
}
atomic_write_json(target / MANIFEST_NAME, manifest)
return manifest
def assignments(project_root: str | Path) -> dict[str, tuple[dict[str, Any], list[str]]]:
"""`{폴더 이름: (항목, 쓰는 구조물 id)}` — 구조물 정본 + 관 지점 정본(id = `pipe@측점`)."""
found: dict[str, tuple[dict[str, Any], list[str]]] = {}
def add(item: dict[str, Any] | None, owner_id: str) -> None:
if item:
found.setdefault(snapshot_folder_name(item), (item, []))[1].append(owner_id)
def assignments(project_root: str | Path) -> dict[str, dict[str, Any]]:
"""`{구조물 id: 항목}` — 구조물 정본 + 관 지점 정본(id = `pipe@측점`) · 지정한 것만."""
found: dict[str, dict[str, Any]] = {}
_, structures = load_structures(str(project_root))
for index, structure in enumerate(structures):
add(structure.summary_item, structure.structure_id or f"#{index}")
if structure.summary_item:
found[structure.structure_id or f"#{index}"] = structure.summary_item
for point in read_pipe_points_file(pipe_points_path_in(Path(project_root))):
add(point.summary_item, f"pipe@{round(float(point.chainage_m), 3)}")
if point.summary_item:
found[f"pipe@{round(float(point.chainage_m), 3)}"] = point.summary_item
return found
def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
"""지정 항목 스냅샷을 맞춤 — `{만듦, 지움, 못만듦}` 폴더 이름."""
base = items_dir(project_root)
def retire(folder: Path) -> str | None:
"""지울 복사본 — 고친 흔적이 있으면 `_지난/{이름}_{때}` 로 옮기고 그 이름 · 아니면 지움."""
manifest = read_json(folder / MANIFEST_NAME)
if isinstance(manifest, dict) and manifest.get("고침"):
past = folder.parent / PAST_DIRNAME
past.mkdir(parents=True, exist_ok=True)
target = past / f"{folder.name}_{datetime.now():%Y%m%d%H%M%S}"
shutil.move(str(folder), str(target))
return target.name
shutil.rmtree(folder)
return None
def migrate_items(project_root: str | Path) -> list[str]:
"""옛 항목 폴더 → 그 항목을 쓰는 구조물마다 복사(고친 값 유지) — 옮겨 받은 구조물 폴더 이름.
다 옮긴 옛 폴더는 지움 · 쓰는 구조물이 없고 고친 흔적이 있으면 `structures/_지난/` 에 둠."""
old = items_dir(project_root)
if not old.is_dir():
return []
base = structures_dir(project_root)
wanted = assignments(project_root)
moved: list[str] = []
for child in sorted(p for p in old.iterdir() if p.is_dir()):
manifest = read_json(child / MANIFEST_NAME)
if not isinstance(manifest, dict):
continue
users = [o for o, item in wanted.items() if snapshot_folder_name(item) == child.name]
for owner in users:
target = base / folder_name(owner)
if (target / MANIFEST_NAME).is_file():
continue # 이미 새 복사본이 있음
shutil.copytree(child, target, dirs_exist_ok=True)
origin = manifest.get("원본") or {}
atomic_write_json(
target / MANIFEST_NAME,
{**manifest, "원본": origin_of(origin), "구조물": [owner], "옮겨받음": now_text()},
)
moved.append(target.name)
if users or not manifest.get("고침"):
shutil.rmtree(child)
else:
(base / PAST_DIRNAME).mkdir(parents=True, exist_ok=True)
shutil.move(str(child), str(base / PAST_DIRNAME / f"items_{child.name}"))
if not any(old.iterdir()):
old.rmdir()
return moved
def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
"""구조물 복사본을 지정에 맞춤 — `{만듦, 지움, 못만듦, 옮겨받음}` 폴더 이름."""
moved = migrate_items(project_root)
base = structures_dir(project_root)
wanted = {
folder_name(owner): (owner, item) for owner, item in assignments(project_root).items()
}
made: list[str] = []
failed: list[str] = []
for folder, (item, users) in sorted(wanted.items()):
removed: list[str] = []
for folder, (owner, item) in sorted(wanted.items()):
target = base / folder
manifest = read_json(target / MANIFEST_NAME)
if isinstance(manifest, dict):
if manifest.get("구조물") != users:
atomic_write_json(target / MANIFEST_NAME, {**manifest, "구조물": users})
if isinstance(manifest, dict) and same_item(manifest, item):
continue
if target.is_dir(): # 다른 항목으로 바뀜 — 옛 것은 지우거나(고침이면) `_지난` 에
retire(target)
removed.append(folder)
try:
write_snapshot(target, item, users, {})
write_snapshot(target, item, owner, {})
except (LookupError, ValueError) as error:
logger.warning("집계 항목 스냅샷을 못 만듦 (%s): %s", folder, error)
logger.warning("집계 항목 복사본을 못 만듦 (%s): %s", folder, error)
failed.append(folder)
continue
made.append(folder)
removed: list[str] = []
if base.is_dir():
for child in sorted(base.iterdir()):
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}
if child.is_dir() and not child.name.startswith("_") and child.name not in wanted:
retire(child)
removed.append(child.name)
return {"만듦": made, "지움": removed, "못만듦": failed, "옮겨받음": moved}
def safe_sync(project_root: str | Path) -> None:
"""저장 뒤 부름 — 실패해도 저장은 그대로(스냅샷은 다음 저장이 다시 맞춤)."""
"""저장 뒤 부름 — 실패해도 저장은 그대로(복사본은 다음 저장이 다시 맞춤)."""
try:
result = sync_snapshots(project_root)
if any(result.values()):
logger.info("집계 항목 스냅샷: %s", result)
logger.info("집계 항목 복사본: %s", result)
except Exception:
logger.exception("집계 항목 스냅샷 맞춤 실패 (저장은 유지): %s", project_root)
logger.exception("집계 항목 복사본 맞춤 실패 (저장은 유지): %s", project_root)
def complete_old(folder: Path) -> bool:
"""옛 스냅샷(21-3 · 새 파일 없음)을 채움 — 입력연결 · 단가는 지금 원본에서 · 값은 빈 것.
"""옛 복사본(21-3 · 새 파일 없음)을 채움 — 입력연결 · 단가는 지금 원본에서 · 값은 빈 것.
옛 내용 파일(집계 · 도면 …)은 그대로 둠. 채웠으면 True(manifest 에 `옮겨받음` 날).
"""
옛 내용 파일(집계 · 도면 …)은 그대로 둠. 채웠으면 True(manifest 에 `옮겨받음` 날)."""
manifest = read_json(folder / MANIFEST_NAME)
if not isinstance(manifest, dict):
return False
@@ -151,15 +234,19 @@ def complete_old(folder: Path) -> bool:
]
if not lacking and "원본판" in manifest:
return False
source = (
build_source(manifest["원본"]) if (LINK_FILE in lacking or PRICE_FILE in lacking) else {}
)
need_source = LINK_FILE in lacking or PRICE_FILE in lacking
try:
source = build_source(manifest["원본"]) if need_source else {}
except LookupError:
source = {} # 원본이 지워짐 — 빈 연결 · 빈 단가로(계산은 멈춤으로 알림)
for name in lacking:
atomic_write_json(folder / name, source.get(name, {}))
atomic_write_json(
folder / name, source.get(name, {"판": None, "줄": []} if name == LINK_FILE else {})
)
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)
if name not in (MANIFEST_NAME, *LOCAL_FILES)
}
atomic_write_json(
folder / MANIFEST_NAME,
@@ -167,13 +254,15 @@ def complete_old(folder: Path) -> bool:
**manifest,
"가져온날": manifest.get("가져온날") or manifest.get("적용일"),
"원본판": manifest.get("원본판") or version_of(kept),
"옮겨받음": now_text(),
"옮겨받음": manifest.get("옮겨받음") or now_text(),
},
)
return True
def read_snapshot(project_root: str | Path, item: dict[str, Any]) -> dict[str, Any] | None:
"""프로젝트 스냅샷의 `집계.json`(`{줄, 열}`) — 없으면 None."""
data = read_json(items_dir(project_root) / snapshot_folder_name(item) / "집계.json")
def read_snapshot(project_root: str | Path, owner_id: str) -> dict[str, Any] | None:
"""그 구조물 복사본의 `집계.json`(`{줄, 열}`) — 없으면 None(옛 항목 폴더는 먼저 옮겨 받음)."""
if items_dir(project_root).is_dir():
migrate_items(project_root)
data = read_json(structures_dir(project_root) / folder_name(owner_id) / "집계.json")
return data if isinstance(data, dict) else None
+15 -4
View File
@@ -54,13 +54,24 @@ def row_of(document: Any, key: str) -> dict[str, Any] | None:
return None
def item_name(item: dict[str, Any]) -> str:
"""층 SS 줄의 보이는 이름(종류 · 분류1~3) — 없으면 LookupError."""
def row_path(row: dict[str, Any]) -> list[str]:
"""SS 줄의 경로(26-1) — `경로` 칸이 있으면 그것 · 옛 줄은 종류 · 분류1~3(빈 분류는 건너뜀)."""
if isinstance(row.get("경로"), list):
return [str(p) for p in row["경로"] if str(p or "").strip()]
return [str(row[k]) for k in ("종류", "분류1", "분류2", "분류3") if row.get(k)]
def item_path(item: dict[str, Any]) -> list[str]:
"""층 SS 줄의 경로 — 없으면 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["키"])
return row_path(row) or [str(row["키"])]
def item_name(item: dict[str, Any]) -> str:
"""층 SS 줄의 보이는 이름(경로를 이은 글) — 없으면 LookupError."""
return " ".join(item_path(item))
def version_of(content: Any) -> str:
+68 -12
View File
@@ -1,12 +1,15 @@
"""프로젝트 집계 항목 복사본 길 — `/api/projects/{id}/items…` (PLAN 25-1b · 25-2b · 25-2e).
"""프로젝트 구조물 집계 항목 복사본 길 — `/api/projects/{id}/items…` (PLAN 25-1b · 26-1 · 26-2b).
GET …/items 목록(이름 · 원본 · 가져온날 · 마스터 바뀜)
POST …/items {layer, owner, key} 미리 가져오기(구조물에 안 이은 항목) → {폴더, 만듦, 가져온날}
GET …/items/{폴더}/{종류} {종류, 판, 내용}
PUT …/items/{폴더}/{종류} {판, 내용} — 판이 다르면 409 · 면적 · 재료는 서버 계산값
POST …/items/{폴더}/refresh 새로 가져오기(입력값 · 고른자재는 같은 키면 살림)
POST …/items/{폴더}/calc 복사본만으로 계산(세 비목 · 계 · 빠진 입력)
종류 = 도면 · 면적 · 재료 · 자재조합 · 입력연결 · 입력값 · 고른자재.
폴더 = 구조물 id(`structure_id` · 관 지점은 `pipe@측점`) — 같은 항목을 고른 구조물도 복사본은 따로.
GET …/items 목록(원본 · 가져온날 · 마스터바뀜 · 원본없음)
POST …/items {structure, layer, owner, key} 미리 가져오기 → {폴더, 만듦, 가져온날}
POST …/items/new {structure, 경로, 이름, 단위} 새 항목(개인 층에도) · 회사 없음 409
GET …/items/{폴더}/{종류} {종류, 판, 내용}
PUT …/items/{폴더}/{종류} {판, 내용} — 판이 다르면 409 · 면적 · 재료는 서버 계산값
POST …/items/{폴더}/reapply 재적용(입력값 · 고른자재는 같은 키면 살림 · `덮임` 알림)
POST …/items/{폴더}/save-as {to, 경로, 이름} 개인 층에 새 항목으로(단가 묶음은 안 가져감)
POST …/items/{폴더}/calc 복사본만으로 계산(세 비목 · 계 · 빠진 입력)
종류 = 도면 · 면적 · 재료 · 자재조합 · 입력연결 · 입력값 · 고른자재 · `refresh` = 옛 이름(재적용).
⚠ 등록은 `main.py`(로그인만) — 프로젝트 권한(같은 회사)은 여기서 `_project` 로 봄(403 · 404).
"""
@@ -23,6 +26,7 @@ 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 import M02_MasterTemplete_Store as store
from M02_MasterTemplete.M02_Item_Source import item_name
router = APIRouter(prefix="/api/projects", tags=["M02 Project Items"])
@@ -36,8 +40,9 @@ class WriteBody(BaseModel):
class PrefetchBody(BaseModel):
"""`{layer, owner, key}`(이름은 층 SS 줄에서) — 옛 모양 `{summary_item}` 도 받음."""
"""`{structure, layer, owner, key}`(이름은 층 SS 줄에서) — 옛 모양 `{summary_item}` 도 받음."""
structure: str | None = None
layer: str | None = None
owner: str | None = None
key: str | None = None
@@ -45,6 +50,28 @@ class PrefetchBody(BaseModel):
summary_item: dict[str, Any] | None = None
class SaveAsBody(BaseModel):
to: Literal["personal"] = "personal"
경로: list[str] = []
이름: str | None = None
class NewBody(BaseModel):
structure: str
경로: list[str] = []
이름: str | None = None
단위: str | None = None
def _me(session: dict[str, Any]) -> tuple[int, int]:
"""개인 층 주인(회사 · 사용자) — 회사 없는 사용자는 409."""
if session.get("company_id") is None:
raise HTTPException(
status_code=409, detail="회사부터 등록 — 개인 층은 회사 연결이 필요합니다."
)
return int(session["company_id"]), int(session["user_id"])
async def _root(session: dict[str, Any], project_id: str):
return (await layer_routes._project(session, project_id))["root"]
@@ -52,6 +79,8 @@ async def _root(session: dict[str, Any], project_id: str):
async def _run(fn, *args):
try:
return await asyncio.to_thread(fn, *args)
except store.StoreError as error:
raise HTTPException(status_code=error.status, detail=error.detail) from error
except copies.Stale as error:
raise HTTPException(
status_code=409, detail={"message": str(error), "판": error.current}
@@ -73,6 +102,8 @@ async def list_items(
async def prefetch_item(
project_id: str, body: PrefetchBody, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
if not body.structure:
raise HTTPException(status_code=400, detail="미리 가져올 구조물(structure)이 필요합니다.")
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 "-"})
@@ -83,7 +114,19 @@ async def prefetch_item(
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)
root = await _root(session, project_id)
return await _run(copies.prefetch, root, body.structure, item)
@router.post("/{project_id}/items/new")
async def new_item(
project_id: str, body: NewBody, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
root = await _root(session, project_id)
company, user = _me(session)
return await _run(
copies.new_item, root, body.structure, company, user, body.경로, body.이름, body.단위
)
@router.get("/{project_id}/items/{folder}/{kind}")
@@ -112,11 +155,24 @@ async def write_item_file(
return await _run(copies.write_file, root, folder, kind, body.판, body.내용, _recalc)
@router.post("/{project_id}/items/{folder}/reapply")
@router.post("/{project_id}/items/{folder}/refresh")
async def refresh_item(
async def reapply_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)
return await _run(copies.reapply, await _root(session, project_id), folder)
@router.post("/{project_id}/items/{folder}/save-as")
async def save_item_as(
project_id: str,
folder: str,
body: SaveAsBody,
session: dict[str, Any] = Depends(verify_session),
) -> dict[str, Any]:
root = await _root(session, project_id)
company, user = _me(session)
return await _run(copies.save_as, root, folder, company, user, body.경로, body.이름)
@router.post("/{project_id}/items/{folder}/calc")
+10 -4
View File
@@ -88,13 +88,15 @@ def collect_items(
) -> list[dict[str, Any]]:
"""설계 정본 → 구조물 목록.
한 건 = `{type_id, placement, chainage_m, start_m, end_m, options, 관연장}`.
한 건 = `{type_id, placement, chainage_m, start_m, end_m, options, 관연장, summary_item,
복사본}` ·
`복사본` = 그 구조물 집계 항목 복사본 폴더의 구조물 id(관 지점은 `pipe@측점` · 26-2b).
"""
types = types or structure_type_map()
lengths = pipe_lengths or {}
items: list[dict[str, Any]] = []
_, structures = load_structures(str(project_root))
for item in structures:
for index, item in enumerate(structures):
definition = types.get(item.type_id)
if definition is not None and definition.managed_by:
continue # 관 지점 정본이 주인 — 옛 저장분이 남아 있어도 두 번 안 셈
@@ -107,6 +109,7 @@ def collect_items(
"end_m": item.end_m,
"options": _with_defaults(definition, dict(item.options or {})),
"summary_item": item.summary_item,
"복사본": item.structure_id or f"#{index}",
}
)
for point in read_pipe_points_file(pipe_points_path_in(Path(project_root))):
@@ -122,6 +125,7 @@ def collect_items(
"options": _with_defaults(types.get(type_id), dict(point.options or {})),
"관연장": _nearest(lengths, chainage) if type_id == "pipe" else None,
"summary_item": point.summary_item,
"복사본": f"pipe@{round(chainage, 3)}",
}
)
return items
@@ -593,7 +597,8 @@ def assign_items(
"""지정 항목 모드 준비 — 구조물마다 스냅샷 열 정의로 고른 열을 정함(PLAN 21-0 · 21-3).
- 미지정 · 스냅샷 없음 = 표에 안 넣음(자동 맞춤 없음) · 알림 한 줄씩
- 시스템 층 항목 = 같은 id 열 자리에 스냅샷 열 정의(원본이 바뀌어도 지정한 때 값)
- 열 정의 = 그 구조물 복사본의 것(26-2b · 원본이 바뀌거나 지워져도 지정한 때 값)
- 시스템 층 항목 = 같은 id 열 자리
- 회사 · 개인 층 항목 = `열id@스냅샷폴더` 새 열(시스템 열과 id · 묶음이 안 겹치게)
돌려주는 것 = (열을 맞춘 문서, `열` · `열묶음` 을 단 구조물, 알림).
"""
@@ -606,7 +611,8 @@ def assign_items(
unassigned = missing = 0
for item in items:
chosen = item.get("summary_item")
column = (read_snapshot(project_root, chosen) or {}).get("열") if chosen else None
owner = item.get("복사본")
column = (read_snapshot(project_root, owner) or {}).get("열") if chosen and owner else None
if not chosen or not isinstance(column, dict) or not column.get("id"):
unassigned += not chosen
missing += bool(chosen)
+202 -20
View File
@@ -1,4 +1,6 @@
"""프로젝트 집계 항목 복사본 완전체 (PLAN 25-2b · 25-2e · 임시 저장소 · 임시 마스터 사본).
"""프로젝트 집계 항목 복사본 완전체 (PLAN 25-2b · 25-2e · 26-2b · 임시 저장소 · 임시 마스터 사본).
복사본은 구조물마다(`templates/structures/{구조물 id}/` · 관은 `pipe@측점`).
- 지정 → [저장] → 폴더에 새 파일(입력연결 · 입력값 · 고른자재 · 단가) · manifest 가져온날 · 원본판
- 단가 묶음이 닫혀 있음 — 조합 37개 · 로직 170개를 마스터와 복사본으로 돌려 같은 값
@@ -32,7 +34,8 @@ 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"
FOLDER = "pipe@40.0" # 관 지점 복사본 폴더 = 구조물 id(26-2b)
PIPE_PATH = ["관공", "(관종·관경마다)", "관매설"]
LABOR_PRICE = "LB000002" # 보통인부 — 되메우기 · 파형강관 부설 호표가 씀
@@ -82,7 +85,7 @@ def _read(path: Path) -> Any:
def _auto_values(root: Path) -> dict[str, Any]:
"""담은 로직마다 자동값을 확인한 값으로 — 사용자가 다 채운 것과 같게."""
labor = _read(root / "templates/items" / FOLDER / "일위대가.json")
labor = _read(root / "templates/structures" / FOLDER / "일위대가.json")
_, whole = m01.loaded()
out: dict[str, Any] = {}
for row in labor["로직"]:
@@ -104,7 +107,7 @@ def _fill_values(root: Path) -> dict[str, Any]:
def test_지정하면_새_파일까지_복사(world):
root = world["root"]
assert _save(root, [_pipe()])["만듦"] == [FOLDER]
folder = root / "templates/items" / FOLDER
folder = root / "templates/structures" / FOLDER
assert _read(folder / "입력연결.json") == {"판": None, "줄": []} # 층 연결이 아직 없음
assert _read(folder / "입력값.json") == {} and _read(folder / "고른자재.json") == {}
price = _read(folder / "단가.json")
@@ -118,7 +121,7 @@ def test_지정하면_새_파일까지_복사(world):
{
"폴더": FOLDER,
"이름": "관공 관매설",
"원본": PIPE,
"원본": {**PIPE, "경로": PIPE_PATH},
"원본판": manifest["원본판"],
"가져온날": manifest["가져온날"],
"마스터바뀜": False,
@@ -221,7 +224,7 @@ def test_마스터_단가를_바꿔도_계산_그대로_새로_가져오면_바
def test_새로_가져오면_없는_키는_버림(world):
root = world["root"]
_save(root, [_pipe()])
folder = root / "templates/items" / FOLDER
folder = root / "templates/structures" / FOLDER
(folder / "입력값.json").write_text(
json.dumps(
{
@@ -267,31 +270,37 @@ def test_쓰기_뒤_다시_읽기_같음_판_다르면_409(world):
copies.read_file(root, "../../etc", "입력값")
def test_고친_복사본은_구조물에서_빠져도_안_지움(world):
def test_고친_복사본은_구조물에서_빠지면_지난에_남김(world):
root = world["root"]
_save(root, [_pipe()])
assert _save(root, [_pipe(None)])["지움"] == [FOLDER] # 안 고쳤으면 지움(지금 규칙)
assert _save(root, [_pipe(None)])["지움"] == [FOLDER] # 안 고쳤으면 지움
assert not (root / "templates/structures/_지난").exists()
_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]["고침"]
assert _save(root, [_pipe(None)])["지움"] == [FOLDER]
assert copies.list_items(root) == []
past = list((root / "templates/structures/_지난").iterdir())
assert len(past) == 1 and past[0].name.startswith("pipe@40.0_")
assert _read(past[0] / "입력값.json") == {"GF000176": {"관경": {"값": 600}}}
def test_미리_가져오기(world):
root = world["root"]
_save(root, [])
made = copies.prefetch(root, PIPE)
made = copies.prefetch(root, FOLDER, 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] # 안 고쳤으면 [저장] 때 지움
assert copies.prefetch(root, FOLDER, PIPE)["만듦"] is False # 같은 항목이면 그대로
assert copies.list_items(root)[0]["구조물"] == [FOLDER]
assert _save(root, [])["지움"] == [FOLDER] # [저장] 뒤에도 안 쓰면 지움
with pytest.raises(ValueError):
copies.prefetch(root, "../x", PIPE)
def test_옛_복사본은_읽을_때_채움(world):
root = world["root"]
_save(root, [_pipe()])
folder = root / "templates/items" / FOLDER
folder = root / "templates/structures" / FOLDER
for name in ("입력연결.json", "입력값.json", "고른자재.json", "단가.json"):
(folder / name).unlink()
manifest = _read(folder / "manifest.json")
@@ -331,12 +340,185 @@ def test_길_권한_남의_프로젝트는_403(world, monkeypatch):
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
wall = {"structure": "st9", "layer": "system", "owner": None}
assert client.post(base, json={**wall, "summary_item": {"layer": "x"}}).status_code == 400
assert client.post(base, json={"layer": "system", "key": "SS000022"}).status_code == 400
made = client.post(base, json={**wall, "key": "SS000022"}).json()
assert made["폴더"] == "st9" 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
assert names["st9"] == "돌쌓기 (찰·메마다) H=(높이마다)" # 층 SS 줄 이름(경로 이음)
assert client.post(base, json={**wall, "key": "SS999999"}).status_code == 404
assert client.post(f"{base}/{FOLDER}/reapply").json()["덮임"] == []
session["company_id"] = 4
assert client.get(base).status_code == 403
assert client.post(f"{base}/{FOLDER}/refresh").status_code == 403
# ── 구조물마다 복사본 (26-2b) ─────────────────────────
MS = {"layer": "system", "owner": None, "key": "SS000022", "name": "돌쌓기"}
def _wall(sid: str, summary: dict | None = MS) -> dict[str, Any]:
return {
"structure_id": sid,
"type_id": "masonry_dry",
"placement": "interval",
"chainage_m": 85,
"start_m": 80,
"end_m": 90,
"options": {"height_m": 2.0, "length_m": 10},
"summary_item": summary,
}
def _save_all(root: Path, walls: list[dict], pipes: list[dict]) -> dict[str, list[str]]:
(root / "B05_Profile/route/structures.json").write_text(
json.dumps({"revision": 1, "structures": walls}, ensure_ascii=False), 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 _memo(root: Path, folder: str, text: str) -> None:
"""면적 시트에 표식 한 칸 — 한쪽만 고쳤는지 보려는 것(계산값 풀이는 건너뜀)."""
area = copies.read_file(root, folder, "면적")
body = {**area["내용"], "메모": text}
copies.write_file(root, folder, "면적", area["판"], body, recalc=lambda doc: doc)
def test_같은_항목_구조물_둘은_따로(world):
root = world["root"]
second = {**_pipe(), "chainage_m": 60.0}
made = _save_all(root, [_wall("a1"), _wall("a2")], [_pipe(), second])["만듦"]
assert made == ["a1", "a2", FOLDER, "pipe@60.0"]
_memo(root, "a1", "a1 만 고침")
assert copies.read_file(root, "a1", "면적")["내용"]["메모"] == "a1 만 고침"
assert "메모" not in copies.read_file(root, "a2", "면적")["내용"] # 다른 쪽 그대로
_fill_values(root) # pipe@40.0 에만 값
assert copies.calc(root, FOLDER)["멈춤"] == [] and copies.calc(root, FOLDER)["계"] > 0
assert len(copies.calc(root, "pipe@60.0")["멈춤"]) == 3 # 계산도 각자
flags = {i["폴더"]: i["고침"] for i in copies.list_items(root)}
assert flags == {"a1": True, "a2": False, FOLDER: True, "pipe@60.0": False}
def test_원본_항목을_지워도_복사본_계산_그대로(world):
root, system = world["root"], world["system"]
_save(root, [_pipe()])
_fill_values(root)
before = copies.calc(root, FOLDER)
combo = _read(system / layers.COMBO_FILE)
combo["줄"] = [r for r in combo["줄"] if r["키"] != PIPE["key"]]
(system / layers.COMBO_FILE).write_text(json.dumps(combo, ensure_ascii=False), "utf-8")
assert copies.calc(root, FOLDER) == before
row = copies.list_items(root)[0]
assert row["원본없음"] is True and row["마스터바뀜"] is False
assert _save(root, [_pipe()])["만듦"] == [] # 다시 저장해도 복사본 그대로
with pytest.raises(LookupError):
copies.reapply(root, FOLDER) # 원본이 없으면 재적용 못 함(404)
def test_재적용은_시트_고침을_덮고_알림_입력값은_살림(world):
root = world["root"]
_save(root, [_pipe()])
values = _fill_values(root)
_memo(root, FOLDER, "x")
assert copies.reapply(root, FOLDER)["덮임"] == ["면적"]
assert "메모" not in copies.read_file(root, FOLDER, "면적")["내용"]
assert copies.read_file(root, FOLDER, "입력값")["내용"] == values
assert copies.reapply(root, FOLDER)["덮임"] == []
def test_개인에_저장하면_개인_층에_항목(world):
root = world["root"]
_save(root, [_pipe()])
rows = [{"로직": "GF000176", "입력": "관경", "자리": "선택"}]
current = copies.read_file(root, FOLDER, "입력연결")
copies.write_file(root, FOLDER, "입력연결", current["판"], {"판": 3, "줄": rows})
made = copies.save_as(root, FOLDER, 3, 12, ["관공", "우리 현장"], "파형강관 매설")
assert made["summary_item"] == {
"layer": "personal",
"owner": "3/12",
"key": made["key"],
"name": "관공 우리 현장 파형강관 매설",
}
folder = layers.personal_dir(3, 12)
row = next(r for r in _read(folder / layers.COMBO_FILE)["줄"] if r["키"] == made["key"])
assert [row["종류"], row["분류1"], row["분류2"]] == ["관공", "우리 현장", "파형강관 매설"]
assert row["일위대가"] == "UA000036" and row["단위"] == "m"
copy = root / "templates/structures" / FOLDER
drawing = _read(folder / "structure" / f"{made['열']}.json")
assert drawing["도면"] == _read(copy / "도면.json")["도면"]
assert _read(folder / "inputs" / f"{made['열']}.json") == {"판": 1, "줄": rows}
assert not list(folder.rglob("단가.json")) # 단가 묶음은 안 가져감
# 개인 층 항목을 다른 구조물에 지정하면 그 내용으로 복사본이 섬
assert _save_all(root, [_wall("b1", made["summary_item"])], [])["만듦"] == ["b1"]
def test_프로젝트에서_새_항목(world):
root = world["root"]
_save_all(root, [_wall("c1")], [])
made = copies.new_item(root, "c1", 3, 12, ["신규 공사"], "돌붙임", "㎡")
assert made["폴더"] == "c1" and made["summary_item"]["layer"] == "personal"
origin = _read(root / "templates/structures/c1/manifest.json")["원본"]
assert origin["key"] == made["key"] and origin["경로"] == ["신규 공사", "돌붙임"]
assert copies.list_items(root)[0]["이름"] == "신규 공사 돌붙임"
# 화면이 새 항목을 지정해 [저장] 하면 그대로 남음
assert _save_all(root, [_wall("c1", made["summary_item"])], [])["만듦"] == []
with pytest.raises(ValueError):
copies.new_item(root, "c2", 3, 12, [], "", None) # 경로 · 이름 빈 값
def test_옛_항목_폴더는_쓰는_구조물마다_옮겨_받음(world):
root = world["root"]
_save_all(root, [_wall("d1"), _wall("d2")], [])
base = root / "templates/structures"
old = root / "templates/items/system_none_SS000022"
old.parent.mkdir(parents=True)
shutil.move(str(base / "d1"), str(old))
shutil.rmtree(base / "d2")
manifest = _read(old / "manifest.json")
manifest.update({"구조물": ["d1", "d2"], "고침": True})
(old / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False), "utf-8")
(old / "입력값.json").write_text(json.dumps({"GF": {"a": {"값": 1}}}), "utf-8")
assert [i["폴더"] for i in copies.list_items(root)] == ["d1", "d2"]
assert not (root / "templates/items").exists()
for sid in ("d1", "d2"):
assert _read(base / sid / "입력값.json") == {"GF": {"a": {"값": 1}}} # 고친 값 유지
assert _read(base / sid / "manifest.json")["구조물"] == [sid]
assert _save_all(root, [_wall("d1"), _wall("d2")], [])["만듦"] == []
def test_지정을_바꾸면_고친_옛_복사본은_지난에(world):
root = world["root"]
_save_all(root, [_wall("e1")], [])
_memo(root, "e1", "y")
rv = {"layer": "system", "owner": None, "key": "SS000001", "name": "돌기슭막이"}
result = _save_all(root, [_wall("e1", rv)], [])
assert result["만듦"] == ["e1"] and result["지움"] == ["e1"]
assert _read(root / "templates/structures/e1/manifest.json")["원본"]["key"] == "SS000001"
past = list((root / "templates/structures/_지난").iterdir())
assert len(past) == 1 and _read(past[0] / "면적.json")["메모"] == "y"
def test_새_항목_길_회사_없으면_409(world, monkeypatch):
root = world["root"]
_save_all(root, [_wall("f1")], [])
async def fake_row(project_id: str) -> dict[str, Any]:
return {"company_id": None, "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": "SYSTEM_ADMIN", "company_id": None, "user_id": 12}
app.dependency_overrides[verify_session] = lambda: session
client = TestClient(app)
body = {"structure": "f1", "경로": ["새"], "이름": "것"}
assert client.post("/api/projects/p1/items/new", json=body).status_code == 409
save_as = client.post("/api/projects/p1/items/f1/save-as", json={"경로": ["a"], "이름": "b"})
assert save_as.status_code == 409
@@ -82,7 +82,8 @@ def _pipe(summary: dict | None) -> dict[str, Any]:
def _items(root: Path) -> Path:
return root / "templates" / "items"
"""구조물 복사본 뿌리(26-2b) — 구조물마다 폴더(`structure_id` · 관은 `pipe@측점`)."""
return root / "templates" / "structures"
# ── 칸 모양 ──────────────────────────────────────────
@@ -126,12 +127,8 @@ def test_옛_파일은_칸_없음_그대로_읽고_씀():
def test_지정하면_폴더가_생기고_내용이_원본과_같음(world):
root = world["root"]
result = _save(root, [_dry(MS), _wet()], [_pipe(PIPE)])
assert result == {
"만듦": ["system_none_SS000011", "system_none_SS000022"],
"지움": [],
"못만듦": [],
}
folder = _items(root) / "system_none_SS000022"
assert result == {"만듦": ["pipe@40.0", "st0"], "지움": [], "못만듦": [], "옮겨받음": []}
folder = _items(root) / "st0"
got = {path.name for path in folder.iterdir()}
assert {"집계.json", "도면.json", "면적.json", "재료.json", "manifest.json"} <= got
total = json.loads((folder / "집계.json").read_text(encoding="utf-8"))
@@ -139,15 +136,12 @@ def test_지정하면_폴더가_생기고_내용이_원본과_같음(world):
drawing = json.loads((world["system"] / "structure/ms.json").read_text(encoding="utf-8"))
assert json.loads((folder / "도면.json").read_text(encoding="utf-8")) == drawing
manifest = json.loads((folder / "manifest.json").read_text(encoding="utf-8"))
assert manifest["원본"] == MS and manifest["구조물"] == ["st0"] and len(manifest["판"]) == 16
pipe = json.loads(
(_items(root) / "system_none_SS000011/manifest.json").read_text(encoding="utf-8")
)
assert manifest["원본"] == {**MS, "경로": ["돌쌓기", "(찰·메마다)", "H=(높이마다)"]}
assert manifest["구조물"] == ["st0"] and len(manifest["판"]) == 16
pipe = json.loads((_items(root) / "pipe@40.0/manifest.json").read_text(encoding="utf-8"))
assert pipe["구조물"] == ["pipe@40.0"]
# 일위대가 UA 줄 + 담은 로직 줄(M01 시스템)
labor = json.loads(
(_items(root) / "system_none_SS000011/일위대가.json").read_text(encoding="utf-8")
)
labor = json.loads((_items(root) / "pipe@40.0/일위대가.json").read_text(encoding="utf-8"))
assert labor["줄"]["키"] == "UA000036"
assert [row["키"] for row in labor["로직"]] + labor["빠진로직"] == [
entry["로직"] for entry in labor["줄"]["담은로직"]
@@ -157,7 +151,7 @@ def test_지정하면_폴더가_생기고_내용이_원본과_같음(world):
def test_원본을_바꿔도_프로젝트_값은_그대로(world):
root, system = world["root"], world["system"]
_save(root, [_dry(MS)], [])
before = (_items(root) / "system_none_SS000022/집계.json").read_text(encoding="utf-8")
before = (_items(root) / "st0/집계.json").read_text(encoding="utf-8")
table = json.loads((system / "table/구조물집계표.json").read_text(encoding="utf-8"))
for column in table["열"]:
if column["id"] == "ms":
@@ -166,7 +160,7 @@ def test_원본을_바꿔도_프로젝트_값은_그대로(world):
json.dumps(table, ensure_ascii=False), encoding="utf-8"
)
assert _save(root, [_dry(MS)], [])["만듦"] == [] # 다시 저장해도 안 다시 씀
after = (_items(root) / "system_none_SS000022/집계.json").read_text(encoding="utf-8")
after = (_items(root) / "st0/집계.json").read_text(encoding="utf-8")
assert after == before
doc = fill.fill_table(root, table)
heads = [c["머리"][0] for c in doc["열"] if c["id"].startswith("ms|")]
@@ -174,16 +168,20 @@ def test_원본을_바꿔도_프로젝트_값은_그대로(world):
def test_지정_바꿈_해제_삭제면_옛_폴더를_지움(world):
"""같은 항목을 고른 구조물 둘 = 폴더 둘 · 한쪽만 바꾸면 그쪽만 새로 · 해제 · 삭제면 지움."""
root = world["root"]
_save(root, [_dry(MS), _dry(MS)], [])
manifest = _items(root) / "system_none_SS000022/manifest.json"
assert json.loads(manifest.read_text(encoding="utf-8"))["구조물"] == ["st0", "st1"]
# 한 구조물만 바꿈 — 같은 항목을 쓰는 구조물이 남으면 폴더 유지 · 목록만 줄어듦
assert sorted(p.name for p in _items(root).iterdir()) == ["st0", "st1"]
result = _save(root, [_dry(RV), _dry(MS)], [])
assert result["만듦"] == ["system_none_SS000001"] and result["지움"] == []
assert json.loads(manifest.read_text(encoding="utf-8"))["구조물"] == ["st1"]
assert result["만듦"] == ["st0"] and result["지움"] == ["st0"] # 안 고쳤으니 옛 것은 지움
assert json.loads((_items(root) / "st0/manifest.json").read_text("utf-8"))["원본"]["key"] == (
"SS000001"
)
assert json.loads((_items(root) / "st1/manifest.json").read_text("utf-8"))["원본"]["key"] == (
"SS000022"
)
# 해제 · 삭제
assert _save(root, [_dry(None)], [])["지움"] == ["system_none_SS000001", "system_none_SS000022"]
assert _save(root, [_dry(None)], [])["지움"] == ["st0", "st1"]
assert list(_items(root).iterdir()) == []
@@ -197,8 +195,8 @@ def test_회사_층_항목은_회사_폴더에서_복사(world):
row["분류3"] = "회사 것"
(company / layers.COMBO_FILE).write_text(json.dumps(combo, ensure_ascii=False), "utf-8")
owned = {**MS, "layer": "company", "owner": "3"}
assert _save(root, [_dry(owned)], [])["만듦"] == ["company_3_SS000022"]
total = json.loads((_items(root) / "company_3_SS000022/집계.json").read_text("utf-8"))
assert _save(root, [_dry(owned)], [])["만듦"] == ["st0"]
total = json.loads((_items(root) / "st0/집계.json").read_text("utf-8"))
assert total["줄"]["분류3"] == "회사 것"
# 채움 — 회사 항목은 시스템 열과 안 겹치는 새 열
doc = fill.fill_table(
@@ -210,7 +208,7 @@ def test_회사_층_항목은_회사_폴더에서_복사(world):
def test_원본에_없는_항목은_못만듦으로_남김(world):
gone = {**MS, "key": "SS999999"}
assert _save(world["root"], [_dry(gone)], [])["못만듦"] == ["system_none_SS999999"]
assert _save(world["root"], [_dry(gone)], [])["못만듦"] == ["st0"]
# ── 채움 ─────────────────────────────────────────────
@@ -245,4 +243,4 @@ def test_스냅샷이_없으면_알림(world):
def test_초기화는_templates_를_안_건드리고_복제는_통째로():
assert not any(tree.startswith("templates") for tree in initial._FILE_TREES)
tool = (config_system.PROJECT_ROOT / "db_management/tools_clone_project.py").read_text("utf-8")
assert '"robocopy",' in tool and '"/E",' in tool # 프로젝트 폴더 전체(templates/items 포함)
assert '"robocopy",' in tool and '"/E",' in tool # 프로젝트 폴더 전체(templates 포함)