eomsangdonandClaude Opus 5.5
b9e6f77dab
- M02_Library_Same_Router — 733-1a 지문(Same 한 벌 · above) 으로 묶음 · 합칠 때 지문 다시 셈(다르면 409) - main.py 에 라우터 등록(로그인만 · 층 쓰기 권한은 길 안에서) - 새 시험 test_733_3_library_same_merge(합치기 참조 옮김 · 금액 같음 · 되돌림 · 지문 · 판 409 · 권한 밖 남김 · 위 층 키 · 일반 사용자 회사 층 403 · SC 수량만 다름) - 새 시험 test_733_3_library_same_screen(부품만 · 날 색 없음 · 나무 메뉴 · 서버 길 짝) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
341 lines
15 KiB
Python
341 lines
15 KiB
Python
"""M02 서고 [중복 찾기] · 합치기 · 되돌림(PLAN 733-3 · 요구서_572 3장 4 · 분석 D-3) —
|
|
단가산출(`sangeun` SK) · 일위대가(`precombo` SC).
|
|
|
|
GET /layers/{층}/{kind}-same — 이 층 + 위 층(회사 · 마스터) 같은 지문 묶음(이 층 문서가 든
|
|
묶음만) · 문서마다 층 · 키 · 이름 · 구분 · 쓰는 곳 수 · 꼬리표 「내용 같음」 ·
|
|
SC 「수량만 다름」(`본뜬.실무수량` 다름) · 합친 것 목록(되돌림)
|
|
POST …/merge {남김, 지움[], 판} — 서버가 지문 다시 셈(다르면 409) · 쓰는 곳 참조를 남김으로
|
|
(쓰기 권한 닿는 곳만 · `store.write` 정본 다시 셈) · 남은 쓰는 곳이 있는 문서 = 안 지움
|
|
「남김 — 쓰는 곳 N」 · 나머지 = `{층}/_합친것/<키>.json` 으로 옮김 · 나무 자리 걷음
|
|
POST …/undo {키} — `_합친것/` 에서 되돌림 · 옮긴 참조도 되돌림
|
|
지움 = 이 층 문서만 · 남김 = 이 층 또는 위 층(위 층과 같은 아래 층 문서 = 남김 = 위 층 키) ·
|
|
판 = 이 층 서고 나무 판(`{종류}_폴더.json`) · 권한 = 그 층 쓰기 · 프로젝트 사본 = 굳힌 값이라
|
|
안 건드림 · 지문 = `M02_Library_Same`(733-1) 읽기만.
|
|
⚠ 등록은 `main.py`(로그인만).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from common_util.common_util_auth import verify_session
|
|
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_Library_Same as same
|
|
from M02_MasterTemplete import M02_MasterTemplete_Store as store
|
|
from M02_MasterTemplete import M02_Precombo_Folders as pre_folders
|
|
from M02_MasterTemplete import M02_Template_Layers as layers
|
|
from M02_MasterTemplete.M02_MasterTemplete_Router_Items import Layer, Session, _root
|
|
|
|
router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Library Same"])
|
|
|
|
Kind = Literal["precombo", "sangeun"]
|
|
GONE = "_합친것"
|
|
USER_KINDS = (("labor", "구조물 조합"), ("precombo", "일위대가"), ("sangeun", "단가산출"))
|
|
_ORDER = {"system": 0, "company": 1, "personal": 2}
|
|
|
|
|
|
class MergeBody(BaseModel):
|
|
"""`남김` = {층, 주인, 키}(이 층 · 위 층) · `지움` = 이 층 키들 · `판` = 이 층 서고 나무 판."""
|
|
|
|
남김: dict[str, Any]
|
|
지움: list[str]
|
|
판: str
|
|
|
|
|
|
class UndoBody(BaseModel):
|
|
키: str
|
|
|
|
|
|
def _bad(status: int, text: Any) -> store.StoreError:
|
|
return store.StoreError(status, text)
|
|
|
|
|
|
def _ref(at: refs.Here, key: str) -> refs.Ref:
|
|
return {"층": at[0], "주인": at[1], "키": key}
|
|
|
|
|
|
def _docs(at: refs.Here, kind: str) -> list[tuple[str, dict[str, Any]]]:
|
|
folder = refs.folder_of(*at) / kind
|
|
out = []
|
|
for path in sorted(folder.glob("[!._]*.json")) if folder.is_dir() else []:
|
|
try:
|
|
doc = json.loads(path.read_bytes())
|
|
except (OSError, ValueError):
|
|
continue
|
|
if isinstance(doc, dict):
|
|
out.append((path.stem, doc))
|
|
return out
|
|
|
|
|
|
# ── 쓰는 곳 ───────────────────────────────────────────────────────────
|
|
def _user_index() -> dict[str, list[tuple[Path, str, str]]]:
|
|
"""모든 층(프로젝트 뺌) 훑기 한 번 — {ref_id: [(파일, 종류, 「어디 이름」)]}."""
|
|
out: dict[str, list[tuple[Path, str, str]]] = {}
|
|
for kind, what in USER_KINDS:
|
|
for where, path in layers.layer_files(kind):
|
|
try:
|
|
doc = json.loads(path.read_bytes())
|
|
except (OSError, ValueError):
|
|
continue
|
|
name = doc.get("일위대가번") or doc.get("이름") or path.stem
|
|
for rid in {refs.ref_id(one) for one in refs.refs_in(doc)}:
|
|
out.setdefault(rid, []).append((path, kind, f"{what} {where} 「{name}」"))
|
|
return out
|
|
|
|
|
|
def _can_write(session: Session, at: refs.Here) -> bool:
|
|
try:
|
|
return access.grant(session, at[0], at[1]).write
|
|
except HTTPException:
|
|
return False
|
|
|
|
|
|
def _retarget(doc: dict[str, Any], olds: set[str], new: refs.Ref) -> dict[str, Any]:
|
|
"""문서의 `olds` 참조를 모두 `new` 로(규격 열 수량 열쇠까지 · `M02_Labor_Ref` 한 벌)."""
|
|
done: dict[str, refs.Ref] = {}
|
|
|
|
def move(ref: refs.Ref, _depth: int) -> refs.Ref:
|
|
if refs.ref_id(ref) in olds:
|
|
done[refs.ref_id(ref)] = new
|
|
return new
|
|
return ref
|
|
|
|
return refs._spec_moved(refs._move_doc(doc, move, 1), done)
|
|
|
|
|
|
def _rewrite(path: Path, kind: str, olds: set[str], new: refs.Ref) -> None:
|
|
"""쓰는 곳 한 파일 — 참조를 바꿔 `store.write`(서버 정본 다시 셈 · 판 검사)."""
|
|
raw = path.read_bytes()
|
|
doc = _retarget(json.loads(raw), olds, new)
|
|
with store.at(path.parent.parent):
|
|
store.write(kind, path.stem, store.version_of(raw), doc)
|
|
|
|
|
|
# ── 목록 ──────────────────────────────────────────────────────────────
|
|
def _qty(doc: dict[str, Any]) -> str:
|
|
return json.dumps((doc.get("본뜬") or {}).get("실무수량") or {}, sort_keys=True)
|
|
|
|
|
|
def _row(at: refs.Here, key: str, doc: dict[str, Any], used: int, mine: refs.Here) -> dict:
|
|
return {
|
|
"층": at[0],
|
|
"주인": at[1],
|
|
"키": key,
|
|
"이름": doc.get("이름") or key,
|
|
"구분": doc.get("구분") or "",
|
|
"상세구분": doc.get("상세구분") or "",
|
|
"쓰는곳수": used,
|
|
"위층": at != mine,
|
|
}
|
|
|
|
|
|
def _gone_rows(folder: Path, kind: str) -> list[dict[str, Any]]:
|
|
out = []
|
|
for path in sorted((folder / GONE).glob("*.json")):
|
|
try:
|
|
rec = json.loads(path.read_bytes())
|
|
except (OSError, ValueError):
|
|
continue
|
|
if rec.get("종류") == kind:
|
|
doc = rec.get("문서") or {}
|
|
out.append(
|
|
{
|
|
"키": rec.get("키"),
|
|
"이름": doc.get("이름") or rec.get("키"),
|
|
"남김": rec.get("남김"),
|
|
"때": rec.get("때"),
|
|
"옮긴곳수": len(rec.get("옮긴곳") or []),
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def find(folder: Path, kind: str) -> dict[str, Any]:
|
|
"""이 층 + 위 층 같은 지문 묶음 — 이 층 문서가 든 묶음 · 둘 이상만."""
|
|
mine = refs.where(folder)
|
|
index = _user_index()
|
|
groups: dict[str, list[tuple[refs.Here, str, dict[str, Any]]]] = {}
|
|
marks = same.Same() # 한 요청 안 참조 지문 메모
|
|
for at in [mine, *same.above(mine)]:
|
|
for key, doc in _docs(at, kind):
|
|
if (mark := marks.finger(kind, doc)) is not None:
|
|
groups.setdefault(mark, []).append((at, key, doc))
|
|
out = []
|
|
for mark, docs in groups.items():
|
|
if len(docs) < 2 or all(at != mine for at, _, _ in docs):
|
|
continue
|
|
docs.sort(key=lambda d: (_ORDER[d[0][0]], d[1]))
|
|
rows = [
|
|
_row(at, key, doc, len(index.get(refs.ref_id(_ref(at, key)), [])), mine)
|
|
for at, key, doc in docs
|
|
]
|
|
qty = kind == "precombo" and len({_qty(doc) for _, _, doc in docs}) > 1
|
|
out.append({"지문": mark, "꼬리표": "수량만 다름" if qty else "내용 같음", "문서": rows})
|
|
out.sort(key=lambda g: (g["꼬리표"] != "내용 같음", g["문서"][0]["이름"]))
|
|
with store.at(folder):
|
|
판 = pre_folders.view(kind)["판"]
|
|
return {"판": 판, "묶음": out, "합친것": _gone_rows(folder, kind)}
|
|
|
|
|
|
# ── 합치기 ────────────────────────────────────────────────────────────
|
|
def merge(session: Session, folder: Path, kind: str, body: MergeBody) -> dict[str, Any]:
|
|
mine = refs.where(folder)
|
|
raw = body.남김 or {}
|
|
keep_at: refs.Here = (str(raw.get("층") or ""), raw.get("주인") or None)
|
|
keep = _ref(keep_at, str(raw.get("키") or ""))
|
|
if keep_at not in [mine, *same.above(mine)]:
|
|
raise _bad(400, "남김은 이 층 또는 위 층 문서")
|
|
if refs.kind_of(keep["키"]) != kind or (keep_doc := refs.read_ref(keep)) is None:
|
|
raise _bad(404, f"없는 {refs.TITLE[kind]} 「{keep['키']}」")
|
|
gone = list(dict.fromkeys(str(k) for k in body.지움))
|
|
if not gone or (keep_at == mine and keep["키"] in gone):
|
|
raise _bad(400, "지울 문서를 고름 — 남김과 달라야 함")
|
|
have = dict(_docs(mine, kind))
|
|
if miss := [k for k in gone if k not in have]:
|
|
raise _bad(404, f"이 층에 없는 문서 「{' · '.join(miss)}」")
|
|
with store.at(folder):
|
|
tree_doc, 판 = pre_folders._read(kind)
|
|
if 판 != body.판:
|
|
raise _bad(409, {"stale": [pre_folders.file_of(kind)], "판": 판})
|
|
marks = same.Same()
|
|
mark = marks.finger(kind, keep_doc)
|
|
if mark is None or any(marks.finger(kind, have[k]) != mark for k in gone):
|
|
raise _bad(409, "내용이 그새 달라짐 — [중복 찾기]를 다시 누름")
|
|
|
|
rids = {refs.ref_id(_ref(mine, k)): k for k in gone}
|
|
by_file: dict[Path, tuple[str, set[str]]] = {}
|
|
labels: dict[Path, str] = {}
|
|
index = _user_index()
|
|
for rid in rids:
|
|
for path, ukind, label in index.get(rid, []):
|
|
by_file.setdefault(path, (ukind, set()))[1].add(rid)
|
|
labels[path] = label
|
|
moved: dict[str, list[dict[str, Any]]] = {k: [] for k in gone}
|
|
left: dict[str, list[str]] = {k: [] for k in gone}
|
|
for path, (ukind, olds) in by_file.items():
|
|
at = refs.where(path.parent.parent)
|
|
why = ""
|
|
if not (_can_write(session, at) and refs.readable(at, keep)):
|
|
why = "권한 밖"
|
|
else:
|
|
try:
|
|
_rewrite(path, ukind, olds, keep)
|
|
except store.StoreError as error: # 같은 참조 두 번 등 — 그 쓰는 곳은 그대로 둠
|
|
why = str(error.detail)
|
|
for rid in olds:
|
|
if why:
|
|
left[rids[rid]].append(f"{labels[path]} — {why}")
|
|
else:
|
|
moved[rids[rid]].append(
|
|
{"층": at[0], "주인": at[1], "종류": ukind, "이름": path.stem}
|
|
)
|
|
|
|
places = dict(tree_doc.get("자리") or {})
|
|
done, kept = [], []
|
|
when = datetime.now().isoformat(timespec="seconds")
|
|
# ponytail: 여러 파일을 차례로 씀(잠금 없음) — 동시에 같은 층을 합치면 앞 것이 이김
|
|
for key in gone:
|
|
if left[key]:
|
|
kept.append({"키": key, "쓰는곳": left[key]})
|
|
continue
|
|
path = folder / kind / f"{key}.json"
|
|
record = {
|
|
"키": key,
|
|
"종류": kind,
|
|
"문서": json.loads(path.read_bytes()),
|
|
"자리": places.pop(key, None),
|
|
"남김": keep,
|
|
"옮긴곳": moved[key],
|
|
"때": when,
|
|
"누가": session.get("name"),
|
|
}
|
|
atomic_write_json(folder / GONE / f"{key}.json", record)
|
|
path.unlink()
|
|
done.append({"키": key, "옮긴곳수": len(moved[key])})
|
|
if done:
|
|
atomic_write_json(folder / pre_folders.file_of(kind), {**tree_doc, "자리": places})
|
|
return {"합침": done, "남김": kept, **find(folder, kind)}
|
|
|
|
|
|
def undo(session: Session, folder: Path, kind: str, key: str) -> dict[str, Any]:
|
|
"""`_합친것/<키>` 되돌림 — 문서 · 나무 자리 · 옮긴 참조(아직 남김을 가리키고 쓸 수 있는 곳)."""
|
|
mine = refs.where(folder)
|
|
rec_path = folder / GONE / f"{key}.json"
|
|
try:
|
|
rec = json.loads(rec_path.read_bytes())
|
|
except (OSError, ValueError):
|
|
raise _bad(404, f"합친 것에 없음 「{key}」") from None
|
|
if rec.get("종류") != kind:
|
|
raise _bad(404, f"합친 것에 없음 「{key}」")
|
|
path = folder / kind / f"{key}.json"
|
|
if path.exists():
|
|
raise _bad(409, f"「{key}」 키를 그새 다른 문서가 씀 — 되돌릴 수 없음")
|
|
atomic_write_json(path, rec["문서"])
|
|
with store.at(folder):
|
|
tree_doc, _ = pre_folders._read(kind)
|
|
if rec.get("자리"):
|
|
places = {**(tree_doc.get("자리") or {}), key: rec["자리"]}
|
|
atomic_write_json(folder / pre_folders.file_of(kind), {**tree_doc, "자리": places})
|
|
keep, back, miss = rec.get("남김") or {}, 0, []
|
|
new = _ref(mine, key)
|
|
# ponytail: 옮긴 뒤 그 쓰는 곳이 남김을 다른 줄에서 따로 담았으면 그 줄까지 되돌림 — 드문 꼴
|
|
for one in rec.get("옮긴곳") or []:
|
|
at = (one.get("층"), one.get("주인"))
|
|
user = refs.folder_of(*at) / str(one.get("종류")) / f"{one.get('이름')}.json"
|
|
try:
|
|
if not _can_write(session, at) or not user.is_file():
|
|
raise store.StoreError(403, "권한 밖 · 없음")
|
|
_rewrite(user, str(one.get("종류")), {refs.ref_id(keep)}, new)
|
|
back += 1
|
|
except store.StoreError as error:
|
|
miss.append(f"{one.get('종류')} {one.get('이름')} — {error.detail}")
|
|
rec_path.unlink()
|
|
return {"되돌림": key, "되돌린곳수": back, "못한곳": miss, **find(folder, kind)}
|
|
|
|
|
|
def _run(fn, *args):
|
|
try:
|
|
return fn(*args)
|
|
except store.StoreError as e:
|
|
raise HTTPException(status_code=e.status, detail=e.detail) from e
|
|
|
|
|
|
@router.get("/layers/{layer}/{kind}-same")
|
|
def library_same(
|
|
layer: Layer, kind: Kind, owner: str | None = None, session: Session = Depends(verify_session)
|
|
) -> dict:
|
|
"""[중복 찾기] — 이 층 + 위 층 같은 내용 묶음 · 합친 것 목록(읽기 = 그 층 쓰기 권한)."""
|
|
return _run(find, _root(session, layer, owner, write=True), kind)
|
|
|
|
|
|
@router.post("/layers/{layer}/{kind}-same/merge")
|
|
def library_same_merge(
|
|
layer: Layer,
|
|
kind: Kind,
|
|
body: MergeBody,
|
|
owner: str | None = None,
|
|
session: Session = Depends(verify_session),
|
|
) -> dict:
|
|
"""합치기 — 쓰는 곳 참조 → 남김 · 지운 문서 `_합친것/` · 판 · 지문 다르면 409."""
|
|
return _run(merge, session, _root(session, layer, owner, write=True), kind, body)
|
|
|
|
|
|
@router.post("/layers/{layer}/{kind}-same/undo")
|
|
def library_same_undo(
|
|
layer: Layer,
|
|
kind: Kind,
|
|
body: UndoBody,
|
|
owner: str | None = None,
|
|
session: Session = Depends(verify_session),
|
|
) -> dict:
|
|
"""되돌림 — `_합친것/<키>` 문서 · 나무 자리 · 옮긴 참조."""
|
|
return _run(undo, session, _root(session, layer, owner, write=True), kind, body.키)
|