# -*- 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{키: 값} · kanjub_stages{차수: 설정} · bdqty_rows[구성행] · qcode_rows[수량코드행](stc_save_extra — 바뀐 것만) · prjlst_rows[공사목록 전체 행 · _recno] — 다른 공사 행도 바뀐 칸만(지금 공사 행은 prjlst_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 from smt_parser import parse_ini_bytes from stc_save_extra import insert_costn_rows, kanjub_smt, save_bdqty, save_qcode 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 _PR_COMP = {"1": "M", "2": "L", "3": "S", "4": "T"} def _pct_rows(tbl: DbfTable, name: str, s: str, log: SaveLog) -> None: """% 줄 다시 셈 — 단가 칸 = 밑수 · 금액 = 밑수 × 율(수량 칸, %) 원 미만 절사. 코드 붙은 줄(조달수수료 M%·T%) = 같은 상위 공종 안 앞 말단 줄 비목 합(server.pct_row_base) PR 줄 = 앞 PR 줄 뒤 표시(PRPT) 줄의 밑수 비목(PRIN) 합 → 대상 비목(PROUT) (calc.js prRow) """ if not all(f in tbl.fields for f in (f"QTY{s}", f"MAMT{s}", f"M_AMT{s}", "UNIT1")): return bds = [tbl.get(i, "BDCODE1") for i in range(len(tbl))] amt = lambda i, c: ( # noqa: E731 sum(_dec(tbl.get(i, f"{k}_AMT{s}")) for k in "MLS") if c == "T" else _dec(tbl.get(i, f"{c}_AMT{s}")) ) marked = {"M": Decimal(0), "L": Decimal(0), "S": Decimal(0)} for i in range(len(tbl)): bd, unit = bds[i], tbl.get(i, "UNIT1").strip() if not _leaf(bd): continue if bd == "PR" and "PRIN" in tbl.fields: bk, ok = ( _PR_COMP.get(tbl.get(i, "PRIN").strip()), _PR_COMP.get(tbl.get(i, "PROUT").strip()), ) if bk and ok: base = sum(marked.values()) if bk == "T" else marked[bk] _set_pct(tbl, name, i, s, "M" if ok == "T" else ok, base, log) marked = {k: Decimal(0) for k in marked} continue elif unit.endswith("%") and unit[:1] in "MLST": base, depth = Decimal(0), [] for j in range(i - 1, -1, -1): b = bds[j] if re.match(r"^E\d$", b): depth.append(b[1]) elif re.match(r"^T\d$", b): if depth and depth[-1] == b[1]: depth.pop() else: break elif _leaf(b): base += amt(j, unit[0]) own = "L" if bd[:1] == "L" else "S" if bd[:1] == "S" else "M" _set_pct(tbl, name, i, s, own, base, log) continue if "PRPT" in tbl.fields and tbl.get(i, "PRPT").strip() not in ("", "0"): for k in marked: marked[k] += _dec(tbl.get(i, f"{k}_AMT{s}")) def _set_pct(tbl: DbfTable, name: str, i: int, s: str, own: str, base: Decimal, log: SaveLog): q = _dec(tbl.get(i, f"QTY{s}")) a = (base * q / 100).quantize(Decimal(1), rounding=ROUND_DOWN) for c in "MLS": log.set(tbl, name, i, f"{c}AMT{s}", base if c == own else 0) log.set(tbl, name, i, f"{c}_AMT{s}", a if c == own else 0) log.set(tbl, name, i, f"T_AMT{s}", a) 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 _pct_rows(tbl, name, s, log) 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} if insert_costn_rows(ct, list(req.get("costn_updates") or []), log): rows = find_rows(ct, "CODE") 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 for r in req.get("prjlst_rows") or []: # 공사목록 전체 행(scr_ui_prj) — 바뀐 칸만 k = int(r["_recno"]) if str(r.get("_recno", "")).isdigit() else -1 vals = {f: v for f, v in r.items() if not f.endswith("_HEX") and f != "PRJ_MEMO"} if k == pi: upd = {**{f: v for f, v in vals.items() if f not in upd}, **upd} elif 0 <= k < len(pt): _apply_fields(pt, "PRJLST.DBF", k, vals, log) _apply_fields(pt, "PRJLST.DBF", pi, upd, log) smt_upd = req.get("smt_updates") or {} stages = req.get("kanjub_stages") or {} if (smt_upd or stages) 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") smt_upd = kanjub_smt(text, stages, smt_upd) if stages else smt_upd 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) # 구성표(BDQTY · 암호) · 수량코드(QCODE) — 화면 행이 불러온 행과 다를 때만 if req.get("bdqty_rows") is not None and "BDQTY.DBF" in files: new_b = save_bdqty(files["BDQTY.DBF"], list(req["bdqty_rows"]), log) if new_b is not None: out["BDQTY.DBF"] = new_b if req.get("qcode_rows") is not None: out.update(save_qcode(files, list(req["qcode_rows"]), log)) # 초기값(Start8x.Ini · STini.MEM) — 있는 키 중 바뀐 값만 그 줄 자리에서 고침 ini_req = req.get("start_ini") or {} for name in ("START8X.INI", "STINI.MEM") if ini_req else (): if name not in files: continue orig = parse_ini_bytes(files[name]) upd = {k: str(v) for k, v in ini_req.items() if k in orig and str(v) != orig[k]} if upd: new_text, keys = ini_update(files[name].decode("cp949", "replace"), upd) out[name] = new_text.encode("cp949", "replace") for k in keys: log.add(name, 0, k, orig.get(k, ""), upd[k]) return out, log.items