eomsangdonandClaude Opus 5.5
e987602f24
- 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 <noreply@anthropic.com>
348 lines
13 KiB
Python
348 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""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
|
|
|
|
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" or f["type"] == "M":
|
|
row[f"{f['name']}_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")
|
|
raw_hex = row.get(f"{f['name']}_HEX")
|
|
if (f["name"] == "IFORM" or ftype == "M") and raw_hex:
|
|
try:
|
|
out.write(bytes.fromhex(raw_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, memo_hex: str = "") -> dict[str, str]:
|
|
"""PRJ_MEMO 포인터(<HII: tag, 길이, 블록번호)가 가리키는 블록만 읽음 — 남은 옛 블록 무시."""
|
|
if len(b) <= 0x200:
|
|
return {}
|
|
body = b[0x200:]
|
|
if len(memo_hex) >= 20:
|
|
_tag, ln, blk = struct.unpack("<HII", bytes.fromhex(memo_hex)[:10])
|
|
bs = struct.unpack("<I", b[4:8])[0] or 64
|
|
if blk and ln and blk * bs + ln <= len(b):
|
|
body = b[blk * bs : blk * bs + ln]
|
|
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], 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]:
|
|
"""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: 글})을 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
|
|
files = {"COSTN.DBF": dbf_bytes, "COSTN.SMT": smt_bytes}
|
|
out, _ = apply_save(files, {"costn_memo_updates": memo_updates})
|
|
return out["COSTN.DBF"], out.get("COSTN.SMT", smt_bytes)
|
|
|
|
|
|
_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
|
|
|
|
|
|
def parse_ini_bytes(b: bytes) -> dict[str, str]:
|
|
"""Start8x.Ini / STini.MEM 키=값(섹션 무시) — 산근 계산 방식(CHGNM) · 차액 보정 순서(OPTNM) 등."""
|
|
out: dict[str, str] = {}
|
|
for ln in b.decode("cp949", "replace").splitlines():
|
|
if "=" in ln and not ln.lstrip().startswith("["):
|
|
k, v = ln.split("=", 1)
|
|
out.setdefault(k.strip(), v.strip())
|
|
return out
|