From 589223289006d74b65ba9f82244bbf52d25325e3 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Tue, 29 Sep 2026 12:00:44 +0900 Subject: [PATCH] =?UTF-8?q?feat(B05=C2=B7M02):=20=EA=B5=AC=EC=A1=B0?= =?UTF-8?q?=EB=AC=BC=20=EC=A7=91=EA=B3=84=20=ED=95=AD=EB=AA=A9=20=EC=B9=B8?= =?UTF-8?q?=20=C2=B7=20=EC=A0=80=EC=9E=A5=20=EC=8A=A4=EB=83=85=EC=83=B7=20?= =?UTF-8?q?=C2=B7=20=EC=A7=80=EC=A0=95=20=ED=95=AD=EB=AA=A9=20=EC=B1=84?= =?UTF-8?q?=EC=9B=80=20(PLAN=2021-3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - summary_item 칸: StructureInstance · pipe_points 점 기록(options 밖) — 모양 검사 한 벌 common_util_summary_item(층 system·company·personal · owner null·{회사}·{회사}/{사용자} · key SS+6자리 · name) · 없으면 null(옛 파일 호환) · 관 요청 모양 틀림 400 - 저장 스냅샷 M02_Item_Snapshot: PUT /route/structures · PUT /drainage/pipe-points 뒤 지정 항목마다 templates/items/{layer}_{owner}_{key}/ 에 집계(SS 줄 + 열 정의) · 도면 · 면적 · 재료 · 자재조합 · 일위대가(UA 줄 + 담은 로직 줄) · manifest(원본 · 판 · 적용일 · 쓰는 구조물) — 있는 폴더는 다시 안 씀(원본이 바뀌어도 프로젝트 값 그대로) · 안 쓰는 폴더 지움 · 실패해도 저장 유지 - 채움 fill_table = 지정 항목 모드: 구조물이 고른 열(묶음)에만 · 미지정 · 복사본 없음은 안 셈 + 알림 · 스냅샷 열 정의 사용 · 회사 · 개인 항목은 열id@폴더 새 열 · 자동 맞춤은 fill_document 기본으로만 남음 - 관 시설 되붙이기가 summary_item 도 옮김 - [초기화]는 templates/ 를 안 건드림 · 복제 도구는 폴더 통째 복사 — 손댈 것 없음(시험으로 지킴) - 시험: test_structures_summary_snapshot 17 · 옛 채움 시험은 자동 맞춤 길(fill_document)로 · 층 라우트 채움 시험은 새 규칙(미지정 관 = 줄 없음) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RrQ65VtGZbae2VKgYVMhmc --- .../B04_PreProcess_Router_Basins.py | 10 +- B05_Profile/B05_Profile_Structures_Router.py | 5 + B05_Profile/B05_Profile_Structures_Schema.py | 11 +- M02_MasterTemplete/M02_Item_Snapshot.py | 197 ++++++++++++++ M02_MasterTemplete/M02_Table_Fill.py | 89 ++++++- common_util/common_util_drainage_pipes.py | 7 + common_util/common_util_summary_item.py | 61 +++++ resources/tester/test_m02_table_fill.py | 12 +- resources/tester/test_m02_template_layers.py | 5 +- .../test_structures_summary_snapshot.py | 248 ++++++++++++++++++ 10 files changed, 632 insertions(+), 13 deletions(-) create mode 100644 M02_MasterTemplete/M02_Item_Snapshot.py create mode 100644 common_util/common_util_summary_item.py create mode 100644 resources/tester/test_structures_summary_snapshot.py diff --git a/B04_PreProcess/B04_PreProcess_Router_Basins.py b/B04_PreProcess/B04_PreProcess_Router_Basins.py index 453e3de1..886996e7 100644 --- a/B04_PreProcess/B04_PreProcess_Router_Basins.py +++ b/B04_PreProcess/B04_PreProcess_Router_Basins.py @@ -269,7 +269,10 @@ async def _resolve( if context is None: return JSONResponse(status_code=404, content={"status": "error", "message": reason}) - requested = parse_pipe_points((payload or {}).get("points")) + try: + requested = parse_pipe_points((payload or {}).get("points")) + except ValueError as error: # 집계 항목(summary_item) 모양이 틀림 + return JSONResponse(status_code=400, content={"status": "error", "message": str(error)}) signature = route_signature(context.vertices) stored = ( load_pipe_points(context.stored_path, signature, context.vertices) if use_stored else None @@ -349,6 +352,11 @@ async def put_pipe_points( _basin_features(context, detail, points), context.crs, ) + # 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지) + from common_util.common_util_storage import resolve_stored_project_path + from M02_MasterTemplete.M02_Item_Snapshot import safe_sync + + await asyncio.to_thread(safe_sync, resolve_stored_project_path(context.stored_path)) result = _payload(project_id, context, detail, points, saved=True) result["saved_count"] = saved return result diff --git a/B05_Profile/B05_Profile_Structures_Router.py b/B05_Profile/B05_Profile_Structures_Router.py index f5396e6e..5e0f3c40 100644 --- a/B05_Profile/B05_Profile_Structures_Router.py +++ b/B05_Profile/B05_Profile_Structures_Router.py @@ -8,6 +8,7 @@ 쪽이 앞의 편집을 조용히 지우지 않게. """ +import asyncio import logging from pathlib import Path from typing import Any @@ -282,6 +283,10 @@ async def write_structures( base_revision=payload.base_revision, max_chainage_m=await _route_length(project_id), ) + # 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지) + from M02_MasterTemplete.M02_Item_Snapshot import safe_sync + + await asyncio.to_thread(safe_sync, root) # 설계에 영향을 주는 변경일 때만 B06 이후를 STALE로 돌린다 — 메모만 고쳐도 # 횡단·수량을 다시 돌리게 만들지 않기 위함이다. 응답 플래그는 실제로 STALE # 전파가 **성공했을 때만** true(실패를 성공처럼 알리지 않는다). diff --git a/B05_Profile/B05_Profile_Structures_Schema.py b/B05_Profile/B05_Profile_Structures_Schema.py index 7e851bba..42ea75e9 100644 --- a/B05_Profile/B05_Profile_Structures_Schema.py +++ b/B05_Profile/B05_Profile_Structures_Schema.py @@ -13,7 +13,9 @@ import os from functools import lru_cache from typing import Any, Literal -from pydantic import BaseModel, ConfigDict, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +from common_util.common_util_summary_item import parse_summary_item # 배치형태: 점형(측점 1개) / 구간형(시~종점) / 부지형(위치+면적). STRUCTURE_PLACEMENTS = ("point", "interval", "site") @@ -156,6 +158,13 @@ class StructureInstance(BaseModel): revision: int = Field(default=0, ge=0) # 부지형 polygon 예약 — 1차 UI는 측점+면적 속성만 쓴다. geometry: dict[str, Any] | None = None + # 구조물 집계표 항목(PLAN 21-1) — 없으면 null(옛 파일 호환) · [저장] 뒤 프로젝트 스냅샷 + summary_item: dict[str, Any] | None = None + + @field_validator("summary_item") + @classmethod + def validate_summary_item(cls, value: Any) -> dict[str, Any] | None: + return parse_summary_item(value) @model_validator(mode="after") def validate_placement_fields(self) -> "StructureInstance": diff --git a/M02_MasterTemplete/M02_Item_Snapshot.py b/M02_MasterTemplete/M02_Item_Snapshot.py new file mode 100644 index 00000000..1c35e542 --- /dev/null +++ b/M02_MasterTemplete/M02_Item_Snapshot.py @@ -0,0 +1,197 @@ +"""구조물 집계표 항목 스냅샷 — 프로젝트 `templates/items/{폴더}/` (PLAN 21-1 · 21-3). + +[저장](`PUT /route/structures` · `PUT /drainage/pipe-points`) 뒤 서버가 맞춤: + - 구조물 · 관 지점이 가리키는 항목(`summary_item`)마다 원본 층에서 내용을 복사 + (`집계.json` SS 줄 + 열 정의 · `도면.json` SD · `면적.json` SA · `재료.json` SM · + `자재조합.json` UM 줄 · `일위대가.json` UA 줄 + 담은 로직 줄 · `manifest.json`) + - 이미 있는 폴더는 내용을 다시 쓰지 않음 — 원본(마스터 · 회사 · 개인)이 바뀌어도 프로젝트 + 값은 지정한 때 그대로 · `manifest` 의 쓰는 구조물 목록만 새로 + - 아무 구조물도 안 쓰는 폴더는 지움(지정 바뀜 · 해제 · 구조물 삭제) +원본 층 자리 = `M02_Template_Layers` 경로(시스템 층은 지금 파일) · 일위대가는 M01 시스템 것. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import shutil +from datetime import datetime +from pathlib import Path +from typing import Any + +from B05_Profile.B05_Profile_Structures_Repository import load_structures +from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file +from common_util.common_util_json import atomic_write_json +from common_util.common_util_summary_item import snapshot_folder_name +from M02_MasterTemplete import M02_Template_Layers as layers + +logger = logging.getLogger(__name__) + +ITEMS_DIRNAME = "items" +MANIFEST_NAME = "manifest.json" +TABLE_NAME = "구조물집계표" +UM_FILE = "자재조합.json" +UA_FILE = "일위대가조합.json" # M01 시스템(`resources/master_data/`) +#: 열 id → 스냅샷 파일(구조물 도면 · 면적산출 · 재료산출) +PART_FILES = (("structure", "도면.json"), ("area", "면적.json"), ("material", "재료.json")) + + +def items_dir(project_root: str | Path) -> Path: + return layers.project_dir(project_root) / ITEMS_DIRNAME + + +def layer_root(item: dict[str, Any]) -> Path: + """항목 원본 층 뿌리 — 시스템 · 회사 · 개인(21-2 층 폴더 함수가 오면 그리로).""" + if item["layer"] == "system": + return layers.system_dir() + if item["layer"] == "company": + return layers.company_dir(item["owner"]) + company, user = str(item["owner"]).split("/") + return layers.personal_dir(company, user) + + +def _read(path: Path) -> Any: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + + +def _row(document: Any, key: str) -> dict[str, Any] | None: + rows = document.get("줄") if isinstance(document, dict) else None + for row in rows or []: + if isinstance(row, dict) and row.get("키") == key: + return row + return None + + +def _labor_rows(ua_key: str) -> dict[str, Any] | None: + """UA 줄 + 담은 로직 줄(M01 시스템 · 없는 로직은 빼고 키만 `빠진로직`).""" + from M01_MasterData import M01_MasterData_Store as m01 + + ua = _row(_read(Path(m01.FOLDER) / UA_FILE), ua_key) + if ua is None: + return None + wanted = [str(entry.get("로직")) for entry in ua.get("담은로직") or [] if entry.get("로직")] + found: dict[str, dict[str, Any]] = {} + raw: dict[str, Any] = {} # M01 적재본은 수가 Decimal — 줄은 원본 파일 JSON 그대로 담음 + for key in wanted: + try: + file = m01.logic(key)["file"] + except m01.StoreError: + continue + if file not in raw: + raw[file] = _read(Path(m01.FOLDER) / file) + row = _row(raw[file], key) + if row is not None: + found[key] = row + return { + "줄": ua, + "로직": [found[key] for key in wanted if key in found], + "빠진로직": [key for key in wanted if key not in found], + } + + +def build_snapshot(item: dict[str, Any]) -> dict[str, Any]: + """원본 층에서 항목 내용을 모음 — `{파일 이름: 내용}` · SS 줄이 없으면 LookupError.""" + root = layer_root(item) + row = _row(_read(root / layers.COMBO_FILE), item["key"]) + if row is None: + raise LookupError(f"집계 항목 {item['key']} 이 {item['layer']} 층에 없음") + column_id = str(row.get("열") or "") + table = _read(root / "table" / f"{TABLE_NAME}.json") or {} + column = next((c for c in table.get("열") or [] if c.get("id") == column_id), None) + files: dict[str, Any] = {"집계.json": {"줄": row, "열": column}} + for kind, name in PART_FILES: + part = _read(root / kind / f"{column_id}.json") if column_id else None + if part is not None: + files[name] = part + if row.get("자재조합"): + um = _row(_read(root / UM_FILE), str(row["자재조합"])) + if um is not None: + files["자재조합.json"] = um + if row.get("일위대가"): + labor = _labor_rows(str(row["일위대가"])) + if labor is not None: + files["일위대가.json"] = labor + return files + + +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) + + _, structures = load_structures(str(project_root)) + for index, structure in enumerate(structures): + add(structure.summary_item, structure.structure_id or f"#{index}") + 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)}") + return found + + +def _version(files: dict[str, Any]) -> str: + raw = json.dumps(files, ensure_ascii=False, sort_keys=True).encode("utf-8") + return hashlib.sha256(raw).hexdigest()[:16] + + +def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]: + """지정 항목 스냅샷을 맞춤 — `{만듦, 지움, 못만듦}` 폴더 이름.""" + base = items_dir(project_root) + wanted = assignments(project_root) + made: list[str] = [] + failed: list[str] = [] + for folder, (item, users) in sorted(wanted.items()): + target = base / folder + manifest = _read(target / MANIFEST_NAME) + if isinstance(manifest, dict): + if manifest.get("구조물") != users: + atomic_write_json(target / MANIFEST_NAME, {**manifest, "구조물": users}) + continue + try: + files = build_snapshot(item) + except (LookupError, ValueError) as error: + logger.warning("집계 항목 스냅샷을 못 만듦 (%s): %s", folder, error) + failed.append(folder) + continue + target.mkdir(parents=True, exist_ok=True) + for name, content in files.items(): + atomic_write_json(target / name, content) + # manifest 는 마지막 — 있으면 다 쓴 폴더(도중에 끊기면 다음 저장이 다시 씀) + atomic_write_json( + target / MANIFEST_NAME, + { + "원본": item, + "판": _version(files), + "적용일": datetime.now().isoformat(timespec="seconds"), + "구조물": users, + }, + ) + made.append(folder) + removed: list[str] = [] + if base.is_dir(): + for child in sorted(base.iterdir()): + if child.is_dir() and child.name not in wanted: + shutil.rmtree(child) + removed.append(child.name) + return {"만듦": made, "지움": removed, "못만듦": failed} + + +def safe_sync(project_root: str | Path) -> None: + """저장 뒤 부름 — 실패해도 저장은 그대로(스냅샷은 다음 저장이 다시 맞춤).""" + try: + result = sync_snapshots(project_root) + if any(result.values()): + logger.info("집계 항목 스냅샷: %s", result) + except Exception: + logger.exception("집계 항목 스냅샷 맞춤 실패 (저장은 유지): %s", project_root) + + +def read_snapshot(project_root: str | Path, item: dict[str, Any]) -> dict[str, Any] | None: + """프로젝트 스냅샷의 `집계.json`(`{줄, 열}`) — 없으면 None.""" + data = _read(items_dir(project_root) / snapshot_folder_name(item) / "집계.json") + return data if isinstance(data, dict) else None diff --git a/M02_MasterTemplete/M02_Table_Fill.py b/M02_MasterTemplete/M02_Table_Fill.py index d0107c4d..812c84d1 100644 --- a/M02_MasterTemplete/M02_Table_Fill.py +++ b/M02_MasterTemplete/M02_Table_Fill.py @@ -106,6 +106,7 @@ def collect_items( "start_m": item.start_m, "end_m": item.end_m, "options": _with_defaults(definition, dict(item.options or {})), + "summary_item": item.summary_item, } ) for point in read_pipe_points_file(pipe_points_path_in(Path(project_root))): @@ -120,6 +121,7 @@ def collect_items( "end_m": None, "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, } ) return items @@ -186,6 +188,27 @@ def _sources(binding: dict[str, Any]) -> list[dict[str, Any]]: return [main, *(binding.get("더함") or [])] +def _home(column: dict[str, Any]) -> str: + """펼친 열이면 마스터 열의 묶음 — 지정 항목이 고른 열(묶음)과 견줌.""" + return _group(_frame(column).get("원열") or column) + + +def _assigned_match(source: dict[str, Any], item: dict[str, Any], column: dict[str, Any]) -> bool: + """지정 항목 모드 — 구조물이 고른 열(묶음)에만 · 종류 · 조건 · 옵션으로 자동 맞추지 않음. + + 값을 읽을 출처 = 바인딩 출처 가운데 그 구조물 종류 것(없으면 주 출처). + """ + if item.get("열묶음") != _home(column): + return False + sources = _sources(_frame(column)) + typed = [entry for entry in sources if entry.get("종류") == item["type_id"]] + return source == (typed[0] if typed else sources[0]) + + +def _auto_match(source: dict[str, Any], item: dict[str, Any], _column: dict[str, Any]) -> bool: + return _matches(source, item) + + def _expansion(source: dict[str, Any], item: dict[str, Any]) -> tuple[str, ...]: values = [] for key in source.get("펼침") or []: @@ -313,6 +336,7 @@ def _expand_columns( items: list[dict[str, Any]], variables: dict[str, Any], choices: dict[str, list[str]], + match: Any = _auto_match, ) -> tuple[list[dict[str, Any]], dict[str, list[tuple[str, str, dict[str, Any]]]]]: """펼침 열 → 값마다 열 · 묶음은 값마다 한데(관 Φ800 관매설 · 커플링 → Φ1000 …). @@ -333,7 +357,7 @@ def _expand_columns( found = groups.setdefault(_group(column), set()) for source in _sources(binding): # 값이 빈 구조물도 열은 세움 — 놓였는데 수량이 빈칸인 것이 보여야 함 - found.update(_expansion(source, item) for item in items if _matches(source, item)) + found.update(_expansion(source, item) for item in items if match(source, item, column)) out: list[dict[str, Any]] = [] made: dict[str, list[tuple[str, str, dict[str, Any]]]] = {} @@ -394,14 +418,19 @@ def fill_document( document: dict[str, Any], items: list[dict[str, Any]], types: dict[str, StructureType] | None = None, + assigned: bool = False, ) -> dict[str, Any]: - """표 문서 + 설계 구조물 목록 → 채운 표 문서(새 사본).""" + """표 문서 + 설계 구조물 목록 → 채운 표 문서(새 사본). + + `assigned` = 지정 항목 모드(`fill_table`) — 구조물마다 고른 열(`열묶음`)에만 넣음. + """ + match = _assigned_match if assigned else _auto_match types = types or structure_type_map() doc = copy.deepcopy(document) variables = dict(doc.get("변수") or {}) interval = _number(variables.get("측점간격_m")) or 20.0 rule_columns = _unexpand(doc.get("열") or []) - columns, made = _expand_columns(rule_columns, items, variables, _choices(types)) + columns, made = _expand_columns(rule_columns, items, variables, _choices(types), match) notes: list[str] = [] # 줄 자리 — 관은 한 줄에 하나(둘째 관부터 줄을 나눔) @@ -427,7 +456,7 @@ def fill_document( if source.get("값") in (None, "식"): continue for item, slot in zip(items, item_slot, strict=True): - if not _matches(source, item): + if not match(source, item, column): continue if expanded is not None and list(_expansion(source, item)) != expanded: continue @@ -558,14 +587,62 @@ def is_fillable(document: Any) -> bool: ) +def assign_items( + project_root: str | Path, document: dict[str, Any], items: list[dict[str, Any]] +) -> tuple[dict[str, Any], list[dict[str, Any]], list[str]]: + """지정 항목 모드 준비 — 구조물마다 스냅샷 열 정의로 고른 열을 정함(PLAN 21-0 · 21-3). + + - 미지정 · 스냅샷 없음 = 표에 안 넣음(자동 맞춤 없음) · 알림 한 줄씩 + - 시스템 층 항목 = 같은 id 열 자리에 스냅샷 열 정의(원본이 바뀌어도 지정한 때 값) + - 회사 · 개인 층 항목 = `열id@스냅샷폴더` 새 열(시스템 열과 id · 묶음이 안 겹치게) + 돌려주는 것 = (열을 맞춘 문서, `열` · `열묶음` 을 단 구조물, 알림). + """ + from common_util.common_util_summary_item import snapshot_folder_name + from M02_MasterTemplete.M02_Item_Snapshot import read_snapshot + + doc = copy.deepcopy(document) + rules = _unexpand(doc.get("열") or []) + out: list[dict[str, Any]] = [] + unassigned = missing = 0 + for item in items: + chosen = item.get("summary_item") + column = (read_snapshot(project_root, chosen) or {}).get("열") if chosen else None + if not chosen or not isinstance(column, dict) or not column.get("id"): + unassigned += not chosen + missing += bool(chosen) + continue + new = copy.deepcopy(column) + if chosen["layer"] != "system": + new["id"] = f"{column['id']}@{snapshot_folder_name(chosen)}" + if isinstance(new.get("바인딩"), dict) and new["바인딩"].get("묶음"): + new["바인딩"]["묶음"] = new["id"] + at = next((i for i, c in enumerate(rules) if c.get("id") == new["id"]), None) + if at is None: + rules.append(new) + else: + rules[at] = new + out.append({**item, "열": new["id"], "열묶음": _group(new)}) + doc["열"] = rules + notes = [] + if unassigned: + notes.append(f"집계 항목 미지정 {unassigned}건 — 표에 안 들어감") + if missing: + notes.append(f"집계 항목 복사본 없음 {missing}건 — [저장] 뒤 채워짐") + return doc, out, notes + + def fill_table( project_root: str | Path, document: dict[str, Any], pipe_lengths: dict[float, float] | None = None, ) -> dict[str, Any]: - """프로젝트 설계로 표 문서를 채움 — 저장 안 함.""" + """프로젝트 설계로 표 문서를 채움 — 저장 안 함 · 지정 항목 모드(미지정은 빈칸).""" types = structure_type_map() - return fill_document(document, collect_items(project_root, pipe_lengths, types), types) + items = collect_items(project_root, pipe_lengths, types) + doc, chosen, notes = assign_items(project_root, document, items) + filled = fill_document(doc, chosen, types, assigned=True) + filled["알림"] = [*notes, *filled.get("알림", [])] + return filled def recalc(document: dict[str, Any]) -> dict[str, Any] | None: diff --git a/common_util/common_util_drainage_pipes.py b/common_util/common_util_drainage_pipes.py index a9aecced..0dcec025 100644 --- a/common_util/common_util_drainage_pipes.py +++ b/common_util/common_util_drainage_pipes.py @@ -35,6 +35,7 @@ from shapely.geometry import LineString, Point from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir from common_util.common_util_json import atomic_write_json from common_util.common_util_route_geometry import RouteVertex, interpolate_vertex +from common_util.common_util_summary_item import parse_summary_item from config.config_system import ( DRAINAGE_CACHE_DIRNAME, DRAINAGE_DETAIL_FILENAME, @@ -89,6 +90,8 @@ class PipePoint: # 관이 실제로 놓인 자리(사업지 CRS, m). 노선이 바뀌어도 이 자리는 그대로다. x: float | None = None y: float | None = None + # 구조물 집계표 항목(PLAN 21-1 · options 밖) — 없으면 None · 모양은 `common_util_summary_item` + summary_item: dict[str, Any] | None = None def as_dict(self) -> dict[str, Any]: # 구 형식 저장분이 확장 필드 없이 그대로 다시 저장되도록 기본값은 생략한다. @@ -110,6 +113,8 @@ class PipePoint: if self.x is not None and self.y is not None: payload["x"] = round(float(self.x), 3) payload["y"] = round(float(self.y), 3) + if self.summary_item: + payload["summary_item"] = self.summary_item return payload @@ -349,6 +354,7 @@ def parse_pipe_points(values: Any) -> list[PipePoint]: if isinstance(options, dict) and options else None ), + summary_item=parse_summary_item(item.get("summary_item")), ) ) points.sort(key=lambda point: point.chainage_m) @@ -399,6 +405,7 @@ def carry_facility_attributes(base: list[PipePoint], reference: list[PipePoint]) point.start_m = ref.start_m point.end_m = ref.end_m point.options = ref.options + point.summary_item = ref.summary_item return base diff --git a/common_util/common_util_summary_item.py b/common_util/common_util_summary_item.py new file mode 100644 index 00000000..08f58c67 --- /dev/null +++ b/common_util/common_util_summary_item.py @@ -0,0 +1,61 @@ +"""구조물 집계표 항목 참조 `summary_item` — 모양 검사 · 스냅샷 폴더 이름 (PLAN 21-1). + + summary_item = {layer: "system"|"company"|"personal", + owner: null|"{회사}"|"{회사}/{사용자}", key: "SS000004", name: "…"} + +구조물 기록(`structures.json` `StructureInstance`) · 관 지점 기록(`pipe_points.json` 점) 두 곳이 +같은 칸을 씀 — 검사는 여기 한 벌. 없으면 null(옛 파일 호환). +""" + +from __future__ import annotations + +import re +from typing import Any + +SUMMARY_LAYERS = ("system", "company", "personal") +_KEY = re.compile(r"^SS\d{6}$") +_OWNER = { + "system": None, + "company": re.compile(r"^\d+$"), + "personal": re.compile(r"^\d+/\d+$"), +} +_NAME_MAX = 200 + + +def parse_summary_item(value: Any) -> dict[str, Any] | None: + """모양이 맞으면 `{layer, owner, key, name}` 새 dict · 빈 값은 None · 틀리면 ValueError.""" + if value is None: + return None + if not isinstance(value, dict): + raise ValueError("집계 항목(summary_item)은 객체여야 합니다.") + extra = set(value) - {"layer", "owner", "key", "name"} + if extra: + raise ValueError(f"집계 항목에 모르는 칸이 있습니다: {', '.join(sorted(extra))}") + layer = value.get("layer") + if layer not in SUMMARY_LAYERS: + raise ValueError("집계 항목 층은 system · company · personal 중 하나입니다.") + owner = value.get("owner") + rule = _OWNER[layer] + if rule is None: + if owner is not None: + raise ValueError("시스템 층 집계 항목의 owner 는 null 이어야 합니다.") + elif not isinstance(owner, str) or not rule.match(owner): + shape = "{회사}" if layer == "company" else "{회사}/{사용자}" + raise ValueError(f"{layer} 층 집계 항목의 owner 는 「{shape}」 숫자 모양입니다.") + key = value.get("key") + if not isinstance(key, str) or not _KEY.match(key): + raise ValueError("집계 항목 key 는 SS 와 숫자 6자리입니다(예: SS000004).") + name = value.get("name") + if not isinstance(name, str) or not name.strip() or len(name) > _NAME_MAX: + raise ValueError(f"집계 항목 name 은 1~{_NAME_MAX}자 글입니다.") + return {"layer": layer, "owner": owner, "key": key, "name": name} + + +def snapshot_folder_name(item: dict[str, Any]) -> str: + """`{layer}_{owner 슬러그}_{key}` — owner 없음 = `none` · `/` = `-`. + + 예: `system_none_SS000004` · `company_3_SS000001` · `personal_3-12_SS000002`. + """ + owner = item.get("owner") + slug = str(owner).replace("/", "-") if owner else "none" + return f"{item['layer']}_{slug}_{item['key']}" diff --git a/resources/tester/test_m02_table_fill.py b/resources/tester/test_m02_table_fill.py index ab0dae5b..ed7b11cb 100644 --- a/resources/tester/test_m02_table_fill.py +++ b/resources/tester/test_m02_table_fill.py @@ -71,8 +71,14 @@ def project(tmp_path: Path) -> Path: return root +def _auto(project: Path, document: dict[str, Any], lengths: dict[float, float]) -> dict[str, Any]: + """바인딩 자동 맞춤(`fill_document` 기본) — 프로젝트 `fill_table` 은 지정 항목 모드(PLAN 21-3)라 + 펼침 · 묶음 · 손 값 규칙은 이 길로 봄.""" + return fill.fill_document(document, fill.collect_items(project, lengths)) + + def _filled(project: Path, document: dict[str, Any] | None = None) -> dict[str, Any]: - return fill.fill_table(project, document or _master(), {85.0: 14.0}) + return _auto(project, document or _master(), {85.0: 14.0}) def _col(doc: dict[str, Any], col_id: str) -> dict[str, Any]: @@ -119,7 +125,7 @@ def test_계산_열은_설계값_딱지_없음(project: Path) -> None: assert "바인딩" not in column and column["펼침틀"]["원열"]["id"] == col_id.split("|")[0] assert _col(doc, "pp_cp|파형강관|1000")["머리"] == ["관공", "파형강관 Φ1000", "커플링밴드"] # 다시 저장 · 채워도 같은 열(계산 열도 원열로 되돌림) - again = fill.fill_table(project, fill.strip_design(doc), {85.0: 14.0}) + again = _auto(project, fill.strip_design(doc), {85.0: 14.0}) assert [c["id"] for c in again["열"]] == [c["id"] for c in doc["열"]] @@ -152,7 +158,7 @@ def test_측점_줄_같은_측점은_한_줄_관은_나눔(project: Path) -> Non assert [row["값"].get("no") for row in doc["줄"][1:]] == [str(i) for i in range(1, 8)] assert "식" not in rows["s200.00~230.00"] # 줄눈 간격이 달라도 열 식이 없으면 줄 식 없음 assert doc["알림"] == [] - bare = fill.fill_table(project, _master(), {}) + bare = _auto(project, _master(), {}) assert bare["알림"] == ["관 연장 없음 2곳 — B06 횡단 설계 전이면 빈칸"] assert not any("pp_len" in key for row in bare["줄"] for key in row["값"]) diff --git a/resources/tester/test_m02_template_layers.py b/resources/tester/test_m02_template_layers.py index c9723a67..e533ac28 100644 --- a/resources/tester/test_m02_template_layers.py +++ b/resources/tester/test_m02_template_layers.py @@ -207,8 +207,9 @@ def test_채운_표는_작업본을_채워_돌려주고_저장_안_함( ) before = layers.version_of(root / "templates/table/구조물집계표.json") got = world["client"].get(f"/api/m02/projects/{P1}/tables/구조물집계표/filled").json() - row = next(r for r in got["문서"]["줄"] if r["id"] == "s40.00") - assert row["값"]["sta"] == "NO.2" and row["값"]["pp_len|파형강관|1000"] == 12 + # 프로젝트 채움 = 지정 항목 모드(PLAN 21-3) — 집계 항목 미지정 관은 표에 안 들어감 + assert not any(r["id"] == "s40.00" for r in got["문서"]["줄"]) + assert "집계 항목 미지정 1건 — 표에 안 들어감" in got["문서"]["알림"] assert got["결과"] == {"계산": {}} and got["판"] == before assert layers.version_of(root / "templates/table/구조물집계표.json") == before diff --git a/resources/tester/test_structures_summary_snapshot.py b/resources/tester/test_structures_summary_snapshot.py new file mode 100644 index 00000000..73773801 --- /dev/null +++ b/resources/tester/test_structures_summary_snapshot.py @@ -0,0 +1,248 @@ +"""구조물 집계표 항목 지정 · 저장 스냅샷 · 채움 (PLAN 21-3 · 임시 저장소). + +- `summary_item` 칸(구조물 · 관 지점) 모양 검사 · 옛 파일(칸 없음) 호환 +- 저장 뒤 스냅샷: 지정 → 폴더 생김 · 원본(마스터) 바꿔도 프로젝트 값 그대로 · 지정 바꿈 → 옛 폴더 + 지움 · 해제 → 지움 · 회사 층 자리 +- 채움(`fill_table`) = 지정 항목 모드 — 고른 열에만 · 미지정은 안 셈 · 스냅샷 열 정의 사용 +- [초기화]는 `templates/` 를 안 건드림 · 프로젝트 복제 도구는 폴더를 통째 복사 +""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path +from typing import Any + +import pytest + +from B05_Profile.B05_Profile_Structures_Schema import StructureInstance +from common_util import common_util_initial_snapshot as initial +from common_util.common_util_drainage_pipes import PipePoint, parse_pipe_points +from common_util.common_util_summary_item import parse_summary_item +from config import config_system +from M02_MasterTemplete import M02_Item_Snapshot as snap +from M02_MasterTemplete import M02_Table_Fill as fill +from M02_MasterTemplete import M02_Template_Layers as layers + +REAL = config_system.PROJECT_ROOT / "resources/master_template" +MS = {"layer": "system", "owner": None, "key": "SS000022", "name": "돌쌓기"} +RV = {"layer": "system", "owner": None, "key": "SS000001", "name": "돌기슭막이 메쌓기 1.2m"} +PIPE = {"layer": "system", "owner": None, "key": "SS000011", "name": "관공 관매설"} + + +def _copy_layer(target: Path) -> None: + target.mkdir(parents=True) + for name in (layers.COMBO_FILE, "자재조합.json"): + shutil.copy2(REAL / name, target / name) + for kind in ("table", "structure", "area", "material"): + shutil.copytree(REAL / kind, target / kind) + + +@pytest.fixture +def world(tmp_path: Path, monkeypatch) -> dict[str, Any]: + system = tmp_path / "master" + _copy_layer(system) + monkeypatch.setattr(layers, "SYSTEM_ROOT", system) + storage = tmp_path / "storage" + monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage)) + root = storage / "3" / "12" / "proj" + (root / "B05_Profile/route").mkdir(parents=True) + (root / "B04_PreProcess/drainage/edits").mkdir(parents=True) + return {"system": system, "storage": storage, "root": root} + + +def _save(root: Path, structures: list[dict], points: list[dict]) -> dict[str, list[str]]: + """화면 [저장] 두 곳(구조물 · 관 지점)이 쓴 뒤 서버가 맞추는 것과 같게.""" + for index, item in enumerate(structures): + item.setdefault("structure_id", f"st{index}") + item.setdefault("placement", "interval" if "start_m" in item else "point") + (root / "B05_Profile/route/structures.json").write_text( + json.dumps({"revision": 1, "structures": structures}, ensure_ascii=False), encoding="utf-8" + ) + (root / "B04_PreProcess/drainage/edits/pipe_points.json").write_text( + json.dumps({"points": points}, ensure_ascii=False), encoding="utf-8" + ) + return snap.sync_snapshots(root) + + +def _dry(summary: dict | None) -> dict[str, Any]: + wall = {"type_id": "masonry_dry", "chainage_m": 85, "start_m": 80, "end_m": 90} + return {**wall, "options": {"height_m": 2.0, "length_m": 10}, "summary_item": summary} + + +def _wet() -> dict[str, Any]: + wall = {"type_id": "masonry_wet", "chainage_m": 120, "start_m": 120, "end_m": 130} + return {**wall, "options": {"height_m": 2.5, "length_m": 10}} + + +def _pipe(summary: dict | None) -> dict[str, Any]: + options = {"pipe_kind": "파형강관", "pipe_diameter_mm": "1000"} + return {"chainage_m": 40.0, "source": "user", "options": options, "summary_item": summary} + + +def _items(root: Path) -> Path: + return root / "templates" / "items" + + +# ── 칸 모양 ────────────────────────────────────────── + + +@pytest.mark.parametrize( + "bad", + [ + {"layer": "project", "owner": None, "key": "SS000001", "name": "x"}, + {"layer": "system", "owner": "3", "key": "SS000001", "name": "x"}, + {"layer": "company", "owner": "3/12", "key": "SS000001", "name": "x"}, + {"layer": "personal", "owner": "3", "key": "SS000001", "name": "x"}, + {"layer": "system", "owner": None, "key": "UA000001", "name": "x"}, + {"layer": "system", "owner": None, "key": "SS000001", "name": ""}, + {"layer": "system", "owner": None, "key": "SS000001", "name": "x", "more": 1}, + "SS000001", + ], +) +def test_모양이_틀린_항목은_거절(bad): + with pytest.raises(ValueError): + parse_summary_item(bad) + with pytest.raises(ValueError): + StructureInstance(type_id="x", placement="point", chainage_m=1, summary_item=bad) + with pytest.raises(ValueError): + parse_pipe_points([{"chainage_m": 1.0, "summary_item": bad}]) + + +def test_옛_파일은_칸_없음_그대로_읽고_씀(): + item = StructureInstance(type_id="x", placement="point", chainage_m=1) + assert item.summary_item is None + point = parse_pipe_points([{"chainage_m": 10.0, "source": "user"}])[0] + assert point.summary_item is None and "summary_item" not in point.as_dict() + owned = {"layer": "personal", "owner": "3/12", "key": "SS000002", "name": "내 옹벽"} + again = parse_pipe_points([PipePoint(chainage_m=5.0, summary_item=owned).as_dict()])[0] + assert again.summary_item == owned + + +# ── 스냅샷 ─────────────────────────────────────────── + + +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" + 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")) + assert total["줄"]["키"] == "SS000022" and total["열"]["id"] == "ms" + 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 pipe["구조물"] == ["pipe@40.0"] + # 일위대가 UA 줄 + 담은 로직 줄(M01 시스템) + labor = json.loads( + (_items(root) / "system_none_SS000011/일위대가.json").read_text(encoding="utf-8") + ) + assert labor["줄"]["키"] == "UA000036" + assert [row["키"] for row in labor["로직"]] + labor["빠진로직"] == [ + entry["로직"] for entry in labor["줄"]["담은로직"] + ] + + +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") + table = json.loads((system / "table/구조물집계표.json").read_text(encoding="utf-8")) + for column in table["열"]: + if column["id"] == "ms": + column["바인딩"]["머리틀"] = ["바뀐 돌쌓기", "{0}", "{1}"] + (system / "table/구조물집계표.json").write_text( + 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") + assert after == before + doc = fill.fill_table(root, table) + heads = [c["머리"][0] for c in doc["열"] if c["id"].startswith("ms|")] + assert heads == ["돌쌓기"] # 채움도 스냅샷 열 정의 + + +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"] + # 한 구조물만 바꿈 — 같은 항목을 쓰는 구조물이 남으면 폴더 유지 · 목록만 줄어듦 + 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 _save(root, [_dry(None)], [])["지움"] == ["system_none_SS000001", "system_none_SS000022"] + assert list(_items(root).iterdir()) == [] + + +def test_회사_층_항목은_회사_폴더에서_복사(world): + root = world["root"] + company = world["storage"] / "3" / "templates" + _copy_layer(company) + combo = json.loads((company / layers.COMBO_FILE).read_text(encoding="utf-8")) + for row in combo["줄"]: + if row["키"] == "SS000022": + 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 total["줄"]["분류3"] == "회사 것" + # 채움 — 회사 항목은 시스템 열과 안 겹치는 새 열 + doc = fill.fill_table( + root, json.loads((world["system"] / "table/구조물집계표.json").read_text("utf-8")) + ) + ids = [c["id"] for c in doc["열"] if "@" in c["id"]] + assert ids == ["ms@company_3_SS000022|메쌓기|2.0"] + + +def test_원본에_없는_항목은_못만듦으로_남김(world): + gone = {**MS, "key": "SS999999"} + assert _save(world["root"], [_dry(gone)], [])["못만듦"] == ["system_none_SS999999"] + + +# ── 채움 ───────────────────────────────────────────── + + +def test_채움은_지정한_열에만_미지정은_안_셈(world): + root = world["root"] + _save(root, [_dry(MS), _wet()], [_pipe(PIPE), {**_pipe(None), "chainage_m": 60.0}]) + table = json.loads((world["system"] / "table/구조물집계표.json").read_text("utf-8")) + doc = fill.fill_table(root, table, {40.0: 12.0, 60.0: 9.0}) + values = {row["id"]: row["값"] for row in doc["줄"] if not row.get("고정")} + assert set(values) == {"s80.00~90.00", "s40.00"} # 미지정 옹벽 · 관은 줄도 없음 + assert values["s80.00~90.00"]["ms|메쌓기|2.0"] == 10 # 메쌓기는 더함 출처(종류)로 + assert values["s40.00"]["pp_len|파형강관|1000"] == 12 + assert not any("찰쌓기" in c["id"] for c in doc["열"]) # 자동 맞춤 없음 + assert "집계 항목 미지정 2건 — 표에 안 들어감" in doc["알림"] + + +def test_스냅샷이_없으면_알림(world): + root = world["root"] + _save(root, [_dry(MS)], []) + shutil.rmtree(_items(root)) + doc = fill.fill_table( + root, json.loads((world["system"] / "table/구조물집계표.json").read_text("utf-8")) + ) + assert "집계 항목 복사본 없음 1건 — [저장] 뒤 채워짐" in doc["알림"] + + +# ── 초기화 · 복제 ──────────────────────────────────── + + +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 포함)