eomsangdonandClaude Opus 5.5
da8bedf115
- test_120 충돌: 물넘이 포장 chute(136-2) + 사본 잇기(143-1) 둘 다 살림 · 도수로/돌붙임 잇기 추가 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015sDZSjFvoiDjdGpgj3muc3
447 lines
20 KiB
Python
447 lines
20 KiB
Python
"""구조물 집계 항목 복사본 — 프로젝트 `templates/structures/{구조물 id}/` (PLAN 25-2b · 26-2b).
|
|
|
|
복사본은 **구조물마다 하나** — 같은 항목을 고른 구조물 둘도 각자 도면 · 면적 ·
|
|
일위대가 · 입력값 · 고른자재 · 단가를 가짐(길이 · 소재 · 높이 · 옵션이 구조물마다 다름).
|
|
폴더 이름 = 구조물 id(`structure_id`) · 관 지점은 `pipe@측점`.
|
|
|
|
[저장](`PUT /route/structures` · `PUT /drainage/pipe-points`) 뒤 서버가 맞춤:
|
|
- 항목을 지정한 구조물마다 원본 층에서 내용을 복사(`M02_Item_Source.build_source`) + 프로젝트 값
|
|
`입력값.json`(첫 값 = 일위대가 문서 입력값 · 확인 거짓 · 자동값은 안 적음) · `고른자재.json` ·
|
|
`manifest.json`
|
|
- 이미 있는 복사본은 다시 쓰지 않음 — 원본(마스터 · 회사 · 개인)이 바뀌거나 지워져도 그대로
|
|
([재적용]은 사용자가)
|
|
- 지정을 다른 항목으로 바꾸면 새로 가져옴 · 구조물을 지우거나 지정을 풀면 복사본을 지움 —
|
|
프로젝트에서 고친 흔적(`manifest.고침`)이 있으면 지우지 않고 `_지난/` 에 옮겨 둠
|
|
- 관 지점을 옮겨도 같은 항목이면 폴더 이름만 바꿈(142-2) · 옵션 그룹 이름 · 가져온이도 굳힘
|
|
- 옛 항목 폴더(`templates/items/{층}_{주인}_{키}/` · 25-2b)는 그 항목을 쓰는 구조물마다 복사해
|
|
옮겨 받음(고친 값 유지)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
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 B05_Profile.B05_Profile_Structures_Schema import STRUCTURE_ID_RULE
|
|
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 GROUP_JOIN, snapshot_folder_name
|
|
from M02_MasterTemplete import M02_Template_Layers as layers
|
|
from M02_MasterTemplete.M02_Item_Source import (
|
|
LABOR_FILE,
|
|
PRICE_FILE,
|
|
build_source,
|
|
item_path,
|
|
read_json,
|
|
version_of,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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@측점` · 옵션 그룹(PLAN 46)은
|
|
#: 뒤에 `__{그룹 id}`
|
|
FOLDER_PATTERN = re.compile(
|
|
rf"^(?:pipe@\d+(?:\.\d+)?(?:__g[0-9a-z]{{1,16}})?|{STRUCTURE_ID_RULE})$"
|
|
)
|
|
#: 관 지점 복사본 — (측점, 그룹 꼬리)
|
|
PIPE_FOLDER = re.compile(r"^pipe@(\d+(?:\.\d+)?)((?:__g[0-9a-z]{1,16})?)$")
|
|
|
|
|
|
def structures_dir(project_root: str | Path) -> Path:
|
|
return layers.project_dir(project_root) / STRUCTURES_DIRNAME
|
|
|
|
|
|
def items_dir(project_root: str | Path) -> Path:
|
|
"""옛 항목 폴더 뿌리(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 safe_folder(base: Path, name: str) -> Path | None:
|
|
"""`base` 바로 아래 복사본 폴더 — 이름이 규칙 밖 · `base` 밖이면 None(지우기 · 쓰기 막음)."""
|
|
target = base / name
|
|
if not FOLDER_PATTERN.match(name) or target.resolve().parent != base.resolve():
|
|
return None
|
|
return target
|
|
|
|
|
|
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 seed_values(labor: Any) -> dict[str, Any]:
|
|
"""일위대가 문서 입력값 → 프로젝트 입력값 첫 값 `{로직: {입력: {값, 확인: false}}}`(39-1c)."""
|
|
doc = labor.get("줄") if isinstance(labor, dict) else None
|
|
out: dict[str, Any] = {}
|
|
for line in (doc or {}).get("담은로직") or []:
|
|
given = line.get("입력값") if isinstance(line, dict) else None
|
|
if isinstance(given, dict) and given and line.get("로직"):
|
|
out[str(line["로직"])] = {k: {"값": v, "확인": False} for k, v in given.items()}
|
|
return out
|
|
|
|
|
|
def seed_picks(labor: Any) -> dict[str, Any]:
|
|
"""일위대가 문서 고름 → 프로젝트 고른자재 첫 값 `{로직: {호표 차례: MT키}}`(103-1)."""
|
|
doc = labor.get("줄") if isinstance(labor, dict) else None
|
|
return {
|
|
str(line["로직"]): dict(line["고름"])
|
|
for line in (doc or {}).get("담은로직") or []
|
|
if isinstance(line, dict) and line.get("로직") and isinstance(line.get("고름"), dict)
|
|
}
|
|
|
|
|
|
def build_snapshot(item: dict[str, Any]) -> dict[str, Any]:
|
|
"""원본 내용 + 프로젝트 값(입력값 · 고른자재 첫 값 = 일위대가 문서 것) — `{파일 이름: 내용}` ·
|
|
SS 줄이 없으면 LookupError."""
|
|
files = build_source(item)
|
|
labor = files.get(LABOR_FILE)
|
|
return {**files, VALUES_FILE: seed_values(labor), PICKS_FILE: seed_picks(labor)}
|
|
|
|
|
|
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)
|
|
stamp = now_text()
|
|
source = version_of(source_files(files))
|
|
manifest = {
|
|
"원본": origin_of(item),
|
|
"판": source,
|
|
"원본판": source,
|
|
"적용일": stamp,
|
|
"가져온날": stamp,
|
|
"구조물": [owner_id],
|
|
**extra,
|
|
}
|
|
atomic_write_json(target / MANIFEST_NAME, manifest)
|
|
return manifest
|
|
|
|
|
|
def assignments(project_root: str | Path) -> dict[str, dict[str, Any]]:
|
|
"""`{구조물 id: 항목}` — 구조물 정본 + 관 지점 정본(id = `pipe@측점`) · 지정한 것만 ·
|
|
옵션 그룹(PLAN 46)에 고른 항목은 `{구조물 id}__{그룹 id}` 로 따로(복사본도 따로)."""
|
|
found: dict[str, dict[str, Any]] = {}
|
|
_, structures = load_structures(str(project_root))
|
|
owners: list[tuple[str, Any]] = [
|
|
(s.structure_id or f"#{index}", s) for index, s in enumerate(structures)
|
|
]
|
|
owners += [
|
|
(f"pipe@{round(float(p.chainage_m), 3)}", p)
|
|
for p in read_pipe_points_file(pipe_points_path_in(Path(project_root)))
|
|
]
|
|
for owner, record in owners:
|
|
if record.summary_item:
|
|
found[owner] = record.summary_item
|
|
for gid, item in (record.group_items or {}).items():
|
|
found[f"{owner}{GROUP_JOIN}{gid}"] = item
|
|
return found
|
|
|
|
|
|
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 = safe_folder(base, folder_name(owner))
|
|
if target is None or (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 rename_moved(base: Path, wanted: dict[str, tuple[str, dict[str, Any]]]) -> list[str]:
|
|
"""관 지점을 옮긴 뒤 [저장] — 새 자리(`pipe@새 측점`)에 복사본이 없고 안 쓰게 된 옛 자리
|
|
복사본이 같은 항목 · 같은 그룹 꼬리면 폴더 이름만 바꿈(가까운 측점 먼저 · 142-2) — 가져온날 ·
|
|
고침 · 입력값 그대로. 바꾼 것 `옛 이름→새 이름`."""
|
|
if not base.is_dir():
|
|
return []
|
|
|
|
def spot(name: str) -> tuple[float, str] | None:
|
|
hit = PIPE_FOLDER.match(name)
|
|
return (float(hit[1]), hit[2]) if hit else None
|
|
|
|
new = [
|
|
(name, at, item)
|
|
for name, (_, item) in wanted.items()
|
|
if (at := spot(name)) and not (base / name / MANIFEST_NAME).is_file()
|
|
]
|
|
old = [
|
|
(child.name, at, manifest)
|
|
for child in base.iterdir()
|
|
if child.is_dir()
|
|
and child.name not in wanted
|
|
and (at := spot(child.name))
|
|
and isinstance(manifest := read_json(child / MANIFEST_NAME), dict)
|
|
]
|
|
# ponytail: 같은 항목 관 둘을 함께 옮기면 가까운 측점끼리 짝 — 고침 · 입력값이 다를 때만 엇갈림
|
|
pairs = sorted(
|
|
(abs(a[0] - b[0]), new_name, old_name)
|
|
for new_name, a, item in new
|
|
for old_name, b, manifest in old
|
|
if a[1] == b[1] and same_item(manifest, item)
|
|
)
|
|
used: set[str] = set()
|
|
out: list[str] = []
|
|
for _, new_name, old_name in pairs:
|
|
target = safe_folder(base, new_name)
|
|
if new_name in used or old_name in used or target is None:
|
|
continue
|
|
if target.is_dir(): # 끊긴 쓰기(manifest 없음) 자리
|
|
retire(target)
|
|
(base / old_name).rename(target)
|
|
manifest = read_json(target / MANIFEST_NAME) or {}
|
|
atomic_write_json(target / MANIFEST_NAME, {**manifest, "구조물": [wanted[new_name][0]]})
|
|
used.update((new_name, old_name))
|
|
out.append(f"{old_name}→{new_name}")
|
|
return out
|
|
|
|
|
|
def group_names() -> dict[str, str]:
|
|
"""옵션 그룹 id → 이름(지금 정의) — 복사본 manifest `그룹` 으로 굳힘(142-2)."""
|
|
from M02_MasterTemplete.StructOptions.M02_SO_Groups import groups_of
|
|
|
|
return {g["id"]: g["name"] for rows in groups_of().values() for g in rows}
|
|
|
|
|
|
def copied_groups(
|
|
project_root: str | Path | None,
|
|
owner: str,
|
|
links: dict[str, Any],
|
|
defined: list[dict[str, Any]],
|
|
) -> list[dict[str, Any]]:
|
|
"""그 구조물의 옵션 그룹 — 지금 정의 + 복사본에 굳힌 정의(이름 · 사라진 그룹도 · 142-2 ·
|
|
층이 바뀌어도 프로젝트 그대로). 굳힌 것 없으면 지금 정의 그대로."""
|
|
if project_root is None:
|
|
return defined
|
|
base = structures_dir(project_root)
|
|
ids = {g["id"] for g in defined}
|
|
out = []
|
|
for group in [*defined, *({"id": gid} for gid in links if gid not in ids)]:
|
|
manifest = read_json(
|
|
base / folder_name(f"{owner}{GROUP_JOIN}{group['id']}") / MANIFEST_NAME
|
|
)
|
|
kept = manifest.get("그룹") if isinstance(manifest, dict) else None
|
|
if isinstance(kept, dict) and kept.get("name"):
|
|
out.append({**group, "name": kept["name"]})
|
|
elif group["id"] in ids:
|
|
out.append(group)
|
|
return out
|
|
|
|
|
|
def sync_snapshots(
|
|
project_root: str | Path, who: dict[str, Any] | None = None
|
|
) -> dict[str, list[str]]:
|
|
"""구조물 복사본을 지정에 맞춤 — `{만듦, 지움, 못만듦, 옮겨받음, 이름바꿈}` 폴더 이름 ·
|
|
`who` = 가져온이(manifest · 142-2)."""
|
|
moved = migrate_items(project_root)
|
|
base = structures_dir(project_root)
|
|
wanted = {
|
|
folder_name(owner): (owner, item) for owner, item in assignments(project_root).items()
|
|
}
|
|
renamed = rename_moved(base, wanted)
|
|
made: list[str] = []
|
|
failed: list[str] = []
|
|
removed: list[str] = []
|
|
names: dict[str, str] | None = None
|
|
for folder, (owner, item) in sorted(wanted.items()):
|
|
target = safe_folder(base, folder)
|
|
if target is None:
|
|
logger.warning("복사본 폴더 이름이 규칙 밖 — 건너뜀: %r", folder)
|
|
failed.append(folder)
|
|
continue
|
|
manifest = read_json(target / MANIFEST_NAME)
|
|
if isinstance(manifest, dict) and same_item(manifest, item):
|
|
if who and not manifest.get("가져온이"): # 미리 가져온 것 — 저장한 이가 가져온이
|
|
atomic_write_json(target / MANIFEST_NAME, {**manifest, "가져온이": who})
|
|
continue
|
|
if target.is_dir(): # 다른 항목으로 바뀜 — 옛 것은 지우거나(고침이면) `_지난` 에
|
|
retire(target)
|
|
removed.append(folder)
|
|
extra: dict[str, Any] = {"가져온이": who} if who else {}
|
|
gid = owner.partition(GROUP_JOIN)[2]
|
|
if gid:
|
|
names = group_names() if names is None else names
|
|
if gid in names:
|
|
extra["그룹"] = {"id": gid, "name": names[gid]}
|
|
try:
|
|
write_snapshot(target, item, owner, extra)
|
|
except (LookupError, ValueError) as error:
|
|
logger.warning("집계 항목 복사본을 못 만듦 (%s): %s", folder, error)
|
|
failed.append(folder)
|
|
continue
|
|
made.append(folder)
|
|
if base.is_dir():
|
|
for child in sorted(base.iterdir()):
|
|
if child.is_dir() and not child.name.startswith("_") and child.name not in wanted:
|
|
retire(child)
|
|
removed.append(child.name)
|
|
out = {"만듦": made, "지움": removed, "못만듦": failed, "옮겨받음": moved}
|
|
return {**out, "이름바꿈": renamed} if renamed else out
|
|
|
|
|
|
def importer(session: dict[str, Any] | None) -> dict[str, Any] | None:
|
|
"""세션 → manifest 가져온이 `{id, 이름}`(142-2)."""
|
|
if not isinstance(session, dict) or session.get("user_id") is None:
|
|
return None
|
|
return {"id": int(session["user_id"]), "이름": session.get("name")}
|
|
|
|
|
|
def safe_sync(project_root: str | Path, who: dict[str, Any] | None = None) -> list[str]:
|
|
"""저장 뒤 부름 — 실패해도 저장은 그대로(복사본은 다음 저장이 다시 맞춤) · 못 만든 복사본
|
|
폴더 이름을 돌려줌(저장 응답 → [저장] 알림 · 142-2 · 통째 실패 = `["*"]`)."""
|
|
try:
|
|
result = sync_snapshots(project_root, who)
|
|
if any(result.values()):
|
|
logger.info("집계 항목 복사본: %s", result)
|
|
return result["못만듦"]
|
|
except Exception:
|
|
logger.exception("집계 항목 복사본 맞춤 실패 (저장은 유지): %s", project_root)
|
|
return ["*"]
|
|
|
|
|
|
def complete_old(folder: Path) -> bool:
|
|
"""옛 복사본(21-3 · 새 파일 없음)을 채움 — 단가는 지금 원본에서 · 값은 빈 것.
|
|
|
|
옛 내용 파일(집계 · 도면 …)은 그대로 둠. 채웠으면 True(manifest 에 `옮겨받음` 날)."""
|
|
manifest = read_json(folder / MANIFEST_NAME)
|
|
if not isinstance(manifest, dict):
|
|
return False
|
|
lacking = [
|
|
name for name in (PRICE_FILE, VALUES_FILE, PICKS_FILE) if not (folder / name).is_file()
|
|
]
|
|
if not lacking and "원본판" in manifest:
|
|
return False
|
|
need_source = 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, {}))
|
|
kept = {
|
|
name: read_json(folder / name)
|
|
for name in sorted(p.name for p in folder.glob("*.json"))
|
|
if name not in (MANIFEST_NAME, *LOCAL_FILES)
|
|
}
|
|
atomic_write_json(
|
|
folder / MANIFEST_NAME,
|
|
{
|
|
**manifest,
|
|
"가져온날": manifest.get("가져온날") or manifest.get("적용일"),
|
|
"원본판": manifest.get("원본판") or version_of(kept),
|
|
"옮겨받음": manifest.get("옮겨받음") or now_text(),
|
|
},
|
|
)
|
|
return True
|
|
|
|
|
|
#: 「연결된 사본」 부분 이름 → 파일(있는 것만 `부분` 에 담음)
|
|
LINKED_PARTS = (("도면", "도면.json"), ("면적", "면적.json"), ("일위대가", LABOR_FILE))
|
|
|
|
|
|
def linked_snapshots(project_root: str | Path) -> dict[str, dict[str, Any]]:
|
|
"""연결된 사본 — 지정 ∩ 같은 항목 사본 `{구조물 id: {항목, 폴더, 부분, 경로}}`(PLAN 143-1).
|
|
|
|
구조물 id = `assignments` 키(관 지점 `pipe@측점` · 옵션 그룹 `{id}__{그룹}`) ·
|
|
미지정 · 사본 없음 · 다른 항목 사본은 안 담음(M02 `assign_items` 규칙) ·
|
|
원본 층과 대조 안 함(빠른 읽기 · 쓰기 없음) ·
|
|
`경로` = manifest 원본 경로(가져온 때 M02 구분 · 이름) · B08 · B09 공용.
|
|
"""
|
|
base = structures_dir(project_root)
|
|
found: dict[str, dict[str, Any]] = {}
|
|
for owner, item in assignments(project_root).items():
|
|
folder = base / folder_name(owner)
|
|
manifest = read_json(folder / MANIFEST_NAME)
|
|
if not same_item(manifest, item):
|
|
continue
|
|
found[owner] = {
|
|
"항목": item,
|
|
"폴더": folder.name,
|
|
"부분": [part for part, name in LINKED_PARTS if (folder / name).is_file()],
|
|
"경로": list(manifest["원본"].get("경로") or []),
|
|
}
|
|
return found
|
|
|
|
|
|
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
|