feat(M02): 집계 항목 복사본을 구조물마다 — templates/structures/{구조물 id} · 옛 항목 폴더 옮겨 받기 · 재적용(덮임 알림) · 개인에 저장 · 프로젝트에서 새 항목 · 지정 바꿈 · 삭제 때 고친 것은 _지난 (PLAN 26-2b)
- 같은 항목을 고른 구조물 둘도 도면 · 면적 · 재료 · 자재조합 · 일위대가 · 입력연결 · 입력값 · 고른자재 · 단가를 따로 가짐 · 폴더 = 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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BU9gbNE3aWFUvE61pS2WGt
This commit is contained in:
Claude Opus 5.5
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user