239 lines
9.4 KiB
Python
239 lines
9.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""조달청 공사원가호환규정 XML → 프로젝트 가져오기 (Conv_XML_ST8 · Conv_Jodal_ST8 대응).
|
|
|
|
근거: 30_조달청_호환규정_XSD.md(문서 뼈대 · 호환규정 ↔ STmate 대응표) · 28번 §3(조달청 ↔ ST8 양방향) ·
|
|
old/30_원자료/조달청_공사원가호환규정.xsd(V1.0.5) · report_pps.js(내보내기 짝)
|
|
대응:
|
|
내역서(내역서번호 두 자리씩 = 깊이) → BOQ Tk 여는 행 · 세부공종 → 말단 행 · Ek/Ck 닫는 행
|
|
공종구분 제요율적용제외공종·관급자재공종 = C 구역(일반 공종 아래 첫 제외 묶음만 Ck 로 닫음)
|
|
단계 번호 k = 가장 깊은 묶음 1 … 바깥으로 갈수록 +1 (XML 에 단계 숫자가 없음 — 깊이로 정함)
|
|
자재 M · 노무 L · 경비항목 S · 일위대가 B · 중기단가산출 D(중기단가산출식 = COSTN.SMT) · 기계경비 X → COSTN
|
|
일위대가상세 · 기계경비상세 → BDQTY 구성(자원코드 · 수량)
|
|
제경비항목 700(총공사비) → PRJLST COSTW1 · 공사명 → PNAME
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from conv_xls_boq import LEVEL_NAME
|
|
from conv_xls_res import stored_unit
|
|
|
|
RES_TAGS = { # 태그 → (머리글자, 코드 태그)
|
|
"자재": ("M", "자재코드"),
|
|
"노무": ("L", "노무코드"),
|
|
"경비항목": ("S", "경비항목코드"),
|
|
"일위대가": ("B", "일위대가코드"),
|
|
"중기단가산출": ("D", "중기단가산출코드"),
|
|
"기계경비": ("X", "기계경비코드"),
|
|
}
|
|
EXCL = ("제요율적용제외공종", "관급자재공종")
|
|
|
|
|
|
def _t(el, tag: str, default: str = "") -> str:
|
|
v = el.findtext(tag) if el is not None else None
|
|
return default if v is None else v.strip()
|
|
|
|
|
|
def _f(el, tag: str) -> float:
|
|
try:
|
|
return float(_t(el, tag) or 0)
|
|
except ValueError:
|
|
return 0.0
|
|
|
|
|
|
def parse_xml(src) -> ET.Element:
|
|
if isinstance(src, (bytes, bytearray)):
|
|
return ET.fromstring(bytes(src))
|
|
if isinstance(src, str) and src.lstrip().startswith("<"):
|
|
return ET.fromstring(src.encode("utf-8"))
|
|
return ET.parse(src).getroot()
|
|
|
|
|
|
def _leaf(it) -> dict:
|
|
u = {c: _f(it, f"{n}단가") for c, n in (("M", "재료비"), ("L", "노무비"), ("S", "경비"))}
|
|
a = {c: _f(it, n) for c, n in (("M", "재료비"), ("L", "노무비"), ("S", "경비"))}
|
|
row = {
|
|
"BDCODE1": _t(it, "세부공종코드"),
|
|
"ITNUM": _t(it, "세부공종번호"),
|
|
"NAME1": _t(it, "세부공종명"),
|
|
"SIZE1": _t(it, "규격"),
|
|
"UNIT1": _t(it, "단위"),
|
|
"QTY1": _f(it, "수량"),
|
|
"BQCODE": _t(it, "비고"),
|
|
"EXIST1": 0,
|
|
}
|
|
row.update({f"{c}AMT1": u[c] for c in "MLS"}, TAMT1=sum(u.values()))
|
|
row.update({f"{c}_AMT1": a[c] for c in "MLS"}, T_AMT1=sum(a.values()))
|
|
return row
|
|
|
|
|
|
def boq_rows(root: ET.Element) -> list[dict]:
|
|
"""내역서 묶음 → BOQ 행(여는 Tk · 말단 · 닫는 Ek/Ck) · SORTCODE↔ENDCODE 짝."""
|
|
groups = list(root.iter("내역서"))
|
|
if not groups:
|
|
return []
|
|
depth = {id(g): max(1, len(_t(g, "내역서번호")) // 2) for g in groups}
|
|
deepest = max(depth.values())
|
|
out: list[dict] = []
|
|
stack: list[tuple[int, dict, int, bool]] = [] # (깊이, 여는 행, 자리, 제외 구역 시작인가)
|
|
|
|
def close_to(d: int):
|
|
while stack and stack[-1][0] >= d:
|
|
dd, head, at, cz = stack.pop()
|
|
lev = deepest - dd + 1
|
|
name = LEVEL_NAME.get(lev, "")
|
|
row = {k: v for k, v in head.items() if "_AMT" in k}
|
|
if cz:
|
|
row.update(BDCODE1=f"C{lev}", ITNUM="", NAME1=f"{name}(제외금액)", EXIST1=1)
|
|
else:
|
|
row.update(BDCODE1=f"E{lev}", ITNUM=name, NAME1="", EXIST1=0)
|
|
row["_pair"], head["_pair"] = at, len(out)
|
|
out.append(row)
|
|
|
|
for g in groups:
|
|
d = depth[id(g)]
|
|
close_to(d)
|
|
kind = _t(g, "공종구분")
|
|
in_excl = any(s[3] or s[1].get("_excl") for s in stack)
|
|
a = {c: _f(g, n) for c, n in (("M", "재료비"), ("L", "노무비"), ("S", "경비"))}
|
|
head = {
|
|
"BDCODE1": f"T{deepest - d + 1}",
|
|
"ITNUM": _t(g, "공종번호"),
|
|
"NAME1": _t(g, "공종명"),
|
|
"EXIST1": 1,
|
|
"_excl": kind in EXCL,
|
|
}
|
|
head.update({f"{c}_AMT1": a[c] for c in "MLS"}, T_AMT1=sum(a.values()))
|
|
stack.append((d, head, len(out), kind in EXCL and not in_excl))
|
|
out.append(head)
|
|
for it in g.findall("세부공종"):
|
|
out.append(_leaf(it))
|
|
close_to(0)
|
|
for i, r in enumerate(out):
|
|
r["SORTCODE"] = 256.0 * (i + 1)
|
|
r.pop("_excl", None)
|
|
for r in out:
|
|
r["ENDCODE"] = out[r.pop("_pair")]["SORTCODE"] if "_pair" in r else 0.0
|
|
return out
|
|
|
|
|
|
def costn_rows(root: ET.Element) -> tuple[list[dict], dict[str, str]]:
|
|
"""자원 목록 → COSTN 행 · 중기단가산출식 → {D코드: 식}."""
|
|
rows, memos, seen = [], {}, set()
|
|
for tag, (pre, ctag) in RES_TAGS.items():
|
|
for el in root.iter(tag):
|
|
code = _t(el, ctag)
|
|
if not code or code in seen or el.find(ctag) is None:
|
|
continue
|
|
size = _t(el, "규격")
|
|
row = {
|
|
"CODE": code,
|
|
"NAME": _t(el, "품명"),
|
|
"SIZE": size,
|
|
"UNIT": stored_unit(code, _t(el, "단위"), size),
|
|
"BQCODE": _t(el, "비고"),
|
|
}
|
|
if pre in "ML":
|
|
p = _f(el, "재료비단가" if pre == "M" else "노무비단가")
|
|
part = {c: (p if c == pre else 0.0) for c in "MLS"}
|
|
elif pre == "S":
|
|
part = {
|
|
c: _f(el, f"{n}단가")
|
|
for c, n in (("M", "재료비"), ("L", "노무비"), ("S", "경비"))
|
|
}
|
|
else:
|
|
part = {c: _f(el, n) for c, n in (("M", "재료비"), ("L", "노무비"), ("S", "경비"))}
|
|
row.update({f"{c}AMT1": part[c] for c in "MLS"}, TAMT1=sum(part.values()))
|
|
rows.append(row)
|
|
seen.add(code)
|
|
if pre == "D":
|
|
text = el.findtext("중기단가산출식") or ""
|
|
if text.strip():
|
|
memos[code] = text.replace("\r\n", "\n").replace("\n", "\r\n")
|
|
return rows, memos
|
|
|
|
|
|
def comp_rows(root: ET.Element) -> list[dict]:
|
|
"""일위대가상세 · 기계경비상세 → BDQTY 구성행(자원코드 · 수량)."""
|
|
out = []
|
|
for tag, ctag, dtag in (
|
|
("일위대가", "일위대가코드", "일위대가상세"),
|
|
("기계경비", "기계경비코드", "기계경비상세"),
|
|
):
|
|
for el in root.iter(tag):
|
|
bd = _t(el, ctag)
|
|
if not bd:
|
|
continue
|
|
for d in el.findall(dtag):
|
|
out.append({"BDCODE": bd, "CODE": _t(d, "자원코드"), "QTY1": _f(d, "수량")})
|
|
return out
|
|
|
|
|
|
def project_info(root: ET.Element) -> dict:
|
|
"""공사명 · 공사관리번호 · 주공사 · 총공사비(제경비 700) · 직접비(100)."""
|
|
info = root.find("공사정보")
|
|
costs = {_t(x, "세부공종코드"): _f(x, "산출금액") for x in root.iter("제경비항목")}
|
|
return {
|
|
"pname": _t(info, "공사명"),
|
|
"pnum": _t(info, "공사관리번호"),
|
|
"main": _t(info, "주공사"),
|
|
"grand_total": costs.get("700", 0.0),
|
|
"direct_total": costs.get("100", 0.0),
|
|
"schema": _t(root, "스키마버전"),
|
|
}
|
|
|
|
|
|
def read_xml(src) -> dict:
|
|
root = parse_xml(src)
|
|
if root.tag != "공사원가호환규정":
|
|
raise ValueError(f"공사원가호환규정 XML 이 아님(뿌리 {root.tag})")
|
|
costn, memos = costn_rows(root)
|
|
return {
|
|
"info": project_info(root),
|
|
"boq": boq_rows(root),
|
|
"costn": costn,
|
|
"memos": memos,
|
|
"comp": comp_rows(root),
|
|
}
|
|
|
|
|
|
def xml_to_project(template: dict, src, pname: str = "") -> tuple[dict, dict]:
|
|
"""XML 한 개 → 새 프로젝트 files(틀 = 다른 프로젝트의 요율·서식·설정) · 보고."""
|
|
from conv_xls_apply import _costn_table, _direct_to_prj, rows_to_table
|
|
from conv_xls_boq import fill_from_costn
|
|
from dbf_rw import DbfTable
|
|
from stc_project import new_project
|
|
from stc_save import SaveLog
|
|
from stc_save_extra import save_bdqty
|
|
|
|
data = read_xml(src)
|
|
info = data["info"]
|
|
out = new_project(template, pname or info["pname"] or "XML 가져온 공사", info["pnum"][:20])
|
|
out["COSTN.DBF"], out["COSTN.SMT"] = _costn_table(
|
|
template["COSTN.DBF"], data["costn"], data["memos"]
|
|
)
|
|
rows = data["boq"]
|
|
fill_from_costn(rows, data["costn"])
|
|
log = SaveLog()
|
|
bt = rows_to_table(template["BOQ11.DBF"], rows)
|
|
direct = _direct_to_prj(out, out, bt, "BOQ11.DBF", log)
|
|
out["BOQ11.DBF"] = bt.to_bytes()
|
|
if data["comp"] and "BDQTY.DBF" in template:
|
|
out["BDQTY.DBF"] = (
|
|
save_bdqty(template["BDQTY.DBF"], data["comp"], log, strict=False)
|
|
or template["BDQTY.DBF"]
|
|
)
|
|
pt = DbfTable(out["PRJLST.DBF"])
|
|
if info["grand_total"] and "COSTW1" in pt.fields:
|
|
pt.set(0, "COSTW1", info["grand_total"])
|
|
out["PRJLST.DBF"] = pt.to_bytes()
|
|
return out, {
|
|
"info": info,
|
|
"boq_count": len(rows),
|
|
"costn_count": len(data["costn"]),
|
|
"memo_count": len(data["memos"]),
|
|
"bdqty_count": len(data["comp"]),
|
|
"direct": direct,
|
|
}
|