feat(M02): 구조물 조합 사본 구멍 메움 — 관 옮김 이름만 · 실패 알림 · 알약 · 그룹 · 가져온이(142-2)

- sync_snapshots: 관 지점을 옮겨도 같은 항목 · 같은 그룹 꼬리면 폴더 이름만 바꿈(가까운 측점 먼저)
- safe_sync 못만듦을 구조물 · 관 지점 저장 응답(snapshot_failed)에 실음 · B05/B06 [저장] 알림 한 줄
- 사전조합 참조사본에 안 로직 알약(배 · 단위) 굳힘 · 사본 셈이 지금 M01 을 안 읽음
- 옵션 그룹 이름을 사본 manifest 에 굳힘 · 집계 채움이 사본 이름으로(정의가 바뀌거나 사라져도)
- manifest 가져온이(세션) · 사본 목록 부분 · 가져온이 · [개인에 저장]이 일위대가 문서도 가져감
- 시험: 관 옮김 → 같은 사본 · 못만듦 · 가져온이 · 그룹 사본 값 · 알약 굳힘 · 개인 저장 일위대가 · 저장 응답

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015sDZSjFvoiDjdGpgj3muc3
This commit is contained in:
2026-10-09 15:05:52 +09:00
co-authored by Claude Opus 5.5
parent f8b38d828b
commit b7beeb292a
15 changed files with 326 additions and 32 deletions
+23 -4
View File
@@ -17,6 +17,7 @@ import {
API_SURFACE_BUILD_TIMEOUT_MS,
API_TIMEOUT_MS,
} from "@config/config_frontend";
import { showToast } from "@ui/ui_template_elements";
import type { GroupItems, SummaryItem } from "../common_util/common_util_summary_item";
/** 지표면 분석 실행 요청 (SurfaceAnalyzeRequest) */
@@ -557,6 +558,22 @@ export interface DetailBasinResponse {
arrow_spacing_m: number;
/** 유역 안쪽 상류 세류망 — 하이라이트 토글용. */
upstream_lonlat: Array<Array<[number, number]>>;
/** [저장] 때 못 만든 구조물 조합 사본 폴더(142-2 · `*` = 통째 실패). */
snapshot_failed?: string[];
}
let lastNote = { text: "", at: 0 };
/** 못 만든 구조물 조합 사본(저장 응답 `snapshot_failed` · 142-2) — [저장] 알림 한 줄 ·
* 관 · 구조물 저장이 같은 맞춤을 두 번 알리면 한 번만. */
export function noteSnapshotFailed(failed: readonly string[] | undefined): void {
if (!failed?.length) return;
const text = failed.includes("*")
? "구조물 조합 사본 맞춤에 실패했습니다 — 다시 [저장]해 주세요."
: `구조물 조합 사본 ${failed.length}곳을 못 만들었습니다(원본 조합 없음) — 조합을 다시 골라 [저장]해 주세요.`;
const now = Date.now();
if (text === lastNote.text && now - lastNote.at < 5000) return;
lastNote = { text, at: now };
showToast(text, "error");
}
/** 저장된 관 매설 지점과 그 세부유역. 저장분이 없으면 백엔드가 자동 생성해 돌려준다. */
@@ -593,8 +610,10 @@ export async function saveDetailPipePoints(
projectId: string,
points: DetailPipeInput[],
): Promise<DetailBasinResponse> {
return requestJson<DetailBasinResponse>(`/projects/${projectId}/drainage/pipe-points`, {
method: "PUT",
body: JSON.stringify({ points }),
});
const saved = await requestJson<DetailBasinResponse>(
`/projects/${projectId}/drainage/pipe-points`,
{ method: "PUT", body: JSON.stringify({ points }) },
);
noteSnapshotFailed(saved.snapshot_failed);
return saved;
}
+10 -4
View File
@@ -20,11 +20,12 @@ import math
from typing import Any
from uuid import UUID
from fastapi import APIRouter
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir
from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
from common_util.common_util_auth import get_optional_session
from common_util.common_util_drainage_context import DrainageContext, load_drainage_context
from common_util.common_util_drainage_detail import DrainageDetail, build_detail
from common_util.common_util_drainage_pipes import (
@@ -403,7 +404,9 @@ async def delete_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
@router.put("/{project_id}/drainage/pipe-points", response_model=None)
async def put_pipe_points(
project_id: UUID, payload: dict[str, Any] | None = None
project_id: UUID,
payload: dict[str, Any] | None = None,
session: dict[str, Any] | None = Depends(get_optional_session), # 가져온이(142-2)
) -> dict[str, Any] | JSONResponse:
"""관 지점을 정본으로 확정하고 세부유역 산출물까지 함께 남긴다(모델 확정 시점)."""
resolved = await _resolve(project_id, payload, use_stored=True)
@@ -424,9 +427,12 @@ async def put_pipe_points(
)
# 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지)
from common_util.common_util_storage import resolve_stored_project_path
from M02_MasterTemplete.M02_Item_Snapshot import safe_sync
from M02_MasterTemplete.M02_Item_Snapshot import importer, safe_sync
await asyncio.to_thread(safe_sync, resolve_stored_project_path(context.stored_path))
failed = await asyncio.to_thread(
safe_sync, resolve_stored_project_path(context.stored_path), importer(session)
)
result = _payload(project_id, context, detail, points, saved=True)
result["saved_count"] = saved
result["snapshot_failed"] = failed # 못 만든 구조물 조합 복사본(142-2) — [저장] 알림
return result
+9 -4
View File
@@ -12,6 +12,7 @@
* ========================================================================== */
import { readState, writeState } from "../A00_Common/b_page_state";
import { noteSnapshotFailed } from "../B04_PreProcess/B04_PreProcess_Api_Fetch";
import type { GroupItems, SummaryItem } from "../common_util/common_util_summary_item";
import { API_BASE_URL, API_TIMEOUT_MS } from "@config/config_frontend";
import {
@@ -150,6 +151,8 @@ export interface StructureSaveResponse {
needs_downstream_invalidation: boolean;
/** 실제로 되돌렸는가. needs와 어긋나면 화면이 사용자에게 알린다. */
invalidated_downstream: boolean;
/** 못 만든 구조물 조합 사본 폴더(142-2). */
snapshot_failed?: string[];
}
/** 다른 창이 먼저 저장해 판번호가 어긋났다 — 화면이 최신본을 다시 받아야 한다. */
@@ -231,10 +234,12 @@ export async function saveStructures(
baseRevision: number,
structures: StructureInstance[],
): Promise<StructureSaveResponse> {
return requestJson<StructureSaveResponse>(`/projects/${projectId}/route/structures`, {
method: "PUT",
body: JSON.stringify({ base_revision: baseRevision, structures }),
});
const saved = await requestJson<StructureSaveResponse>(
`/projects/${projectId}/route/structures`,
{ method: "PUT", body: JSON.stringify({ base_revision: baseRevision, structures }) },
);
noteSnapshotFailed(saved.snapshot_failed);
return saved;
}
/** 구 비정규 측점(자유 텍스트)을 구조물 정본으로 이관한다. 멱등 — 배관은 관 지점
+8 -4
View File
@@ -14,7 +14,7 @@ from pathlib import Path
from typing import Any
from uuid import UUID
from fastapi import APIRouter
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
@@ -34,6 +34,7 @@ from B05_Profile.B05_Profile_Structures_Schema import (
StructureSaveRequest,
StructureSaveResponse,
)
from common_util.common_util_auth import get_optional_session
from common_util.common_util_drainage_pipes import (
PipePoint,
append_pipe_points_file,
@@ -264,7 +265,9 @@ async def migrate_structures(project_id: UUID, payload: StructureMigrateRequest)
@router.put("/{project_id}/route/structures", response_model=StructureSaveResponse)
async def write_structures(
project_id: UUID, payload: StructureSaveRequest
project_id: UUID,
payload: StructureSaveRequest,
session: dict[str, Any] | None = Depends(get_optional_session), # 가져온이(142-2)
) -> StructureSaveResponse | JSONResponse:
"""구조물 목록을 정본에 덮어쓴다(판번호 불일치 시 409, 타입 오류 시 400)."""
try:
@@ -279,9 +282,9 @@ async def write_structures(
max_chainage_m=await _route_length(project_id),
)
# 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지)
from M02_MasterTemplete.M02_Item_Snapshot import safe_sync
from M02_MasterTemplete.M02_Item_Snapshot import importer, safe_sync
await asyncio.to_thread(safe_sync, root)
failed = await asyncio.to_thread(safe_sync, root, importer(session))
# 설계에 영향을 주는 변경일 때만 B06 이후를 STALE로 돌린다 — 메모만 고쳐도
# 횡단·수량을 다시 돌리게 만들지 않기 위함이다. 응답 플래그는 실제로 STALE
# 전파가 **성공했을 때만** true(실패를 성공처럼 알리지 않는다).
@@ -293,6 +296,7 @@ async def write_structures(
count=len(payload.structures),
needs_downstream_invalidation=needs_invalidation,
invalidated_downstream=invalidated,
snapshot_failed=failed,
)
except LookupError:
return JSONResponse(status_code=404, content=_PROJECT_PATH_MISSING)
@@ -279,6 +279,8 @@ class StructureSaveResponse(BaseModel):
# 둘이 어긋나면(필요했는데 못 했다) 화면이 사용자에게 알린다.
needs_downstream_invalidation: bool = False
invalidated_downstream: bool = False
# 못 만든 구조물 조합 복사본 폴더(142-2) — 화면이 [저장] 알림 한 줄
snapshot_failed: list[str] = []
class StructureListResponse(BaseModel):
+4
View File
@@ -46,6 +46,7 @@ from M02_MasterTemplete.M02_Item_Source import (
)
#: 길의 종류 이름 → 복사본 파일
PARTS = ("도면", "면적", "일위대가") # 목록 `부분` — 구조물 조합 사본이 가진 것
KINDS = {
"도면": "도면.json",
"면적": "면적.json",
@@ -115,6 +116,9 @@ def list_items(project_root: str | Path) -> list[dict[str, Any]]:
"구조물": manifest.get("구조물") or [],
"고침": bool(manifest.get("고침")),
"값있음": bool(manifest.get("값있음")),
"가져온이": manifest.get("가져온이"),
# 부분(142-2) — 복사본에 있는 파일만(원본 대조 없이)
"부분": [name for name in PARTS if (child / KINDS[name]).is_file()],
}
)
return out
+8 -2
View File
@@ -1,7 +1,7 @@
"""프로젝트에서 개인 층으로 — [개인에 저장](save-as) · 프로젝트에서 새 항목(new) (PLAN 26-2b).
개인 층 새 항목 = 구조물집계표 새 열 하나 + 빈 도면 · 면적산출(번호 · SS 키는 저장소가
줌). [개인에 저장]은 구조물 복사본의 열 정의 · 도면 · 면적을 담음 —
줌). [개인에 저장]은 구조물 복사본의 열 정의 · 도면 · 면적 · 일위대가 문서를 담음 —
단가 묶음 · 입력값 · 고른자재는 프로젝트 몫이라 안 가져감. 쓰기는 지금 층 저장소 함수(`store.at` ·
`store.write`) — 나무 고침(`M02_MasterTemplete_Tree`)과
같은 집계표 저장 한 줄기(경로 · 머리 줄 수 맞춤 · 번호 · SS 장부)를 탐(PLAN 26-2a).
@@ -19,7 +19,7 @@ from M02_MasterTemplete import M02_MasterTemplete_Items as items_store
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_MasterTemplete_Tree as tree
from M02_MasterTemplete import M02_Template_Layers as layers
from M02_MasterTemplete.M02_Item_Source import read_json
from M02_MasterTemplete.M02_Item_Source import LABOR_FILE, read_json
#: 복사본 파일 → 층 양식 종류
PART_KINDS = (("도면.json", "structure"), ("면적.json", "area"))
@@ -89,6 +89,12 @@ def create_item(
body = store._server_values(body)
with store._LOCK:
atomic_write_json(store._path(kind, new_id), body)
labor = (read_json(copy / LABOR_FILE) or {}).get("줄") if copy else None
# 일위대가 문서도(142-2) — 옛 UA 조합 줄은 문서가 아니라 안 담음 · 번호 · 열은 새 열 것
# ponytail: 참조(단가승격 · 사전조합)는 키 그대로 — 개인 층이 못 읽는 참조는 계산 때
# 까닭으로 멈춤(참조사본까지 옮기려면 `Labor_Ref.carry` 길)
if isinstance(labor, dict) and labor.get("종류") == "일위대가":
store.copy_from("labor", new_id, new_id, body=labor)
ss = next(r for r in store.read_combos()["줄"] if r.get("열") == new_id)
return {"key": ss["키"], "열": new_id, "이름": tree.name_of(heads)}
+113 -7
View File
@@ -12,6 +12,7 @@
([재적용]은 사용자가)
- 지정을 다른 항목으로 바꾸면 새로 가져옴 · 구조물을 지우거나 지정을 풀면 복사본을 지움 —
프로젝트에서 고친 흔적(`manifest.고침`)이 있으면 지우지 않고 `_지난/` 에 옮겨 둠
- 관 지점을 옮겨도 같은 항목이면 폴더 이름만 바꿈(142-2) · 옵션 그룹 이름 · 가져온이도 굳힘
- 옛 항목 폴더(`templates/items/{층}_{주인}_{키}/` · 25-2b)는 그 항목을 쓰는 구조물마다 복사해
옮겨 받음(고친 값 유지)
"""
@@ -54,6 +55,8 @@ LOCAL_FILES = (VALUES_FILE, PICKS_FILE)
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:
@@ -221,16 +224,100 @@ def migrate_items(project_root: str | Path) -> list[str]:
return moved
def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
"""구조물 복사본을 지정에 맞춤 — `{만듦, 지움, 못만듦, 옮겨받음}` 폴더 이름."""
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:
@@ -239,12 +326,20 @@ def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
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, {})
write_snapshot(target, item, owner, extra)
except (LookupError, ValueError) as error:
logger.warning("집계 항목 복사본을 못 만듦 (%s): %s", folder, error)
failed.append(folder)
@@ -255,17 +350,28 @@ def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
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}
out = {"만듦": made, "지움": removed, "못만듦": failed, "옮겨받음": moved}
return {**out, "이름바꿈": renamed} if renamed else out
def safe_sync(project_root: str | Path) -> None:
"""저장 뒤 부름 — 실패해도 저장은 그대로(복사본은 다음 저장이 다시 맞춤)."""
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)
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:
+15 -2
View File
@@ -416,9 +416,11 @@ class Expand:
from M02_MasterTemplete import M02_Labor_Doc as labor
lines = [one for one in doc.get("담은로직") or [] if isinstance(one, dict)]
kept = doc.get("알약") # 프로젝트 참조사본 = 가져온 때 굳힌 알약(142-2) · 층 = 지금 M01
pills = kept if isinstance(kept, list) else None
by_logic = {
code: pill
for pill in combo_store.combo_pills({"담은로직": lines})["물량"]
for pill in pills or combo_store.combo_pills({"담은로직": lines})["물량"]
for code in pill["로직"]
}
inner, rows, why = [], [], None
@@ -472,13 +474,24 @@ def frozen(doc: dict[str, Any]) -> dict[str, dict[str, Any]]:
continue
got = read_ref(ref)
if got is not None:
out[ref_id(ref)] = got
out[ref_id(ref)] = with_pills(got)
walk(got.get("담은로직"), depth + 1)
walk(doc.get("담은로직"), 1)
return out
def with_pills(doc: dict[str, Any]) -> dict[str, Any]:
"""사전조합 굳힘 — 안 로직 줄 알약(배 · 단위)도 그때 M01 값으로(142-2 · 사본 셈이 지금 M01 을
안 읽게)."""
if doc.get("종류") != TITLE["precombo"]:
return doc
from M01_MasterData import M01_MasterData_Store_Combo as combo_store
lines = [one for one in doc.get("담은로직") or [] if isinstance(one, dict)]
return {**doc, "알약": combo_store.combo_pills({"담은로직": lines})["물량"]}
def frozen_logics(copies: dict[str, Any]) -> list[str]:
"""굳힌 단가승격 · 사전조합이 담은 로직 키 — 프로젝트 단가 묶음에 넣음."""
return [
+7 -5
View File
@@ -140,14 +140,15 @@ def collect_items(
"group_items": point.group_items,
}
)
return _with_group_items(items)
return _with_group_items(items, project_root)
def _with_group_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
def _with_group_items(items: list[dict[str, Any]], project_root=None) -> list[dict[str, Any]]:
"""옵션 그룹(M02 PLAN 46)마다 한 건 더 — 같은 구조물 · 옵션 · 자리에 그 그룹이 고른 항목 ·
복사본 `{구조물 id}__{그룹 id}`. 정의된 그룹만(미지정 그룹 = 「미지정」 알림에 셈) · 정의가 사라진
그룹 연결은 기록에만 남고 표에 안 들어감."""
복사본 `{구조물 id}__{그룹 id}`. 그룹 = 지금 정의 + 복사본에 굳힌 정의(142-2 · 이름 · 사라진
그룹도) · 미지정 그룹 = 「미지정」 알림에 셈."""
from common_util.common_util_summary_item import GROUP_JOIN
from M02_MasterTemplete.M02_Item_Snapshot import copied_groups
from M02_MasterTemplete.M02_Table_Fill_Spec import container_rows
from M02_MasterTemplete.StructOptions.M02_SO_Groups import groups_of
@@ -156,7 +157,8 @@ def _with_group_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
for item in items:
links = item.pop("group_items", None) or {}
out.append(item)
for group in defined.get(item["type_id"], []):
mine = defined.get(item["type_id"], [])
for group in copied_groups(project_root, item["복사본"], links, mine):
out.append(
{
**item,
@@ -83,6 +83,17 @@ def test_save_then_read_roundtrip(client):
assert read.json()["structures"][0]["structure_id"]
def test_save_response_carries_snapshot_failed(client, monkeypatch):
"""142-2 — 못 만든 구조물 조합 사본을 저장 응답에 실음(화면 [저장] 알림) · 로그인 세션 없으면
가져온이 없음."""
from M02_MasterTemplete import M02_Item_Snapshot as snap
seen = []
monkeypatch.setattr(snap, "safe_sync", lambda root, who: seen.append(who) or ["st9"])
saved = client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
assert saved.json()["snapshot_failed"] == ["st9"] and seen == [None]
def test_stale_revision_returns_409(client):
client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
conflict = client.put(
@@ -130,6 +130,8 @@ def test_지정하면_새_파일까지_복사(world):
"구조물": ["pipe@40.0"],
"고침": False,
"값있음": False,
"가져온이": None,
"부분": ["도면", "면적", "일위대가"], # 142-2 — 사본이 가진 부분
}
]
@@ -444,6 +446,19 @@ def test_개인에_저장하면_개인_층에_항목(world):
assert _save_all(root, [_wall("b1", made["summary_item"])], [])["만듦"] == ["b1"]
def test_개인에_저장은_일위대가_문서도_가져감_142_2(world):
root = world["root"]
_save(root, [_pipe()])
copy = root / "templates/structures" / FOLDER
doc = {"종류": "일위대가", "열": "old", "단위": "m", "담은로직": [{"로직": "GF000176"}]}
labor = {**_read(copy / "일위대가.json"), "줄": doc}
(copy / "일위대가.json").write_text(json.dumps(labor, ensure_ascii=False), "utf-8")
made = copies.save_as(root, FOLDER, 3, 12, ["관공"], "문서 담음")
got = _read(layers.personal_dir(3, 12) / "labor" / f"{made['열']}.json")
assert got["열"] == made["열"] and got["일위대가번"] # 번호 · 열은 새 열 것
assert [line["로직"] for line in got["담은로직"]] == ["GF000176"]
def test_프로젝트에서_새_항목(world):
root = world["root"]
_save_all(root, [_wall("c1")], [])
+25
View File
@@ -434,3 +434,28 @@ def test_서고_구분_상세구분_연결_승격은_원_로직_사전조합은_
refs.summaries(world["system"], rows)
assert (rows[0]["구분"], rows[0]["상세구분"]) == (row["구분"], row["상세구분"])
assert (rows[1]["구분"], rows[1]["상세구분"], rows[1]["연결"]) == ("가", "나", None)
def test_사전조합_굳힘은_안_로직_알약도_사본_값_142_2(world, monkeypatch):
"""프로젝트 참조사본 사전조합 = 가져온 때 알약(배 10) — 그 뒤 M01 이 바뀌어도 사본 셈 그대로 ·
층 문서(알약 없음)는 지금 M01 을 읽음."""
from M01_MasterData import M01_MasterData_Store_Combo as combo_store
from M02_MasterTemplete import M02_Item_Prices as prices
from M02_MasterTemplete import M02_Item_Source as source
sc = _combo(world["system"], [{"로직": TEN}], "열배")
ref = {**SYS, "키": sc}
with store.at(world["system"]):
doc = labor.clean({"열": "rv", "담은로직": [{"참조": ref, "수량": {"손": 1}}]})
rows = source.labor_doc_rows(doc)
kept = rows["참조사본"][refs.ref_id(ref)]["알약"]
assert [(p["로직"], p.get("배")) for p in kept] == [([TEN], 10)]
assert "알약" not in _read(world["system"], "precombo", sc) # 층 문서는 그대로
bundle = prices.to_json(prices.collect([TEN]))
before = prices.calc(rows, bundle, {}, {}, None)["줄"][0]["계"]
changed = {"물량": [{"키": "x|㎡", "이름": "x", "단위": "㎡", "로직": [TEN]}]} # 배 1 로 바뀜
monkeypatch.setattr(combo_store, "combo_pills", lambda row: changed)
assert prices.calc(rows, bundle, {}, {}, None)["줄"][0]["계"] == pytest.approx(before)
with store.at(world["system"]):
live = labor.calc(doc, {})["줄"][0]["기준"]["계"]
assert live == pytest.approx(before * 10) # 층 셈은 지금 M01(배 1)
@@ -95,3 +95,30 @@ def test_채움은_정의된_그룹마다_한_건_더(world, monkeypatch):
items = fill.collect_items(root)
assert [i["복사본"] for i in items] == ["st0", "st0__g1", "st1", "st1__g1"]
assert items[1]["그룹"] == "기슭막이" and items[1]["summary_item"] == RV
def test_그룹_정의도_사본_값_정의가_바뀌거나_사라져도_그대로(world, monkeypatch):
"""142-2 — 그룹 연결 사본 manifest `그룹` = 가져온 때 정의(이름) · 이름을 바꾸거나 그룹을
지워도 프로젝트 집계 줄은 사본 이름 그대로 · 연결 없는 구조물은 지금 정의만."""
root, system = world["root"], world["system"]
_define(system, monkeypatch, {"masonry_dry": [{"id": "g1", "이름": "기슭막이", "옵션": []}]})
linked = {**_dry(MS), "group_items": {"g1": RV}}
bare = {**_dry(None), "chainage_m": 125, "start_m": 120, "end_m": 130}
_save(root, [linked, bare], [])
manifest = json.loads((_items(root) / "st0__g1/manifest.json").read_text("utf-8"))
assert manifest["그룹"] == {"id": "g1", "name": "기슭막이"}
_define(system, monkeypatch, {"masonry_dry": [{"id": "g1", "이름": "바뀐 이름", "옵션": []}]})
items = fill.collect_items(root)
assert [(i["복사본"], i.get("그룹")) for i in items] == [
("st0", None),
("st0__g1", "기슭막이"),
("st1", None),
("st1__g1", "바뀐 이름"),
]
_define(system, monkeypatch, {})
items = fill.collect_items(root)
assert [(i["복사본"], i.get("그룹")) for i in items] == [
("st0", None),
("st0__g1", "기슭막이"),
("st1", None),
]
@@ -218,6 +218,55 @@ def test_원본에_없는_항목은_못만듦으로_남김(world):
assert _save(world["root"], [_dry(gone)], [])["못만듦"] == ["st0"]
def _manifest(root: Path, folder: str) -> dict[str, Any]:
return json.loads((_items(root) / folder / "manifest.json").read_text("utf-8"))
def test_관_옮겨_저장해도_같은_사본_이름만_바꿈(world):
"""142-2 — 같은 항목 관을 옮기면 폴더 이름만(가져온날 · 고침 · 입력값 그대로 · 원본을 다시
안 읽음) · 같은 항목 관 둘 = 가까운 측점끼리 · 다른 항목 자리는 새로."""
root, system = world["root"], world["system"]
_save(root, [], [_pipe(PIPE), {**_pipe(PIPE), "chainage_m": 100.0}])
for folder in ("pipe@40.0", "pipe@100.0"):
manifest = {**_manifest(root, folder), "고침": True, "가져온날": f"옛날-{folder}"}
(_items(root) / folder / "manifest.json").write_text(json.dumps(manifest), "utf-8")
(_items(root) / "pipe@40.0/입력값.json").write_text('{"L": {"x": {"값": 7}}}', "utf-8")
(system / layers.COMBO_FILE).unlink() # 원본이 사라져도 이름만 바꾸므로 못만듦 없음
moved = [{**_pipe(PIPE), "chainage_m": 98.0}, {**_pipe(PIPE), "chainage_m": 41.5}]
result = _save(root, [], moved)
assert sorted(result["이름바꿈"]) == ["pipe@100.0→pipe@98.0", "pipe@40.0→pipe@41.5"]
assert result["만듦"] == result["지움"] == result["못만듦"] == []
assert sorted(p.name for p in _items(root).iterdir()) == ["pipe@41.5", "pipe@98.0"]
near = _manifest(root, "pipe@41.5")
assert near["가져온날"] == "옛날-pipe@40.0" and near["고침"] is True
assert near["구조물"] == ["pipe@41.5"]
assert json.loads((_items(root) / "pipe@41.5/입력값.json").read_text("utf-8"))["L"]["x"]
assert _manifest(root, "pipe@98.0")["가져온날"] == "옛날-pipe@100.0"
def test_관_옮김_다른_항목이면_이름_안_바꿈(world):
root = world["root"]
_save(root, [], [_pipe(PIPE)])
result = _save(root, [], [{**_pipe(MS), "chainage_m": 41.5}])
assert "이름바꿈" not in result
assert result["만듦"] == ["pipe@41.5"] and result["지움"] == ["pipe@40.0"]
assert _manifest(root, "pipe@41.5")["원본"]["key"] == MS["key"]
def test_못만듦은_저장_응답으로_가져온이는_manifest_에(world):
"""142-2 — `safe_sync` 가 못 만든 폴더를 돌려줌(저장 응답 → [저장] 알림) · 가져온이 =
저장한 세션 · 미리 가져온 사본(가져온이 없음)에도 [저장] 때 채움."""
root = world["root"]
_save(root, [_dry({**MS, "key": "SS999999"}), _dry(MS)], [])
who = snap.importer({"user_id": 42, "name": "검증"})
assert who == {"id": 42, "이름": "검증"} and snap.importer(None) is None
assert snap.safe_sync(root, who) == ["st0"]
assert _manifest(root, "st1").get("가져온이") == who # 앞 저장(가져온이 없음) 사본도 채움
shutil.rmtree(_items(root) / "st1")
snap.safe_sync(root, who)
assert _manifest(root, "st1")["가져온이"] == who
# ── 채움 ─────────────────────────────────────────────