diff --git a/B09_Estimation/B09_Estimation_Estx.py b/B09_Estimation/B09_Estimation_Estx.py new file mode 100644 index 000000000..0989656b9 --- /dev/null +++ b/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(" 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(" 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) diff --git a/resources/tester/test_707_5_estx.py b/resources/tester/test_707_5_estx.py new file mode 100644 index 000000000..887c2101c --- /dev/null +++ b/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())