feat(B09 설계내역서): .estx(EST 원본) 바로 읽기 모듈 — 내역 · 호표 목록 · 노임 · 자재 · 기계 평문 표 (PLAN 707-5)

- B09_Estimation_Estx: ZIP 안 DBISAM 고정 레코드 표를 메모리에서 읽음 · 엑셀 읽기와 같은 꼴(707-1 내역 · 707-3 단가표)
- 구성 줄 → 호표 귀속 · 제N호표 번호 · 원가계산서 행 경계는 안 읽음(UNKNOWN · 데스크탑 REA 필요)
- 시험 test_707_5_estx: 거창 = 엑셀과 줄마다 같음 · 직접비 384,916,227 · 장수 · 진안 줄 수 · 합 기록

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

B09_Estimation/B09_Estimation_Estx.py

@@ -0,0 +1,244 @@
+"""B09 EST 원본(.estx) 바로 읽기(PLAN 707-5) — 메모리에서 풀어 평문 표만 · 저장 안 함.
+
+.estx = 보통 ZIP · 표마다 DBISAM `.dat`(고정 레코드) · `.idx` · `.blb`
+(지식DB `original/원가계산/다산소프트/ESTX_파일구조.md` · `ESTX_테이블사전.md`):
+    머리 0x29 레코드 수 · 0x2D 레코드 길이 · 0x2F 필드 수 ·
+    필드 설명자 = 0x200 + n×0x300(+0x02 이름 길이 · +0xA4 자료형 · +0xA9 길이 · +0xAC 자리) ·
+    데이터 = 512 + 필드 수×768 · 필드 = 존재 바이트 1 + 값(0 = 미저장 = None)
+    자료형 1 CP949 글 · 2 날짜 서수 · 4 int16 · 6 int32 · 7 double · 3 BLOB(안 읽음)
+금액 넷(`REA 분석/ESTPlus2_REA_최종보고서_20261030.md:31`) = N 노무 · G 재료 · J 경비 · S 실적
+    (단가 NU · GU · JU · SU · 금액 NC · GC · JC · SC · 합 TU · TC · 수량 GN) — S 는 셋에 안 넣음.
+    밑줄 칸(`TC_` …) = 비교 스냅샷(방향 미확정) — 안 읽음.
+
+돌려주는 꼴 = 엑셀 읽기와 같은 dict — 뒤에 고리만 이으면 같은 그릇:
+    `read_bill`   = `B09_Estimation_Bill_Excel.read_bill`(707-1) — EstMainDat
+        묶음 = Level 있음(클수록 바깥) · 항목 = Level 없음 + 이름 · 이름 없음 = 빈 줄 ·
+        LinkIndex 54 · 55 · 59(순공사비 · 도급액 · 경계) = 합계 줄(안 읽음) ·
+        57 = 조달수수료(율 = GN · 줄로 안 실음) · 그 밖 % 줄 = 1 식(단가 = 금액) ·
+        제외 = 순공사비 뒤 · 이름 「관급」 맨 위 묶음 ·
+        호표 = LinkIndex 103 대가 · 104 산근 · 5 자재 · 6 노임 · 105 기계 + 코드
+    `read_prices` = `B09_Estimation_Import_Prices.read_prices`(707-3) — LaborcDat · StuffcDat ·
+        MachTitleDat(+ 자재 출처 6칸 `sources`)
+    `read_hopyo`  = 호표 목록(HeadTitleDat 대가 · CostTitleDat 산근 — 코드 · 이름 · 단가 셋)
+안 읽음(데스크탑 REA 필요 · 같은 공사 엑셀로 메움) = `UNKNOWN`.
+"""
+
+from __future__ import annotations
+
+import io
+import struct
+import zipfile
+from decimal import Decimal
+from typing import Any
+
+from B09_Estimation.B09_Estimation_Bill_Excel import BillExcelError, _plain, _unwrap
+
+UNKNOWN = (
+    "일위대가 · 단가산출 구성 줄 → 호표 귀속(HeadMainDat · CostMainDat)",
+    "화면 「제N호표」 번호 자리",
+    "원가계산서 행 경계(WongaDat)",
+)
+#: 가격자료 6칸 이름 — `ProjectDataInfo.Ini` TitleName1..6 이 없을 때(`ESTX_데이터관계.md`)
+SOURCES = ("가격정보", "물가자료", "유통물가", "물가정보", "거래가격", "견적가")
+PARTS = (("N", "labor"), ("G", "material"), ("J", "expense"))
+LINK = {103: "대가", 104: "산근", 5: "자재", 6: "노임", 105: "기계"}
+TOTAL_LINKS = {54, 55, 59}
+FEE_LINK = 57
+_SIZE = {1: None, 2: 4, 4: 2, 6: 4, 7: 8}
+
+
+class EstxError(BillExcelError):
+    """못 읽는 .estx — 길이 400 으로 답함."""
+
+
+def _zip(data: bytes) -> zipfile.ZipFile:
+    if data[:4] != b"PK\x03\x04":
+        raise EstxError(".estx 가 아님 — ZIP 머리 없음")
+    try:
+        book = zipfile.ZipFile(io.BytesIO(data))
+    except zipfile.BadZipFile as error:
+        raise EstxError("깨진 .estx") from error
+    if any(i.flag_bits & 1 for i in book.infolist()):
+        raise EstxError("암호 걸린 .estx — 읽을 수 없음")
+    return book
+
+
+def table(book: zipfile.ZipFile, name: str, want: tuple[str, ...]) -> list[dict[str, Any]]:
+    """DBISAM `{name}.dat` → 레코드 목록(파일 차례) · `want` 칸만."""
+    try:
+        d = book.read(f"{name}.dat")
+    except KeyError as error:
+        raise EstxError(f".estx 안에 {name} 표가 없음") from error
+    count, size, nf = struct.unpack_from("<IHH", d, 0x29)
+    cols = []
+    for n in range(nf):
+        at = 0x200 + n * 0x300
+        field = d[at + 3 : at + 3 + d[at + 2]].decode("ascii", "replace")
+        kind = d[at + 0xA4]
+        if field in want and kind in _SIZE:
+            width = struct.unpack_from("<H", d, at + 0xA9)[0]
+            cols.append((field, kind, width, struct.unpack_from("<I", d, at + 0xAC)[0]))
+    start = 512 + nf * 768
+    if start + count * size > len(d):
+        raise EstxError(f"{name} 표 길이가 머리와 다름")
+    out = []
+    for r in range(count):
+        base = start + r * size
+        row: dict[str, Any] = dict.fromkeys(want)
+        for field, kind, width, off in cols:
+            if not d[base + off]:
+                continue
+            at = base + off + 1
+            if kind == 1:
+                row[field] = d[at : at + width].split(b"\0")[0].decode("cp949", "replace").strip()
+            elif kind == 7:
+                row[field] = struct.unpack_from("<d", d, at)[0]
+            else:
+                row[field] = struct.unpack_from("<h" if kind == 4 else "<i", d, at)[0]
+        out.append(row)
+    return out
+
+
+def _dec(value: Any) -> Decimal:
+    return Decimal(repr(value)) if isinstance(value, float) else Decimal(value or 0)
+
+
+def _parts(row: dict[str, Any], tail: str) -> dict[str, Decimal]:
+    return {key: _dec(row.get(f"{p}{tail}")) for p, key in PARTS}
+
+
+_BILL = ("Code", "Name", "Size", "Unite", "GN", "Level", "LinkIndex")
+_BILL += tuple(f"{p}{t}" for p, _ in PARTS for t in "UC")
+
+
+def read_bill(name: str, data: bytes) -> dict[str, Any]:
+    """.estx → 엑셀 `read_bill` 과 같은 꼴(`행` = EstMainDat 차례 1~)."""
+    rows = table(_zip(data), "EstMainDat", _BILL)
+    lines: list[dict[str, Any]] = []
+    stack: list[tuple[int, dict[str, Any]]] = []  # (Level, 묶음 줄)
+    skipped, fee, pct_lines, after_net = 0, None, 0, False
+    for at, row in enumerate(rows, 1):
+        item_name, link, level = row["Name"] or "", row["LinkIndex"], row["Level"]
+        if link in TOTAL_LINKS:
+            skipped += 1
+            after_net = after_net or link == 54
+            continue
+        if not item_name:
+            continue
+        group = level is not None
+        if group:
+            while stack and stack[-1][0] <= level:
+                stack.pop()
+        up = stack[-1][1] if stack else None
+        code = row["Code"] or ""
+        line = {
+            "행": at,
+            "묶음": group,
+            "위": up["행"] if up else None,
+            "공종": "",
+            "이름": item_name,
+            "규격": row["Size"] or "",
+            "단위": row["Unite"] or "",
+            "수량": None if group else _dec(row["GN"]),
+            "단가": _parts(row, "U"),
+            "금액": _parts(row, "C"),
+            "비고": "",
+            "호표": {"종류": LINK[link], "코드": code} if not group and link in LINK else None,
+            "제외": (up or {}).get("제외", False)
+            or (up is None and (after_net or "관급" in item_name)),
+        }
+        if group:
+            stack.append((level, line))
+        elif line["단위"] == "%":
+            if link == FEE_LINK:
+                fee = fee if fee is not None else line["수량"]
+                continue
+            pct_lines += 1
+            line.update({"단가": line["금액"], "수량": Decimal(1), "단위": "식"})
+        lines.append(line)
+    _unwrap(lines)
+    total = {"직접": Decimal(0), "제외": Decimal(0)}
+    for line in lines:
+        if not line["묶음"]:
+            total["제외" if line["제외"] else "직접"] += sum(line["금액"].values())
+    return _plain(
+        {
+            "꼴": "estx",
+            "시트": "EstMainDat",
+            "줄": lines,
+            "합계": total,
+            "조달수수료율": fee,
+            "합계줄": skipped,
+            "율줄": pct_lines,
+        }
+    )
+
+
+def _item(row: dict[str, Any]) -> dict[str, Any]:
+    return {"name": row["Name"] or "", "spec": row["Size"] or "", "unit": row["Unite"] or ""}
+
+
+def _titles(book: zipfile.ZipFile) -> tuple[str, ...]:
+    try:
+        text = book.read("ProjectDataInfo.Ini").decode("cp949", "replace")
+    except KeyError:
+        return SOURCES
+    got = dict(line.split("=", 1) for line in text.splitlines() if "=" in line)
+    return tuple(got.get(f"TitleName{n}", "").strip() or SOURCES[n - 1] for n in range(1, 7))
+
+
+def read_prices(name: str, data: bytes) -> dict[str, Any]:
+    """.estx → 707-3 `read_prices` 꼴 {labor, material, machine, *_sheet} ·
+    자재 줄 + `sources` [{출처, 단가, 쪽}](값 있는 칸만)."""
+    book = _zip(data)
+    head = ("Code", "Name", "Size", "Unite", "TC")
+    cost = tuple(f"Cost{n}" for n in range(1, 7)) + tuple(f"Page{n}" for n in range(1, 7))
+    titles = _titles(book)
+    labor = [{**_item(r), "price": r["TC"]} for r in table(book, "LaborcDat", head) if r["Name"]]
+    material = []
+    for r in table(book, "StuffcDat", head + cost):
+        if not r["Name"] or r["Unite"] == "%":
+            continue
+        sources = [
+            {"출처": titles[n - 1], "단가": r[f"Cost{n}"], "쪽": r[f"Page{n}"] or ""}
+            for n in range(1, 7)
+            if r[f"Cost{n}"]
+        ]
+        material.append({**_item(r), "price": r["TC"], "sources": sources})
+    machine = [
+        {**_item(r), **{key: r[f"{p}C"] or 0 for p, key in PARTS}}
+        for r in table(book, "MachTitleDat", head + tuple(f"{p}C" for p, _ in PARTS))
+        if r["Name"]
+    ]
+    return _plain(
+        {
+            "labor": labor,
+            "labor_sheet": "LaborcDat",
+            "material": material,
+            "material_sheet": "StuffcDat",
+            "machine": machine,
+            "machine_sheet": "MachTitleDat",
+        }
+    )
+
+
+def read_hopyo(name: str, data: bytes) -> dict[str, list[dict[str, Any]]]:
+    """.estx → {대가: HeadTitleDat, 산근: CostTitleDat} 목록
+    [{코드, 이름, 규격, 단위, 단가 {labor, material, expense}, 합계}] — 구성 줄은 `UNKNOWN`."""
+    book = _zip(data)
+    want = ("Code", "Name", "Size", "Unite", "TC") + tuple(f"{p}C" for p, _ in PARTS)
+    out = {}
+    for kind, name_ in (("대가", "HeadTitleDat"), ("산근", "CostTitleDat")):
+        out[kind] = [
+            {
+                "코드": r["Code"] or "",
+                "이름": r["Name"],
+                "규격": r["Size"] or "",
+                "단위": r["Unite"] or "",
+                "단가": _parts(r, "C"),
+                "합계": _dec(r["TC"]),
+            }
+            for r in table(book, name_, want)
+            if r["Name"]
+        ]
+    return _plain(out)

resources/tester/test_707_5_estx.py

@@ -0,0 +1,127 @@
+"""707-5 EST 원본(.estx) 바로 읽기 — `B09_Estimation_Estx` (읽기만 · 저장 안 함).
+
+- 실무 .estx 셋(거창 · 장수 · 진안 — 암호 없음) = 내역 줄 수 · 직접비 합 = 파일 안 순공사비 줄 ·
+  조달수수료 = 율 · 노임 · 자재 · 기계 · 호표 목록 줄 수
+- 거창 = 같은 공사 엑셀(707-1)과 줄마다 같음(이름 · 수량 · 금액 · 제외 · 묶음 나무) ·
+  직접비 384,916,227 · 707-1 짜기 그릇에 그대로 부음
+- 장수 · 진안 = 엑셀이 DRM — 대조 대신 줄 수 · 금액 합만 기록
+- 깨진 파일 · 표 없음 = 막힘
+"""
+
+import io
+import sys
+import time
+import zipfile
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from B09_Estimation import B09_Estimation_Bill_Import as imp  # noqa: E402
+from B09_Estimation import B09_Estimation_Estx as ex  # noqa: E402
+from B09_Estimation import B09_Estimation_Store as store  # noqa: E402
+from B09_Estimation.B09_Estimation_Bill_Excel import read_bill as read_excel  # noqa: E402
+from B09_Estimation.B09_Estimation_Calc_Bill import compute_estimation  # noqa: E402
+
+K = ROOT / "resources" / "knowledge" / "original" / "실무문서"
+FILES = {
+    "거창": K / "01.산불거창(거창북상병곡)/01.내역서/00.25년_산불_거창(거창,북상,병곡,산8외) _단가변경.estx",  # noqa: E501
+    "장수": K / "02.산불 장수/01.내역서/00.단가변경_26년_산불_무주(장수,장수,덕산,산54-1).estx",
+    "진안": K / "03.산불진안/01.내역서/00.단가변경_26년_산불_무주(진안,백운,백암,산1-2).estx",
+}  # fmt: skip
+EXCEL = (
+    K / "01.산불거창(거창북상병곡)/01.내역서/00.25년_산불_거창(거창,북상,병곡,산8외)내역+수량.xlsx"
+)
+#: (항목 수, 직접비, 제외, 노임, 자재, 기계, 대가, 산근) — 장수 · 진안 = 기록(엑셀 DRM)
+WANT = {
+    "거창": (60, 384916227, 74634214, 17, 52, 22, 14, 61),
+    "장수": (58, 496306311, 61483490, 16, 45, 22, 3, 65),
+    "진안": (72, 465955223, 56991694, 17, 55, 22, 14, 74),
+}
+EMPTY = {"materials": [], "bill": {"items": []}, "hopyo": {}}
+
+
+@pytest.fixture(scope="module")
+def data():
+    return {k: p.read_bytes() for k, p in FILES.items()}
+
+
+@pytest.mark.parametrize("name", list(FILES))
+def test_실무_estx_줄_수_합계_순공사비(data, name):
+    t = time.perf_counter()
+    bill = ex.read_bill(FILES[name].name, data[name])
+    prices = ex.read_prices(FILES[name].name, data[name])
+    hopyo = ex.read_hopyo(FILES[name].name, data[name])
+    print(f"{name} 읽기 {time.perf_counter() - t:.3f} s")
+    got = (
+        sum(1 for x in bill["줄"] if not x["묶음"]),
+        bill["합계"]["직접"],
+        bill["합계"]["제외"],
+        *(len(prices[k]) for k in ("labor", "material", "machine")),
+        len(hopyo["대가"]),
+        len(hopyo["산근"]),
+    )
+    assert got == WANT[name]
+    assert bill["조달수수료율"] == 0.54 and bill["합계줄"] == 1
+    # 직접비 = 파일 안 순공사비 줄(LinkIndex 54) 금액
+    net = ex.table(ex._zip(data[name]), "EstMainDat", ("LinkIndex", "TC"))
+    assert [r["TC"] for r in net if r["LinkIndex"] == 54] == [bill["합계"]["직접"]]
+    # 항목 금액 = 성분 셋 · 호표 연결 = 코드
+    for line in bill["줄"]:
+        if not line["묶음"]:
+            assert line["호표"] and line["호표"]["코드"], line["이름"]
+    assert all(r["price"] > 0 for r in prices["labor"])
+    assert all(h["합계"] == sum(h["단가"].values()) for h in hopyo["대가"] + hopyo["산근"])
+
+
+def test_거창_엑셀과_줄마다_같음(data):
+    mine = ex.read_bill("a.estx", data["거창"])
+    excel = read_excel(EXCEL.name, EXCEL.read_bytes())
+
+    def key(line):
+        return (line["이름"].replace(" ", ""), line["수량"], line["금액"], line["제외"])
+
+    items = [[key(x) for x in got["줄"] if not x["묶음"]] for got in (mine, excel)]
+    assert items[0] == items[1]
+
+    def tree(got):
+        names = {x["행"]: x["이름"] for x in got["줄"]}
+        return [(x["이름"], names.get(x["위"])) for x in got["줄"] if x["묶음"]]
+
+    assert tree(mine) == tree(excel)
+    assert mine["합계"] == excel["합계"]
+
+
+def test_거창_707_1_그릇에_부음(data):
+    got = ex.read_bill("a.estx", data["거창"])
+    rows = compute_estimation(EMPTY, {}, None, {})["bill"]["rows"]
+    made, report = imp.apply(got, {}, rows, "abcd1234")
+    made = store.clean_user(made)
+    bill = compute_estimation(EMPTY, made, None, {})["bill"]
+    assert report["실음"] == len(got["줄"])
+    assert bill["totals"]["direct"]["sum"] == 384916227
+    assert bill["totals"]["direct"]["blank"] == 0
+    assert imp.differences(got, report, bill["rows"]) == []
+    assert made["estimate"]["procurement_fee_pct"] == 0.54
+
+
+def test_자재_출처_칸(data):
+    stuff = ex.read_prices("a.estx", data["거창"])["material"]
+    wire = next(x for x in stuff if x["name"].startswith("아연도철선"))
+    assert wire["price"] == 2480
+    assert {"출처": "물가자료", "단가": 2480, "쪽": "27(26.1)"} in wire["sources"]
+
+
+def test_깨진_파일_표_없음_막힘():
+    with pytest.raises(ex.EstxError):
+        ex.read_bill("a.estx", b"not a zip")
+    with pytest.raises(ex.EstxError):
+        ex.read_bill("a.estx", b"PK\x03\x04 broken")
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w") as z:
+        z.writestr("Other.dat", b"")
+    with pytest.raises(ex.EstxError, match="EstMainDat"):
+        ex.read_bill("a.estx", buf.getvalue())