376 lines
13 KiB
Python
376 lines
13 KiB
Python
# -*- 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
|