From 4cd4fdb3a39e46847024e9fa30e5e983fa647513 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sat, 31 Oct 2026 15:53:33 +0900 Subject: [PATCH] =?UTF-8?q?feat(B09=20=EB=B6=88=EB=9F=AC=EC=98=A4=EA=B8=B0?= =?UTF-8?q?):=20=EC=9B=90=EA=B0=80=EA=B3=84=EC=82=B0=EC=84=9C=20=EC=9A=94?= =?UTF-8?q?=EC=9C=A8=20=C2=B7=20=EA=B8=B0=EC=B4=88=20=EB=8B=A8=EA=B0=80?= =?UTF-8?q?=ED=91=9C=20=EC=9D=BD=EA=B8=B0=20=C2=B7=20=EC=9D=B4=20=EA=B3=B5?= =?UTF-8?q?=EC=82=AC=20=EB=8B=A8=EA=B0=80=20=EC=B1=85=20=EC=9D=B4=EC=9D=8C?= =?UTF-8?q?=20=C2=B7=20=EC=98=AC=EB=A6=B0=20=ED=8C=8C=EC=9D=BC=20=EC=97=B4?= =?UTF-8?q?=EC=87=A0=20=EA=B8=B8=20(PLAN=20707-2=20=C2=B7=20707-3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - POST estimation/import/rates · import/prices (file | file_key · 저장 안 함 · patch 만) - 요율 = 행 이름 → 줄 키 · 손입력 칸(안전 기초액 · 이윤 보정 · 부가세 조정 · 조달수수료율) · 엑셀 대조 - 단가표 = 노임 · 자재 · 기계 → labor_cover · machine_cover · materials.applied · 짝 없음 · 짝 여럿 Co-Authored-By: Claude Opus 5.5 --- .../B09_Estimation_Import_Prices.py | 186 ++++++++++++ B09_Estimation/B09_Estimation_Import_Rates.py | 278 ++++++++++++++++++ B09_Estimation/B09_Estimation_Router.py | 5 +- .../B09_Estimation_Router_Import.py | 144 +++++++++ .../test_707_b09_import_rates_prices.py | 174 +++++++++++ 5 files changed, 786 insertions(+), 1 deletion(-) create mode 100644 B09_Estimation/B09_Estimation_Import_Prices.py create mode 100644 B09_Estimation/B09_Estimation_Import_Rates.py create mode 100644 B09_Estimation/B09_Estimation_Router_Import.py create mode 100644 resources/tester/test_707_b09_import_rates_prices.py diff --git a/B09_Estimation/B09_Estimation_Import_Prices.py b/B09_Estimation/B09_Estimation_Import_Prices.py new file mode 100644 index 000000000..ae2a337e0 --- /dev/null +++ b/B09_Estimation/B09_Estimation_Import_Prices.py @@ -0,0 +1,186 @@ +"""B09 경쟁사 기초 단가표 불러오기(PLAN 707-3) — 서버 단독 · 저장 안 함 +(덮개 제안 → 화면 캐시 → [저장]). + +읽는 시트(꼴마다 첫 것) — 머리 = 「명칭」 · 「품명」 칸이 든 줄: + 노임 = STmate 「노무비목록표」 · EST 「노임단가」 → 단가(「적용단가」 › 「단가」) + 자재 = STmate 「재료비목록표」 › 「자재단가대비표」 · EST 「자재단가」 → 적용 단가(단위 % 줄 뺌) + 기계 = STmate 「중기목록표」(= 「중기사용료」 블록 합계) · + EST 「기계경비총괄표」(= 「기계경비」 합) → 시간 단가 노무비 · 재료비 · 경비 +이음 = 이 공사 단가 책 줄(바탕 셈 `compute_base` 노무비목록표 · 중기목록표 · 자재단가대비표)에 + 이름 · 규격 · 단위로 — 705-3 이음 낱말 그대로(노임 = `M02_Hopyo_Shares.trade` 직종 이름 · + 기계 = 이름 + 규격 같음 › `eq_key` 갈래 · 크기 · 자재 = 「별도계상」 글 뗀 이름 · 규격 · 단위 › + 이름 · 단위) · 한 줄에만 맞을 때만 · 기계 갈래 · 크기는 표지 글(타이어 · 브레이커 · 할증 …) + 까지 · 한 책 줄에 엑셀 줄 여럿 = 「짝 여럿」(못 고름). +덮개 = 노임 `labor_cover` · 기계 `machine_cover`(새 · `B09_Estimation_Calc_Base`) · + 자재 = 지금 덮개 `materials[키].applied`(고른 출처 `pick` 은 뺌 — 적용 차례가 pick 먼저) · + 책 값과 같으면 덮지 않음(옛 덮개 걷음). +못 이은 줄 = 「짝 없음 n」(이름 목록). +""" + +from __future__ import annotations + +import re +import unicodedata +from collections import Counter +from typing import Any + +from B09_Estimation.B09_Estimation_Import_Rates import flat, money, read_sheets +from M02_MasterTemplete import M02_Hopyo_Material as hmat +from M02_MasterTemplete import M02_Hopyo_Shares as shares +from M02_MasterTemplete.M02_Hopyo_Excel import ExcelError + +KINDS = { + "labor": ("노무비목록표", "노임단가"), + "material": ("재료비목록표", "자재단가대비표", "자재단가"), + "machine": ("중기목록표", "기계경비총괄표"), +} +PARTS = ("labor", "material", "expense") +_MACHINE = {"labor": "노무비", "material": "재료비", "expense": "경비"} +LIST_MAX = 40 +#: 같은 갈래 · 크기라도 다른 기계 — 표지 글이 같아야 이음 +_MARK = re.compile(r"타이어|브레|할증|암석|\(암\)|집게|진동|탑재|크레인|살수") + + +def _n(text: Any) -> str: + return flat(unicodedata.normalize("NFKC", str(text or ""))).lower() + + +def table(rows: list[list[Any]], machine: bool) -> list[dict[str, Any]]: + """목록 시트 → [{name, spec, unit, price | labor · material · expense}].""" + for at, row in enumerate(rows[:12]): + cols = [flat(v) for v in row] + name = next((cols.index(h) for h in ("명칭", "품명") if h in cols), None) + if name is None: + continue + spec, unit = (cols.index(h) if h in cols else None for h in ("규격", "단위")) + if machine: + price = {p: cols.index(h) for p, h in _MACHINE.items() if h in cols} + if len(price) < 3: + continue + else: + price = next((cols.index(h) for h in ("적용단가", "단가") if h in cols), None) + if price is None: + continue + out = [] + for line in rows[at + 1 :]: + + def cell(col: int | None, line: list[Any] = line) -> Any: + return line[col] if col is not None and col < len(line) else None + + item = {"name": str(cell(name) or "").strip(), "spec": str(cell(spec) or "").strip(), + "unit": str(cell(unit) or "").strip()} # fmt: skip + if machine: + item.update({p: money(cell(c)) for p, c in price.items()}) + ok = None not in (item[p] for p in PARTS) + else: + item["price"] = money(cell(price)) + ok = item["price"] is not None and item["unit"] != "%" + if item["name"] and ok: + out.append(item) + return out + return [] + + +def read_prices(name: str, data: bytes) -> dict[str, list[dict[str, Any]]]: + """파일 → {labor, material, machine: 줄 목록} · 시트 없음 = 빈 목록 · 셋 다 없으면 막힘.""" + sheets = read_sheets(name, data, tuple(s for names in KINDS.values() for s in names)) + out = {} + for kind, names in KINDS.items(): + sheet = next((s for s in names if s in sheets), None) + out[kind] = table(sheets[sheet], kind == "machine") if sheet else [] + out[f"{kind}_sheet"] = sheet + if not any(out[k] for k in KINDS): + raise ExcelError("노무비목록표 · 재료비목록표 · 중기목록표(STmate) · 노임단가 · 자재단가 · " + "기계경비총괄표(EST) 시트가 없음") # fmt: skip + return out + + +def _labor_keys(row: dict[str, Any]) -> list[str]: + return [_n(shares.trade(row.get("name")) or row.get("name"))] + + +def _machine_keys(row: dict[str, Any]) -> list[str]: + """[이름 + 규격 같음, 갈래 · 크기 + 표지 글(타이어 · 브레이커 · 할증 · 암 …)].""" + text = f"{row.get('name') or ''} {row.get('spec') or ''}" + kind = shares.eq_key(text) + marks = "+".join(sorted(set(_MARK.findall(_n(text))))) + return [_n(text), f"{kind}#{marks}" if kind else ""] + + +def _material_keys(row: dict[str, Any]) -> list[str]: + me, _ = hmat.clean({"이름": row.get("name"), "규격": row.get("spec")}) + unit = _n(row.get("unit")) + return [f"{_n(me['이름'])}|{_n(me['규격'])}|{unit}", f"{_n(me['이름'])}||{unit}"] + + +def _match(book: list[dict[str, Any]], keys_of, item: dict[str, Any]) -> dict[str, Any] | None: + """책 줄 하나 — 단계마다 한 줄에만 맞으면 그 줄 · 여럿 · 없음이면 다음 단계.""" + for step, want in enumerate(keys_of(item)): + hits = [r for r in book if keys_of(r)[step] == want] if want else [] + if len(hits) == 1: + return hits[0] + return None + + +def _same(a: Any, b: Any) -> bool: + if isinstance(a, dict): + return all(_same(a[p], (b or {}).get(p)) for p in PARTS) + return a is not None and b is not None and abs(a - b) < 0.5 + + +def propose(read: dict[str, Any], sheets: dict[str, Any], user: dict[str, Any]) -> dict[str, Any]: + """읽은 단가표 + 이 공사 단가 책(덮개 뺀 바탕 셈) → 덮개 patch · 이은 줄 · 짝 없음.""" + base = {"labor": sheets.get("labor") or [], "machine": sheets.get("machine") or [], + "material": sheets.get("compare") or []} # fmt: skip + keys = {"labor": _labor_keys, "machine": _machine_keys, "material": _material_keys} + labor_cover = dict(user.get("labor_cover") or {}) + machine_cover = dict(user.get("machine_cover") or {}) + materials = {k: dict(v) for k, v in (user.get("materials") or {}).items()} + linked: list[dict[str, Any]] = [] + missing: dict[str, list[str]] = {k: [] for k in KINDS} + for kind in KINDS: + items = read.get(kind) or [] + rows = [_match(base[kind], keys[kind], item) for item in items] + hits = Counter(r["key"] for r in rows if r) + for item, row in zip(items, rows, strict=True): + label = " ".join(x for x in (item["name"], item["spec"]) if x) + if row is None or hits[row["key"]] > 1: # 한 책 줄에 엑셀 줄 여럿 = 못 고름 + missing[kind].append(label if row is None else f"{label}(짝 여럿)") + continue + if kind == "machine": + got = {p: item[p] for p in PARTS} + book = {p: row.get(p) for p in PARTS} + else: + got, book = item["price"], row.get("price" if kind == "labor" else "applied") + linked.append({"kind": kind, "key": row["key"], "name": row.get("name") or "", + "from": label, "excel": got, "book": book, + "same": _same(got, book)}) # fmt: skip + same = linked[-1]["same"] # 책 값 = 엑셀 → 옛 덮개 걷음(값 = 책) + if kind in ("labor", "machine"): + cover = labor_cover if kind == "labor" else machine_cover + cover.pop(row["key"], None) + if not same: + cover[row["key"]] = got + else: + drop = ("pick", "applied") if same else ("pick",) + cell = {k: v for k, v in materials.get(row["key"], {}).items() if k not in drop} + materials[row["key"]] = cell if same else {**cell, "applied": got} + return { + "patch": { + "labor_cover": labor_cover, + "machine_cover": machine_cover, + "materials": materials, + }, # fmt: skip + "linked": linked, + "missing": {k: v[:LIST_MAX] for k, v in missing.items()}, + "counts": { + k: { + "read": len(read.get(k) or []), + "linked": sum(x["kind"] == k for x in linked), + "covered": sum(x["kind"] == k and not x["same"] for x in linked), + "book": len(base[k]), + "missing": len(missing[k]), + } + for k in KINDS + }, # fmt: skip + } diff --git a/B09_Estimation/B09_Estimation_Import_Rates.py b/B09_Estimation/B09_Estimation_Import_Rates.py new file mode 100644 index 000000000..014687a04 --- /dev/null +++ b/B09_Estimation/B09_Estimation_Import_Rates.py @@ -0,0 +1,278 @@ +"""B09 경쟁사 원가계산서 요율 불러오기(PLAN 707-2) — 서버 단독 · 저장 안 함 +(제안만 → 화면 캐시 → [저장]). + +읽는 시트 = STmate 「공사원가계산서」(율 = 비고 글 「직접노무비 x 14.5%」) · + EST 「원가계산서」(율 = 「요율」 칸 · 비면 산출근거 글). +머리 = 「금액」 칸이 든 줄 · 행 이름 = 금액 칸 앞 마지막 한글 두 글자 넘는 글 + (EST 세로 머리 · 구분 칸 뺌). +행 이름 → 원가계산서 줄 키(`ROWS` · RT 코드 = `B09_Estimation_Calc_Total` 줄 표) → + 율 덮개 `estimate.rates{줄 키: %}` · + 손입력 칸 = 지금 칸(`B09_Estimation_Store.ESTIMATE_FIELDS`): + safety_base = 산업안전보건관리비 글 「% + 5,349,000」(글 있고 + 없으면 0 · 글 없으면 안 실음) · + indirect_material · scrap = 그 줄 금액 · + profit_adjust = ROUNDDOWN((순공사원가 − 재료비 + 일반관리비) × 율) − 엑셀 이윤 · + vat_adjust = 엑셀 부가세 − ROUNDDOWN(총원가 × 율) · + procurement_fee_pct = 관급 조달수수료 ÷ 순자재대(소수 2~4 자리 중 수수료를 그대로 내는 첫 값). +대조(`check`) = 엑셀 비목 합 + 실은 값으로 우리 식(`compute_total`)을 돌려 줄마다 엑셀 금액과 견줌 — + 1 원 넘게 다른 줄 = 「식 다름」(노인장기요양 · 산업안전보건관리비 A/B 등 식 꼴이 다른 비목) · + 우리 표에 없는 행(폐기물처리 · 국가지점번호판 …) · 셈 없는 현행 비목(퇴직공제부금비 …) + = 「짝 없음」 · 「셈 없음」. +""" + +from __future__ import annotations + +import io +import re +from typing import Any + +from B09_Estimation.B09_Estimation_Calc_Base import num, rounddown +from B09_Estimation.B09_Estimation_Calc_Total import ( + CURRENT_ONLY, + RATE_LINES, + TAIL_LINES, + compute_total, +) +from M02_MasterTemplete.M02_Hopyo_Excel import DRM, ExcelError + +SHEETS = ("공사원가계산서", "원가계산서") +#: 행 이름 앞머리(빈칸 뺀 글) → 키 — 율 줄 = 원가계산서 줄 키 · 그 밖 = 비목 합 · 대조용 +ROWS: tuple[tuple[str, str], ...] = ( + ("직접재료비", "material_direct"), + ("간접재료비", "indirect_material"), + ("작업설", "scrap"), + ("직접노무비", "labor"), + ("간접노무비", "indirect_labor"), + ("산출경비", "expense"), + ("산재보험", "accident"), + ("고용보험", "employment"), + ("건강보험", "health"), + ("노인장기요양", "care"), + ("연금보험", "pension"), + ("산업안전보건관리비", "safety"), + ("기타경비", "other"), + ("환경보전비", "environment"), + ("건설기계대여금", "machine"), + ("퇴직공제", "retirement"), + ("임금채권", "wage_claim"), + ("석면", "asbestos"), + ("하도급대금", "subcontract"), + ("공사이행보증", "performance"), + ("일반관리비", "overhead"), + ("이윤", "profit"), + ("총원가", "supply_price"), + ("부가가치세", "vat"), + ("도급", "contract"), + ("관급자재대", "supply"), + ("총공사비", "grand"), + ("순공사원가", "net_cost"), + ("소계", "subtotal"), +) +RATE_KEYS = {key: (name, rt) for key, name, rt, _ in (*RATE_LINES, *TAIL_LINES)} +NO_CALC = {key: (name, rt) for key, name, rt in CURRENT_ONLY} +_PCT = re.compile(r"[x×*]\s*([\d.]+)\s*%") +_BASE = re.compile(r"%\s*\+\s*([\d,]+)") +_HANGUL = re.compile(r"[가-힣]") +_FEE = re.compile(r"원자재대\s*:\s*([\d,]+).*?조달수수료\s*:\s*([\d,]+)") + + +def flat(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")) + + +def read_sheets(name: str, data: bytes, wanted: tuple[str, ...]) -> dict[str, list[list[Any]]]: + """시트 이름 → 줄 목록(`wanted` 만 · 이름 앞뒤 빈칸 뺌) — xlsx · xls · DRM 막힘.""" + if not data: + raise ExcelError("빈 파일") + if b"DRMONE" in data[:256] and data[:4] not in (b"PK\x03\x04", b"\xd0\xcf\x11\xe0"): + raise ExcelError(DRM) + try: + if data[:4] == b"PK\x03\x04" or name.lower().endswith(".xlsx"): + import openpyxl + + book = openpyxl.load_workbook(io.BytesIO(data), read_only=True, data_only=True) + out = { + ws.title.strip(): [list(row) for row in ws.iter_rows(values_only=True)] + for ws in book.worksheets + if ws.title.strip() in wanted + } + book.close() + return out + import xlrd + + book = xlrd.open_workbook(file_contents=data, logfile=io.StringIO()) + return { + sh.name.strip(): [sh.row_values(r) for r in range(sh.nrows)] + for sh in book.sheets() + if sh.name.strip() in wanted + } + except Exception as error: # 라이브러리마다 다른 예외 — 깨진 파일 · 엑셀 아님 + raise ExcelError("엑셀 파일을 읽지 못함 — xls · xlsx 가 아니거나 깨진 파일") from error + + +def money(value: Any) -> float | None: + """칸 → 수(「80,165,010」 글도) · 아니면 None.""" + if isinstance(value, str): + value = value.replace(",", "").strip() + try: + return float(value) if value else None + except ValueError: + return None + return num(value) + + +def _pct(value: Any) -> float | None: + """요율 칸 — 「16.5%」 글 · 엑셀 % 수(0.165) → %.""" + if isinstance(value, str): + return money(value.replace("%", "")) if "%" in value else None + got = num(value) + return None if got is None else round(got * 100, 6) + + +def _key(name: str) -> str | None: + return next((key for head, key in ROWS if name.startswith(head)), None) + + +def parse(rows: list[list[Any]]) -> dict[str, Any]: + """원가계산서 줄 → [{name, key, amount, rate, text, more}] — 머리 = 「금액」 칸 든 줄.""" + head = next((i for i, r in enumerate(rows[:12]) if "금액" in [flat(v) for v in r]), None) + if head is None: + raise ExcelError("원가계산서 머리(금액 칸)를 못 찾음") + cols = [flat(v) for v in rows[head]] + at = cols.index("금액") + rate_col = cols.index("요율") if "요율" in cols else None + text_col = next((cols.index(h) for h in ("산출근거", "비고") if h in cols), None) + out = [] + for row in rows[head + 1 :]: + names = [flat(v) for v in row[:at] if len(_HANGUL.findall(str(v or ""))) > 1] + amount = money(row[at]) if at < len(row) else None + if not names: + continue + text = str(row[text_col] or "") if text_col is not None and text_col < len(row) else "" + rate = _pct(row[rate_col]) if rate_col is not None and rate_col < len(row) else None + if rate is None and (m := _PCT.search(text)): + rate = float(m.group(1)) + tail = row[max(at, text_col or 0) + 1 :] + more = [v for v in (money(x) for x in tail) if v is not None] + out.append({"name": names[-1], "key": _key(names[-1]), "amount": amount, + "rate": rate, "text": text.strip(), "more": more}) # fmt: skip + return {"rows": out} + + +def _fee(row: dict[str, Any]) -> tuple[float, float] | None: + """관급 줄 → (순자재대, 조달수수료) — STmate 글 「원자재대: X (조달수수료: Y 포함)」 · + EST 뒤 칸 둘.""" + if m := _FEE.search(row["text"].replace("\n", " ")): + raw, fee = (float(g.replace(",", "")) for g in m.groups()) + return raw - fee, fee + if len(row["more"]) >= 2 and row["more"][0] > row["more"][1] > 0: + return row["more"][0] - row["more"][1], row["more"][1] + return None + + +def _fee_pct(net: float, fee: float) -> float | None: + for places in (2, 3, 4): + pct = round(fee / net * 100, places) + if rounddown(net * pct / 100) == fee: + return pct + return None + + +def read_rates(name: str, data: bytes) -> dict[str, Any]: + """파일 → {kind, rows, estimate(실을 값), excel(키: 금액), notes}.""" + sheets = read_sheets(name, data, SHEETS) + sheet = next((s for s in SHEETS if s in sheets), None) + if sheet is None: + raise ExcelError("「공사원가계산서」(STmate) · 「원가계산서」(EST) 시트가 없음") + rows = parse(sheets[sheet])["rows"] + excel: dict[str, float] = {} + subtotals: list[float] = [] + rates: dict[str, float] = {} + estimate: dict[str, Any] = {} + notes: list[str] = [] + for row in rows: + key, amount = row["key"], row["amount"] + if key == "subtotal": + subtotals.append(amount or 0.0) + continue + if key is None: + if amount: + notes.append( + f"짝 없음 — 「{row['name']}」 {amount:,.0f} 원(우리 원가계산서에 줄 없음)" + ) + continue + if key == "safety" and key in excel: # EST A · B 두 줄 = 작은 쪽(우리 식과 같음) + excel[key] = min(excel[key], amount or excel[key]) + continue + excel[key] = amount if amount is not None else excel.get(key, 0.0) + if key in RATE_KEYS or key in NO_CALC: + if row["rate"] is not None: + rates[key] = row["rate"] + elif amount: + notes.append(f"율 없음 — 「{row['name']}」 산출근거 빈칸 · 지금 율 그대로") + if key in NO_CALC and amount: + notes.append( + f"셈 없음 — 「{row['name']}」 {amount:,.0f} 원 · 율 {row['rate']}% 만 실음" + "(우리 원가계산서 줄만 · 총액에 안 듦)" + ) + if key == "safety" and row["text"]: + got = _BASE.search(row["text"]) + estimate["safety_base"] = float(got.group(1).replace(",", "")) if got else 0.0 + if key in ("indirect_material", "scrap") and amount: + estimate[key] = amount + if key == "supply" and (fee := _fee(row)): + excel["supply_net"], excel["supply_fee"] = fee + if (pct := _fee_pct(*fee)) is not None: + estimate["procurement_fee_pct"] = pct + else: + notes.append( + f"조달수수료 {fee[1]:,.0f} 원 — 한 율로 안 맞음(품목마다 셈) · 수수료율 손입력" + ) + excel["material"] = subtotals[0] if subtotals else excel.get("material_direct") + net = excel.get("net_cost") or (sum(subtotals[:3]) if len(subtotals) >= 3 else None) + if net is not None and excel.get("overhead") is not None and "profit" in rates: + base = net - (excel["material"] or 0.0) + excel["overhead"] + adjust = rounddown(base * rates["profit"] / 100) - (excel.get("profit") or 0.0) + if adjust: + estimate["profit_adjust"] = adjust + if excel.get("supply_price") is not None and "vat" in rates and "vat" in excel: + adjust = excel["vat"] - rounddown(excel["supply_price"] * rates["vat"] / 100) + if adjust: + estimate["vat_adjust"] = adjust + kind = "STmate" if sheet == "공사원가계산서" else "EST" + return {"kind": kind, "sheet": sheet, "rows": rows, "rates": rates, + "estimate": estimate, "excel": excel, "notes": notes} # fmt: skip + + +NAMES = {"supply": "관급자재대", "supply_price": "총원가(공급가액)", "contract": "도급공사비", + "grand": "총공사비"} # fmt: skip +#: 대조 줄 — 엑셀 키 → 우리 셈 자리 +_OURS = { + **{key: ("lines", key) for key in RATE_KEYS}, + "supply": ("lines", "supply"), + "supply_price": ("subtotals", "supply_price"), + "contract": ("subtotals", "contract"), + "grand": ("subtotals", "grand"), +} + + +def check(read: dict[str, Any], estimate: dict[str, Any] | None, table: Any) -> dict[str, Any]: + """엑셀 비목 합 + (지금 사용자 값 위에 실은 값)으로 우리 식 → 줄마다 엑셀과 견줌.""" + excel = read["excel"] + mine = dict(estimate or {}) + est = {**mine, **read["estimate"], "rates": {**(mine.get("rates") or {}), **read["rates"]}} + net, fee = excel.get("supply_net", 0.0), excel.get("supply_fee", 0.0) + if "procurement_fee_pct" not in read["estimate"] and net: # 한 율로 못 맞춘 수수료 = 그 비율째 + est["procurement_fee_pct"] = fee / net * 100 + sums = {"material": excel.get("material"), "labor": excel.get("labor"), + "expense": excel.get("expense"), "supply_net": net} # fmt: skip + got = compute_total(sums, est, table) + lines = [] + for key, (part, at) in _OURS.items(): + if key not in excel: + continue + ours = got[part].get(at) + same = ours is not None and abs(ours - excel[key]) < 1 + name = RATE_KEYS[key][0] if key in RATE_KEYS else NAMES[key] + lines.append({"key": key, "name": name, "excel": excel[key], "ours": ours, "same": same}) + return {"lines": lines, "grand": got["subtotals"]["grand"], "excel_grand": excel.get("grand"), + "differ": [ln["name"] for ln in lines if not ln["same"]]} # fmt: skip diff --git a/B09_Estimation/B09_Estimation_Router.py b/B09_Estimation/B09_Estimation_Router.py index d15bd3353..944f39764 100644 --- a/B09_Estimation/B09_Estimation_Router.py +++ b/B09_Estimation/B09_Estimation_Router.py @@ -41,6 +41,7 @@ 지난 공사 일위대가 → 이 공사 서고 사본(덮개째) → {폴더, 만듦, key, cover, 덮개, payload} · 덮개 = 그 공사 저장 칸(화면 캐시 cover 가 비었을 때만 얹음 · 같은 꼴 블록이 먼저 있으면 빈 것) · 같은 회사만(남의 회사 403) +불러오기(707-2 요율 · 707-3 단가표) = `B09_Estimation_Router_Import`. 운반 사본 연결 = 셈 때마다 거리로 사본 다시 고름(`Bill_Library.resolve_hauls` · `fix_keys`). 저장 자리 = 프로젝트 안 `B09_Estimation/v1/`(`B09_Estimation_Store`) · @@ -85,6 +86,7 @@ from B09_Estimation.B09_Estimation_Calc_Total import compute_total from B09_Estimation.B09_Estimation_Labor_Source import daega_inputs from B09_Estimation.B09_Estimation_Linked import linked_resource_keys from B09_Estimation.B09_Estimation_Rates import rate_candidates, rate_table +from B09_Estimation.B09_Estimation_Router_Import import router as import_router from B09_Estimation.B09_Estimation_Router_Quotes import router as quotes_router from B09_Estimation.B09_Estimation_Summary import compute_summary, hopyo_lines, used_materials from common_util.common_util_auth import verify_session @@ -554,8 +556,9 @@ async def past_import( return {**made, "key": key, "cover": cover, "payload": out} -# 166-4 견적서 파일 길 — 같은 머리 · 같은 권한 +# 166-4 견적서 파일 길 · 707-2 · 3 불러오기 길 — 같은 머리 · 같은 권한 router.include_router(quotes_router) +router.include_router(import_router) # 302 초안 길(draft/*_Router.py 의 router · 폴더째 지우면 길도 없음) — 같은 머리 · 같은 권한 diff --git a/B09_Estimation/B09_Estimation_Router_Import.py b/B09_Estimation/B09_Estimation_Router_Import.py new file mode 100644 index 000000000..aefb067c7 --- /dev/null +++ b/B09_Estimation/B09_Estimation_Router_Import.py @@ -0,0 +1,144 @@ +"""B09 경쟁사 결과물 불러오기 길 — 요율(707-2) · 기초 단가표(707-3) · 저장 안 함(제안 → 화면 캐시). + + POST /api/projects/{id}/estimation/import/rates form: file | file_key · user?(화면 캐시 JSON) + 원가계산서 요율 → {file_key, name, kind, rows, rates, notes, check, patch{estimate}} + POST /api/projects/{id}/estimation/import/prices form: file | file_key · user? + 노임 · 자재 · 기계 단가표 → 이 공사 단가 책에 이음 → + {file_key, name, counts, linked, missing, patch{labor_cover, machine_cover, materials}} + +올린 파일 = 서버 임시 폴더에 내용 지문(`file_key`)으로 한 시간 둠 — 같은 파일을 요율 · 단가표 · +내역 줄(707-1 · sub5)이 다시 올리지 않고 `file_key` 로 씀(`load` · `remember`). +화면은 `patch` 를 캐시에 얹고 [저장] 때 서버가 정본으로 다시 셈(5장). +`B09_Estimation_Router` 가 끼워 넣음(같은 머리 · 같은 권한). +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import re +import tempfile +import time +from pathlib import Path +from typing import Any +from uuid import UUID + +from fastapi import APIRouter, File, Form, HTTPException, UploadFile + +from B03_FileInput.B03_FileInput_Engine import save_upload_stream +from B08_Quantity.B08_Quantity_Router_Hydro import _project_root +from B09_Estimation import B09_Estimation_Import_Prices as import_prices +from B09_Estimation import B09_Estimation_Import_Rates as import_rates +from B09_Estimation import B09_Estimation_Prices as prices +from B09_Estimation import B09_Estimation_Store as store +from B09_Estimation.B09_Estimation_Calc_Base import compute_base +from B09_Estimation.B09_Estimation_Rates import rate_table +from M02_MasterTemplete.M02_Hopyo_Excel import ExcelError + +router = APIRouter() +PATH = "/{project_id}/estimation/import" +KEEP_S = 3600 +_KEY = re.compile(r"^[0-9a-f]{24}$") + + +def _dir() -> Path: + return Path(tempfile.gettempdir()) / "aislo_b09_import" + + +def load(file_key: str) -> tuple[str, bytes]: + """지문 → (파일 이름, 내용) — 없거나 지난 것 = 404.""" + if not _KEY.match(file_key or ""): + raise HTTPException(400, "file_key 꼴이 아님") + data, name = _dir() / f"{file_key}.bin", _dir() / f"{file_key}.name" + if not data.is_file() or time.time() - data.stat().st_mtime > KEEP_S: + raise HTTPException(404, "올린 파일이 지났음 — 다시 올림") + return name.read_text(encoding="utf-8"), data.read_bytes() + + +async def remember(upload: UploadFile) -> tuple[str, str, bytes]: + """올린 파일 → (지문, 이름, 내용) · 한 시간 지난 파일 지움.""" + folder = _dir() + folder.mkdir(parents=True, exist_ok=True) + for old in folder.glob("*.bin"): + if time.time() - old.stat().st_mtime > KEEP_S: + old.unlink(missing_ok=True) + old.with_suffix(".name").unlink(missing_ok=True) + temp = folder / f"up-{time.time_ns()}.part" + try: + await save_upload_stream(upload, temp) + except ValueError as error: + raise HTTPException(400, str(error)) from error + data = temp.read_bytes() + key = hashlib.sha256(data).hexdigest()[:24] + temp.replace(folder / f"{key}.bin") + name = store.safe_name(upload.filename or "") or "불러온 파일.xlsx" + (folder / f"{key}.name").write_text(name, encoding="utf-8") + return key, name, data + + +async def _file(file: UploadFile | None, file_key: str | None) -> tuple[str, str, bytes]: + if file is not None: + return await remember(file) + if file_key: + return (file_key, *load(file_key)) + raise HTTPException(400, "file 또는 file_key 가 필요함") + + +def _draft(user: str | None) -> dict[str, Any] | None: + try: + raw = json.loads(user) if user else None + except ValueError as error: + raise HTTPException(400, "user 가 JSON 이 아님") from error + return store.clean_user(raw) if isinstance(raw, dict) else None + + +@router.post(PATH + "/rates") +async def import_rate_sheet( + project_id: UUID, + file: UploadFile | None = File(None), + file_key: str | None = Form(None), + user: str | None = Form(None), +) -> dict[str, Any]: + root = await _project_root(project_id) + key, name, data = await _file(file, file_key) + draft = _draft(user) + mine = (draft if draft is not None else await asyncio.to_thread(store.read_user, root))[ + "estimate" + ] + try: + read = await asyncio.to_thread(import_rates.read_rates, name, data) + except ExcelError as error: + raise HTTPException(400, str(error)) from error + check = await asyncio.to_thread(import_rates.check, read, mine, rate_table()) + estimate = {**mine, **read["estimate"], "rates": {**(mine.get("rates") or {}), **read["rates"]}} + keep = ("kind", "sheet", "rows", "rates", "notes") + return {"file_key": key, "name": name, **{k: read[k] for k in keep}, "check": check, + "patch": {"estimate": estimate}} # fmt: skip + + +@router.post(PATH + "/prices") +async def import_price_sheets( + project_id: UUID, + file: UploadFile | None = File(None), + file_key: str | None = Form(None), + user: str | None = Form(None), +) -> dict[str, Any]: + from B09_Estimation import B09_Estimation_Router as main # 서로 부름 — 부를 때만 + + root = await _project_root(project_id) + key, name, data = await _file(file, file_key) + draft = _draft(user) + mine = draft if draft is not None else await asyncio.to_thread(store.read_user, root) + try: + read = await asyncio.to_thread(import_prices.read_prices, name, data) + except ExcelError as error: + raise HTTPException(400, str(error)) from error + keys = await main._demand(root, draft) + b08 = await main._b08_for(project_id, [], True) + book = await asyncio.to_thread(store.read_book, root) + inputs = await asyncio.to_thread(prices.build_inputs, book, keys, b08["materials"]) + base = compute_base(inputs, {}) # 덮개 뺀 이 공사 단가 책 + out = import_prices.propose(read, base, mine) + sheets = {k: read[f"{k}_sheet"] for k in import_prices.KINDS} + return {"file_key": key, "name": name, "sheets": sheets, **out} diff --git a/resources/tester/test_707_b09_import_rates_prices.py b/resources/tester/test_707_b09_import_rates_prices.py new file mode 100644 index 000000000..5b328bc48 --- /dev/null +++ b/resources/tester/test_707_b09_import_rates_prices.py @@ -0,0 +1,174 @@ +"""707-2 원가계산서 요율 · 707-3 기초 단가표 불러오기 — 실무 xlsx 넷(EST 거창 · STmate 영월 · 울진 대흥 · +울진 소광) · 이음 규칙 · 덮개 지킴(새 판 [업데이트] 뒤). + +총액 대조 = 엑셀 비목 합 + 실은 요율 · 손입력으로 우리 식(`compute_total`) → 엑셀 총공사비: + 대흥 = 같음 · 영월 = 「국가지점번호판검증수수료」 762,000 원만 다름(우리 표에 줄 없음) · + 거창 = 퇴직공제부금비(셈 없는 현행 비목)만큼 · 소광 = 산출근거 빈칸 율 넷 · 폐기물처리 줄. +""" + +import sys +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_Import_Prices as ip # noqa: E402 +from B09_Estimation import B09_Estimation_Import_Rates as ir # noqa: E402 +from B09_Estimation import B09_Estimation_Store as store # noqa: E402 +from B09_Estimation.B09_Estimation_Calc_Base import compute_base # noqa: E402 +from B09_Estimation.B09_Estimation_Rates import rate_table # noqa: E402 + +K = ROOT / "resources" / "knowledge" / "original" / "실무문서" +FILES = { + "거창": "01.산불거창(거창북상병곡)/01.내역서/00.25년_산불_거창(거창,북상,병곡,산8외)내역+수량.xlsx", + "영월": "05. 동부청 영월/내역서/2024년 임도시설사업2차(산불진화_기번6)-영월.xlsx", + "대흥": "2024년 간선임도(기번3-울진.대흥)(공통)/1. 설계원본/01 내역자료/2024년 간선임도사업(기번3).xlsx", + "소광": "25년 산불진화임도(기번8)(울진소광)/" + "2025년 산불진화임도 신설사업(기번8 울진.금강송.소광.산29외).xlsx", +} + + +def _file(tag): + path = K / FILES[tag] + if not path.is_file(): + pytest.skip(f"실무 xlsx 없음 — {tag}") + return path.name, path.read_bytes() + + +@pytest.fixture(scope="module") +def rates(): + return {tag: ir.read_rates(*_file(tag)) for tag in FILES} + + +def test_707_2_요율_짝(rates): + dh = rates["대흥"] + assert dh["kind"] == "STmate" and rates["거창"]["kind"] == "EST" + assert len(dh["rates"]) == 13 # 율 줄 10 + 일반관리비 · 이윤 · 부가세 + assert (dh["rates"]["indirect_labor"], dh["rates"]["safety"], dh["rates"]["profit"]) == ( + 14.5, + 1.86, + 15.0, + ) + assert dh["estimate"]["safety_base"] == 5349000 # 「x1.86% + 5,349,000」 + assert dh["estimate"]["procurement_fee_pct"] == 0.54 # 375,160 ÷ 69,474,220 + gc = rates["거창"] # EST 요율 칸 · 건강보험 글 「13.595%」 대신 칸 3.595 + assert (gc["rates"]["health"], gc["rates"]["care"], gc["rates"]["retirement"]) == ( + 3.595, + 13.14, + 2.3, + ) + assert gc["estimate"]["safety_base"] == 0 # 글에 + 기초액 없음 + assert any("퇴직공제부금비" in n and "셈 없음" in n for n in gc["notes"]) + sk = rates["소광"] + assert sum("율 없음" in n for n in sk["notes"]) == 4 # 산출근거 빈칸 — 지금 율 그대로 + assert any("페기물처리" in n for n in sk["notes"]) + + +def test_707_2_총액_대조(rates): + got = {tag: ir.check(r, {}, rate_table()) for tag, r in rates.items()} + assert got["대흥"]["grand"] == got["대흥"]["excel_grand"] == 1201879000 + assert got["대흥"]["differ"] == [] + assert got["영월"]["excel_grand"] - got["영월"]["grand"] == 762000 # 국가지점번호판 줄 몫 + assert set(got["영월"]["differ"]) == {"도급공사비", "총공사비"} + lines = {ln["key"]: ln for ln in got["거창"]["lines"]} + rate_keys = ("indirect_labor", "accident", "employment", "health", "care", "pension") + assert all(lines[k]["same"] for k in (*rate_keys, "safety", "other", "environment")) + assert not lines["overhead"]["same"] # 퇴직공제부금비 5,603,907 이 순공사원가에 없음 + assert lines["overhead"]["excel"] - lines["overhead"]["ours"] == 448312 # = 5,603,907 × 8 % + + +def test_707_3_단가표_읽기(): + counts = {} + for tag in FILES: + got = ip.read_prices(*_file(tag)) + counts[tag] = tuple(len(got[k]) for k in ip.KINDS) + assert counts == {"거창": (17, 52, 22), "영월": (16, 55, 29), "대흥": (12, 39, 25), + "소광": (14, 35, 24)} # fmt: skip + + +BASE = { + "labor": [ + {"key": "LB1", "name": "석공", "spec": "", "unit": "인", "price": 268908}, + {"key": "LB2", "name": "보통인부", "spec": "", "unit": "인", "price": 172068}, + ], + "machine": [ + { + "key": "EQ1", + "name": "굴착기(무한궤도)", + "spec": "0.7㎥", + "labor": 1, + "material": 2, + "expense": 3, + }, + { + "key": "EQ2", + "name": "굴착기(무한궤도)", + "spec": "0.2㎥", + "labor": 1, + "material": 2, + "expense": 3, + }, + ], # fmt: skip + "compare": [ + {"key": "MO1", "name": "휘발유", "spec": "전국평균", "unit": "L", "applied": 1863.25}, + {"key": "MT1", "name": "레미콘", "spec": "25-18-80", "unit": "㎥", "applied": 92330}, + ], +} + + +def _m(name, spec, n): + return {"name": name, "spec": spec, "unit": "hr", "labor": n, "material": n, "expense": n} + + +READ = { + "labor": [ + {"name": "석 공", "spec": "", "unit": "인", "price": 258935}, + {"name": "보통인부", "spec": "", "unit": "인", "price": 172068}, # 책과 같음 + {"name": "벌 목 부", "spec": "", "unit": "인", "price": 253243}, + ], + "machine": [ + _m("굴삭기(무한궤도)", "0.7㎥", 10), + _m("굴삭기(타이어형)", "0.7㎥", 11), # 타이어 = 다른 기계 + _m("굴삭기+브레이카", "0.7㎥", 12), + _m("굴삭기(무한궤도)", "0.2㎥", 13), + _m("굴삭기(무한궤도)", "0.2㎥", 14), # 같은 책 줄에 둘 = 못 고름 + ], + "material": [ + {"name": "휘 발 유", "spec": "", "unit": "ℓ", "price": 1473}, # 이름 · 단위(ℓ = L) + {"name": "레미콘(관급)", "spec": "25-18-80", "unit": "m3", "price": 92330}, + ], +} + + +def test_707_3_이음_규칙(): + user = {"labor_cover": {"LB2": 1}, "materials": {"MO1": {"pick": "유가", "supply": "관급"}}} + out = ip.propose(READ, BASE, user) + patch = out["patch"] + assert patch["labor_cover"] == {"LB1": 258935} # 같은 값 = 옛 덮개 걷음 + assert patch["machine_cover"] == {"EQ1": {"labor": 10, "material": 10, "expense": 10}} + assert patch["materials"]["MO1"] == {"supply": "관급", "applied": 1473} # pick 뺌 + assert "MT1" not in patch["materials"] # 「(관급)」 이름 다름 = 짝 없음 + assert "굴삭기(무한궤도) 0.2㎥(짝 여럿)" in out["missing"]["machine"] + assert out["counts"]["machine"] == {"read": 5, "linked": 1, "covered": 1, "missing": 4, + "book": 2} # fmt: skip + assert out["counts"]["labor"]["linked"] == 2 and out["missing"]["labor"] == ["벌 목 부"] + + +def test_707_3_덮개_새_판_뒤_남음(tmp_path): + """[업데이트] = 단가 책만 새로 굳힘 — 사용자 값의 노임 · 기계 덮개는 그대로 · 셈은 덮개.""" + cover = {"labor_cover": {"LB1": 258935, "": None}, + "machine_cover": {"EQ1": {"labor": 10, "material": 10, "expense": 10}, + "EQ2": {"labor": 1}}} # fmt: skip + saved = store.write_user(tmp_path, cover) + assert saved["labor_cover"] == {"LB1": 258935} # 빈 값 버림 + assert saved["machine_cover"] == {"EQ1": {"labor": 10, "material": 10, "expense": 10}} + for price in (268908, 280000): # 옛 판 → 새 판 책 + store.write_book(tmp_path, {"rows": {"LB1": {"key": "LB1", "price": price}}}) + user = store.read_user(tmp_path) + assert user["labor_cover"] == saved["labor_cover"] + inputs = {"labor": [{"key": "LB1", "name": "석공", "price": price}]} + row = compute_base(inputs, user)["labor"][0] + assert (row["price"], row["book"], row["covered"]) == (258935, price, True)