Aislo/B09_Estimation/B09_Estimation_Bill_Excel.py

"""B09 경쟁사 내역 엑셀 읽기(PLAN 707-1) — 파일 → 내역 줄(묶음 · 항목) · 엑셀 합계. 저장 안 함.

꼴 둘(`_분석_705-4_불러오기_범위.md` ②):
    STmate 「설계내역서」(없으면 「착공내역서」 · 「설계변경내역서」) —
        숨은 열 = 줄 코드 · 부모 코드(`B00194_1` · `E2_1` — 비고 열 뒤 첫 둘) ·
        묶음 = 코드 E · C(C = 제외금액 맨 위) ·
        변경 차수(`_2` 뒤) 줄 = 안 읽음 · 부모 = 열린 묶음 가운데 부모 코드가 같은 가장 안쪽 ·
        호표 = 코드 B 대가 · D 산근 · M 자재(번 = 비고 「대가 5호표」 · 「산근 13호표」 숫자)
    EST 「내역서」 — 숨은 열 없음 · 묶음 = 공종 칸 있음 + 수량 없음 · 깊이 = 공종 번호
        (Ⅰ · Ⅱ = 1 · 「4.1」 = 점 둘레 수 + 1) · 공종 · 수량 없이 금액만 = 합계 줄(안 읽음) ·
        호표 = 비고 「단산46참조」 · 「일위8참조」 · 「관급」
열 = 머리 글(앞 10 줄 · 빈칸 뺌) 공종 · 명칭|품명 · 규격 · 수량 · 단위 · 합계 ·
    노무비 · 재료비 · 경비 · 비고 — 돈 머리 = 단가 열 · 다음 열 = 금액
    (`M02_Hopyo_Excel` 의 칸 읽기 재사용) · 빈 단가 = 0.
제외 = 맨 위 묶음 가운데 STmate 코드 C · 비고 「제외」 · 이름 「관급」 — 직접비 합에서 뺌.
조달수수료 = 제외 묶음 안 단위 % · 이름 「수수료」 줄 → `조달수수료율`
    (줄로 안 실음 · 우리 조달수수료 줄이 셈).
그 밖 % 줄 = 1 식 · 단가 = 엑셀 금액(금액 그대로).
감싼 묶음 = 직접 맨 위 묶음이 하나뿐(공사 이름)이면 벗김(깊이 5 한도 · 레벨 1 = 공종).
DRM 파일 = 「읽을 수 없음」(`M02_Hopyo_Excel.DRM`).
"""

from __future__ import annotations

import io
import re
import warnings
from decimal import Decimal
from typing import Any

from M02_MasterTemplete import M02_Hopyo_Excel as hx

SHEETS = ("설계내역서", "내역서", "착공내역서", "설계변경내역서")
HEADS = {
    "공종": "공종",
    "명칭": "품명",
    "품명": "품명",
    "규격": "규격",
    "수량": "수량",
    "단위": "단위",
    "합계": "합계",
    "노무비": "노무비",
    "재료비": "재료비",
    "경비": "경비",
    "비고": "비고",
}
PARTS = (("노무비", "labor"), ("재료비", "material"), ("경비", "expense"))
_CODE = re.compile(r"^([A-Z]+\d*)_(\d+)$")
_ROMAN = re.compile(r"^[ⅠⅡⅢⅣⅤⅥⅦⅧⅨⅩ]+$|^[IVX]+\.?$")
_ST_NOTE = re.compile(r"^(대가|산근|자재)(\d+)")
_EST_NOTE = re.compile(r"^(단산|일위)(\d+)")
NO_SHEET = "내역서 시트(설계내역서 · 내역서) 없음"


class BillExcelError(ValueError):
    """못 읽는 파일 — 길이 400 으로 답함."""


def _sheet(name: str, data: bytes) -> tuple[str, list[list[Any]]]:
    if not data:
        raise BillExcelError("빈 파일")
    if b"DRMONE" in data[:256] and data[:4] not in (b"PK\x03\x04", b"\xd0\xcf\x11\xe0"):
        raise BillExcelError(f"읽을 수 없음 — {hx.DRM}")
    try:
        if data[:4] == b"PK\x03\x04" or name.lower().endswith(".xlsx"):
            import openpyxl

            with warnings.catch_warnings():  # 인쇄 영역 이름 경고(STmate) — 읽기와 무관
                warnings.simplefilter("ignore")
                book = openpyxl.load_workbook(io.BytesIO(data), read_only=True, data_only=True)
            titles = {ws.title.strip(): ws for ws in book.worksheets}
            pick = next((t for t in SHEETS if t in titles), None)
            rows = [list(r) for r in titles[pick].iter_rows(values_only=True)] if pick else []
            book.close()
        else:
            import xlrd

            book = xlrd.open_workbook(file_contents=data, logfile=io.StringIO())
            titles = {sh.name.strip(): sh for sh in book.sheets()}
            pick = next((t for t in SHEETS if t in titles), None)
            sh = titles.get(pick or "")
            rows = [sh.row_values(r) for r in range(sh.nrows)] if sh else []
    except Exception as error:  # 라이브러리마다 다른 예외 — 깨진 파일 · 엑셀 아님
        raise BillExcelError("엑셀 파일을 읽지 못함 — xls · xlsx 가 아니거나 깨진 파일") from error
    if pick is None:
        raise BillExcelError(NO_SHEET)
    return pick, rows


def _columns(rows: list[list[Any]]) -> tuple[int, dict[str, int]]:
    for at, row in enumerate(rows[:10]):
        flat = [hx._flat(v) for v in row]
        if "공종" not in flat or not ({"명칭", "품명"} & set(flat)):
            continue
        cols: dict[str, int] = {}
        for col, head in enumerate(flat):
            if head in HEADS and HEADS[head] not in cols:
                cols[HEADS[head]] = col
        below = [hx._flat(v) for v in rows[at + 1]] if at + 1 < len(rows) else []
        return (at + 2 if "단가" in below or "금액" in below else at + 1), cols
    raise BillExcelError("내역서 머리 글(공종 · 명칭)을 못 찾음")


def _codes(row: list[Any], after: int) -> tuple[str, str] | None:
    """STmate 숨은 열 (줄 코드, 부모 코드) — 비고 뒤 「X_n」 꼴 첫 둘 ·
    변경 차수(_2 뒤)면 ("", "")."""
    got = [m for v in row[after + 1 :] if (m := _CODE.match(hx._text(v)))]
    if not got:
        return None
    if got[0].group(2) != "1":
        return "", ""
    return got[0].group(1), (got[1].group(1) if len(got) > 1 else "")


def _depth(head: str) -> int:
    if _ROMAN.match(head):
        return 1
    parts = [p for p in head.split(".") if p.strip()]
    return len(parts) + 1 if parts and all(p.strip().isdigit() for p in parts) else 1


def _money(row: list[Any], cols: dict[str, int], offset: int) -> dict[str, Decimal]:
    got = hx._money(row, cols, offset)
    return {key: got.get(head) or Decimal(0) for head, key in PARTS}


def _hopyo(note: str, code: str) -> dict[str, Any] | None:
    flat = hx._flat(note)
    if hit := _ST_NOTE.match(flat):
        return {"종류": hit.group(1), "번": int(hit.group(2)), "코드": code}
    if hit := _EST_NOTE.match(flat):
        return {"종류": "산근" if hit.group(1) == "단산" else "대가", "번": int(hit.group(2))}
    if code[:1] in ("B", "D", "M"):
        return {"종류": {"B": "대가", "D": "산근", "M": "자재"}[code[0]], "코드": code}
    return None


def read_bill(name: str, data: bytes) -> dict[str, Any]:
    """엑셀 → `{꼴, 시트, 줄 [{행, 묶음, 위, 공종, 이름, 규격, 단위, 수량, 단가 {labor, material,
    expense}, 금액 {…}, 비고, 호표, 제외}], 합계 {직접, 제외}, 조달수수료율, 합계줄, 율줄}` —
    `위` = 부모 줄 행(None = 맨 위) · 수 = JSON 수."""
    title, rows = _sheet(name, data)
    start, cols = _columns(rows)
    lines: list[dict[str, Any]] = []
    stack: list[tuple[Any, dict[str, Any]]] = []  # (코드 | 깊이, 묶음 줄)
    stmate, skipped, fee, pct_lines = False, 0, None, 0
    for at, row in enumerate(rows[start:], start + 1):
        cell = lambda h: hx._cell(row, cols.get(h))  # noqa: E731
        head, item_name = hx._text(cell("공종")), hx._text(cell("품명"))
        if not item_name:
            continue
        qty = hx._num(cell("수량"))
        codes = _codes(row, cols.get("비고", 13))
        stmate = stmate or codes is not None
        if codes == ("", ""):
            continue
        if codes:
            own, parent = codes
            group = own[:1] in ("E", "C")
            at_up = next((n for n in range(len(stack) - 1, -1, -1) if stack[n][0] == parent), None)
            stack = stack[: at_up + 1] if at_up is not None else []
        else:
            own = ""
            group = bool(head) and qty is None
            if not group and qty is None:
                skipped += 1  # 「순 공 사 비」 같은 합계 줄
                continue
            if group:
                depth = _depth(head)
                while stack and stack[-1][0] >= depth:
                    stack.pop()
        up = stack[-1][1] if stack else None
        note = hx._text(cell("비고"))
        line = {
            "행": at,
            "묶음": group,
            "위": up["행"] if up else None,
            "공종": head,
            "이름": item_name,
            "규격": hx._text(cell("규격")),
            "단위": hx._text(cell("단위")),
            "수량": qty,
            "단가": _money(row, cols, 0),
            "금액": _money(row, cols, 1),
            "비고": note,
            "호표": None if group else _hopyo(note, own),
            "제외": (up or {}).get("제외", False)
            or (up is None and (own[:1] == "C" or "제외" in note or "관급" in item_name)),
        }
        if group:
            stack.append((own if codes else _depth(head), line))
        elif hx._flat(line["단위"]) == "%":
            if line["제외"] and "수수료" in item_name:
                fee = fee if fee is not None else qty
                continue
            pct_lines += 1
            line.update({"단가": line["금액"], "수량": Decimal(1), "단위": "식"})
            line["규격"] = " ".join(t for t in (line["규격"], f"({hx._out(qty)} %)") if t)
        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(
        {
            "꼴": "STmate" if stmate else "EST",
            "시트": title,
            "줄": lines,
            "합계": total,
            "조달수수료율": fee,
            "합계줄": skipped,
            "율줄": pct_lines,
        }
    )


def _unwrap(lines: list[dict[str, Any]]) -> None:
    """직접 맨 위 묶음이 하나뿐이면(공사 이름) 벗김 — 자식이 맨 위로(되풀이)."""
    while True:
        tops = [x for x in lines if x["위"] is None and not x["제외"]]
        if len(tops) != 1 or not tops[0]["묶음"]:
            return
        wrap = tops[0]["행"]
        lines.remove(tops[0])
        for line in lines:
            if line["위"] == wrap:
                line["위"] = None


def _plain(value: Any) -> Any:
    if isinstance(value, Decimal):
        return hx._out(value)
    if isinstance(value, dict):
        return {k: _plain(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_plain(v) for v in value]
    return value