Files
Aislo/tools/stmate-test/stc_save.py
T
eomsangdonandClaude Opus 5.5 086750b6d4 feat(stmate-test): 2차 화면·코드관리 합침 · % 줄(조달수수료) 계산·저장 재계산 · 줄 고정 단가 보존 · 장부
- 코드 등록·삭제·변경·병합·데이타병합·일괄변경·검색·동일코드·풀어넣기 끝전 등(ui_code · ui_merge · api_ui_code)
- 코드 붙은 % 줄 = 같은 상위 공종 앞 말단 줄 비목 합 × 율(server pct_row_base · calc.js) · 저장 때 다시 셈(stc_save _pct_rows) — 실무 7건 저장값 그대로
- 내역 줄 단가 ≠ 단가목록이면 줄 고정 단가(fixedUnit)
- 메뉴 94개 눌러 오류 0 · 시험 502 통과

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-13 20:10:16 +09:00

434 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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