"""M02 서고 나무 지우기 = 경고 뒤 지움 · 쓰는 곳 그 줄도 지움(PLAN 735-1 · 요구 574) — 일위대가(`precombo` SC) · 단가산출(`sangeun` SK) 한 벌. 미리 보기 `POST /layers/{층}/{kind}-tree/delete-check {대상}` → `{문서, 쓰는곳: [줄]}` — 줄 = {층, 주인, 어디, 종류, 문서, 이름, 키, 줄, 쓰기} · 종류 = 구조물 조합 · 일위대가 · 단가산출 · 줄 = 담은로직 「N째」 · 「더한줄 N째」 · 「바탕」 · 모든 층 훑음(프로젝트 뺌 · 지우는 것끼리 뺌). 지우기(나무 고침 `op: delete` · `M02_Precombo_Folders.edit`) — 권한 밖 쓰는 곳 = 409 `{까닭, 막힘}` · 쓰는 곳이 있는데 `확인` 없음 = 409 `{까닭, 쓰는곳}` · 확인 = 쓰는 문서의 그 줄 지움(`store.write` = 서버 정본 다시 셈) → 문서 지움 · 기록 `{층}/_합친것/<첫 키>.json`(733-3 자리 · `지움` = 지운 문서 · 자리 · `고친곳` = 고친 문서 앞판 · 뒤판) · 못 쓰면 앞판으로 되돌리고 409. 되돌림(`undo` · `M02_Library_Same_Router` undo 길이 부름) = 문서 · 자리 · 고친 문서 앞판(그새 안 고친 것만). 프로젝트 사본 · B09 공사 = 굳힌 값이라 안 건드림. 바탕 지움(단가산출) = 바탕 없음 + 바탕 줄 덮개 · 뺀줄 · 고름 · 자재표시 · 열린입력 걷음 · 더한줄 지움 = 뒤 `add:#k` 열쇠 당김. """ from __future__ import annotations import json import re from datetime import datetime from pathlib import Path from typing import Any from fastapi import HTTPException from common_util.common_util_json import atomic_write_json from M02_MasterTemplete import M02_Labor_Ref as refs from M02_MasterTemplete import M02_Layer_Access as access 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 GONE = "_합친것" USER_KINDS = (("labor", "구조물 조합"), ("precombo", "일위대가"), ("sangeun", "단가산출")) _ADD = re.compile(r"^add:#(\d+)(.*)$") _PATCH = ("덮개", "고름", "자재표시") def can_write(session: dict[str, Any] | None, at: refs.Here) -> bool: if session is None: return False try: return access.grant(session, at[0], at[1]).write except HTTPException: return False def _hits(doc: dict[str, Any], rids: dict[str, str]) -> list[tuple[str, str]]: """문서 안 지울 참조 줄 — [(줄, 키)].""" out = [] for field, head in (("담은로직", ""), ("더한줄", "더한줄 ")): for at, one in enumerate(doc.get(field) or [], 1): ref = one.get("참조") if isinstance(one, dict) else None if isinstance(ref, dict) and refs.ref_id(ref) in rids: out.append((f"{head}{at}째", rids[refs.ref_id(ref)])) base = doc.get("바탕") if isinstance(base, dict) and isinstance(base.get("참조"), dict): if refs.ref_id(base["참조"]) in rids: out.append(("바탕", rids[refs.ref_id(base["참조"])])) return out def _columns(folder: Path, memo: dict[Path, dict[str, str]]) -> dict[str, str]: """층 구조물 열 id → 구조물 이름(구조물 조합 줄 이름 · 층마다 한 번).""" if folder not in memo: try: with store.at(folder): cols = store._structure_columns() memo[folder] = {c["id"]: tree.name_of(tree.path_of(c)) for c in cols} except (store.StoreError, OSError, ValueError): memo[folder] = {} return memo[folder] def users(session: dict[str, Any] | None, kind: str, gone: list[str]) -> list[dict[str, Any]]: """지금 층(`store.at`) 지울 문서를 쓰는 줄 — 모든 층(프로젝트 뺌 · 지우는 것끼리 뺌).""" mine = refs.here() rids = {refs.ref_id({"층": mine[0], "주인": mine[1], "키": k}): k for k in gone} skip = {(store.root() / kind / f"{k}.json").resolve() for k in gone} rows, memo = [], {} for ukind, what in USER_KINDS: for where, path in layers.layer_files(ukind): if path.resolve() in skip: continue try: doc = json.loads(path.read_bytes()) except (OSError, ValueError): continue hits = _hits(doc, rids) if isinstance(doc, dict) else [] if not hits: continue at = refs.where(path.parent.parent) ok = can_write(session, at) name = doc.get("일위대가번") or doc.get("이름") or path.stem if ukind == "labor" and (title := _columns(path.parent.parent, memo).get(path.stem)): name = f"{title} · {name}" for line, key in hits: rows.append( { "층": at[0], "주인": at[1], "어디": where, "종류": what, "문서종류": ukind, "문서": path.stem, "이름": name, "키": key, "줄": line, "쓰기": ok, } ) return rows def _shift(key: str, dropped: list[int]) -> str | None: """`add:#k…` 열쇠 — 지운 더한줄이면 None · 뒤 줄은 당김 · 그 밖 그대로.""" hit = _ADD.match(key) if not hit: return key k = int(hit[1]) if k in dropped: return None return f"add:#{k - sum(1 for d in dropped if d < k)}{hit[2]}" def strip(doc: dict[str, Any], rids: set[str]) -> dict[str, Any]: """문서에서 `rids` 참조 줄을 뺌(순수) — 담은로직 · 더한줄(뒤 `add:#k` 당김) · 바탕.""" def hit(one: Any) -> bool: ref = one.get("참조") if isinstance(one, dict) else None return isinstance(ref, dict) and refs.ref_id(ref) in rids out = {**doc} if "담은로직" in doc: out["담은로직"] = [one for one in doc.get("담은로직") or [] if not hit(one)] added = doc.get("더한줄") or [] dropped = [at for at, one in enumerate(added) if hit(one)] base_gone = hit(doc.get("바탕")) if "더한줄" in doc: out["더한줄"] = [one for one in added if not hit(one)] if base_gone: out["바탕"], out["열린입력"] = None, [] if not (dropped or base_gone): return out def keep(key: str) -> str | None: if not key.startswith("add:"): return None if base_gone else key # 바탕 줄 열쇠 — 바탕이 사라지면 걷음 return _shift(key, dropped) for field in _PATCH: if isinstance(doc.get(field), dict): moved = {keep(str(k)): v for k, v in doc[field].items()} out[field] = {k: v for k, v in moved.items() if k is not None} if isinstance(doc.get("뺀줄"), list): out["뺀줄"] = [k for k in (keep(str(x)) for x in doc["뺀줄"]) if k is not None] return out def _restore(done: list[dict[str, Any]]) -> None: for one in reversed(done): Path(one["파일"]).write_bytes(one["앞"]) def remove( session: dict[str, Any] | None, kind: str, gone: list[str], places: dict[str, Any], sure: bool, ) -> str: """지금 층 `gone` 지움 — 막힘 · 확인 없음 409 · 쓰는 줄 지움 → 문서 지움 · 기록 → 기록 키.""" rows = users(session, kind, gone) if blocked := [r for r in rows if not r["쓰기"]]: names = " · ".join(f"{r['어디']} 「{r['이름']}」 ← {r['키']}" for r in blocked) raise store.StoreError( 409, {"까닭": f"권한 밖 쓰는 곳이 있어 못 지움 — {names}", "막힘": blocked} ) if rows and not sure: raise store.StoreError(409, {"까닭": f"쓰는 곳 {len(rows)} — 확인 뒤 지움", "쓰는곳": rows}) mine = refs.here() rids = {refs.ref_id({"층": mine[0], "주인": mine[1], "키": k}) for k in gone} files = {} for r in rows: at = refs.folder_of(r["층"], r["주인"]) / r["문서종류"] / f"{r['문서']}.json" files[at] = r done: list[dict[str, Any]] = [] # ponytail: 여러 파일을 차례로 씀(잠금 없음) — 실패하면 앞서 고친 파일을 앞판 바이트로 되돌림 for path, r in files.items(): raw = path.read_bytes() try: with store.at(path.parent.parent): store.write( r["문서종류"], path.stem, store.version_of(raw), strip(json.loads(raw), rids) ) except store.StoreError as error: _restore(done) raise store.StoreError( 409, {"까닭": f"{r['어디']} 「{r['이름']}」 줄을 못 지움 — {error.detail}"} ) from error done.append( {"파일": str(path), "앞": raw, "r": r, "뒤판": store.version_of(path.read_bytes())} ) root = store.root() docs = [ {"키": k, "문서": json.loads(p.read_bytes()), "자리": places.get(k)} for k in gone if (p := root / kind / f"{k}.json").is_file() ] for one in docs: (root / kind / f"{one['키']}.json").unlink() name = _free(root / GONE, gone[0]) atomic_write_json( root / GONE / f"{name}.json", { "키": name, "종류": kind, "지움": docs, "고친곳": [ { "층": d["r"]["층"], "주인": d["r"]["주인"], "종류": d["r"]["문서종류"], "이름": d["r"]["문서"], "문서": json.loads(d["앞"]), "뒤판": d["뒤판"], } for d in done ], "때": datetime.now().isoformat(timespec="seconds"), "누가": (session or {}).get("name"), }, ) return name def _free(folder: Path, key: str) -> str: """기록 이름 — 그 키 기록이 이미 있으면(지운 키를 다시 씀) 「키~2」 …""" name, n = key, 1 while (folder / f"{name}.json").exists(): n += 1 name = f"{key}~{n}" return name def undo(session: dict[str, Any], folder: Path, kind: str, rec: dict[str, Any]) -> dict[str, Any]: """지움 기록 되돌림 — 문서(키를 그새 남이 쓰면 409) · 나무 자리 · 고친 문서 앞판 (그새 안 고친 것 · 쓸 수 있는 곳만 · 못한 곳 = 까닭).""" from M02_MasterTemplete import M02_Precombo_Folders as pre_folders docs = rec.get("지움") or [] if taken := [d["키"] for d in docs if (folder / kind / f"{d['키']}.json").exists()]: raise store.StoreError( 409, f"「{' · '.join(taken)}」 키를 그새 다른 문서가 씀 — 되돌릴 수 없음" ) for d in docs: atomic_write_json(folder / kind / f"{d['키']}.json", d["문서"]) with store.at(folder): tree_doc, _ = pre_folders._read(kind) places = {**(tree_doc.get("자리") or {}), **{d["키"]: d["자리"] for d in docs if d.get("자리")}} atomic_write_json(folder / pre_folders.file_of(kind), {**tree_doc, "자리": places}) back, miss = 0, [] for one in rec.get("고친곳") or []: at = (one.get("층"), one.get("주인")) path = refs.folder_of(*at) / str(one.get("종류")) / f"{one.get('이름')}.json" try: if not can_write(session, at) or not path.is_file(): raise store.StoreError(403, "권한 밖 · 없음") raw = path.read_bytes() if store.version_of(raw) != one.get("뒤판"): raise store.StoreError(409, "그새 고침 — 그 줄은 손으로 다시 담음") with store.at(path.parent.parent): store.write(str(one["종류"]), path.stem, store.version_of(raw), one["문서"]) back += 1 except store.StoreError as error: miss.append(f"{one.get('종류')} {one.get('이름')} — {error.detail}") return {"되돌린곳수": back, "못한곳": miss}