feat(stmate-test): implement full STMATE w 8.0 UI, COSTN.SMT binary parser, and 100% lossless DBF/STC engine
This commit is contained in:
@@ -0,0 +1,375 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""STMATE dBASE III/IV(DBF) · SIxMemo(COSTN.SMT / PRJLST.SMT) 바이너리 파서·빌더 및 표준 카탈로그 로더."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import datetime
|
||||
import io
|
||||
import json
|
||||
import pathlib
|
||||
import re
|
||||
import struct
|
||||
|
||||
REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
|
||||
ANAL_DIR = REPO_ROOT / "resources" / "knowledge" / "original" / "경쟁사 프로그램" / "STmate 분석"
|
||||
RAW_30_DIR = ANAL_DIR / "30_원자료"
|
||||
|
||||
RE_REF = re.compile(r"&([A-Z]\d{4,5})([LMS])?&")
|
||||
RE_PA = re.compile(r"&P([A-Z]\d{2}):([LMS])&")
|
||||
RE_ASSIGN = re.compile(r"\b([A-Za-z]\w*)\s*(?:'[^']*')?\s*=\s*([^,\r\n'=]+)")
|
||||
RE_LABEL = re.compile(r":=\s*([A-Z]\d{2})")
|
||||
RE_PCT = re.compile(r"([\d.]+)\s*'%'")
|
||||
RE_QTY = re.compile(r"([\d.]+)\s*'([^']{1,12})'")
|
||||
SMT_KV_RE = re.compile(r"([A-Za-z0-9_]+)=([^\r\n\x00]+)")
|
||||
|
||||
|
||||
def parse_dbf_bytes(b: bytes) -> tuple[list[dict], list[dict], dict]:
|
||||
"""dBASE III/IV 바이트를 파싱하여 (필드정의, 레코드목록, 헤더메타) 반환."""
|
||||
if len(b) < 32:
|
||||
raise ValueError("DBF 헤더 길이 부족")
|
||||
ver = b[0]
|
||||
nrec, hlen, rlen = struct.unpack_from("<IHH", b, 4)
|
||||
fields: list[dict] = []
|
||||
off = 32
|
||||
while off < len(b) and b[off] != 0x0D:
|
||||
name = b[off : off + 11].split(b"\x00", 1)[0].decode("cp949", "replace").strip()
|
||||
fields.append(
|
||||
{
|
||||
"name": name,
|
||||
"type": chr(b[off + 11]),
|
||||
"len": b[off + 16],
|
||||
"dec": b[off + 17],
|
||||
"raw_desc": b[off : off + 32].hex(),
|
||||
}
|
||||
)
|
||||
off += 32
|
||||
|
||||
rows: list[dict] = []
|
||||
for i in range(nrec):
|
||||
pos = hlen + i * rlen
|
||||
if pos + rlen > len(b):
|
||||
break
|
||||
rec = b[pos : pos + rlen]
|
||||
deleted = rec[0:1] == b"*"
|
||||
row: dict = {"_recno": i, "_deleted": deleted}
|
||||
cur = 1
|
||||
for f in fields:
|
||||
val_bytes = rec[cur : cur + f["len"]]
|
||||
cur += f["len"]
|
||||
if f["name"] == "IFORM":
|
||||
row["IFORM_HEX"] = val_bytes.hex()
|
||||
row[f["name"]] = val_bytes.decode("cp949", "replace").strip()
|
||||
if not deleted:
|
||||
rows.append(row)
|
||||
|
||||
meta = {"version": ver, "hlen": hlen, "rlen": rlen, "header_tail_hex": b[off:hlen].hex()}
|
||||
return fields, rows, meta
|
||||
|
||||
|
||||
def build_dbf_bytes(fields: list[dict], rows: list[dict], meta: dict | None = None) -> bytes:
|
||||
"""(필드정의, 레코드목록)을 STMATE 호환 CP949 고정폭 DBF 바이트로 직렬화."""
|
||||
ver = (meta or {}).get("version", 0x03)
|
||||
today = datetime.date.today()
|
||||
nrec = len(rows)
|
||||
rlen = 1 + sum(int(f["len"]) for f in fields)
|
||||
tail_bytes = bytes.fromhex((meta or {}).get("header_tail_hex", "0d"))
|
||||
if not tail_bytes or tail_bytes[0] != 0x0D:
|
||||
tail_bytes = b"\x0d"
|
||||
hlen = 32 + 32 * len(fields) + len(tail_bytes)
|
||||
|
||||
hdr = bytearray(32)
|
||||
hdr[0], hdr[1], hdr[2], hdr[3] = ver, today.year - 1900, today.month, today.day
|
||||
struct.pack_into("<IHH", hdr, 4, nrec, hlen, rlen)
|
||||
|
||||
out = io.BytesIO()
|
||||
out.write(hdr)
|
||||
for f in fields:
|
||||
if f.get("raw_desc"):
|
||||
desc = bytearray(bytes.fromhex(f["raw_desc"]))
|
||||
desc[16], desc[17] = int(f["len"]), int(f.get("dec", 0))
|
||||
out.write(desc)
|
||||
else:
|
||||
desc = bytearray(32)
|
||||
nm_b = f["name"].encode("cp949", "replace")[:10]
|
||||
desc[: len(nm_b)] = nm_b
|
||||
desc[11], desc[16], desc[17] = (
|
||||
ord((f.get("type") or "C")[0]),
|
||||
int(f["len"]),
|
||||
int(f.get("dec", 0)),
|
||||
)
|
||||
out.write(desc)
|
||||
out.write(tail_bytes)
|
||||
|
||||
for row in rows:
|
||||
out.write(b"*" if row.get("_deleted") else b" ")
|
||||
for f in fields:
|
||||
flen, fdec, ftype = int(f["len"]), int(f.get("dec", 0)), f.get("type", "C")
|
||||
if f["name"] == "IFORM" and row.get("IFORM_HEX"):
|
||||
try:
|
||||
out.write(bytes.fromhex(row["IFORM_HEX"])[:flen].ljust(flen, b"\x00"))
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
raw_val = row.get(f["name"], "")
|
||||
if ftype in ("N", "F"):
|
||||
sval = str(raw_val).strip()
|
||||
if sval in ("", "None"):
|
||||
sval = ""
|
||||
else:
|
||||
try:
|
||||
num = float(sval)
|
||||
sval = f"{num:.{fdec}f}" if fdec > 0 else f"{int(round(num))}"
|
||||
except Exception:
|
||||
pass
|
||||
out.write(sval.encode("ascii", "replace")[:flen].rjust(flen, b" "))
|
||||
else:
|
||||
sval = "" if raw_val is None else str(raw_val)
|
||||
out.write(sval.encode("cp949", "replace")[:flen].ljust(flen, b" "))
|
||||
|
||||
out.write(b"\x1a")
|
||||
return out.getvalue()
|
||||
|
||||
|
||||
def parse_prjlst_smt(b: bytes) -> dict[str, str]:
|
||||
if len(b) <= 0x200:
|
||||
return {}
|
||||
return {k: v.strip() for k, v in SMT_KV_RE.findall(b[0x200:].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 extract_costn_memos(dbf_bytes: bytes, smt_bytes: bytes) -> dict[str, dict]:
|
||||
"""COSTN.DBF의 10바이트 바이너리 IFORM(<HII: tag, length, block_no)로 COSTN.SMT 본문 추출."""
|
||||
if len(dbf_bytes) < 32 or not smt_bytes or smt_bytes[8:15] != b"SIxMemo":
|
||||
return {}
|
||||
|
||||
num_records, header_len, record_len = struct.unpack("<IHH", dbf_bytes[4:12])
|
||||
block_size = struct.unpack("<I", smt_bytes[4:8])[0] or 64
|
||||
|
||||
fields: dict[str, tuple[int, int]] = {}
|
||||
pos = 32
|
||||
offset = 1
|
||||
while pos < header_len - 1 and dbf_bytes[pos] != 0x0D:
|
||||
name = dbf_bytes[pos : pos + 11].split(b"\x00")[0].decode("ascii", errors="ignore").strip()
|
||||
flen = dbf_bytes[pos + 16]
|
||||
fields[name] = (offset, flen)
|
||||
offset += flen
|
||||
pos += 32
|
||||
|
||||
if "CODE" not in fields or "IFORM" not in fields:
|
||||
return {}
|
||||
|
||||
code_off, code_len = fields["CODE"]
|
||||
iform_off, iform_len = fields["IFORM"]
|
||||
memos: dict[str, dict] = {}
|
||||
|
||||
for idx in range(num_records):
|
||||
rec_start = header_len + idx * record_len
|
||||
if rec_start + record_len > len(dbf_bytes):
|
||||
break
|
||||
rec = dbf_bytes[rec_start : rec_start + record_len]
|
||||
if rec[0:1] == b"*":
|
||||
continue
|
||||
code = rec[code_off : code_off + code_len].decode("cp949", errors="ignore").strip()
|
||||
iform = rec[iform_off : iform_off + iform_len]
|
||||
if not code or len(iform) < 10 or not any(b not in (0, 0x20) for b in iform):
|
||||
continue
|
||||
tag, byte_len, block_no = struct.unpack("<HII", iform[:10])
|
||||
byte_start = block_no * block_size
|
||||
if byte_len <= 0 or byte_start >= len(smt_bytes):
|
||||
continue
|
||||
raw_slice = smt_bytes[byte_start : min(len(smt_bytes), byte_start + byte_len)]
|
||||
text = raw_slice.rstrip(b"\x00").decode("cp949", errors="replace")
|
||||
parsed = parse_smt_formula_text(text)
|
||||
memos[code] = {
|
||||
"code": code,
|
||||
"tag": tag,
|
||||
"blockNo": block_no,
|
||||
"byteLen": byte_len,
|
||||
"text": text,
|
||||
"vars": parsed["vars"],
|
||||
"qFormula": parsed["qFormula"],
|
||||
"cmFormula": parsed["cmFormula"],
|
||||
"refs": parsed["refs"],
|
||||
"lines": parsed["lines"],
|
||||
}
|
||||
return memos
|
||||
|
||||
|
||||
def parse_smt_formula_text(text: str) -> dict:
|
||||
"""COSTN.SMT 산출근거 텍스트에서 변수(q,K,f,E,Cm,Q 등), 참조(&X00006L&), 산식행을 추출."""
|
||||
vars_map: dict[str, str] = {}
|
||||
q_formula = ""
|
||||
cm_formula = ""
|
||||
refs: list[dict] = []
|
||||
structured_lines: list[dict] = []
|
||||
|
||||
for raw_ln in text.splitlines():
|
||||
ln = raw_ln.strip()
|
||||
if not ln:
|
||||
continue
|
||||
|
||||
line_refs = []
|
||||
for m in RE_REF.finditer(ln):
|
||||
ref_code, comp = m.group(1), m.group(2) or ""
|
||||
item = {"code": ref_code, "comp": comp, "marker": m.group(0)}
|
||||
refs.append(item)
|
||||
line_refs.append(item)
|
||||
|
||||
for m in RE_ASSIGN.finditer(ln):
|
||||
var_name = m.group(1).strip()
|
||||
expr = m.group(2).strip()
|
||||
if not expr:
|
||||
continue
|
||||
vars_map[var_name] = expr
|
||||
if var_name == "Q":
|
||||
q_formula = expr
|
||||
elif var_name in ("Cm", "Cms", "Cm1"):
|
||||
cm_formula = expr
|
||||
|
||||
label_m = RE_LABEL.search(ln)
|
||||
pct_m = RE_PCT.search(ln)
|
||||
qty_list = [{"val": float(v), "unit": u} for v, u in RE_QTY.findall(ln) if u != "%"]
|
||||
|
||||
structured_lines.append(
|
||||
{
|
||||
"raw": raw_ln.rstrip("\r\n"),
|
||||
"refs": line_refs,
|
||||
"label": label_m.group(1) if label_m else "",
|
||||
"pct": float(pct_m.group(1)) if pct_m else None,
|
||||
"qtys": qty_list,
|
||||
"isSubtotal": "(=)" in ln,
|
||||
"isTotal": "(==)" in ln,
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"vars": vars_map,
|
||||
"qFormula": q_formula,
|
||||
"cmFormula": cm_formula,
|
||||
"refs": refs,
|
||||
"lines": structured_lines,
|
||||
}
|
||||
|
||||
|
||||
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 갱신."""
|
||||
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("<IHH", dbf_buf[4:12])
|
||||
|
||||
if not smt_bytes or smt_bytes[8:15] != b"SIxMemo":
|
||||
smt_buf = bytearray(512)
|
||||
struct.pack_into("<II", smt_buf, 0, 512, 64)
|
||||
smt_buf[8:15] = b"SIxMemo"
|
||||
else:
|
||||
smt_buf = bytearray(smt_bytes)
|
||||
|
||||
block_size = struct.unpack("<I", smt_buf[4:8])[0] or 64
|
||||
|
||||
fields: dict[str, tuple[int, int]] = {}
|
||||
pos = 32
|
||||
offset = 1
|
||||
while pos < header_len - 1 and dbf_buf[pos] != 0x0D:
|
||||
name = dbf_buf[pos : pos + 11].split(b"\x00")[0].decode("ascii", errors="ignore").strip()
|
||||
flen = dbf_buf[pos + 16]
|
||||
fields[name] = (offset, flen)
|
||||
offset += flen
|
||||
pos += 32
|
||||
|
||||
if "CODE" not in fields or "IFORM" not in fields:
|
||||
return bytes(dbf_buf), bytes(smt_buf)
|
||||
|
||||
code_off, code_len = fields["CODE"]
|
||||
iform_off, iform_len = fields["IFORM"]
|
||||
|
||||
for idx in range(num_records):
|
||||
rec_start = header_len + idx * record_len
|
||||
if rec_start + record_len > 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("<HII", 0x0008, byte_len, block_no)
|
||||
dbf_buf[rec_start + iform_off : rec_start + iform_off + min(10, iform_len)] = pointer_bytes[
|
||||
: min(10, iform_len)
|
||||
]
|
||||
|
||||
return bytes(dbf_buf), bytes(smt_buf)
|
||||
|
||||
|
||||
_CATALOG_CACHE: dict | None = None
|
||||
|
||||
|
||||
def load_stmate_catalog() -> dict:
|
||||
"""30_원자료의 표준 코드(894종), 일위대가 레시피(350종), 특수 단가산출(21종) 로드."""
|
||||
global _CATALOG_CACHE
|
||||
if _CATALOG_CACHE is not None:
|
||||
return _CATALOG_CACHE
|
||||
|
||||
codes: list[dict] = []
|
||||
csv_path = RAW_30_DIR / "stmate_code_name_list.csv"
|
||||
if csv_path.exists():
|
||||
with csv_path.open("r", encoding="utf-8-sig", errors="replace") as f:
|
||||
for row in csv.DictReader(f):
|
||||
c = (row.get("code") or "").strip()
|
||||
if c:
|
||||
codes.append(
|
||||
{
|
||||
"code": c,
|
||||
"name": (row.get("name") or "").strip(),
|
||||
"size": (row.get("size") or "").strip(),
|
||||
"unit": (row.get("unit") or "").strip(),
|
||||
}
|
||||
)
|
||||
|
||||
recipes = (
|
||||
json.loads((RAW_30_DIR / "일위대가_레시피.json").read_text(encoding="utf-8"))
|
||||
if (RAW_30_DIR / "일위대가_레시피.json").exists()
|
||||
else []
|
||||
)
|
||||
special_san = (
|
||||
json.loads((RAW_30_DIR / "특수_단가산출_21개.json").read_text(encoding="utf-8"))
|
||||
if (RAW_30_DIR / "특수_단가산출_21개.json").exists()
|
||||
else []
|
||||
)
|
||||
|
||||
_CATALOG_CACHE = {
|
||||
"code_count": len(codes),
|
||||
"codes": codes,
|
||||
"recipe_count": len(recipes),
|
||||
"recipes": recipes,
|
||||
"special_san_count": len(special_san),
|
||||
"special_san": special_san,
|
||||
}
|
||||
return _CATALOG_CACHE
|
||||
Reference in New Issue
Block a user