From e987602f24de7ada18d3fe8f779e9ada9ff1fa69 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Fri, 13 Nov 2026 18:42:04 +0900 Subject: [PATCH] =?UTF-8?q?feat(stmate-test):=20=EB=AC=B4=EC=86=90?= =?UTF-8?q?=EC=8B=A4=20=EC=A0=80=EC=9E=A5=20=E2=80=94=20=EC=86=90=EB=8C=80?= =?UTF-8?q?=EC=A7=80=20=EC=95=8A=EC=9D=80=20=EC=A0=80=EC=9E=A5=20=3D=20?= =?UTF-8?q?=EC=9B=90=EB=B3=B8=20STC=20=EB=B0=94=EC=9D=B4=ED=8A=B8=20?= =?UTF-8?q?=EB=8F=99=EC=9D=BC=20=C2=B7=20=EA=B3=A0=EC=B9=9C=20=EC=B9=B8?= =?UTF-8?q?=EB=A7=8C=20=EB=8B=A4=EC=8B=9C=20=EC=94=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - DBF 고친 칸만 다시 씀(한글 깨짐 · 2칸을 1칸으로 · 3칸 87.745% 덮어쓰기 없앰) · 단가 → 내역 행 → 소계 → 직공비 다시 셈 · 행 끼우기·지우기 - SIxMemo(COSTN.SMT · PRJLST.SMT) 블록 구조 유지 · 넘치면 뒤에 붙이고 가리키는 값 갱신 - STC 다시 묶기(같은 파일은 원본 바이트) · 검사 · 자동 백업·복원 · 새 공사 · 복사등록(차수 인계) · 해당분 압축 - BDQTY 역변환(쓰기 준비) · 모듈 dbf_rw · dbf_memo · dbf_crypt · stc_archive · stc_save · stc_read · stc_project · api_file · scr_file - 지킴 시험 test_stmate_file_roundtrip · file_project · file_payload — 실무 STC 7건 왕복 Co-Authored-By: Claude Opus 5.5 --- .../tester/helper_stmate_file_payload.cjs | 24 ++ resources/tester/test_stmate_file_payload.py | 59 ++++ resources/tester/test_stmate_file_project.py | 173 +++++++++ .../tester/test_stmate_file_roundtrip.py | 261 ++++++++++++++ tools/stmate-test/api_file.py | 140 ++++++++ tools/stmate-test/dbf_crypt.py | 42 +++ tools/stmate-test/dbf_memo.py | 193 ++++++++++ tools/stmate-test/dbf_rw.py | 194 ++++++++++ tools/stmate-test/scr_file.js | 146 ++++++++ tools/stmate-test/server.py | 170 ++------- tools/stmate-test/smt_parser.py | 92 ++--- tools/stmate-test/stc_archive.py | 282 +++++++++++++++ tools/stmate-test/stc_project.py | 267 ++++++++++++++ tools/stmate-test/stc_read.py | 108 ++++++ tools/stmate-test/stc_save.py | 334 ++++++++++++++++++ tools/stmate-test/view_prj_boq.js | 227 ++++++------ 16 files changed, 2402 insertions(+), 310 deletions(-) create mode 100644 resources/tester/helper_stmate_file_payload.cjs create mode 100644 resources/tester/test_stmate_file_payload.py create mode 100644 resources/tester/test_stmate_file_project.py create mode 100644 resources/tester/test_stmate_file_roundtrip.py create mode 100644 tools/stmate-test/api_file.py create mode 100644 tools/stmate-test/dbf_crypt.py create mode 100644 tools/stmate-test/dbf_memo.py create mode 100644 tools/stmate-test/dbf_rw.py create mode 100644 tools/stmate-test/scr_file.js create mode 100644 tools/stmate-test/stc_archive.py create mode 100644 tools/stmate-test/stc_project.py create mode 100644 tools/stmate-test/stc_read.py create mode 100644 tools/stmate-test/stc_save.py diff --git a/resources/tester/helper_stmate_file_payload.cjs b/resources/tester/helper_stmate_file_payload.cjs new file mode 100644 index 000000000..321437232 --- /dev/null +++ b/resources/tester/helper_stmate_file_payload.cjs @@ -0,0 +1,24 @@ +// STMATE 재현 앱 저장 요청(view_prj_boq.js buildSavePayload)을 api_load 응답(stdin)으로 만들어 냄 +// 입력 {load, edit?: {recno, qty}} — edit 는 app.js 수량 칸 고침과 같게 leaf.qty · drow.QTY1 을 바꿈 +globalThis.window = globalThis; +const Calc = require("../../tools/stmate-test/calc.js"); +require("../../tools/stmate-test/view_prj_boq.js"); +let buf = ""; +process.stdin.on("data", (d) => (buf += d)); +process.stdin.on("end", () => { + const { load, edit } = JSON.parse(buf); + const ctx = { + state: load.sim_state, + dbfData: load, + currentSourcePath: load.source_path, + uiState: { editedSmtMap: {} }, + }; + if (edit) { + const leaf = ctx.state.boqItems.find((b) => b._recno === edit.recno); + const drow = load.boq_rows.find((r) => r._recno === edit.recno); + if (leaf) leaf.qty = edit.qty; + if (drow) drow.QTY1 = String(edit.qty); + } + ctx.computed = Calc.calculateAll(ctx.state); + process.stdout.write(JSON.stringify(window.STMATE_VIEW_PRJ_BOQ.buildSavePayload(ctx))); +}); diff --git a/resources/tester/test_stmate_file_payload.py b/resources/tester/test_stmate_file_payload.py new file mode 100644 index 000000000..d8f53cb74 --- /dev/null +++ b/resources/tester/test_stmate_file_payload.py @@ -0,0 +1,59 @@ +"""STMATE 재현 앱 화면 저장 요청(view_prj_boq.js buildSavePayload) — 고치지 않으면 저장해도 한 칸도 안 바뀜. + +옛 요청은 2차 칸을 1차로, 3차(계약) 칸을 고정 낙찰율로 덮었음 — 이제 바뀐 칸만 나감을 실무 STC 7건으로 확인. +""" + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "tools/stmate-test")) +import server # noqa: E402 +from stc_save import _leaf, apply_save # noqa: E402 + +HELPER = Path(__file__).with_name("helper_stmate_file_payload.cjs") +STCS = sorted((ROOT / "resources/knowledge/original/실무문서").rglob("*.STC")) +IDS = [p.stem[:16] for p in STCS] + + +def _payload(load, edit=None): + raw = json.dumps({"load": load, "edit": edit}, default=str).encode() + return json.loads(subprocess.check_output(["node", str(HELPER)], input=raw)) + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_screen_payload_without_edits_changes_nothing(stc): + load = server.api_load(server.LoadReq(source_path=str(stc))) + p = _payload(load) + by = {r["_recno"]: r for r in load["boq_rows"]} + for u in p["boq_updates"]: # 2·3차 단가는 불러온 값 그대로(1차 복사·고정 낙찰율 아님) + for k in ("MAMT2", "LAMT2", "MAMT3", "LAMT3", "SAMT3"): + if k in u and u.get("_recno") in by: + assert float(u[k] or 0) == float(by[u["_recno"]][k] or 0) + files = server.read_project_files(str(stc)) + out, log = apply_save(files, p) + assert log == [] and all(out[k] is files[k] for k in files) + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_screen_qty_edit_touches_slot1_only(stc): + load = server.api_load(server.LoadReq(source_path=str(stc))) + leaf = next( + r + for r in load["boq_rows"] + if _leaf(r["BDCODE1"].strip()) + and float(r["TAMT1"] or 0) > 0 + and not r["UNIT1"].strip().endswith("%") + ) + q = float(leaf["QTY1"]) + 3 + p = _payload(load, {"recno": leaf["_recno"], "qty": q}) + files = server.read_project_files(str(stc)) + _out, log = apply_save(files, p) + fields = {(x["table"], x["field"]) for x in log} + assert ("BOQ11.DBF", "QTY1") in fields + assert all(f[1][-1] == "1" or f[1].startswith("COSTW") for f in fields), fields + assert {x["recno"] for x in log if x["field"] == "QTY1"} == {leaf["_recno"]} diff --git a/resources/tester/test_stmate_file_project.py b/resources/tester/test_stmate_file_project.py new file mode 100644 index 000000000..ecd86794d --- /dev/null +++ b/resources/tester/test_stmate_file_project.py @@ -0,0 +1,173 @@ +"""STMATE 재현 앱 공사 단위 파일 작업 — STC 검사 · 해당분만 압축 · 복사등록(차수/계약 변환) · 새 공사 · 백업/복원. + +정답은 실무 STC 자체 — 소광 신설(BOQ11+BOQ12 통합형) → 변경1차(분리형) 짝으로 차수 인계 규칙(16번 §4.6) 대조. +쓰기는 임시 폴더(server.WORK_DBF_ROOT 바꿈)에서만. +""" + +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "tools/stmate-test")) +import api_file # noqa: E402 +import server # noqa: E402 +from dbf_memo import SixMemo, ini_items # noqa: E402 +from dbf_rw import DbfTable # noqa: E402 +from stc_archive import build_stc, check_stc, files_of, read_entries # noqa: E402 +from stc_project import copy_register, new_project, pack_selected # noqa: E402 +from stc_save import _leaf, boq_spans # noqa: E402 + +DOCS = ROOT / "resources/knowledge/original/실무문서" +STCS = sorted(DOCS.rglob("*.STC")) +IDS = [p.stem[:16] for p in STCS] +SOGWANG = DOCS / "25년 산불진화임도(기번8)(울진소광)" +JOINT = next(SOGWANG.glob("2025년*.STC")) # BOQ11(당초) + BOQ12(현재안) +SPLIT = next(SOGWANG.glob("변경*.STC")) # 분리형 변경1차 + + +def _files(p): + return files_of(read_entries(p.read_bytes())) + + +def _memo_text(files, i): + t = DbfTable(files["PRJLST.DBF"]) + return SixMemo(files["PRJLST.SMT"]).read(t.raw(i, "PRJ_MEMO")).decode("cp949") + + +@pytest.fixture +def work(tmp_path, monkeypatch): + monkeypatch.setattr(server, "WORK_DBF_ROOT", tmp_path) + return tmp_path + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_check_stc_accepts_real_files(stc): + res = check_stc(stc.read_bytes()) + assert res["ok"], res["problems"] + assert res["tables"]["PRJLST.DBF"]["nrec"] >= 1 + assert not check_stc(b"PK-not-a-zip")["ok"] + + +def test_pack_selected_carries_current_plan_into_split_file(): + src = _files(JOINT) + packed = pack_selected(src, "12") + assert packed["BOQ11.DBF"] == src["BOQ12.DBF"] and "BOQ12.DBF" not in packed + pt, st = DbfTable(packed["PRJLST.DBF"]), DbfTable(src["PRJLST.DBF"]) + assert len(pt) == 1 and pt.get(0, "PCODE") == "11" + for f in pt.order: + if f not in ("PCODE", "PRJ_MEMO"): + assert pt.raw(0, f) == st.raw(1, f), f + assert _memo_text(packed, 0) == _memo_text(src, 1) + assert SixMemo(packed["PRJLST.SMT"]).next_free == -(-len(packed["PRJLST.SMT"]) // 64) + # 16번 §4.6 — 변경1차 파일의 차수1 직접비 = 앞 파일 BOQ12 공사 직접비 + real = DbfTable(_files(SPLIT)["PRJLST.DBF"]) + for c in ("MCOST1", "LCOST1", "SCOST1", "TCOST1"): + assert float(pt.get(0, c)) == float(real.get(0, c)), c + assert check_stc(build_stc(read_entries(JOINT.read_bytes()), packed))["ok"] + + +def test_copy_register_change_mode_opens_next_stage(): + base = pack_selected(_files(JOINT), "12") + out, pcode, _log = copy_register(base, "11", "change") + assert pcode == "12" and out["BOQ11.DBF"] == base["BOQ11.DBF"] + a, b = DbfTable(base["BOQ11.DBF"]), DbfTable(out["BOQ12.DBF"]) + assert len(a) == len(b) + for i in range(len(a)): + for f in ("QTY", "MAMT", "LAMT", "SAMT", "TAMT", "M_AMT", "T_AMT", "BDCODE"): + assert b.raw(i, f"{f}2") == a.raw(i, f"{f}1"), (i, f) + assert b.raw(i, f"{f}1") == a.raw(i, f"{f}1"), (i, f) + pt = DbfTable(out["PRJLST.DBF"]) + real = DbfTable(_files(SPLIT)["PRJLST.DBF"]) + assert len(pt) == 2 and pt.get(1, "PCODE") == "12" + assert pt.get(1, "K_ORD") == real.get(0, "K_ORD") == "1" + assert pt.get(1, "TCOST1") == pt.get(1, "TCOST2") == real.get(0, "TCOST1") + secs = {s for s, _k, _v in ini_items(_memo_text(out, 1))} + assert {"1_KANJUB", "2_KANJUB"} <= secs + one = {k[2:]: v for s, k, v in ini_items(_memo_text(out, 1)) if s == "1_KANJUB"} + two = {k[2:]: v for s, k, v in ini_items(_memo_text(out, 1)) if s == "2_KANJUB"} + assert one == two + assert _memo_text(out, 0) == _memo_text(base, 0) # 원래 공사 메모 그대로 + + +def test_copy_register_contract_mode_applies_rates_except_excluded(): + src = _files(STCS[0]) + same, _, log = copy_register(src, "11", "contract", rates={"M": 100, "L": 100, "S": 100}) + assert DbfTable(same["BOQ12.DBF"]).recs == DbfTable(src["BOQ11.DBF"]).recs and not log + out, pcode, _ = copy_register(src, "11", "contract", rates={"M": 87.745, "L": 90, "S": 0}) + a, b = DbfTable(src["BOQ11.DBF"]), DbfTable(out[f"BOQ{pcode}.DBF"]) + spans, excl = boq_spans(a) + for i in range(len(a)): + if not _leaf(a.get(i, "BDCODE1")): + continue + if i in excl: + assert b.recs[i] == a.recs[i] + continue + assert int(float(b.get(i, "MAMT1"))) == int(float(a.get(i, "MAMT1") or 0) * 0.87745 + 1e-9) + assert float(b.get(i, "SAMT1") or 0) == 0 # 0% = 공내역 + for r, x, y, isx in spans: + tot = sum( + int(float(b.get(k, "T_AMT1") or 0)) + for k in range(x, y + 1) + if _leaf(b.get(k, "BDCODE1")) and (isx or k not in excl) + ) + assert int(float(b.get(r, "T_AMT1") or 0)) == tot + direct = sum( + int(float(b.get(k, "T_AMT1") or 0)) + for k in range(len(b)) + if _leaf(b.get(k, "BDCODE1")) and k not in excl + ) + assert int(float(DbfTable(out["PRJLST.DBF"]).get(1, "TCOST1"))) == direct + + +def test_new_project_opens_empty(work): + out = new_project(_files(STCS[0]), "새 임도 공사", "2026-01") + p = work / "new.STC" + p.write_bytes(build_stc(read_entries(STCS[0].read_bytes()), out)) + assert check_stc(p.read_bytes())["ok"] + load = server.api_load(server.LoadReq(source_path=str(p))) + assert load["costn_count"] == 0 and load["boq_count"] == 0 + assert load["prjlst"]["PNAME"] == "새 임도 공사" and load["prjlst"]["PCODE"] == "11" + assert float(load["prjlst"]["TCOST1"] or 0) == 0 + smt = load["prjlst_smt"] + assert smt and all(float(v) == 0 for k, v in smt.items() if k.startswith("1_C_")) + assert any(k.startswith("1_R_") and float(v) > 0 for k, v in smt.items()) # 요율 설정은 틀에서 + + +def test_file_api_endpoints(work): + src = str(JOINT) + assert api_file.api_file_check(api_file.PathReq(source_path=src))["ok"] + r = api_file.api_file_pack(api_file.PackReq(source_path=src, pcode="12", target_name="pk")) + assert Path(r["saved_stc"]).parent == work and check_stc(Path(r["saved_stc"]).read_bytes())["ok"] + r2 = api_file.api_file_copy_register( + api_file.CopyRegReq(source_path=r["saved_stc"], mode="change", target_folder_name="cr") + ) + assert r2["new_pcode"] == "12" and (work / "cr" / "BOQ12.DBF").exists() + load = server.api_load(server.LoadReq(source_path=str(work / "cr"))) + assert sorted(load["boq_tables"]) == ["BOQ.DBF", "BOQ11.DBF", "BOQ12.DBF"] + r3 = api_file.api_file_new( + api_file.NewPrjReq(template_path=src, pname="새 공사", target_folder_name="nw") + ) + assert Path(r3["saved_stc"]).exists() + api_file.api_file_new(api_file.NewPrjReq(template_path=src, pname="새 공사2", target_folder_name="nw")) + bks = api_file.api_file_backups("nw")["backups"] + assert len(bks) == 1 + api_file.api_file_restore(api_file.RestoreReq(backup_path=bks[0]["path"], target_folder_name="nw")) + assert server.api_load(server.LoadReq(source_path=str(work / "nw")))["prjlst"]["PNAME"] == "새 공사" + with pytest.raises(Exception): + api_file.api_file_restore(api_file.RestoreReq(backup_path=src, target_folder_name="x")) + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_load_reads_side_tables_and_all_settings(stc): + load = server.api_load(server.LoadReq(source_path=str(stc))) + assert load["kongjong_rows"] and all("CODE" in r for r in load["kongjong_rows"]) + assert load["costa_count"] == 0 and load["qcode_rows"] == [] + rows = load["prjlst_smt_all"] + assert len(rows) == len(load["prjlst_all"]) + first = rows[0] + assert first["key_count"] >= len(load["prjlst_smt"]) > 100 and "KANJUB" in first["sections"] + for k, v in load["prjlst_smt"].items(): # 첫 공사 읽기(parse_prjlst_smt)와 같은 값 + assert any(s.get(k) == v for s in first["sections"].values()), k + assert "START" in load["start_ini_sections"] diff --git a/resources/tester/test_stmate_file_roundtrip.py b/resources/tester/test_stmate_file_roundtrip.py new file mode 100644 index 000000000..aa9142cbe --- /dev/null +++ b/resources/tester/test_stmate_file_roundtrip.py @@ -0,0 +1,261 @@ +"""STMATE 재현 앱 파일 저장 무손실 — 실무 STC 7건 열기 → 저장 왕복과 고친 칸만 바뀌는지. + +정답은 원본 STC 자체(코드에 값을 박지 않음). 쓰기는 임시 폴더(server.WORK_DBF_ROOT 바꿈)에서만. +근거: STC_왕복검증.md R2·R3 · SIxMemo_구조.md · 16_골든셋_회귀_결과.md §4 +""" + +import struct +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "tools/stmate-test")) +import server # noqa: E402 +from dbf_crypt import BdqtyCodec, covers # noqa: E402 +from dbf_memo import SixMemo, ptr_unpack, ptr_valid # noqa: E402 +from dbf_rw import DbfTable # noqa: E402 +from smt_parser import extract_costn_memos, parse_dbf_bytes, parse_prjlst_smt # noqa: E402 +from stc_archive import build_stc, files_of, read_entries # noqa: E402 +from stc_save import _leaf, apply_save, boq_spans # noqa: E402 + +DOCS = ROOT / "resources/knowledge/original/실무문서" +STCS = sorted(DOCS.rglob("*.STC")) +IDS = [p.stem[:16] for p in STCS] + + +def _files(p): + return files_of(read_entries(p.read_bytes())) + + +def _full_payload(files): + """화면이 보내는 꼴 — 불러온 값 전부(글자 칸은 strip 된 값).""" + _, crows, _ = parse_dbf_bytes(files["COSTN.DBF"]) + _, brows, _ = parse_dbf_bytes(files["BOQ11.DBF"]) + _, prows, _ = parse_dbf_bytes(files["PRJLST.DBF"]) + clean = lambda r: {k: v for k, v in r.items() if not k.endswith("_HEX") and k not in ("IFORM", "PRJ_MEMO", "XCODE")} # noqa: E731,E501 + memos = extract_costn_memos(files["COSTN.DBF"], files["COSTN.SMT"]) + return { + "costn_updates": [clean(r) for r in crows], + "boq_updates": [clean(r) for r in brows], + "prjlst_updates": {k: v for k, v in clean(prows[0]).items() if not k.startswith("_")}, + "smt_updates": parse_prjlst_smt(files["PRJLST.SMT"], prows[0]["PRJ_MEMO_HEX"]), + "costn_memo_updates": {c: m["text"] for c, m in memos.items()}, + } + + +@pytest.fixture +def work(tmp_path, monkeypatch): + monkeypatch.setattr(server, "WORK_DBF_ROOT", tmp_path) + return tmp_path + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_stc_rebuild_is_byte_identical(stc): + b = stc.read_bytes() + ents = read_entries(b) + assert build_stc(ents, files_of(ents)) == b + for name, data in files_of(ents).items(): + if name.endswith(".DBF"): + assert DbfTable(data).to_bytes() == data, name + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_api_save_without_edits_reproduces_stc(stc, work): + req = server.SaveDbfReq(source_path=str(stc), target_folder_name="rt", **_full_payload(_files(stc))) + res = server.api_save(req) + assert res["change_count"] == 0 and res["changed_files"] == [] + assert Path(res["saved_stc"]).read_bytes() == stc.read_bytes() + for e in read_entries(stc.read_bytes()): + assert (work / "rt" / e["name"]).read_bytes() == e["data"], e["name"] + server.api_save(req) # 두 번째 저장 — 앞 작업본 백업 + assert list((work / "_backup").glob("rt__*.STC")) + + +def _changed_cells(a: bytes, b: bytes): + ta, tb = DbfTable(a), DbfTable(b) + assert len(ta) == len(tb) + out = set() + for i in range(len(ta)): + for f in ta.order: + if ta.raw(i, f) != tb.raw(i, f): + out.add((i, f)) + return out + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_boq_qty_edit_changes_only_derived_cells(stc): + f = _files(stc) + bt = DbfTable(f["BOQ11.DBF"]) + i = next( + k + for k in range(len(bt)) + if _leaf(bt.get(k, "BDCODE1")) and float(bt.get(k, "TAMT1") or 0) > 0 + and not bt.get(k, "UNIT1").endswith("%") + ) + q = float(bt.get(i, "QTY1") or 0) + 1 + out, log = apply_save(f, {"boq_updates": [{"_recno": i, "QTY1": q, "NAME1": bt.get(i, "NAME1")}]}) + spans, _excl = boq_spans(bt) + span_rows = {r for r, a, b, _x in spans if a <= i <= b} + allowed = {(i, c) for c in ("QTY1", "M_AMT1", "L_AMT1", "S_AMT1", "T_AMT1")} + allowed |= {(r, c) for r in span_rows for c in ("M_AMT1", "L_AMT1", "S_AMT1", "T_AMT1")} + cells = _changed_cells(f["BOQ11.DBF"], out["BOQ11.DBF"]) + assert (i, "QTY1") in cells and cells <= allowed + nt = DbfTable(out["BOQ11.DBF"]) + assert float(nt.get(i, "QTY1")) == q + assert nt.get(i, "NAME1") == bt.get(i, "NAME1") # 한글 그대로 + # 소계 행 = 구간 말단 합 · PRJLST TCOST1 = 직접비 + for r, a, b, isx in spans: + tot = sum( + int(float(nt.get(k, "T_AMT1") or 0)) + for k in range(a, b + 1) + if _leaf(nt.get(k, "BDCODE1")) and (isx or k not in _excl) + ) + assert int(float(nt.get(r, "T_AMT1") or 0)) == tot + pt, pt0 = DbfTable(out["PRJLST.DBF"]), DbfTable(f["PRJLST.DBF"]) + direct = sum( + int(float(nt.get(k, "T_AMT1") or 0)) + for k in range(len(nt)) + if _leaf(nt.get(k, "BDCODE1")) and k not in _excl + ) + assert int(float(pt.get(0, "TCOST1"))) == direct + assert {c for _r, c in _changed_cells(f["PRJLST.DBF"], out["PRJLST.DBF"])} <= { + "MCOST1", "LCOST1", "SCOST1", "TCOST1" + } + assert pt.get(0, "PNAME") == pt0.get(0, "PNAME") + for k in f: # 다른 파일은 그대로 + if k not in ("BOQ11.DBF", "PRJLST.DBF"): + assert out[k] == f[k], k + assert {x["table"] for x in log} <= {"BOQ11.DBF", "PRJLST.DBF"} + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_costn_price_edit_propagates_to_linked_boq_rows(stc): + f = _files(stc) + ct, bt = DbfTable(f["COSTN.DBF"]), DbfTable(f["BOQ11.DBF"]) + crow = {ct.get(i, "CODE"): i for i in range(len(ct))} + linked = [ + k + for k in range(len(bt)) + if bt.get(k, "BDCODE1") in crow + and all( + float(bt.get(k, f"{c}AMT1") or 0) == float(ct.get(crow[bt.get(k, "BDCODE1")], f"{c}AMT1") or 0) + for c in "MLS" + ) + and float(bt.get(k, "MAMT1") or 0) > 0 + ] + if not linked: + pytest.skip("단가가 같은 연결 행 없음") + code = bt.get(linked[0], "BDCODE1") + m1 = float(ct.get(crow[code], "MAMT1")) + 10 + out, _log = apply_save(f, {"costn_updates": [{"CODE": code, "MAMT1": m1}]}) + cc = _changed_cells(f["COSTN.DBF"], out["COSTN.DBF"]) + assert cc == {(crow[code], "MAMT1")} + nt = DbfTable(out["BOQ11.DBF"]) + for k in range(len(bt)): + if bt.get(k, "BDCODE1") == code: + want = m1 if k in linked else float(bt.get(k, "MAMT1") or 0) + assert float(nt.get(k, "MAMT1")) == want + k = linked[0] + assert int(float(nt.get(k, "M_AMT1"))) == int(float(nt.get(k, "QTY1")) * m1 + 1e-9) + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_costn_memo_edit_keeps_sixmemo_layout(stc): + f = _files(stc) + memos = extract_costn_memos(f["COSTN.DBF"], f["COSTN.SMT"]) + codes = sorted(memos, key=lambda c: memos[c]["blockNo"]) + grow, shrink = codes[0], codes[-1] + upd = {grow: memos[grow]["text"] + "\r\n'추가 줄' 1'개' =" * 40, shrink: memos[shrink]["text"][:20]} + out, _ = apply_save(f, {"costn_memo_updates": upd}) + smt, smt0 = out["COSTN.SMT"], f["COSTN.SMT"] + assert smt[4:16] == smt0[4:16] + m = SixMemo(smt) + assert m.next_free == -(-len(smt) // 64) # 다음 빈 블록 = 파일 끝 블록 + new = extract_costn_memos(out["COSTN.DBF"], smt) + for c, v in memos.items(): + nl = lambda t: t.replace("\r\n", "\n").replace("\r", "\n") # noqa: E731 + assert nl(new[c]["text"]) == nl(upd.get(c, v["text"])), c + # 늘어난 메모는 뒤에 붙고 줄어든 메모는 제자리 · 다른 메모 블록 바이트 그대로 + assert new[grow]["blockNo"] >= len(smt0) // 64 + assert new[shrink]["blockNo"] == memos[shrink]["blockNo"] + for c, v in memos.items(): + if c not in upd: + st = v["blockNo"] * 64 + assert smt[st : st + v["byteLen"]] == smt0[st : st + v["byteLen"]] + cc = _changed_cells(f["COSTN.DBF"], out["COSTN.DBF"]) + assert {c for _r, c in cc} == {"IFORM"} + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_prjlst_smt_update_writes_valid_memo(stc): + f = _files(stc) + _, prows, _ = parse_dbf_bytes(f["PRJLST.DBF"]) + kv = parse_prjlst_smt(f["PRJLST.SMT"], prows[0]["PRJ_MEMO_HEX"]) + key = next(k for k in kv if k.startswith("1_R_")) + upd = {key: float(kv[key]) + 0.5, "1_Z_FILE_TEST": 7} + out, _ = apply_save(f, {"smt_updates": upd}) + pt = DbfTable(out["PRJLST.DBF"]) + ptr = pt.raw(0, "PRJ_MEMO") + assert ptr_valid(ptr) and ptr_unpack(ptr)[0] == ptr_unpack(bytes.fromhex(prows[0]["PRJ_MEMO_HEX"]))[0] + kv2 = parse_prjlst_smt(out["PRJLST.SMT"], ptr.hex()) + assert float(kv2[key]) == float(kv[key]) + 0.5 and kv2["1_Z_FILE_TEST"] == "7" + assert {k: v for k, v in kv2.items() if k not in upd} == {k: v for k, v in kv.items() if k not in upd} + assert struct.unpack(" 1: # 둘째 공사 메모는 그대로 + m0, m1 = SixMemo(f["PRJLST.SMT"]), SixMemo(out["PRJLST.SMT"]) + assert m0.read(DbfTable(f["PRJLST.DBF"]).raw(1, "PRJ_MEMO")) == m1.read(pt.raw(1, "PRJ_MEMO")) + + +def test_korean_edit_and_unencodable_text(work): + stc = STCS[0] + f = _files(stc) + out, _ = apply_save(f, {"prjlst_updates": {"PNAME": "임도 시설 공사(변경)"}}) + assert DbfTable(out["PRJLST.DBF"]).get(0, "PNAME") == "임도 시설 공사(변경)" + with pytest.raises(Exception): + server.api_save( + server.SaveDbfReq(source_path=str(stc), target_folder_name="bad", prjlst_updates={"PNAME": "工事🙂"}) + ) + + +def test_boq_insert_and_delete_rows(): + stc = STCS[0] + f = _files(stc) + _, brows, _ = parse_dbf_bytes(f["BOQ11.DBF"]) + leaf = next(r for r in brows if _leaf(r["BDCODE1"]) and float(r["TAMT1"] or 0) > 0) + k = leaf["_recno"] + ups = [] + for r in brows: + ups.append({"_recno": r["_recno"]}) + if r["_recno"] == k: + ups.append({"_new": True, "BDCODE1": leaf["BDCODE1"], "NAME1": "끼운 행", "QTY1": 2}) + out, log = apply_save(f, {"boq_updates": ups, "boq_deleted": []}) + nt = DbfTable(out["BOQ11.DBF"]) + assert len(nt) == len(brows) + 1 and nt.get(k + 1, "NAME1") == "끼운 행" + s0, s1, s2 = (float(nt.get(j, "SORTCODE")) for j in (k, k + 1, k + 2)) + assert s0 < s1 < s2 + assert int(float(nt.get(k + 1, "M_AMT1"))) == int(2 * float(leaf["MAMT1"])) + out2, _ = apply_save(out, {"boq_deleted": [k + 1]}) + assert DbfTable(out2["BOQ11.DBF"]).recs == DbfTable(f["BOQ11.DBF"]).recs + assert any(x["field"] == "_insert" for x in log) + + +@pytest.mark.parametrize("stc", STCS, ids=IDS) +def test_bdqty_codec_is_inverse(stc): + b = _files(stc)["BDQTY.DBF"] + t, codec = DbfTable(b), BdqtyCodec() + n = 0 + for i in range(min(len(t), 60)): + for name in t.order: + fd = t.fields[name] + if covers(fd.off, fd.len): + c = bytes(t.recs[i][fd.off : fd.off + fd.len]) + assert codec.encode(codec.decode(c, fd.off), fd.off) == c + n += 1 + assert n > 0 + ct = DbfTable(b, codec=codec) + name = next(x for x in ct.order if ct.fields[x].type == "N" and covers(ct.fields[x].off, ct.fields[x].len)) + v = float(ct.get(0, name) or 0) + assert ct.set(0, name, v + 1) and float(ct.get(0, name)) == v + 1 + assert ct.set(0, name, v) and ct.to_bytes()[32:] == b[32:] diff --git a/tools/stmate-test/api_file.py b/tools/stmate-test/api_file.py new file mode 100644 index 000000000..0e00fafa5 --- /dev/null +++ b/tools/stmate-test/api_file.py @@ -0,0 +1,140 @@ +# -*- coding: utf-8 -*- +"""파일 입출력 끼워 넣기 API — STC 검사 · 복사등록 · 해당분만 압축 · 새 공사 · 백업 목록/복원. + +근거: 24_모듈구조_전수.md(ST_U_Archive Chk_STC_Ver·ComP_Data · wM_D_Restore) · 08번 §2 · 16번 §4.6 · 27번 §2 +쓰기는 tools/stmate-test/work_dbf/ 아래만(server.WORK_DBF_ROOT). +""" + +from __future__ import annotations + +import pathlib + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel +from stc_archive import build_stc, check_stc +from stc_project import ( + copy_register, + list_backups, + new_project, + pack_selected, + safe_name, + write_output, +) +from stc_read import Source + +router = APIRouter() + + +def _root() -> pathlib.Path: + import server # 실행 중 바뀔 수 있는 작업 폴더(시험은 임시 폴더로 바꿈) + + return server.WORK_DBF_ROOT + + +def _src(path: str) -> Source: + try: + return Source(path) + except FileNotFoundError: + raise HTTPException(404, f"경로 없음: {path}") + except ValueError as e: + raise HTTPException(400, str(e)) + + +def _guard(fn, *a, **k): + try: + return fn(*a, **k) + except (ValueError, KeyError) as e: + raise HTTPException(400, str(e)) + + +class PathReq(BaseModel): + source_path: str + + +class CopyRegReq(BaseModel): + source_path: str + src_pcode: str = "11" + mode: str = "copy" # copy · change · contract + pname: str = "" + rates: dict = {} # contract — {"M": %, "L": %, "S": %} + exclude_recnos: list[int] = [] + target_folder_name: str = "" + + +class PackReq(BaseModel): + source_path: str + pcode: str = "11" + target_name: str = "" + + +class NewPrjReq(BaseModel): + template_path: str + pname: str + pnum: str = "" + target_folder_name: str = "" + + +class RestoreReq(BaseModel): + backup_path: str + target_folder_name: str + + +@router.post("/api/file/check") +def api_file_check(req: PathReq): + """Chk_STC_Ver 대응 — ZIP·필수 엔트리·DBF 머리·SMT 서명·판 추정.""" + s = _src(req.source_path) + b = s.path.read_bytes() if s.kind == "stc" else build_stc([], s.files) + return {"kind": s.kind, **check_stc(b)} + + +@router.post("/api/file/copy-register") +def api_file_copy_register(req: CopyRegReq): + s = _src(req.source_path) + files, pcode, changes = _guard( + copy_register, + s.files, + req.src_pcode, + req.mode, + req.pname, + req.rates, + req.exclude_recnos, + ) + target = req.target_folder_name or f"{s.path.stem}_복사{pcode}" + names = dict(s.names, **{f"BOQ{pcode}.DBF": f"BOQ{pcode}.DBF"}) + res = write_output(_root(), target, files, names, s.entries) + return {**res, "new_pcode": pcode, "change_count": len(changes)} + + +@router.post("/api/file/pack") +def api_file_pack(req: PackReq): + """해당분만 압축 — 고른 공사(PCODE)만 담은 STC 를 work_dbf/ 에 만듦.""" + s = _src(req.source_path) + files = _guard(pack_selected, s.files, req.pcode) + name = safe_name(req.target_name or f"{s.path.stem}_{req.pcode}") + out = _root() / f"{name}.STC" + out.parent.mkdir(parents=True, exist_ok=True) + out.write_bytes(build_stc(s.entries, files)) + return {"ok": True, "saved_stc": str(out), "stc_size": out.stat().st_size} + + +@router.post("/api/file/new") +def api_file_new(req: NewPrjReq): + s = _src(req.template_path) + files = _guard(new_project, s.files, req.pname, req.pnum) + res = write_output(_root(), req.target_folder_name or req.pname, files, s.names, s.entries) + return res + + +@router.get("/api/file/backups") +def api_file_backups(name: str = ""): + return {"backups": list_backups(_root(), safe_name(name) if name else "")} + + +@router.post("/api/file/restore") +def api_file_restore(req: RestoreReq): + """wM_D_Restore 대응 — 백업 STC 를 작업본으로 되돌림(지금 작업본은 다시 백업).""" + p = pathlib.Path(req.backup_path).resolve() + if _root().resolve() not in p.parents or not p.exists(): + raise HTTPException(400, "작업 폴더 백업만 복원") + s = _src(str(p)) + return write_output(_root(), req.target_folder_name, s.files, s.names, s.entries) diff --git a/tools/stmate-test/dbf_crypt.py b/tools/stmate-test/dbf_crypt.py new file mode 100644 index 000000000..b295ee332 --- /dev/null +++ b/tools/stmate-test/dbf_crypt.py @@ -0,0 +1,42 @@ +# -*- coding: utf-8 -*- +"""암호 DBF(BDQTY.DBF) 자리별 변환 — 복호(bdqty._dec)의 역변환으로 다시 숨김. + +근거: STC_난독화_상태.md · 29_암호화자료의_뼈대.md · bdqty.py(자리별 표 PARAMS) + A 암호 = rol(평문, r) + k B 암호 = rol(평문 + k, r) X 암호 = rol(평문, r) ^ k +표에 없는 자리(글자 칸 등)는 풀 수도 숨길 수도 없음 — 그 자리를 건드리는 쓰기는 오류. +""" + +from __future__ import annotations + +from bdqty import PARAMS, _dec + + +def _rol(v: int, r: int) -> int: + return ((v << r) | (v >> (8 - r))) & 0xFF if r else v + + +def enc_byte(p: int, param: tuple[str, int, int]) -> int: + kind, r, k = param + if kind == "A": + return (_rol(p, r) + k) & 0xFF + if kind == "B": + return _rol((p + k) & 0xFF, r) + return _rol(p, r) ^ k + + +def covers(off: int, n: int) -> bool: + return all(j in PARAMS for j in range(off, off + n)) + + +class BdqtyCodec: + """DbfTable codec — off = 레코드 안 칸 시작 자리.""" + + def decode(self, cipher: bytes, off: int) -> bytes: + if not covers(off, len(cipher)): + raise ValueError(f"BDQTY 자리 {off}~{off + len(cipher) - 1} 은 복호 표에 없음") + return bytes(_dec(c, PARAMS[off + i]) for i, c in enumerate(cipher)) + + def encode(self, plain: bytes, off: int) -> bytes: + if not covers(off, len(plain)): + raise ValueError(f"BDQTY 자리 {off}~{off + len(plain) - 1} 은 암호 표에 없음") + return bytes(enc_byte(p, PARAMS[off + i]) for i, p in enumerate(plain)) diff --git a/tools/stmate-test/dbf_memo.py b/tools/stmate-test/dbf_memo.py new file mode 100644 index 000000000..4a2d1604b --- /dev/null +++ b/tools/stmate-test/dbf_memo.py @@ -0,0 +1,193 @@ +# -*- coding: utf-8 -*- +"""SIxMemo(COSTN.SMT · PRJLST.SMT · QCODE.SMT) 무손실 읽기·고쳐 쓰기 + PRJLST.SMT 키=값 본문 갱신. + +근거: SIxMemo_구조.md · 25_설정키_전수와_식템플릿.md §1 · 실무 STC 7건 실측 +구조(실측): + [0..4) 다음 빈 블록 번호(LE) = ceil(파일 길이 / 블록 크기) + [4..8) 블록 크기 = 64 + [8..16) "SIxMemo\\0" · 나머지 0 → 512바이트(8블록) 머리 + 메모 = 블록 경계에서 시작 · 블록 패딩 없이 다음 메모는 다음 블록부터 · 마지막 메모 뒤 패딩 없음 + DBF 메모 칸 10바이트 = · 빈 메모 = 공백 10 +고쳐 쓰기(SIx 동작과 같게): 새 본문이 차지하던 블록 수 안에 들면 그 자리에 덮어씀(남는 옛 바이트 유지 — + 실무 COSTN.SMT 패딩에 옛 글자가 남은 것과 같음) · 넘치면 '다음 빈 블록'에 붙이고 머리 갱신. +""" + +from __future__ import annotations + +import re +import struct + +SIG = b"SIxMemo\x00" +BLANK_PTR = b" " * 10 +HEAD_BLOCKS = 8 +DEFAULT_TAG = 8 + + +def ptr_valid(ptr: bytes) -> bool: + if len(ptr) < 10 or ptr[:10] == BLANK_PTR or not ptr[:10].strip(b" \x00"): + return False + _tag, ln, blk = struct.unpack(" 0 and blk >= HEAD_BLOCKS + + +def ptr_unpack(ptr: bytes) -> tuple[int, int, int]: + return struct.unpack(" bytes: + return struct.pack(" int: + return struct.unpack_from(" int: + return struct.unpack_from(" int: + return max(1, -(-n // self.bs)) + + def read(self, ptr: bytes) -> bytes: + if not ptr_valid(ptr): + return b"" + _tag, ln, blk = ptr_unpack(ptr) + return bytes(self.buf[blk * self.bs : blk * self.bs + ln]) + + def write(self, ptr: bytes, data: bytes) -> bytes: + """본문 data 를 쓰고 새 10바이트 포인터 반환(같으면 원래 포인터).""" + if not data: + return BLANK_PTR + valid = ptr_valid(ptr) + tag = ptr_unpack(ptr)[0] if valid else DEFAULT_TAG + if valid and self.read(ptr) == data: + return bytes(ptr[:10]) + self.dirty = True + if valid: + _t, ln, blk = ptr_unpack(ptr) + if self._blocks(len(data)) <= self._blocks(ln): + st = blk * self.bs + if len(self.buf) < st + len(data): + self.buf.extend(b"\x00" * (st + len(data) - len(self.buf))) + self.buf[st : st + len(data)] = data + return ptr_pack(tag, len(data), blk) + blk = max(self.next_free, -(-len(self.buf) // self.bs)) + self.buf.extend(b"\x00" * (blk * self.bs - len(self.buf))) + self.buf.extend(data) + struct.pack_into(" bytes: + return bytes(self.buf) + + +def memo_text_bytes(text: str) -> bytes: + """화면 글(\\n) → 저장 본문(\\r\\n · cp949 엄격).""" + norm = str(text).replace("\r\n", "\n").replace("\r", "\n").replace("\n", "\r\n") + try: + return norm.encode("cp949") + except UnicodeEncodeError as e: + raise ValueError(f"메모에 cp949 로 쓸 수 없는 글자: {e.object[e.start : e.end]!r}") from e + + +def same_memo(a: bytes, b: bytes) -> bool: + n = lambda x: x.rstrip(b"\x00").replace(b"\r\n", b"\n").replace(b"\r", b"\n") # noqa: E731 + return n(a) == n(b) + + +# ── PRJLST.SMT 본문 = [섹션] 과 키=값 줄(CRLF) ── +_KEY_LINE = re.compile(r"^([A-Za-z0-9_]+)=(.*)$") + + +def ini_items(text: str) -> list[tuple[str, str, str]]: + """(섹션, 키, 값) 목록 — 본문 순서 그대로.""" + out, sec = [], "" + for ln in text.splitlines(): + s = ln.strip() + if s.startswith("[") and s.endswith("]"): + sec = s[1:-1] + continue + m = _KEY_LINE.match(s) + if m: + out.append((sec, m.group(1), m.group(2).strip())) + return out + + +def delphi_num(v: float) -> str: + """Delphi FloatToStr 꼴(유효숫자 15) — 실측 '47149628.1818182' · '16.5' · '0'.""" + if float(v) == int(v) and abs(v) < 1e15: + return str(int(v)) + return f"{v:.15g}" + + +def _same_ini_value(old: str, new: object) -> bool: + try: + return float(old) == float(delphi_num(float(new))) # type: ignore[arg-type] + except (TypeError, ValueError): + return old.strip() == str(new).strip() + + +def ini_value_text(new: object) -> str: + if isinstance(new, bool): + return "1" if new else "0" + if isinstance(new, (int, float)): + return delphi_num(float(new)) + return str(new).strip() + + +def ini_update(text: str, updates: dict) -> tuple[str, list[str]]: + """키 값을 그 줄 자리에서 바꿈 · 없는 키는 접두 차수의 [n_KANJUB] 끝(없으면 [KANJUB])에 덧붙임.""" + lines = text.split("\r\n") + changed: list[str] = [] + index = {} + sec_end: dict[str, int] = {} + sec = "" + for i, ln in enumerate(lines): + s = ln.strip() + if s.startswith("[") and s.endswith("]"): + sec = s[1:-1] + continue + m = _KEY_LINE.match(s) + if m: + index.setdefault(m.group(1), i) + sec_end[sec] = i + for k, v in updates.items(): + if v is None: + continue + if k in index: + old = _KEY_LINE.match(lines[index[k]].strip()).group(2) + if _same_ini_value(old, v): + continue + lines[index[k]] = f"{k}={ini_value_text(v)}" + else: + m = re.match(r"^(\d+)_", k) + want = f"{m.group(1)}_KANJUB" if m else "KANJUB" + if want in sec_end: + at = sec_end[want] + 1 + else: + tail = 1 if lines and lines[-1] == "" else 0 + at = len(lines) - tail + lines.insert(at, f"[{want}]") + at += 1 + lines.insert(at, f"{k}={ini_value_text(v)}") + for s2, e in sec_end.items(): + if e >= at: + sec_end[s2] = e + 1 + sec_end[want] = at + index = {kk: (ii + 1 if ii >= at else ii) for kk, ii in index.items()} + index[k] = at + changed.append(k) + return "\r\n".join(lines), changed diff --git a/tools/stmate-test/dbf_rw.py b/tools/stmate-test/dbf_rw.py new file mode 100644 index 000000000..567673ab4 --- /dev/null +++ b/tools/stmate-test/dbf_rw.py @@ -0,0 +1,194 @@ +# -*- coding: utf-8 -*- +"""STMATE DBF 무손실 읽기·고쳐 쓰기 — 손대지 않은 칸은 원래 바이트 그대로. + +근거: STC_컨테이너_구조.md §3·§4(필드명 기반 매핑 · 미지원 칸 원문 보존) · STC_필드사전_핵심.md +규칙: + - 헤더(필드 설명자·끝 0x0D 뒤 꼬리 포함)·레코드·파일 끝(0x1A) 원문 보존 + - 칸 값 비교는 뜻으로 — 숫자는 자리수 맞춰, 글자는 앞뒤 공백 무시, 빈 숫자 = 0 + - 같은 값이면 원래 바이트 유지(앞 공백·빈칸 표기 보존) · 다른 값만 그 칸 폭에 맞춰 씀 + - 글자는 cp949 엄격 인코딩(못 쓰는 글자는 오류 — '?' 로 바꾸지 않음) · 두 바이트 글자 중간에서 자르지 않음 + - 숫자는 실무 STC 7건 전 칸과 같은 꼴(소수 dec 자리 고정 · 오른쪽 맞춤) + - 고친 표만 헤더 날짜(1~3바이트)를 오늘로 — dBase 마지막 갱신일 + - 암호 표(버전 0x06 · BDQTY)는 dbf_crypt 로 자리별 변환 — 표에 없는 자리는 쓰지 않음 +""" + +from __future__ import annotations + +import datetime +import struct +from dataclasses import dataclass + +ENC = "cp949" + + +@dataclass(frozen=True) +class Field: + name: str + type: str + len: int + dec: int + off: int + + +def _num_text(v: float, dec: int) -> str: + return f"{v:.{dec}f}" if dec > 0 else str(int(round(v))) + + +def _to_float(v: object) -> float | None: + if v is None: + return None + s = str(v).strip().replace(",", "") + if s in ("", "None", "null"): + return None + return float(s) + + +def cut_cp949(b: bytes, n: int) -> bytes: + """n 바이트 안으로 자르되 두 바이트 글자 중간을 피함.""" + if len(b) <= n: + return b + i = 0 + while i < n: + step = 2 if b[i] >= 0x81 else 1 + if i + step > n: + break + i += step + return b[:i] + + +class DbfTable: + """DBF 한 개 — 레코드는 bytearray 로 들고 칸 단위로만 고침.""" + + def __init__(self, b: bytes, codec=None): + if len(b) < 32: + raise ValueError("DBF 헤더 길이 부족") + self.version = b[0] + nrec, hlen, rlen = struct.unpack_from(" body_end else b"\x1a" + self.codec = codec + self.dirty = False + + # ── 읽기 ── + def __len__(self) -> int: + return len(self.recs) + + def raw(self, i: int, name: str) -> bytes: + f = self.fields[name] + val = bytes(self.recs[i][f.off : f.off + f.len]) + return self.codec.decode(val, f.off) if self.codec else val + + def get(self, i: int, name: str) -> str: + return self.raw(i, name).decode(ENC, "replace").strip() + + def deleted(self, i: int) -> bool: + return self.recs[i][0:1] == b"*" + + def row(self, i: int) -> dict: + return {n: self.get(i, n) for n in self.order} + + # ── 쓰기 ── + def set_raw(self, i: int, name: str, val: bytes) -> bool: + f = self.fields[name] + val = bytes(val[: f.len]).ljust(f.len, b" ") + if self.codec: + val = self.codec.encode(val, f.off) + if bytes(self.recs[i][f.off : f.off + f.len]) == val: + return False + self.recs[i][f.off : f.off + f.len] = val + self.dirty = True + return True + + def encode_value(self, name: str, value: object) -> bytes: + f = self.fields[name] + if f.type in ("N", "F"): + v = _to_float(value) + if v is None: + return b" " * f.len + txt = _num_text(v, f.dec) + if len(txt) > f.len: + raise ValueError(f"{name} 칸 폭 {f.len} 넘는 숫자: {txt}") + return txt.encode("ascii").rjust(f.len) + if f.type == "L": + s = str(value).strip().upper()[:1] if value is not None else "" + s = {"1": "T", "0": "F", "Y": "T", "N": "F"}.get(s, s) + return (s or " ").encode("ascii").ljust(f.len) + s = "" if value is None else str(value) + try: + enc = s.encode(ENC) + except UnicodeEncodeError as e: + raise ValueError(f"{name} 칸에 cp949 로 쓸 수 없는 글자: {s!r}") from e + return cut_cp949(enc, f.len).ljust(f.len, b" ") + + def same_value(self, i: int, name: str, value: object) -> bool: + f = self.fields[name] + old = self.raw(i, name) + if f.type in ("N", "F"): + ov = _to_float(old.decode("ascii", "replace")) + nv = _to_float(value) + if ov is None or nv is None: + return (ov or 0.0) == (nv or 0.0) + return _num_text(ov, f.dec) == _num_text(nv, f.dec) + if f.type == "L": + return self.encode_value(name, value).strip() == old.strip() + new = "" if value is None else str(value) + return old.decode(ENC, "replace").strip() == new.strip() + + def set(self, i: int, name: str, value: object) -> bool: + """값이 뜻으로 다를 때만 씀 — 바뀌었으면 True.""" + f = self.fields[name] + if f.type == "M" or (f.type not in ("C", "N", "F", "L", "D")): + raise ValueError(f"{name}({f.type}) 칸은 set_raw 로만 씀") + if self.same_value(i, name, value): + return False + return self.set_raw(i, name, self.encode_value(name, value)) + + def blank_record(self) -> bytearray: + rec = bytearray(b" " * self.rlen) + if self.codec: + for f in self.fields.values(): + rec[f.off : f.off + f.len] = self.codec.encode(b" " * f.len, f.off) + return rec + + def insert(self, pos: int, values: dict | None = None, template: int | None = None) -> int: + """pos 자리에 레코드를 끼움(template 레코드 복사 또는 빈칸) — 끼운 자리 반환.""" + rec = bytearray(self.recs[template]) if template is not None else self.blank_record() + rec[0:1] = b" " + pos = max(0, min(pos, len(self.recs))) + self.recs.insert(pos, rec) + self.dirty = True + for k, v in (values or {}).items(): + if k in self.fields and self.fields[k].type != "M": + self.set_raw(pos, k, self.encode_value(k, v)) + return pos + + def to_bytes(self, today: datetime.date | None = None) -> bytes: + head = bytearray(self.head) + struct.pack_into(" dict[str, int]: + """name 칸 값 → 레코드 자리(삭제 레코드 제외 · 같은 값이면 첫 자리).""" + out: dict[str, int] = {} + for i in range(len(tbl)): + if not tbl.deleted(i): + out.setdefault(tbl.get(i, name), i) + return out diff --git a/tools/stmate-test/scr_file.js b/tools/stmate-test/scr_file.js new file mode 100644 index 000000000..cd67dcb1e --- /dev/null +++ b/tools/stmate-test/scr_file.js @@ -0,0 +1,146 @@ +/** + * STMATE w 8.0 — 파일 작업 끼워 넣기 (scr_file.js) · 공사목록표 단추와 1번 메뉴 + * 복사등록(차수/단계 변환) · 해당분만 압축(.STC) · 새 공사 · STC 구조·버전 검사(Chk_STC_Ver) · 백업·복원(wM_D_Restore) + * 서버: api_file.py — 쓰기는 work_dbf/ 아래만 · 단추는 data-file-act 로 문서 전체에서 받음 + */ +(function (global) { + "use strict"; + + const PLUG = (global.STMATE_PLUGINS = global.STMATE_PLUGINS || { + screens: {}, + actions: {}, + menus: [], + }); + const app = () => global.STMATE_APP; + + async function post(url, body) { + const res = await fetch(url, { + method: body === undefined ? "GET" : "POST", + headers: { "Content-Type": "application/json" }, + body: body === undefined ? undefined : JSON.stringify(body), + }); + if (!res.ok) throw new Error(await res.text()); + return res.json(); + } + + const curPath = (ctx) => { + if (!ctx.currentSourcePath) throw new Error("먼저 프로젝트를 불러오세요."); + return ctx.currentSourcePath; + }; + const pickedPcode = (ctx) => { + const el = document.querySelector('input[name="prj-pcode"]:checked'); + const pc = el ? el.value : ctx.uiState.filePcode || "11"; + ctx.uiState.filePcode = pc; + return pc; + }; + + async function openSaved(path) { + const a = app(); + const src = await post("/api/sources"); + if (src && src.projects) a.ctx.sourcesData = src; + const sel = document.getElementById("top-project-select"); + if (sel && src && src.projects) + sel.innerHTML = src.projects + .map((p) => ``) + .join(""); + await a.loadProjectFromServer(path, "wPrjList"); + } + + const ACTS = { + async check(ctx) { + const r = await post("/api/file/check", { source_path: curPath(ctx) }); + const t = Object.entries(r.tables || {}) + .map(([n, h]) => `${n} v${h.ver.toString(16)} ${h.nfield}칸 ${h.nrec}행`) + .join("\n"); + alert( + `[STC 검사] ${r.ok ? "이상 없음" : "문제 있음"} · ${r.edition}\n${(r.problems || []).join("\n")}\n\n${t}`, + ); + }, + async pack(ctx) { + const pc = pickedPcode(ctx); + const name = prompt( + `BOQ${pc} 공사만 담은 STC 이름:`, + `${ctx.state.projectMeta.pname || "공사"}_${pc}`, + ); + if (!name) return; + const r = await post("/api/file/pack", { + source_path: curPath(ctx), + pcode: pc, + target_name: name, + }); + alert(`[해당분만 압축] ${r.saved_stc} (${r.stc_size}B)`); + }, + async copy(ctx) { + const pc = pickedPcode(ctx); + const mode = prompt( + `BOQ${pc} 복사등록 방식 — copy(그대로) · change(변경 차수 변환) · contract(계약 변환):`, + "change", + ); + if (!mode) return; + const body = { source_path: curPath(ctx), src_pcode: pc, mode: mode.trim() }; + if (body.mode === "contract") { + const r = prompt("단가 적용율 % — 재료,노무,경비:", "87.745,87.745,87.745"); + if (!r) return; + const [M, L, S] = r.split(",").map((x) => parseFloat(x) || 0); + body.rates = { M, L, S }; + } + const res = await post("/api/file/copy-register", body); + alert(`[복사등록] 새 공사 BOQ${res.new_pcode} · ${res.saved_dir}`); + await openSaved(res.saved_dir); + }, + async new(ctx) { + const pname = prompt("새 공사명:", ""); + if (!pname) return; + const res = await post("/api/file/new", { template_path: curPath(ctx), pname }); + alert(`[새 공사] 틀(요율·서식·설정)은 지금 공사에서, 내역·단가는 빈 채로 · ${res.saved_dir}`); + await openSaved(res.saved_dir); + }, + async backups(ctx) { + const nm = (ctx.state.projectMeta.pname || "").slice(0, 28); + const r = await post(`/api/file/backups?name=${encodeURIComponent(nm)}`); + const list = r.backups || []; + if (!list.length) return alert("[백업] 이 공사 이름의 작업본 백업이 없습니다."); + const pick = prompt( + `[백업 ${list.length}벌] 되돌릴 번호(취소 = 닫기):\n` + + list.map((b, i) => `${i + 1}. ${b.stamp} (${b.size}B)`).join("\n"), + "1", + ); + const b = list[parseInt(pick, 10) - 1]; + if (!b) return; + const res = await post("/api/file/restore", { + backup_path: b.path, + target_folder_name: b.name, + }); + await openSaved(res.saved_dir); + }, + }; + + async function run(act) { + try { + await ACTS[act](app().ctx); + } catch (err) { + alert("파일 작업 실패: " + err.message); + } + } + + document.addEventListener("click", (e) => { + const el = e.target && e.target.closest && e.target.closest("[data-file-act]"); + if (!el || !ACTS[el.getAttribute("data-file-act")]) return; + e.preventDefault(); + run(el.getAttribute("data-file-act")); + }); + document.addEventListener("change", (e) => { + if (e.target && e.target.name === "prj-pcode") app().ctx.uiState.filePcode = e.target.value; + }); + + [ + ["Act_FileCopyReg", "복사등록 — 차수/단계 변환", "copy"], + ["Act_FilePack", "해당분만 압축하기 (.STC)", "pack"], + ["Act_FileNew", "새 공사 만들기", "new"], + ["Act_FileCheck", "STC 구조·버전 검사 (Chk_STC_Ver)", "check"], + ["Act_FileRestore", "작업본 백업·복원 (wM_D_Restore)", "backups"], + ].forEach(([act, label, key]) => { + PLUG.actions[act] = () => run(key); + PLUG.menus.push({ menu: "m1", label, act }); + }); +})(window); diff --git a/tools/stmate-test/server.py b/tools/stmate-test/server.py index 90fc135d8..b20c67822 100644 --- a/tools/stmate-test/server.py +++ b/tools/stmate-test/server.py @@ -6,27 +6,25 @@ from __future__ import annotations import importlib import pathlib import re -import zipfile -from decimal import Decimal, ROUND_DOWN + +from bdqty import decode_bdqty +from excel_builder import build_stmate_full_xlsx from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import Response from fastapi.staticfiles import StaticFiles -from pydantic import BaseModel - -from bdqty import decode_bdqty -from excel_builder import build_stmate_full_xlsx from kanjub_cfg import build_kanjub_config, stage_slots +from pydantic import BaseModel from smt_parser import ( - build_dbf_bytes, extract_costn_memos, load_stmate_catalog, parse_dbf_bytes, parse_ini_bytes, parse_prjlst_smt, - rebuild_costn_with_smt, - update_prjlst_smt_bytes, ) +from stc_project import safe_name, write_output +from stc_read import Source, extra_tables +from stc_save import apply_save HERE = pathlib.Path(__file__).resolve().parent REPO_ROOT = HERE.parent.parent @@ -93,22 +91,18 @@ def discover_sources() -> dict: return {"projects": projects, "jbr_files": jbr_files, "work_root": str(WORK_DBF_ROOT)} -def read_project_files(source_path: str) -> dict[str, bytes]: - p = pathlib.Path(source_path) - if not p.exists(): +def open_source(source_path: str) -> Source: + try: + return Source(source_path) + except FileNotFoundError: raise HTTPException(status_code=404, detail=f"경로를 찾을 수 없음: {source_path}") - files: dict[str, bytes] = {} - if p.is_file() and p.suffix.upper() == ".STC": - with zipfile.ZipFile(p, "r") as zf: - for nm in zf.namelist(): - files[pathlib.Path(nm).name.upper()] = zf.read(nm) - elif p.is_dir(): - for fp in p.iterdir(): - if fp.is_file(): - files[fp.name.upper()] = fp.read_bytes() - else: - raise HTTPException(status_code=400, detail="지원하지 않는 경로 형식") - return files + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + +def read_project_files(source_path: str) -> dict[str, bytes]: + """STC(ZIP)·폴더 → 대문자 이름 → 바이트 (stc_read.Source).""" + return open_source(source_path).files def _make_syn_resource(res_id: str, name: str, spec: str, unit: str, kind: str, val: float) -> dict: @@ -394,6 +388,7 @@ class SaveDbfReq(BaseModel): prjlst_updates: dict = {} smt_updates: dict = {} costn_memo_updates: dict[str, str] = {} + boq_table: str = "" @app.get("/api/sources") @@ -457,6 +452,7 @@ def api_load(req: LoadReq): "bdqty_rows": decode_bdqty(files.get("BDQTY.DBF", b"")), "start_ini": parse_ini_bytes(files.get("START8X.INI") or files.get("STINI.MEM") or b""), "sim_state": sim_state, + **extra_tables(files), } @@ -502,119 +498,19 @@ def api_load_jbr(req: LoadJbrReq): @app.post("/api/save") def api_save(req: SaveDbfReq): - files = read_project_files(req.source_path) - safe_name = re.sub(r"[^\w\-가-힣]", "_", req.target_folder_name or "stmate_edited").strip("_") - out_dir = WORK_DBF_ROOT / (safe_name or "stmate_edited") - out_dir.mkdir(parents=True, exist_ok=True) - - if "COSTN.DBF" in files: - flds, rows, meta = parse_dbf_bytes(files["COSTN.DBF"]) - upd_map = {str(u.get("CODE", "")).strip(): u for u in req.costn_updates if u.get("CODE")} - for r in rows: - cd = (r.get("CODE") or "").strip() - if cd in upd_map: - for k, v in upd_map[cd].items(): - if k in r: - r[k] = str(v) - dbf_b = build_dbf_bytes(flds, rows, meta) - if req.costn_memo_updates: - dbf_b, smt_b = rebuild_costn_with_smt( - dbf_b, files.get("COSTN.SMT", b""), req.costn_memo_updates - ) - files["COSTN.SMT"] = smt_b - files["COSTN.DBF"] = dbf_b - - boq_name = "BOQ11.DBF" if "BOQ11.DBF" in files else "BOQ12.DBF" - if boq_name in files: - flds, rows, meta = parse_dbf_bytes(files[boq_name]) - upd_by_rec = {int(u["_recno"]): u for u in req.boq_updates if "_recno" in u} - for r in rows: - rno = int(r.get("_recno", -1)) - if rno in upd_by_rec: - u = upd_by_rec[rno] - for k in ( - "ITNUM", - "NAME1", - "SIZE1", - "UNIT1", - "QTY1", - "QTY2", - "MAMT1", - "LAMT1", - "SAMT1", - "TAMT1", - "MAMT2", - "LAMT2", - "SAMT2", - "TAMT2", - "MAMT3", - "LAMT3", - "SAMT3", - "TAMT3", - ): - if k in u and k in r: - r[k] = str(u[k]) - bd = (r.get("BDCODE1") or "").strip() - if bd and not STRUCT_RE.match(bd): - for s_idx in ("1", "2", "3"): - q_k = ( - f"QTY{s_idx}" if f"QTY{s_idx}" in r and r.get(f"QTY{s_idx}") else "QTY1" - ) - q = Decimal(str(fnum(r.get(q_k)))) - m_a = int( - (q * Decimal(str(fnum(r.get(f"MAMT{s_idx}"))))).quantize( - Decimal(1), rounding=ROUND_DOWN - ) - ) - l_a = int( - (q * Decimal(str(fnum(r.get(f"LAMT{s_idx}"))))).quantize( - Decimal(1), rounding=ROUND_DOWN - ) - ) - s_a = int( - (q * Decimal(str(fnum(r.get(f"SAMT{s_idx}"))))).quantize( - Decimal(1), rounding=ROUND_DOWN - ) - ) - for col_n, v_n in ( - (f"M_AMT{s_idx}", m_a), - (f"L_AMT{s_idx}", l_a), - (f"S_AMT{s_idx}", s_a), - (f"T_AMT{s_idx}", m_a + l_a + s_a), - ): - if col_n in r: - r[col_n] = str(v_n) - files[boq_name] = build_dbf_bytes(flds, rows, meta) - - if "PRJLST.DBF" in files and req.prjlst_updates: - flds, rows, meta = parse_dbf_bytes(files["PRJLST.DBF"]) - if rows: - for k, v in req.prjlst_updates.items(): - if k in rows[0]: - rows[0][k] = str(v) - files["PRJLST.DBF"] = build_dbf_bytes(flds, rows, meta) - - if "PRJLST.SMT" in files and req.smt_updates: - files["PRJLST.SMT"] = update_prjlst_smt_bytes(files["PRJLST.SMT"], req.smt_updates) - - written_files = [] - for fname, content in files.items(): - (out_dir / fname).write_bytes(content) - written_files.append({"name": fname, "size": len(content)}) - - stc_path = WORK_DBF_ROOT / f"{out_dir.name}.STC" - with zipfile.ZipFile(stc_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: - for fname, content in files.items(): - zf.writestr(fname, content) - - return { - "ok": True, - "saved_dir": str(out_dir), - "saved_stc": str(stc_path), - "stc_size": stc_path.stat().st_size, - "written_count": len(written_files), - "written_files": written_files, - } + """바뀐 칸만 고쳐 work_dbf/<이름>/ 와 <이름>.STC 로 씀(stc_save · stc_project) — 무변경이면 원본과 같은 바이트.""" + src = open_source(req.source_path) + try: + files, changes = apply_save(src.files, req.model_dump()) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + res = write_output( + WORK_DBF_ROOT, safe_name(req.target_folder_name), files, src.names, src.entries + ) + res["changed_files"] = sorted(k for k in files if files[k] is not src.files.get(k)) + res["changes"] = changes[:500] + res["change_count"] = len(changes) + return res class ExportXlsxReq(BaseModel): diff --git a/tools/stmate-test/smt_parser.py b/tools/stmate-test/smt_parser.py index 062f55e26..46262cc07 100644 --- a/tools/stmate-test/smt_parser.py +++ b/tools/stmate-test/smt_parser.py @@ -1,5 +1,8 @@ # -*- coding: utf-8 -*- -"""STMATE dBASE III/IV(DBF) · SIxMemo(COSTN.SMT / PRJLST.SMT) 바이너리 파서·빌더 및 표준 카탈로그 로더.""" +"""STMATE dBASE III/IV(DBF) · SIxMemo(COSTN.SMT / PRJLST.SMT) 바이너리 파서·빌더 및 표준 카탈로그 로더. + +무손실 고쳐 쓰기는 dbf_rw(DBF 칸) · dbf_memo(SIxMemo 블록) · stc_save(저장) 를 씀 — build_dbf_bytes 는 새 표 만들 때만. +""" from __future__ import annotations @@ -145,14 +148,19 @@ def parse_prjlst_smt(b: bytes, memo_hex: str = "") -> dict[str, str]: return {k: v.strip() for k, v in SMT_KV_RE.findall(body.decode("cp949", "replace"))} -def update_prjlst_smt_bytes(orig_b: bytes, updates: dict[str, str | int | float]) -> bytes: - """PRJLST.SMT의 SIxMemo 헤더(0x200)를 보존하며 키=값 본문을 갱신.""" - header = bytearray(orig_b[:0x200]) if len(orig_b) >= 0x200 else bytearray(b"\x00" * 0x200) - txt = orig_b[0x200:].decode("cp949", "replace") if len(orig_b) > 0x200 else "" - for k, v in updates.items(): - pat = re.compile(rf"(^|\r?\n)({re.escape(k)}=)([^\r\n\x00]*)", re.M) - txt = pat.sub(rf"\g<1>\g<2>{v}", txt) if pat.search(txt) else txt + f"\r\n{k}={v}" - return bytes(header) + txt.encode("cp949", "replace") +def update_prjlst_smt_bytes( + orig_b: bytes, updates: dict[str, str | int | float], memo_hex: str = "" +) -> tuple[bytes, str]: + """PRJ_MEMO 포인터가 가리키는 메모의 키=값만 갱신 — (새 SMT, 새 포인터 hex). SIxMemo 블록 규칙은 dbf_memo.""" + from dbf_memo import SixMemo, ini_update, memo_text_bytes + + memo = SixMemo(orig_b) + ptr = bytes.fromhex(memo_hex) if memo_hex else b" " * 10 + text, keys = ini_update(memo.read(ptr).decode("cp949", "replace"), updates) + if not keys: + return orig_b, ptr.hex() + new_ptr = memo.write(ptr, memo_text_bytes(text)) + return memo.to_bytes(), new_ptr.hex() def extract_costn_memos(dbf_bytes: bytes, smt_bytes: bytes) -> dict[str, dict]: @@ -272,68 +280,14 @@ def parse_smt_formula_text(text: str) -> dict: def rebuild_costn_with_smt( dbf_bytes: bytes, smt_bytes: bytes, memo_updates: dict[str, str] ) -> tuple[bytes, bytes]: - """수정된 단가산출 텍스트(memo_updates: {CODE: text})를 COSTN.SMT에 64B 정렬로 추가하고 IFORM 갱신.""" + """단가산출 글(memo_updates: {CODE: 글})을 COSTN.SMT 에 쓰고 IFORM 갱신 — 제자리 덮기/뒤에 붙이기(dbf_memo).""" + from stc_save import apply_save + if not memo_updates or len(dbf_bytes) < 32: return dbf_bytes, smt_bytes - - dbf_buf = bytearray(dbf_bytes) - num_records, header_len, record_len = struct.unpack(" len(dbf_buf): - break - code = ( - dbf_buf[rec_start + code_off : rec_start + code_off + code_len] - .decode("cp949", errors="ignore") - .strip() - ) - if code not in memo_updates: - continue - new_text = memo_updates[code] - encoded = ( - new_text.replace("\r\n", "\n").replace("\n", "\r\n").encode("cp949", errors="replace") - ) - byte_len = len(encoded) - pad_head = (-len(smt_buf)) % block_size - if pad_head: - smt_buf.extend(b"\x00" * pad_head) - block_no = len(smt_buf) // block_size - smt_buf.extend(encoded) - pad_tail = (-len(smt_buf)) % block_size - if pad_tail: - smt_buf.extend(b"\x00" * pad_tail) - pointer_bytes = struct.pack(" tuple[int, int]: + return (dt.hour << 11) | (dt.minute << 5) | (dt.second // 2), ((dt.year - 1980) << 9) | ( + dt.month << 5 + ) | dt.day + + +def read_entries(b: bytes) -> list[dict]: + """중앙 목록 차례대로 {name, meta, data(풀린), raw_local(로컬 헤더+압축 바이트)}.""" + at = b.rfind(EOCD) + if at < 0: + raise ValueError("ZIP 끝 레코드 없음 — STC 아님") + _sig, _d, _cd, _n1, n, _size, cd_off, _clen = struct.unpack_from("<4sHHHHIIH", b, at) + out, p = [], cd_off + for _ in range(n): + f = struct.unpack_from(_CEN_FMT, b, p) + (_s, vm, vn, flag, meth, tm, dt, crc, csz, usz, nl, xl, cl, _dk, ia, ea, lho) = f + raw_name = b[p + 46 : p + 46 + nl] + extra = b[p + 46 + nl : p + 46 + nl + xl] + comment = b[p + 46 + nl + xl : p + 46 + nl + xl + cl] + p += 46 + nl + xl + cl + lf = struct.unpack_from(_LOC_FMT, b, lho) + lnl, lxl = lf[9], lf[10] + dstart = lho + 30 + lnl + lxl + comp = b[dstart : dstart + csz] + data = zlib.decompress(comp, -15) if meth == 8 else comp + if zlib.crc32(data) != crc: + raise ValueError(f"CRC 불일치: {raw_name!r}") + enc = "utf-8" if flag & 0x800 else "cp949" + out.append( + { + "name": raw_name.decode(enc, "replace"), + "raw_name": raw_name, + "meta": { + "ver_made": vm, + "ver_need": vn, + "flag": flag, + "method": meth, + "time": tm, + "date": dt, + "iattr": ia, + "eattr": ea, + "extra": extra, + "comment": comment, + }, + "data": data, + "raw_local": b[lho : dstart + csz], + "crc": crc, + "csize": csz, + } + ) + return out + + +def files_of(entries: list[dict]) -> dict[str, bytes]: + return {pathlib.Path(e["name"]).name.upper(): e["data"] for e in entries} + + +def build_stc(entries: list[dict], files: dict[str, bytes], now=None) -> bytes: + """entries(원본 차례·머리 값)에 files(대문자 이름 → 내용)를 얹어 STC 바이트로. + + files 에 없는 엔트리는 뺌 · entries 에 없는 files 는 STD_ORDER 자리에 기본 머리 값으로 더함. + """ + now = now or datetime.datetime.now() + by_up = {pathlib.Path(e["name"]).name.upper(): e for e in entries} + names = [pathlib.Path(e["name"]).name.upper() for e in entries if e.get("name")] + names = [n for n in names if n in files] + std_up = [s.upper() for s in STD_ORDER] + for n in sorted(files, key=lambda x: (std_up.index(x) if x in std_up else 99, x)): + if n not in names: + pos = len(names) + if n in std_up: + rank = std_up.index(n) + pos = next( + (k for k, m in enumerate(names) if m in std_up and std_up.index(m) > rank), + len(names), + ) + names.insert(pos, n) + body, cen = bytearray(), bytearray() + for n in names: + data = files[n] + e = by_up.get(n) + off = len(body) + if e is not None and e["data"] == data: + local = e["raw_local"] + m, crc, csz = e["meta"], e["crc"], e["csize"] + raw_name = e["raw_name"] + else: + m = dict(e["meta"]) if e else dict(DEFAULT_META, extra=b"", comment=b"") + if not e: + proper = next((s for s in STD_ORDER if s.upper() == n), n) + raw_name = proper.encode("cp949") + else: + raw_name = e["raw_name"] + m["time"], m["date"] = dos_time(now) + comp = data + if m["method"] == 8: + co = zlib.compressobj(9, zlib.DEFLATED, -15) + comp = co.compress(data) + co.flush() + crc, csz = zlib.crc32(data), len(comp) + local = ( + struct.pack( + _LOC_FMT, + LOC, + m["ver_need"], + m["flag"] & ~0x08, + m["method"], + m["time"], + m["date"], + crc, + csz, + len(data), + len(raw_name), + 0, + ) + + raw_name + + comp + ) + body += local + cen += struct.pack( + _CEN_FMT, + CEN, + m["ver_made"], + m["ver_need"], + m["flag"] & ~0x08, + m["method"], + m["time"], + m["date"], + crc, + csz, + len(data), + len(raw_name), + len(m.get("extra", b"")), + len(m.get("comment", b"")), + 0, + m["iattr"], + m["eattr"], + off, + ) + cen += raw_name + m.get("extra", b"") + m.get("comment", b"") + eocd = struct.pack("<4sHHHHIIH", EOCD, 0, 0, len(names), len(names), len(cen), len(body), 0) + return bytes(body + cen + eocd) + + +# ── 구조·버전 검사 ── +REQUIRED = ("PRJLST.DBF", "PRJLST.SMT", "COSTN.DBF", "COSTN.SMT", "BOQ11.DBF") +SCHEMA_VARIANTS = { # STC_컨테이너_구조.md §4 — 표본에 있는 필드 수 + "BOQ11.DBF": (83, 94), + "COSTN.DBF": (99, 101), + "PRJLST.DBF": (47, 48, 55), + "BDQTY.DBF": (37, 38), +} + + +def _dbf_head(b: bytes) -> dict: + n, hlen, rlen = struct.unpack_from(" list[str]: + """메모 포인터가 SMT 안(머리 뒤 · 파일 끝 안)을 가리키는지.""" + from dbf_memo import ptr_unpack, ptr_valid + from dbf_rw import DbfTable + + t = DbfTable(dbf) + if fld not in t.fields or smt[8:15] != b"SIxMemo": + return [] + bs = struct.unpack_from(" len(smt) + return [f"{name}.{fld} 메모 포인터 {bad}개가 SMT 밖"] if bad else [] + + +def check_stc(b: bytes) -> dict: + """ZIP·필수 엔트리·DBF 머리·SMT 서명·메모 포인터 범위 검사와 판(구형/2020 확장) 추정.""" + res: dict = {"ok": True, "problems": [], "tables": {}, "edition": ""} + try: + ents = read_entries(b) + except Exception as e: # noqa: BLE001 + return {"ok": False, "problems": [f"ZIP 열기 실패: {e}"], "tables": {}, "edition": ""} + files = files_of(ents) + res["entries"] = [e["name"] for e in ents] + for r in REQUIRED: + if r not in files: + res["problems"].append(f"필수 엔트리 없음: {r}") + for n, data in files.items(): + if n.endswith(".DBF"): + h = _dbf_head(data) + res["tables"][n] = {k: v for k, v in h.items() if k != "fields"} + if not h["ok"]: + res["problems"].append(f"{n} 머리 길이 불일치") + if n in SCHEMA_VARIANTS and h["nfield"] not in SCHEMA_VARIANTS[n]: + res["problems"].append(f"{n} 필드 수 {h['nfield']} — 표본에 없던 판") + elif n.endswith(".SMT") and data[8:15] != b"SIxMemo": + res["problems"].append(f"{n} SIxMemo 서명 없음") + for dbf, smt, fld in ( + ("COSTN.DBF", "COSTN.SMT", "IFORM"), + ("PRJLST.DBF", "PRJLST.SMT", "PRJ_MEMO"), + ): + if dbf in files and smt in files: + res["problems"] += _memo_problems(files[dbf], files[smt], fld, dbf) + prj = _dbf_head(files["PRJLST.DBF"])["fields"] if "PRJLST.DBF" in files else [] + res["edition"] = "8.0 2020 확장(5차 칸)" if "MCOST5" in prj else "8.0 구형(4차 칸)" + res["ok"] = not res["problems"] + return res diff --git a/tools/stmate-test/stc_project.py b/tools/stmate-test/stc_project.py new file mode 100644 index 000000000..f53140d7d --- /dev/null +++ b/tools/stmate-test/stc_project.py @@ -0,0 +1,267 @@ +# -*- coding: utf-8 -*- +"""공사 단위 파일 작업 — 작업본 쓰기·자동 백업/복원 · 복사등록(차수/단계 변환) · 해당분만 압축 · 새 공사. + +근거: 08_규칙6_8_9_재계산_변경설계_일괄변경.md §2(통합형 BOQ11+BOQ12 · 분리형 새 STC) · + 16_골든셋_회귀_결과.md §4.5·§4.6(새 파일 BOQ11 차수1 = 앞 파일 BOQ12) · 27_계약_실행_기성_단계.md §2(계약 변환) · + 24_모듈구조_전수.md(ST_U_Archive ComP_Data · wM_D_Restore) · SIxMemo_구조.md +규칙: + - PRJLST 한 행 = 공사 하나 · PCODE '1n' ↔ BOQ1n.DBF · PRJ_MEMO = 그 공사의 원가 설정(PRJLST.SMT) + - 복사등록: 원본 BOQ1n 을 새 BOQ1m 으로 복사 + PRJLST 행·메모 복사(m = 빈 번호) + copy 그대로 + change 변경 차수 변환 — K_ORD k → k+1 · 내역 (k+1)차 칸 → (k+2)차 칸 복사 · [k+1_KANJUB] → [k+2_KANJUB] + contract 계약 변환 — 1차 단가 × 비목별 적용율(%) 버림 · 제외 행(C* 구역 + 고른 행) 그대로 · 0% = 공내역 + - 해당분만 압축: 고른 공사만 BOQ11·PRJLST 한 행·PRJLST.SMT 한 메모(8블록부터)로 새 STC + - 저장마다 같은 이름 작업본 STC 가 있으면 _backup/ 에 시각 붙여 남김(최근 10벌) +""" + +from __future__ import annotations + +import datetime +import pathlib +import re +import shutil +from decimal import ROUND_DOWN + +from dbf_memo import BLANK_PTR, SixMemo, ini_items, memo_text_bytes +from dbf_rw import DbfTable +from stc_archive import build_stc +from stc_save import SaveLog, _dec, _leaf, _row_amounts, boq_spans, recompute_boq + +BACKUP_KEEP = 10 +SLOT_COL = re.compile( + r"^(QTY|TAMT|LAMT|MAMT|SAMT|T_AMT|L_AMT|M_AMT|S_AMT|EXIST|BDCODE|AAMT|A_AMT)(\d)$" +) + + +def safe_name(name: str) -> str: + return re.sub(r"[^\w\-가-힣]", "_", name or "stmate_edited").strip("_") or "stmate_edited" + + +def backup_dir(root: pathlib.Path) -> pathlib.Path: + return root / "_backup" + + +def make_backup(root: pathlib.Path, name: str) -> str: + stc = root / f"{name}.STC" + if not stc.exists(): + return "" + bd = backup_dir(root) + bd.mkdir(parents=True, exist_ok=True) + stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f") + dst = bd / f"{name}__{stamp}.STC" + shutil.copy2(stc, dst) + olds = sorted(bd.glob(f"{name}__*.STC")) + for p in olds[:-BACKUP_KEEP]: + p.unlink() + return str(dst) + + +def list_backups(root: pathlib.Path, name: str = "") -> list[dict]: + bd = backup_dir(root) + if not bd.exists(): + return [] + pat = f"{name}__*.STC" if name else "*__*.STC" + return [ + { + "name": p.stem.split("__")[0], + "path": str(p), + "size": p.stat().st_size, + "stamp": p.stem.split("__")[1], + } + for p in sorted(bd.glob(pat), reverse=True) + ] + + +def write_output(root: pathlib.Path, target: str, files: dict, names: dict, entries: list) -> dict: + """작업본 폴더(원래 이름 대소문자) + 같은 이름 STC — 덮기 전 백업.""" + name = safe_name(target) + out_dir = root / name + out_dir.mkdir(parents=True, exist_ok=True) + backup = make_backup(root, name) + written = [] + for up, data in files.items(): + fname = names.get(up, up) + (out_dir / fname).write_bytes(data) + written.append({"name": fname, "size": len(data)}) + stc_path = root / f"{name}.STC" + stc_path.write_bytes(build_stc(entries, files)) + return { + "ok": True, + "saved_dir": str(out_dir), + "saved_stc": str(stc_path), + "stc_size": stc_path.stat().st_size, + "backup": backup, + "written_count": len(written), + "written_files": written, + } + + +# ── 공사(PRJLST 행) 다루기 ── +def _prj_index(prj: DbfTable, pcode: str) -> int: + for i in range(len(prj)): + if prj.get(i, "PCODE") == pcode: + return i + raise ValueError(f"PRJLST 에 PCODE {pcode} 공사 없음") + + +def _free_pcode(files: dict, prj: DbfTable) -> str: + used = {prj.get(i, "PCODE") for i in range(len(prj))} | { + k[3:5] for k in files if re.match(r"^BOQ1\d\.DBF$", k) + } + for n in range(11, 20): + if str(n) not in used: + return str(n) + raise ValueError("빈 공사 번호(BOQ11~19) 없음") + + +def _rename_section(text: str, src: int, dst: int) -> str: + """[src_KANJUB] 섹션을 [dst_KANJUB] 로 복사해 덧붙임(이미 있으면 그대로).""" + secs = {} + for sec, k, v in ini_items(text): + secs.setdefault(sec, []).append((k, v)) + if f"{dst}_KANJUB" in secs or f"{src}_KANJUB" not in secs: + return text + body = [f"[{dst}_KANJUB]"] + [ + f"{re.sub(rf'^{src}_', f'{dst}_', k)}={v}" for k, v in secs[f"{src}_KANJUB"] + ] + sep = "" if text.endswith("\r\n") or not text else "\r\n" + return text + sep + "\r\n".join(body) + + +def _copy_slot(tbl: DbfTable, a: str, b: str) -> None: + cols = [] + for c in tbl.order: + m = SLOT_COL.match(c) + if m and m.group(2) == a and f"{m.group(1)}{b}" in tbl.fields: + cols.append((c, f"{m.group(1)}{b}")) + for i in range(len(tbl)): + for x, y in cols: + tbl.set_raw(i, y, tbl.raw(i, x)) + + +def _contract(tbl: DbfTable, name: str, rates: dict, exclude: set[int], log: SaveLog) -> None: + _spans, excl = boq_spans(tbl) + for i in range(len(tbl)): + bd = tbl.get(i, "BDCODE1") + if not _leaf(bd) or i in excl or i in exclude: + continue + parts = [] + for c in "MLS": + r = _dec(rates.get(c, 100)) / 100 + v = (_dec(tbl.get(i, f"{c}AMT1")) * r).to_integral_value(rounding=ROUND_DOWN) + log.set(tbl, name, i, f"{c}AMT1", v) + parts.append(v) + log.set(tbl, name, i, "TAMT1", sum(parts)) + _row_amounts(tbl, name, i, "1", {}, log) + + +def copy_register( + files: dict, src_pcode: str, mode: str = "copy", pname: str = "", rates=None, exclude=() +) -> tuple[dict, str, list]: + """복사등록 — (새 files, 새 PCODE, 바뀐 칸 목록).""" + out = dict(files) + prj = DbfTable(files["PRJLST.DBF"]) + si = _prj_index(prj, src_pcode) + src_tbl = f"BOQ{src_pcode}.DBF" + if src_tbl not in files: + raise ValueError(f"{src_tbl} 없음") + new = _free_pcode(files, prj) + new_tbl = f"BOQ{new}.DBF" + log = SaveLog() + bt = DbfTable(files[src_tbl]) + bt.dirty = True + k = int(float(prj.get(si, "K_ORD") or 0)) if "K_ORD" in prj.fields else 0 + memo = SixMemo(files.get("PRJLST.SMT")) + text = memo.read(prj.raw(si, "PRJ_MEMO")).decode("cp949", "replace") + ni = prj.insert(len(prj), {}, template=si) + prj.set(ni, "PCODE", new) + if pname: + prj.set(ni, "PNAME", pname) + if mode == "change": + a, b = str(k + 1), str(k + 2) + _copy_slot(bt, a, b) + for c in ("MCOST", "LCOST", "SCOST", "TCOST", "ACOST", "COSTW"): + if f"{c}{a}" in prj.fields and f"{c}{b}" in prj.fields: + prj.set_raw(ni, f"{c}{b}", prj.raw(si, f"{c}{a}")) + prj.set(ni, "K_ORD", k + 1) + text = _rename_section(text, k + 1, k + 2) + elif mode == "contract": + _contract(bt, new_tbl, rates or {}, set(exclude), log) + direct = recompute_boq(bt, new_tbl, {"1"}, log) + for c, v in zip(("MCOST1", "LCOST1", "SCOST1", "TCOST1"), direct.get("1", [])): + prj.set(ni, c, v) + elif mode != "copy": + raise ValueError(f"모르는 복사등록 방식: {mode}") + ptr = memo.write(BLANK_PTR, memo_text_bytes(text)) if text else BLANK_PTR + prj.set_raw(ni, "PRJ_MEMO", ptr) + out[new_tbl] = bt.to_bytes() + out["PRJLST.DBF"] = prj.to_bytes() + out["PRJLST.SMT"] = memo.to_bytes() + return out, new, log.items + + +def pack_selected(files: dict, pcode: str) -> dict: + """해당분만 압축 — 고른 공사만 BOQ11 · PRJLST 한 행(PCODE 11) · 그 메모 하나.""" + out = {k: v for k, v in files.items() if not re.match(r"^BOQ1\d\.DBF$", k)} + prj = DbfTable(files["PRJLST.DBF"]) + si = _prj_index(prj, pcode) + src = files.get(f"BOQ{pcode}.DBF") + if src is None: + raise ValueError(f"BOQ{pcode}.DBF 없음") + out["BOQ11.DBF"] = src + memo = SixMemo(files.get("PRJLST.SMT")) + body = memo.read(prj.raw(si, "PRJ_MEMO")) + fresh = SixMemo(None, memo.bs) + prj.recs = [prj.recs[si]] + prj.dirty = True + prj.set(0, "PCODE", "11") + prj.set_raw(0, "PRJ_MEMO", fresh.write(BLANK_PTR, body) if body else BLANK_PTR) + out["PRJLST.DBF"] = prj.to_bytes() + out["PRJLST.SMT"] = fresh.to_bytes() + return out + + +def _empty(b: bytes) -> bytes: + t = DbfTable(b) + t.recs, t.dirty = [], True + return t.to_bytes() + + +def new_project(template: dict, pname: str, pnum: str = "") -> dict: + """새 공사 — 틀 프로젝트의 표 구조·요율표·서식·설정은 두고 내역·단가·구성·수량은 비움.""" + out = dict(template) + for k in list(out): + if re.match(r"^BOQ1[2-9]\.DBF$", k): + del out[k] + for k in ("BOQ11.DBF", "COSTN.DBF", "BDQTY.DBF", "QCODE.DBF", "COSTA.DBF"): + if k in out: + out[k] = _empty(out[k]) + out["COSTN.SMT"] = SixMemo(None).to_bytes() + out["QCODE.SMT"] = SixMemo(None).to_bytes() + prj = DbfTable(template["PRJLST.DBF"]) + memo = SixMemo(template.get("PRJLST.SMT")) + text = memo.read(prj.raw(0, "PRJ_MEMO")).decode("cp949", "replace") + keep, sec = [], "" + for ln in text.split("\r\n"): + s = ln.strip() + if s.startswith("["): + sec = s[1:-1] + if sec in ("KANJUB", "1_KANJUB") and ln: + m = re.match(r"^(1_C_\w+)=", s) + keep.append(f"{m.group(1)}=0" if m else ln) + prj.recs = [prj.recs[0]] + prj.dirty = True + for f in prj.order: + fd = prj.fields[f] + if fd.type in ("N", "F") and re.match(r"^[MLSTA]COST\d$|^COSTW\d$", f): + prj.set(0, f, 0) + prj.set(0, "PNAME", pname) + prj.set(0, "PNUM", pnum) + prj.set(0, "PCODE", "11") + if "K_ORD" in prj.fields: + prj.set(0, "K_ORD", 0) + fresh = SixMemo(None) + prj.set_raw(0, "PRJ_MEMO", fresh.write(BLANK_PTR, memo_text_bytes("\r\n".join(keep)))) + out["PRJLST.DBF"] = prj.to_bytes() + out["PRJLST.SMT"] = fresh.to_bytes() + return out diff --git a/tools/stmate-test/stc_read.py b/tools/stmate-test/stc_read.py new file mode 100644 index 000000000..f95f80915 --- /dev/null +++ b/tools/stmate-test/stc_read.py @@ -0,0 +1,108 @@ +# -*- coding: utf-8 -*- +"""STMATE 프로젝트 원천 읽기 — STC/폴더 → (대문자 이름 → 바이트) · 보조 표(KONGJONG·COSTA·QCODE) · 설정 전수. + +근거: STC_컨테이너_구조.md §2·§5 · 02_데이터셋과_파일형식.md §3 · 25_설정키_전수와_식템플릿.md §1 · + 20_설정파일과_보고서_카탈로그.md §2(Start8x.Ini) +""" + +from __future__ import annotations + +import pathlib + +from dbf_memo import SixMemo, ini_items +from dbf_rw import DbfTable +from stc_archive import files_of, read_entries + + +class Source: + """원천 한 곳 — files(대문자 이름 → 바이트) · names(대문자 → 원래 이름) · entries(STC 머리 값).""" + + def __init__(self, path: str): + p = pathlib.Path(path) + if not p.exists(): + raise FileNotFoundError(path) + self.path = p + self.entries: list[dict] = [] + self.names: dict[str, str] = {} + if p.is_file() and p.suffix.upper() == ".STC": + self.entries = read_entries(p.read_bytes()) + self.files = files_of(self.entries) + self.names = {pathlib.Path(e["name"]).name.upper(): e["name"] for e in self.entries} + elif p.is_dir(): + self.files = {} + for fp in sorted(p.iterdir()): + if fp.is_file(): + self.files[fp.name.upper()] = fp.read_bytes() + self.names[fp.name.upper()] = fp.name + else: + raise ValueError("지원하지 않는 경로 형식") + self.kind = "stc" if self.entries else "folder" + + +def table_rows(b: bytes) -> list[dict]: + """평문 DBF 행(삭제 제외) — 암호 표(0x06)는 빈 목록.""" + if not b or b[0] == 0x06: + return [] + t = DbfTable(b) + return [dict(t.row(i), _recno=i) for i in range(len(t)) if not t.deleted(i)] + + +def ini_sections(text: str) -> dict[str, dict[str, str]]: + out: dict[str, dict[str, str]] = {} + for sec, k, v in ini_items(text): + out.setdefault(sec, {})[k] = v + return out + + +def prjlst_smt_all(files: dict[str, bytes]) -> list[dict]: + """PRJLST 행마다 PRJ_MEMO 본문 전체(섹션별 키=값) — PCODE 로 BOQ1n 과 짝.""" + if "PRJLST.DBF" not in files: + return [] + t = DbfTable(files["PRJLST.DBF"]) + memo = SixMemo(files.get("PRJLST.SMT")) + out = [] + for i in range(len(t)): + ptr = t.raw(i, "PRJ_MEMO") if "PRJ_MEMO" in t.fields else b"" + text = memo.read(ptr).decode("cp949", "replace") + out.append( + { + "recno": i, + "pcode": t.get(i, "PCODE") if "PCODE" in t.fields else "", + "sections": ini_sections(text), + "key_count": len(ini_items(text)), + } + ) + return out + + +def qcode_rows(files: dict[str, bytes]) -> list[dict]: + """QCODE.DBF 행 + IFORM·XCODE 메모 본문(QCODE.SMT).""" + b = files.get("QCODE.DBF") + if not b: + return [] + t = DbfTable(b) + memo = SixMemo(files.get("QCODE.SMT")) + rows = [] + for i in range(len(t)): + if t.deleted(i): + continue + r = dict(t.row(i), _recno=i) + for f in ("IFORM", "XCODE"): + if f in t.fields: + r[f] = memo.read(t.raw(i, f)).decode("cp949", "replace") + rows.append(r) + return rows + + +def extra_tables(files: dict[str, bytes]) -> dict: + """api_load 덧붙임 — KONGJONG 공종 · COSTA(암호면 건수만) · QCODE · PRJLST.SMT 전수 · Start8x.Ini 섹션.""" + costa = files.get("COSTA.DBF", b"") + ini = files.get("START8X.INI") or files.get("STINI.MEM") or b"" + return { + "kongjong_rows": table_rows(files.get("KONGJONG.DBF", b"")), + "costa_rows": table_rows(costa), + "costa_count": len(DbfTable(costa)) if costa else 0, + "qcode_rows": qcode_rows(files), + "prjlst_smt_all": prjlst_smt_all(files), + "start_ini_sections": ini_sections(ini.decode("cp949", "replace")), + } diff --git a/tools/stmate-test/stc_save.py b/tools/stmate-test/stc_save.py new file mode 100644 index 000000000..890facee7 --- /dev/null +++ b/tools/stmate-test/stc_save.py @@ -0,0 +1,334 @@ +# -*- coding: utf-8 -*- +"""STMATE 프로젝트 저장 — 바뀐 칸만 고쳐 씀(무손실) · 단가 → 내역 → 소계 → PRJLST 합계 파생 갱신. + +근거: STC_왕복검증.md §3(export ① 원본을 틀로 알려진 칸만 갱신) · SIxMemo_구조.md · + 16_골든셋_회귀_결과.md §4(행 금액 = ROUNDDOWN(수량×단가) · 계층 집계 · C* 제외 구역 · TCOSTn = 직접비) +요청 키(SaveDbfReq): costn_updates[{CODE, 칸…}] · costn_memo_updates{CODE: 글} · + boq_table · boq_updates[{_recno | _new+_after, 칸…}] · prjlst_updates{칸} · smt_updates{키: 값} +파생 규칙(실무 STC 7건 전 행에서 다시 셈 = 저장값 확인): + - 단가 전파: COSTN n차 단가가 바뀐 코드 → 그 코드 내역 행 중 옛 단가가 COSTN 옛 단가와 같던 행만 새 단가 + - 행 금액: 그 행 n차 수량·단가가 바뀌었을 때만 ROUNDDOWN(QTYn × ?AMTn) · T_AMTn = 재+노+경 + - 소계: 금액이 바뀐 차수만 SORTCODE↔ENDCODE 짝 구간 합(C* 구역은 C 행·구역 안 행만 넣음) + - PRJLST ?COSTn: 그 차수 직접비(C* 구역 밖 말단 행 합) +""" + +from __future__ import annotations + +import re +from decimal import ROUND_DOWN, Decimal, InvalidOperation + +from dbf_memo import SixMemo, ini_update, memo_text_bytes, same_memo +from dbf_rw import DbfTable, find_rows + +STRUCT_RE = re.compile(r"^(T\d|E\d|C\d|:)$") +SLOTS = ("1", "2", "3", "4", "5") + + +def _dec(s: str) -> Decimal: + try: + return Decimal(str(s).strip() or "0") + except InvalidOperation: + return Decimal(0) + + +def pick_boq_table(files: dict[str, bytes], want: str = "") -> str: + """api_load 와 같은 고름 — 지정 > BOQ11 > 첫 BOQ 표.""" + want = (want or "").upper() + if want and want in files: + return want + boqs = sorted(k for k in files if k.startswith("BOQ") and k.endswith(".DBF")) + return "BOQ11.DBF" if "BOQ11.DBF" in files else (boqs[0] if boqs else "") + + +def prj_row_for(prj: DbfTable, boq_name: str) -> int: + """BOQ1n 표 ↔ PRJLST 행(PCODE = 1n) — 없으면 첫 행.""" + pc = boq_name[3:5] if boq_name.startswith("BOQ") else "" + for i in range(len(prj)): + if "PCODE" in prj.fields and prj.get(i, "PCODE") == pc: + return i + return 0 + + +class SaveLog: + def __init__(self): + self.items: list[dict] = [] + + def add(self, table: str, recno: int, field: str, old: str, new: str): + self.items.append({"table": table, "recno": recno, "field": field, "old": old, "new": new}) + + def set(self, tbl: DbfTable, table: str, i: int, field: str, value) -> bool: + old = tbl.get(i, field) + if tbl.set(i, field, value): + self.add(table, i, field, old, tbl.get(i, field)) + return True + return False + + +def _leaf(bd: str) -> bool: + return bool(bd) and not STRUCT_RE.match(bd) + + +def boq_spans(tbl: DbfTable) -> tuple[list[tuple[int, int, int, bool]], set[int]]: + """(행, 구간 시작, 구간 끝, 제외 구역 합인가) 목록과 C* 제외 구역 행 집합.""" + n = len(tbl) + bds = [tbl.get(i, "BDCODE1") for i in range(n)] + sort_at = {} + for i in range(n): + sort_at.setdefault(tbl.get(i, "SORTCODE"), []).append(i) + excl: set[int] = set() + partner = {} + for i in range(n): + if not bds[i] or not STRUCT_RE.match(bds[i]) or bds[i] == ":": + continue + j = next((j for j in sort_at.get(tbl.get(i, "ENDCODE"), []) if j != i), None) + if j is not None: + partner[i] = j + if bds[i].startswith("C"): + excl |= set(range(min(i, j), max(i, j) + 1)) + spans = [ + (i, min(i, j), max(i, j), bds[i].startswith("C") or i in excl) for i, j in partner.items() + ] + return spans, excl + + +def recompute_boq(tbl: DbfTable, name: str, slots: set[str], log: SaveLog) -> dict[str, list]: + """바뀐 차수의 소계 행 다시 셈 · 차수별 직접비(재·노·경·계) 반환.""" + spans, excl = boq_spans(tbl) + bds = [tbl.get(i, "BDCODE1") for i in range(len(tbl))] + direct: dict[str, list] = {} + for s in sorted(slots): + cols = [f"{c}_AMT{s}" for c in "MLST"] + if not all(c in tbl.fields for c in cols): + continue + vals = [[_dec(tbl.get(i, c)) for c in cols] for i in range(len(tbl))] + for i, a, b, isx in spans: + tot = [Decimal(0)] * 4 + for k in range(a, b + 1): + if _leaf(bds[k]) and (isx or k not in excl): + tot = [x + y for x, y in zip(tot, vals[k])] + for c, v in zip(cols, tot): + log.set(tbl, name, i, c, v) + direct[s] = [ + sum( + (vals[i][n] for i in range(len(tbl)) if _leaf(bds[i]) and i not in excl), Decimal(0) + ) + for n in range(4) + ] + return direct + + +def _row_amounts(tbl: DbfTable, name: str, i: int, s: str, given, log: SaveLog) -> None: + """given = 요청이 직접 바꾼 칸(그 칸은 다시 셈하지 않음).""" + if f"QTY{s}" not in tbl.fields: + return + q = _dec(tbl.get(i, f"QTY{s}")) + parts = [] + for c in "MLS": + if f"{c}AMT{s}" not in tbl.fields: + return + v = (q * _dec(tbl.get(i, f"{c}AMT{s}"))).quantize(Decimal(1), rounding=ROUND_DOWN) + parts.append(v) + if f"{c}_AMT{s}" not in given: + log.set(tbl, name, i, f"{c}_AMT{s}", v) + if f"T_AMT{s}" not in given: + log.set(tbl, name, i, f"T_AMT{s}", sum(parts)) + + +def _apply_fields(tbl: DbfTable, name: str, i: int, upd: dict, log: SaveLog) -> set[str]: + done = set() + for k, v in upd.items(): + if k.startswith("_") or k not in tbl.fields or tbl.fields[k].type == "M": + continue + if log.set(tbl, name, i, k, v): + done.add(k) + return done + + +def _sortcode_between(tbl: DbfTable, prev_i: int | None, next_i: int | None) -> Decimal: + a = _dec(tbl.get(prev_i, "SORTCODE")) if prev_i is not None else Decimal(0) + b = _dec(tbl.get(next_i, "SORTCODE")) if next_i is not None else a + 512 + return ((a + b) / 2).quantize(Decimal("0.0001")) + + +def _arrange(bt: DbfTable, ups: list[dict], deleted: set[int]) -> tuple[list, bool]: + """행 차례 정하기 — (원래 자리 | None(새 행), 요청) 목록 · 차례·끼움·지움이 없으면 False.""" + n0 = len(bt) + seq, seen = [], set() + for u in ups: + r = int(u.get("_recno", -1)) if str(u.get("_recno", "")).lstrip("-").isdigit() else -1 + if u.get("_new") or not (0 <= r < n0): + seq.append((None, u)) + elif r not in deleted and r not in seen: + seq.append((r, u)) + seen.add(r) + olds = [r for r, _u in seq if r is not None] + moved = olds != sorted(olds) or any(r is None for r, _u in seq) + moved = moved or bool(deleted & set(range(n0))) + if not moved: + by = {r: u for r, u in seq} + return [(r, by.get(r, {})) for r in range(n0)], False + for r in range(n0): # 요청에 없던 원래 행 — 앞 원래 행 바로 뒤 + if r in seen or r in deleted: + continue + at = 0 + for k, (q, _u) in enumerate(seq): + if q is not None and q < r: + at = k + 1 + seq.insert(at, (r, {})) + return seq, True + + +def _save_boq(b: bytes, name: str, req: dict, cctx, log: SaveLog, today): + """내역 표 저장 — (새 바이트 | None, 차수별 직접비).""" + costn_old, costn_new, costn_slots = cctx + bt = DbfTable(b) + deleted = {int(x) for x in (req.get("boq_deleted") or [])} + seq, moved = _arrange(bt, list(req.get("boq_updates") or []), deleted) + if moved: # 끼움·지움·차례 바꿈 — 레코드 다시 늘어놓음(원래 행 바이트는 그대로 옮김) + tmpl = {bt.get(i, "BDCODE1"): i for i in range(len(bt))} + recs = [] + for r, u in seq: + if r is not None: + recs.append(bt.recs[r]) + continue + code = str(u.get("BDCODE1", "")).strip() + src = tmpl.get(code) + rec = bytearray(bt.recs[src]) if src is not None else bt.blank_record() + rec[0:1] = b" " + recs.append(rec) + for r in sorted(deleted): + if 0 <= r < len(bt): + log.add(name, r, "_delete", bt.get(r, "BDCODE1"), "") + bt.recs, bt.dirty = recs, True + touched: dict[int, set[str]] = {} + given: dict[int, set] = {} # 요청이 직접 바꾼 칸 — 새 행은 비움(금액 늘 다시 셈) + for i, (r, u) in enumerate(seq): + if r is None: + vals = {k: v for k, v in u.items() if not k.startswith("_")} + code = str(vals.get("BDCODE1", "")).strip() + for c in "TMLS": # 새 행 1차 단가 = COSTN 단가(주지 않은 칸) + if f"{c}AMT1" not in vals and f"{c}AMT1" in costn_new.get(code, {}): + vals[f"{c}AMT1"] = costn_new[code][f"{c}AMT1"] + for k, v in vals.items(): + if k in bt.fields and bt.fields[k].type != "M": + bt.set_raw(i, k, bt.encode_value(k, v)) + if "SORTCODE" not in vals and "SORTCODE" in bt.fields: + prev = i - 1 if i > 0 else None + nxt = i + 1 if i + 1 < len(seq) and seq[i + 1][0] is not None else None + bt.set(i, "SORTCODE", _sortcode_between(bt, prev, nxt)) + touched[i] = set(vals) | {f"QTY{s}" for s in SLOTS} + given[i] = set() + log.add(name, i, "_insert", "", code) + elif u: + touched[i] = _apply_fields(bt, name, i, u, log) + given[i] = set(touched[i]) + for i in range(len(bt)): # 단가 전파 + code = bt.get(i, "BDCODE1") + for s in costn_slots.get(code, ()): + old = costn_old[code] + if any(f"{c}AMT{s}" in given.get(i, set()) for c in "MLST"): + continue + cols = [f"{c}AMT{s}" for c in "MLS" if f"{c}AMT{s}" in bt.fields] + if all(_dec(bt.get(i, c)) == _dec(old.get(c, "0")) for c in cols): + for c in "TMLS": + col = f"{c}AMT{s}" + if col in bt.fields and log.set(bt, name, i, col, costn_new[code].get(col)): + touched.setdefault(i, set()).add(col) + slots: set[str] = set(SLOTS) if moved else set() + for i, done in touched.items(): # 행 금액 · 바뀐 차수 + if not _leaf(bt.get(i, "BDCODE1")): + continue + g = given.get(i, set()) + for s in SLOTS: + units = {f"MAMT{s}", f"LAMT{s}", f"SAMT{s}"} + if done & (units | {f"QTY{s}"}): + if done & units and f"TAMT{s}" not in g and f"TAMT{s}" in bt.fields: + t = sum(_dec(bt.get(i, f"{c}AMT{s}")) for c in "MLS") + log.set(bt, name, i, f"TAMT{s}", t) + _row_amounts(bt, name, i, s, g, log) + slots.add(s) + if done & {f"{c}_AMT{s}" for c in "MLST"}: + slots.add(s) + direct = recompute_boq(bt, name, slots, log) if slots else {} + return (bt.to_bytes(today) if bt.dirty else None), direct + + +def apply_save(files: dict[str, bytes], req: dict, today=None) -> tuple[dict[str, bytes], list]: + """files(대문자 이름 → 바이트)에 저장 요청을 반영한 새 files 와 바뀐 칸 목록.""" + out = dict(files) + log = SaveLog() + costn_old: dict[str, dict[str, str]] = {} + costn_slots: dict[str, set[str]] = {} + costn_new: dict[str, dict[str, str]] = {} + + if "COSTN.DBF" in files: + ct = DbfTable(files["COSTN.DBF"]) + rows = find_rows(ct, "CODE") + price_cols = [f"{c}AMT{s}" for s in SLOTS for c in "TMLS" if f"{c}AMT{s}" in ct.fields] + for code, i in rows.items(): + costn_old[code] = {c: ct.get(i, c) for c in price_cols} + for u in req.get("costn_updates") or []: + code = str(u.get("CODE", "")).strip() + if code not in rows: + continue + done = _apply_fields(ct, "COSTN.DBF", rows[code], u, log) + for f in done: + m = re.match(r"^[TMLS]AMT(\d)$", f) + if m: + costn_slots.setdefault(code, set()).add(m.group(1)) + memo_upd = req.get("costn_memo_updates") or {} + if memo_upd and "IFORM" in ct.fields: + memo = SixMemo(files.get("COSTN.SMT")) + for code, text in memo_upd.items(): + i = rows.get(str(code).strip()) + if i is None or text is None: + continue + ptr = ct.raw(i, "IFORM") + data = memo_text_bytes(text) + if same_memo(memo.read(ptr), data): + continue + new_ptr = memo.write(ptr, data) + if ct.set_raw(i, "IFORM", new_ptr): + log.add("COSTN.DBF", i, "IFORM", ptr.hex(), new_ptr.hex()) + log.add("COSTN.SMT", i, code, "", f"{len(data)}B") + if memo.dirty: + out["COSTN.SMT"] = memo.to_bytes() + costn_new = {code: {c: ct.get(i, c) for c in price_cols} for code, i in rows.items()} + if ct.dirty: + out["COSTN.DBF"] = ct.to_bytes(today) + + boq_name = pick_boq_table(files, req.get("boq_table", "")) + direct: dict[str, list] = {} + if boq_name: + ctx = (costn_old, costn_new, costn_slots) + new_b, direct = _save_boq(files[boq_name], boq_name, req, ctx, log, today) + if new_b is not None: + out[boq_name] = new_b + + if "PRJLST.DBF" in files: + pt = DbfTable(files["PRJLST.DBF"]) + pi = prj_row_for(pt, boq_name) if len(pt) else None + if pi is not None: + upd = dict(req.get("prjlst_updates") or {}) + for s, (m, lab, sv, t) in direct.items(): + for c, v in (("MCOST", m), ("LCOST", lab), ("SCOST", sv), ("TCOST", t)): + if f"{c}{s}" in pt.fields and f"{c}{s}" not in upd: + upd[f"{c}{s}"] = v + _apply_fields(pt, "PRJLST.DBF", pi, upd, log) + smt_upd = req.get("smt_updates") or {} + if smt_upd and "PRJ_MEMO" in pt.fields: + memo = SixMemo(files.get("PRJLST.SMT")) + ptr = pt.raw(pi, "PRJ_MEMO") + text = memo.read(ptr).decode("cp949", "replace") + new_text, keys = ini_update(text, smt_upd) + if keys: + new_ptr = memo.write(ptr, memo_text_bytes(new_text)) + if pt.set_raw(pi, "PRJ_MEMO", new_ptr): + log.add("PRJLST.DBF", pi, "PRJ_MEMO", ptr.hex(), new_ptr.hex()) + for k in keys: + log.add("PRJLST.SMT", pi, k, "", str(smt_upd[k])) + if memo.dirty: + out["PRJLST.SMT"] = memo.to_bytes() + if pt.dirty: + out["PRJLST.DBF"] = pt.to_bytes(today) + return out, log.items diff --git a/tools/stmate-test/view_prj_boq.js b/tools/stmate-test/view_prj_boq.js index d426a25f7..a57c9cae9 100644 --- a/tools/stmate-test/view_prj_boq.js +++ b/tools/stmate-test/view_prj_boq.js @@ -172,8 +172,11 @@ - - + + + + + @@ -242,6 +245,8 @@ + ${renderPrjRows(ctx)} + ${ showPreview ? ` @@ -295,6 +300,18 @@ `; } + // 열린 프로젝트의 공사(PRJLST 행 · PCODE 1n ↔ BOQ1n) — 복사등록·해당분 압축 대상 고름(scr_file.js) + function renderPrjRows(ctx) { + const rows = (ctx.dbfData && ctx.dbfData.prjlst_all) || []; + if (!rows.length) return ""; + const pick = ctx.uiState.filePcode || String(rows[0].PCODE || "11").trim(); + const tr = (r) => { + const pc = String(r.PCODE || "").trim(); + return `BOQ${pc}${r.PNAME || ""}${KORD_BADGES[num(r.K_ORD)] || r.K_ORD || ""}${fmt(num(r.TCOST1))}${fmt(num(r.TCOST2))}${fmt(num(r.COSTW1))}`; + }; + return `
공사 (PRJLST ${rows.length}행) — 고른 공사로 복사등록·해당분 압축
${rows.map(tr).join("")}
고름표공사명단계(K_ORD)직공비 1차직공비 2차총공사비 1차
`; + } + /** * 2. wBoq (계층 설계내역서 편집 화면) * - BOQ11~14.DBF의 모든 행(T1~T6 공종제목, E1~E6 소계/계/총계/총합계, C* 제외구역, 말단 공종행) 100% 표시 @@ -549,114 +566,116 @@ }); } + /** + * 저장 요청(/api/save) — 지금 값을 그대로 보내고 서버(stc_save.py)가 원본과 뜻으로 견줘 바뀐 칸만 씀 + * 보내지 않는 것: 1차 단가로 2차 칸 덮기 · 고정 낙찰율 3차 칸 · 소계·PRJLST 직접비(서버가 다시 셈) + * 근거: STC_왕복검증.md §3(알려진 칸만 갱신) · 16번 §4(행 금액·계층 집계·TCOSTn) + */ + const COSTN_SAVE_COL = /^(NAME|SIZE|UNIT|AMT_\d1|[TMLS]AMT[1-5])$/; + const isPctRow = (r) => + String(r.UNIT1 || "") + .trim() + .endsWith("%") && num(r.T_AMT1) > 0; + const plainFields = (r) => + Object.fromEntries( + Object.entries(r).filter( + ([k, v]) => + !k.startsWith("_") && !k.endsWith("_HEX") && k !== "IFORM" && typeof v !== "object", + ), + ); + + function itemQtyFields(b, raw) { + const u = {}; + if (!b) return u; + if (isPctRow(raw)) { + (b.pctQtys || []).forEach((q, k) => q != null && (u[`QTY${k + 1}`] = q)); + return u; + } + if (b.qty !== undefined) u.QTY1 = b.qty; + const q = Array.isArray(b.qtys) ? b.qtys : null; + for (let n = 2; n <= 4; n++) if (q && q[n - 1] !== undefined) u[`QTY${n}`] = q[n - 1]; + if (q && num(q[1]) > 0 && b.changedQty !== undefined && num(b.changedQty) !== num(q[1])) + u.QTY2 = b.changedQty; // 변경 수량 칸만 고친 화면(rev-edit-qty2) + Object.entries(b.units || {}).forEach(([n, mls]) => { + if (num(n) < 2 || !Array.isArray(mls)) return; + ["M", "L", "S"].forEach((c, k) => (u[`${c}AMT${n}`] = mls[k])); + }); + return u; + } + + function newItemRow(b) { + const q = Array.isArray(b.qtys) ? b.qtys : [b.qty, b.changedQty, 0, 0]; + const u = { _new: true, BDCODE1: b.code || b.refId, NAME1: b.name, SIZE1: b.spec }; + Object.assign(u, { UNIT1: b.unit, ITNUM: b.itnum || "", ENDCODE: 0 }); + q.forEach((v, k) => (u[`QTY${k + 1}`] = v || 0)); + return Object.assign(u, itemQtyFields(b, u)); + } + function buildSavePayload(ctx) { - const kj = ctx.computed.kanjubComputed; - const revKj = ctx.computed.revComputed.revKj; - const bt = ctx.computed.boqComputed.totals; - const revBt = ctx.computed.revComputed.revBoq.totals; - const fullBoq = evaluateFullBoqRows(ctx); - const rts = ctx.state.kanjubConfig.rates; - const [bm, bl, bs] = [ - (ctx.uiState.bidRateM || 87.745) / 100, - (ctx.uiState.bidRateL || 87.745) / 100, - (ctx.uiState.bidRateS || 87.745) / 100, - ]; - const revByRec = Object.fromEntries( - (ctx.state.boqItems || []).map((b) => [b._recno, b.changedQty ?? b.qty]), + const d = ctx.dbfData || {}; + const comp = ctx.computed || {}; + const items = (ctx.state.boqItems || []).filter((b) => b._recno !== undefined && !b.isNew); + const byRec = new Map(items.map((b) => [b._recno, b])); + const fresh = (ctx.state.boqItems || []).filter((b) => b.isNew && b._recno === undefined); + const boqUpdates = []; + (d.boq_rows || []).forEach((r) => { + const isNew = r._new || r.SORTCODE === undefined; + const u = Object.assign(plainFields(r), itemQtyFields(byRec.get(r._recno), r)); + boqUpdates.push( + isNew ? Object.assign(u, { _new: true }) : Object.assign(u, { _recno: r._recno }), + ); + fresh + .filter((b) => b.parentRecno === r._recno) + .forEach((b) => boqUpdates.push(newItemRow(b))); + }); + fresh.filter((b) => b.parentRecno == null).forEach((b) => boqUpdates.push(newItemRow(b))); + + const prj = d.prjlst || {}; + const prjUpd = { PNAME: ctx.state.projectMeta.pname, PNUM: ctx.state.projectMeta.pnum }; + Object.entries(comp.stages || {}).forEach(([n, st]) => { + const g = st && st.kanjub && st.kanjub.grandTotal; + if ( + g !== undefined && + (n === "1" || num(prj[`COSTW${n}`]) > 0 || num(prj.K_ORD) >= num(n) - 1) + ) + prjUpd[`COSTW${n}`] = g; + }); + if (!comp.stages && comp.kanjubComputed) prjUpd.COSTW1 = comp.kanjubComputed.GRAND_TOTAL; + + const smt = d.prjlst_smt || {}; + const rts = (ctx.state.kanjubConfig && ctx.state.kanjubConfig.rates) || {}; + const smtCand = { + "1_R_KANNO": rts.KANNO, + "1_R_SANBO": rts.SANBO, + "1_R_KOBO": rts.KOBO, + "1_R_KUNBO": rts.KUNBO, + "1_R_SLVBO": rts.SLVBO, + "1_R_YUNBO": rts.YUNBO, + "1_R_ANJUN": rts.ANJUN, + "1_R_T_KITA": rts.KITA, + "1_R_HWAN": rts.HWAN, + "1_R_EQIPSU": rts.EQIP, + "1_R_ILBAN": rts.ILBAN, + "1_R_IYUN": rts.IYUN, + "1_C_ANJUN_G": ctx.state.kanjubConfig && ctx.state.kanjubConfig.anjunBaseAdd, + }; + const smtUpd = Object.fromEntries( + Object.entries(smtCand).filter(([k, v]) => k in smt && v !== undefined && v !== null), ); return { source_path: ctx.currentSourcePath, target_folder_name: (ctx.state.projectMeta.pname || "stmate_edited").slice(0, 28), - costn_updates: (ctx.dbfData.costn_rows || []).map((r) => { - const m1 = num(r.MAMT1), - l1 = num(r.LAMT1), - s1 = num(r.SAMT1), - t1 = m1 + l1 + s1 || num(r.TAMT1); - const m3 = Math.floor(m1 * bm), - l3 = Math.floor(l1 * bl), - s3 = Math.floor(s1 * bs); - return { - CODE: r.CODE, - MAMT1: m1, - LAMT1: l1, - SAMT1: s1, - TAMT1: t1, - MAMT2: m1, - LAMT2: l1, - SAMT2: s1, - TAMT2: t1, - MAMT3: m3, - LAMT3: l3, - SAMT3: s3, - TAMT3: m3 + l3 + s3, - AMT_61: t1, - }; - }), - boq_updates: fullBoq.map((r) => ({ - _recno: r._recno, - ITNUM: r.ITNUM || "", - NAME1: r.NAME1 || "", - SIZE1: r.SIZE1 || "", - UNIT1: r.UNIT1 || "", - QTY1: r.QTY1, - QTY2: revByRec[r._recno] !== undefined ? revByRec[r._recno] : r.QTY1, - MAMT1: r.MAMT1, - LAMT1: r.LAMT1, - SAMT1: r.SAMT1, - TAMT1: r.TAMT1, - MAMT2: r.MAMT1, - LAMT2: r.LAMT1, - SAMT2: r.SAMT1, - TAMT2: r.TAMT1, - MAMT3: Math.floor(num(r.MAMT1) * bm), - LAMT3: Math.floor(num(r.LAMT1) * bl), - SAMT3: Math.floor(num(r.SAMT1) * bs), - TAMT3: - Math.floor(num(r.MAMT1) * bm) + - Math.floor(num(r.LAMT1) * bl) + - Math.floor(num(r.SAMT1) * bs), - })), - prjlst_updates: { - PNAME: ctx.state.projectMeta.pname, - PNUM: ctx.state.projectMeta.pnum, - MCOST1: bt.directM, - LCOST1: bt.directL, - SCOST1: bt.directS, - TCOST1: bt.directT, - COSTW1: kj.GRAND_TOTAL, - MCOST2: revBt.directM, - LCOST2: revBt.directL, - SCOST2: revBt.directS, - TCOST2: revBt.directT, - COSTW2: revKj.GRAND_TOTAL, - TCOST3: Math.floor(bt.directT * bm), - COSTW3: Math.floor(kj.GRAND_TOTAL * bm), - }, - smt_updates: { - "1_R_KANNO": rts.KANNO, - "1_R_SANBO": rts.SANBO, - "1_R_KOBO": rts.KOBO, - "1_R_KUNBO": rts.KUNBO, - "1_R_SLVBO": rts.SLVBO, - "1_R_YUNBO": rts.YUNBO, - "1_R_ANJUN": rts.ANJUN, - "1_R_T_KITA": rts.KITA, - "1_R_HWAN": rts.HWAN, - "1_R_EQIPSU": rts.EQIP, - "1_R_ILBAN": rts.ILBAN, - "1_R_IYUN": rts.IYUN, - "1_C_ANJUN_G": ctx.state.kanjubConfig.anjunBaseAdd, - "1_M_MAT": bt.directM, - "1_M_LAB": bt.directL + kj.KANNO, - "1_M_SAN": bt.directS + (kj.IND_EXP - kj.KANNO), - "1_M_ILBAN": kj.ILBAN, - "1_M_IYUN": kj.IYUN, - "1_M_VAT": kj.VAT, - "1_M_TOT": kj.FINAL_SUPPLY, - "1_M_KWNGP": kj.GOV_MAT, - "2_M_TOT": revKj.FINAL_SUPPLY, - }, + boq_table: d.boq_table_name || "", + costn_updates: (d.costn_rows || []).map((r) => + Object.fromEntries( + Object.entries(r).filter(([k]) => k === "CODE" || COSTN_SAVE_COL.test(k)), + ), + ), costn_memo_updates: ctx.uiState.editedSmtMap || {}, + boq_updates: boqUpdates, + boq_deleted: d.boq_deleted || ctx.uiState.boqDeleted || [], + prjlst_updates: prjUpd, + smt_updates: smtUpd, }; }