From fbb214cd9468d947cf7fe7e5a287320fcc25f9bb Mon Sep 17 00:00:00 2001 From: umsangdon Date: Wed, 30 Sep 2026 07:45:20 +0900 Subject: [PATCH] =?UTF-8?q?feat(M02):=20=EC=A7=91=EA=B3=84=20=ED=95=AD?= =?UTF-8?q?=EB=AA=A9=20=EB=B3=B5=EC=82=AC=EB=B3=B8=EC=9D=84=20=EA=B5=AC?= =?UTF-8?q?=EC=A1=B0=EB=AC=BC=EB=A7=88=EB=8B=A4=20=E2=80=94=20templates/st?= =?UTF-8?q?ructures/{=EA=B5=AC=EC=A1=B0=EB=AC=BC=20id}=20=C2=B7=20?= =?UTF-8?q?=EC=98=9B=20=ED=95=AD=EB=AA=A9=20=ED=8F=B4=EB=8D=94=20=EC=98=AE?= =?UTF-8?q?=EA=B2=A8=20=EB=B0=9B=EA=B8=B0=20=C2=B7=20=EC=9E=AC=EC=A0=81?= =?UTF-8?q?=EC=9A=A9(=EB=8D=AE=EC=9E=84=20=EC=95=8C=EB=A6=BC)=20=C2=B7=20?= =?UTF-8?q?=EA=B0=9C=EC=9D=B8=EC=97=90=20=EC=A0=80=EC=9E=A5=20=C2=B7=20?= =?UTF-8?q?=ED=94=84=EB=A1=9C=EC=A0=9D=ED=8A=B8=EC=97=90=EC=84=9C=20?= =?UTF-8?q?=EC=83=88=20=ED=95=AD=EB=AA=A9=20=C2=B7=20=EC=A7=80=EC=A0=95=20?= =?UTF-8?q?=EB=B0=94=EA=BF=88=20=C2=B7=20=EC=82=AD=EC=A0=9C=20=EB=95=8C=20?= =?UTF-8?q?=EA=B3=A0=EC=B9=9C=20=EA=B2=83=EC=9D=80=20=5F=EC=A7=80=EB=82=9C?= =?UTF-8?q?=20(PLAN=2026-2b)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 같은 항목을 고른 구조물 둘도 도면 · 면적 · 재료 · 자재조합 · 일위대가 · 입력연결 · 입력값 · 고른자재 · 단가를 따로 가짐 · 폴더 = structure_id 또는 pipe@측점 - 옛 templates/items/{층}_{주인}_{키} 는 읽을 때 그 항목을 쓰는 구조물마다 복사(고친 값 유지) · 안 쓰이고 고친 것은 structures/_지난 - 길: 미리 가져오기 {structure, layer, owner, key} · POST …/{폴더}/reapply(refresh 옛 이름) · …/save-as {to, 경로, 이름} · …/items/new {structure, 경로, 이름, 단위} · 회사 없음 409 - manifest 원본에 경로(26-1 · 경로 칸 우선 · 옛 종류 · 분류는 건너뛰며 받음) · 원본이 바뀌거나 지워져도 복사본 · 계산 그대로(목록 마스터바뀜 · 원본없음) - 채움(M02_Table_Fill)은 구조물 복사본의 열 정의를 읽음 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BU9gbNE3aWFUvE61pS2WGt --- M02_MasterTemplete/M02_Item_Copy.py | 114 +++++++-- M02_MasterTemplete/M02_Item_Personal.py | 120 ++++++++++ M02_MasterTemplete/M02_Item_Snapshot.py | 215 ++++++++++++----- M02_MasterTemplete/M02_Item_Source.py | 19 +- .../M02_Project_Items_Router.py | 80 ++++++- M02_MasterTemplete/M02_Table_Fill.py | 14 +- resources/tester/test_item_copy_complete.py | 222 ++++++++++++++++-- .../test_structures_summary_snapshot.py | 50 ++-- 8 files changed, 683 insertions(+), 151 deletions(-) create mode 100644 M02_MasterTemplete/M02_Item_Personal.py diff --git a/M02_MasterTemplete/M02_Item_Copy.py b/M02_MasterTemplete/M02_Item_Copy.py index f7e3cd8c..fc40d21f 100644 --- a/M02_MasterTemplete/M02_Item_Copy.py +++ b/M02_MasterTemplete/M02_Item_Copy.py @@ -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) diff --git a/M02_MasterTemplete/M02_Item_Personal.py b/M02_MasterTemplete/M02_Item_Personal.py new file mode 100644 index 00000000..8b5ca230 --- /dev/null +++ b/M02_MasterTemplete/M02_Item_Personal.py @@ -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) diff --git a/M02_MasterTemplete/M02_Item_Snapshot.py b/M02_MasterTemplete/M02_Item_Snapshot.py index cd583500..3cb25ea8 100644 --- a/M02_MasterTemplete/M02_Item_Snapshot.py +++ b/M02_MasterTemplete/M02_Item_Snapshot.py @@ -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 diff --git a/M02_MasterTemplete/M02_Item_Source.py b/M02_MasterTemplete/M02_Item_Source.py index 134c62ce..895dda70 100644 --- a/M02_MasterTemplete/M02_Item_Source.py +++ b/M02_MasterTemplete/M02_Item_Source.py @@ -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: diff --git a/M02_MasterTemplete/M02_Project_Items_Router.py b/M02_MasterTemplete/M02_Project_Items_Router.py index 4ca0e2f7..6214c696 100644 --- a/M02_MasterTemplete/M02_Project_Items_Router.py +++ b/M02_MasterTemplete/M02_Project_Items_Router.py @@ -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") diff --git a/M02_MasterTemplete/M02_Table_Fill.py b/M02_MasterTemplete/M02_Table_Fill.py index 812c84d1..b8f71e67 100644 --- a/M02_MasterTemplete/M02_Table_Fill.py +++ b/M02_MasterTemplete/M02_Table_Fill.py @@ -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) diff --git a/resources/tester/test_item_copy_complete.py b/resources/tester/test_item_copy_complete.py index 1546f92c..7e064363 100644 --- a/resources/tester/test_item_copy_complete.py +++ b/resources/tester/test_item_copy_complete.py @@ -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 diff --git a/resources/tester/test_structures_summary_snapshot.py b/resources/tester/test_structures_summary_snapshot.py index 73773801..260aa92a 100644 --- a/resources/tester/test_structures_summary_snapshot.py +++ b/resources/tester/test_structures_summary_snapshot.py @@ -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 포함)