eomsangdonandClaude Opus 5.5
550c0adcce
- 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>
245 lines
10 KiB
Python
245 lines
10 KiB
Python
"""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)
|