From 63ba44a717b83a52e6d4e85bbd15451d71b0c6f9 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Wed, 14 Oct 2026 22:03:59 +0900 Subject: [PATCH] =?UTF-8?q?feat(=EC=9D=BC=EC=9C=84=EB=8C=80=EA=B0=80?= =?UTF-8?q?=EC=8B=A4=EB=AC=B4):=20324-1=20=EC=8B=A4=EB=AC=B4=EC=9E=90?= =?UTF-8?q?=EB=A3=8C=20=EC=97=91=EC=85=80=20=EC=9D=BD=EA=B8=B0=20=C2=B7=20?= =?UTF-8?q?=EC=A0=80=EC=9E=A5=20=C2=B7=20=EC=98=AC=EB=A6=AC=EA=B8=B0=20?= =?UTF-8?q?=EA=B8=B8=20=E2=80=94=20=E3=80=8C=EC=9D=BC=EC=9C=84=EB=8C=80?= =?UTF-8?q?=EA=B0=80=E3=80=8D=20=EC=8B=9C=ED=8A=B8=20=E2=86=92=20=ED=98=B8?= =?UTF-8?q?=ED=91=9C=20=C2=B7=20=EC=A4=84=20=C2=B7=20=EA=B0=88=EB=9E=98=20?= =?UTF-8?q?=C2=B7=20=EC=B0=B8=EC=A1=B0=20=C2=B7=20=EA=B2=80=EC=82=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - M02_Hopyo_Excel: xls(xlrd · logfile 막음) · xlsx(openpyxl) · 머리 글로 시트 · 열 찾기 · 제N호표 · 구성 줄 · 갈래(단가산출 · 자재 · 기계 · 호표 · % · 별도계상 · 기타) · 단산N/일위N 참조 · 검산(줄 금액 floor · 합계 단가 · % 줄 · 호표 비목 · 총괄표) · 규격 원문 보관 · 시트 없음 = 「일위대가 시트 없음」 - M02_Hopyo_Store: 층 폴더 실무자료/HP-######.json · 맞춤 기억 실무자료/맞춤.json · 잠금 + 원자 쓰기 · 양식 종류 밖(목록 · 복사 · 프로젝트 씨앗에 안 섞임) - M02_Labor_Router: GET/POST /layers/{층}/hopyo · GET/DELETE /layers/{층}/hopyo/{키} · 20 MB · 엉터리 · 시트 없음 400 · 권한 = 층 읽기 · 쓰기 규칙 - requirements: xlrd==2.0.2 - 시험 test_m02_hopyo_excel 15 통과(북부청 68 · 사방 27 · 검산 0 · 역할 표 · sha256) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Ji4nuyLVHcxrXWHTdb5nRD --- M02_MasterTemplete/M02_Hopyo_Excel.py | 269 +++++++++++++++++++++++ M02_MasterTemplete/M02_Hopyo_Store.py | 116 ++++++++++ M02_MasterTemplete/M02_Labor_Router.py | 66 +++++- requirements.txt | 1 + resources/tester/test_m02_hopyo_excel.py | 209 ++++++++++++++++++ 5 files changed, 659 insertions(+), 2 deletions(-) create mode 100644 M02_MasterTemplete/M02_Hopyo_Excel.py create mode 100644 M02_MasterTemplete/M02_Hopyo_Store.py create mode 100644 resources/tester/test_m02_hopyo_excel.py diff --git a/M02_MasterTemplete/M02_Hopyo_Excel.py b/M02_MasterTemplete/M02_Hopyo_Excel.py new file mode 100644 index 000000000..38d11a81e --- /dev/null +++ b/M02_MasterTemplete/M02_Hopyo_Excel.py @@ -0,0 +1,269 @@ +"""M02 일위대가 실무자료 읽기(PLAN 324-1) — 설계내역서 엑셀 「일위대가」 시트 → +호표 · 줄 · 갈래 · 검산. + +읽기 = .xls `xlrd`(경고가 stdout 으로 안 나가게 logfile) · .xlsx `openpyxl`(빈 칸 None) · +시트 · 열은 머리 글(빈칸 뺌)로 찾음 — 호표 · 품명 · 규격 · 수량 · 단위 · 합계 · 노무비 · +재료비 · 경비 · 비고(돈 머리 = 단가 열 · 다음 열 = 금액). +호표 줄 = 「제N호표」 · 호표 1 단위 값 = 금액 열 · 구성 줄 = 호표 칸 빔 · 품명 있음 · +끝 = 다음 호표. +갈래 = % · 호표(비고 일위N참조) · 단가산출(단산N참조) · 기계(단위 hr) · 별도계상(값 없음) · +자재(재료비만) · 기타 — 규격은 원문 그대로 둠. +검산 = 줄 금액 floor(단가 × 수량 × 10) / 10 · 합계 단가 = 셋 합 · +% 줄 = floor(바탕 × 율 / 100 × 10) / 10 · 호표 비목 = floor(줄 금액 합) · 총괄표 합계 — +어긋남은 글 목록(막지 않음). +""" + +from __future__ import annotations + +import io +import re +from decimal import ROUND_FLOOR, Decimal, InvalidOperation +from typing import Any + +SHEET = "일위대가" +SUMMARY = "일위대가총괄표" +COSTS = ("노무비", "재료비", "경비") +HEADS = ("호표", "품명", "규격", "수량", "단위", "비고") +NO_SHEET = "일위대가 시트 없음" +_HOPYO = re.compile(r"제\s*(\d+)\s*호표") +_REF = re.compile(r"^(단산|일위)\s*(\d+)\s*참조$") +_MACHINE_UNITS = ("hr", "시간", "h") + + +class ExcelError(ValueError): + """엉터리 파일 · 시트 없음 — 길이 400 으로 답함.""" + + +def _flat(value: Any) -> str: + return re.sub(r"\s+", "", str(value or "")) + + +def _text(value: Any) -> str: + if isinstance(value, float) and value == int(value): + value = int(value) + return str(value if value is not None else "").strip() + + +def _num(value: Any) -> Decimal | None: + if value is None or isinstance(value, bool) or value == "": + return None + try: + got = Decimal(str(value).replace(",", "").strip()) + except InvalidOperation: + return None + return got if got.is_finite() else None + + +def _floor(value: Decimal, places: int = 1) -> Decimal: + step = Decimal(1).scaleb(-places) + return value.quantize(step, rounding=ROUND_FLOOR) + + +def _out(value: Decimal | None) -> int | float | None: + if value is None: + return None + return int(value) if value == value.to_integral_value() else float(value) + + +# ── 파일 → 시트 표(줄마다 값 목록) ──────────────────────────────────────── +def _sheets(name: str, data: bytes) -> dict[str, list[list[Any]]]: + """시트 이름 → 줄 목록 — 일위대가 · 총괄표만 읽음.""" + wanted = (SHEET, SUMMARY) + if not data: + raise ExcelError("빈 파일") + 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 ExcelError: + raise + except Exception as error: # 라이브러리마다 다른 예외 — 깨진 파일 · 엑셀 아님 + raise ExcelError("엑셀 파일을 읽지 못함 — xls · xlsx 가 아니거나 깨진 파일") from error + + +def _columns(rows: list[list[Any]], money: tuple[str, ...]) -> tuple[int, dict[str, int]]: + """(자료 첫 줄, {머리: 열}) — 머리 글은 앞 10 줄 안 「호표」 · 「품명」 이 든 줄.""" + for at, row in enumerate(rows[:10]): + flat = [_flat(v) for v in row] + if "호표" not in flat or "품명" not in flat: + continue + cols = {head: flat.index(head) for head in (*HEADS, *money) if head in flat} + below = [_flat(v) for v in rows[at + 1]] if at + 1 < len(rows) else [] + start = at + 2 if "단가" in below or "금액" in below else at + 1 + return start, cols + raise ExcelError(f"{SHEET} 시트 머리 글(호표 · 품명)을 못 찾음") + + +def _cell(row: list[Any], col: int | None) -> Any: + return row[col] if col is not None and col < len(row) else None + + +def _money(row: list[Any], cols: dict[str, int], offset: int) -> dict[str, Decimal | None]: + """비목 넷(합계 = 계) — `offset` 0 = 단가 열 · 1 = 금액 열.""" + return { + ("계" if head == "합계" else head): _num(_cell(row, None if c is None else c + offset)) + for head in ("합계", *COSTS) + for c in [cols.get(head)] + } + + +# ── 갈래 · 검산 ───────────────────────────────────────────────────────── +def kind_of(line: dict[str, Any]) -> str: + """구성 줄 갈래 — 단가산출 · 자재 · 기계 · 호표 · % · 별도계상 · 기타.""" + ref = line.get("참조") or {} + unit = _flat(line.get("단위")).lower() + price = line["단가"] + if unit == "%": + return "%" + if ref.get("종류") == "일위": + return "호표" + if ref.get("종류") == "단산": + return "단가산출" + if unit in _MACHINE_UNITS: + return "기계" + if not any(price.get(k) for k in ("계", *COSTS)): + return "별도계상" + if price.get("재료비") and not price.get("노무비") and not price.get("경비"): + return "자재" + return "기타" + + +def _check_line(where: str, line: dict[str, Any]) -> list[str]: + qty, price, amount = line["수량"], line["단가"], line["금액"] + bad = [] + if line["갈래"] == "%": + base = price.get("계") + want = _floor(base * qty / 100) if base is not None and qty else None + got = sum((amount.get(k) or Decimal(0)) for k in COSTS) + if want is not None and got != want: + bad.append(f"{where} · % 금액 {got} ≠ {want}") + return bad + if qty is None: + return bad + for k in COSTS: + if price.get(k) is None: + continue + want = _floor(price[k] * qty) + if (amount.get(k) or Decimal(0)) != want: + bad.append(f"{where} · {k} 금액 {amount.get(k)} ≠ {want}") + parts = [price.get(k) for k in COSTS] + if price.get("계") is not None and any(p is not None for p in parts): + total = sum((p or Decimal(0)) for p in parts) + if total != price["계"]: + bad.append(f"{where} · 합계 단가 {price['계']} ≠ {total}") + return bad + + +def _check_hopyo(hopyo: dict[str, Any], lines: list[dict[str, Any]]) -> list[str]: + where = f"제{hopyo['번']}호표" + bad = [] + for line in lines: + bad += _check_line(f"{where} {line['차례'] + 1}째 줄 「{line['이름']}」", line) + if not lines: + return bad + for k in COSTS: + total = sum((line["금액"].get(k) or Decimal(0)) for line in lines) + want, got = _floor(total, 0), hopyo["금액"].get(k) or Decimal(0) + if got != want: + bad.append(f"{where} · {k} {got} ≠ 줄 합 {want}") + return bad + + +def _summary(rows: list[list[Any]] | None) -> dict[int, Decimal]: + """총괄표 {호표 번: 합계} — 머리 글이 없으면 빈 것.""" + if not rows: + return {} + try: + start, cols = _columns(rows, ("합계",)) + except ExcelError: + return {} + out = {} + for row in rows[start:]: + hit = _HOPYO.fullmatch(_flat(_cell(row, cols.get("호표")))) + total = _num(_cell(row, cols.get("합계"))) + if hit and total is not None: + out[int(hit.group(1))] = total + return out + + +# ── 읽기 ────────────────────────────────────────────────────────────── +def read_book(name: str, data: bytes) -> dict[str, Any]: + """엑셀 → `{시트, 호표: [...], 검산: {어긋남: [글 …], 총괄표: 맞춘 수}}` — 값은 JSON 수.""" + sheets = _sheets(name, data) + rows = sheets.get(SHEET) + if rows is None: + raise ExcelError(NO_SHEET) + start, cols = _columns(rows, ("합계", *COSTS)) + hopyos: list[dict[str, Any]] = [] + raw: list[list[dict[str, Any]]] = [] + for row in rows[start:]: + head = _flat(_cell(row, cols.get("호표"))) + name_ = _text(_cell(row, cols.get("품명"))) + hit = _HOPYO.fullmatch(head) + if hit: + hopyos.append( + { + "번": int(hit.group(1)), + "이름": name_, + "규격": _text(_cell(row, cols.get("규격"))), + "단위": _text(_cell(row, cols.get("단위"))), + "금액": _money(row, cols, 1), + } + ) + raw.append([]) + continue + if head or not name_ or not hopyos: + continue + note = _text(_cell(row, cols.get("비고"))) + ref = _REF.match(note.replace(" ", "")) + line = { + "차례": len(raw[-1]), + "이름": name_, + "규격": _text(_cell(row, cols.get("규격"))), + "수량": _num(_cell(row, cols.get("수량"))), + "단위": _text(_cell(row, cols.get("단위"))), + "단가": _money(row, cols, 0), + "금액": _money(row, cols, 1), + "비고": note, + "참조": {"종류": ref.group(1), "번": int(ref.group(2))} if ref else None, + } + line["갈래"] = kind_of(line) + raw[-1].append(line) + if not hopyos: + raise ExcelError(f"{SHEET} 시트에 호표(제N호표)가 없음") + wrong: list[str] = [] + totals = _summary(sheets.get(SUMMARY)) + for hopyo, lines in zip(hopyos, raw, strict=True): + wrong += _check_hopyo(hopyo, lines) + total = hopyo["금액"].get("계") + if hopyo["번"] in totals and total is not None and totals[hopyo["번"]] != total: + wrong.append(f"제{hopyo['번']}호표 · 총괄표 합계 {totals[hopyo['번']]} ≠ {total}") + out = [ + {**_json(hopyo), "줄": [_json(line) for line in lines]} + for hopyo, lines in zip(hopyos, raw, strict=True) + ] + return {"시트": SHEET, "호표": out, "검산": {"어긋남": wrong, "총괄표": len(totals)}} + + +def _json(value: Any) -> Any: + if isinstance(value, Decimal): + return _out(value) + if isinstance(value, dict): + return {k: _json(v) for k, v in value.items()} + return value diff --git a/M02_MasterTemplete/M02_Hopyo_Store.py b/M02_MasterTemplete/M02_Hopyo_Store.py new file mode 100644 index 000000000..c632a409d --- /dev/null +++ b/M02_MasterTemplete/M02_Hopyo_Store.py @@ -0,0 +1,116 @@ +"""M02 일위대가 실무자료 저장(PLAN 324-1) — 층 폴더 `실무자료/<키>.json` · +맞춤 기억 `실무자료/맞춤.json`. + +파일 = 읽은 결과만(원본 엑셀 안 둠) `{키, 파일, 올린때, 올린이, 시트, 호표, 검산}` · 키 `HP-######` +(층마다 다음 번호) · 맞춤 기억 = `{줄: {줄 열쇠: {로직?|자재?|빼기?, 입력값, 환산, 확정}}}`. +층 종류(`layers.KINDS`) 밖 폴더 — 양식 목록 · 복사 · 프로젝트 씨앗에 안 섞임 · 권한은 길 몫. +""" + +from __future__ import annotations + +import json +import re +import threading +from datetime import datetime +from pathlib import Path +from typing import Any + +from common_util.common_util_json import atomic_write_json +from M02_MasterTemplete import M02_MasterTemplete_Store as store + +DIRNAME = "실무자료" +MEMORY = "맞춤" +_KEY = re.compile(r"^HP-(\d{6})$") +# ponytail: 프로세스 안 잠금 — 서버를 여럿 띄우면 파일 잠금으로 +_LOCK = threading.Lock() + + +def _dir(folder: str | Path) -> Path: + return Path(folder) / DIRNAME + + +def _path(folder: str | Path, key: str) -> Path: + if not _KEY.match(str(key)): + raise store.StoreError(404, f"없는 실무자료 「{key}」") + return _dir(folder) / f"{key}.json" + + +def _load(path: Path) -> Any: + try: + return json.loads(path.read_bytes()) + except (OSError, ValueError): + return None + + +def brief(doc: dict[str, Any]) -> dict[str, Any]: + """목록 한 줄 — `{키, 파일, 올린때, 올린이, 시트, 호표수, 어긋남}`.""" + keep = ("키", "파일", "올린때", "올린이", "시트") + return { + **{k: doc.get(k) for k in keep}, + "호표수": len(doc.get("호표") or []), + "어긋남": len((doc.get("검산") or {}).get("어긋남") or []), + } + + +def rows(folder: str | Path) -> list[dict[str, Any]]: + """층 실무자료 목록(키 차례).""" + out = [] + for path in sorted(_dir(folder).glob("HP-*.json")): + doc = _load(path) + if isinstance(doc, dict) and _KEY.match(path.stem): + out.append(brief(doc)) + return out + + +def read(folder: str | Path, key: str) -> dict[str, Any]: + doc = _load(_path(folder, key)) + if not isinstance(doc, dict): + raise store.StoreError(404, f"없는 실무자료 「{key}」") + return doc + + +def save(folder: str | Path, name: str, parsed: dict[str, Any], who: Any) -> dict[str, Any]: + """읽은 결과 → 새 키로 씀 → 목록 한 줄.""" + with _LOCK: + have = [ + int(m.group(1)) for p in _dir(folder).glob("HP-*.json") if (m := _KEY.match(p.stem)) + ] + key = f"HP-{max(have, default=0) + 1:06d}" + doc = { + "키": key, + "파일": name, + "올린때": datetime.now().isoformat(timespec="seconds"), + "올린이": who, + **parsed, + } + atomic_write_json(_path(folder, key), doc) + return brief(doc) + + +def delete(folder: str | Path, key: str) -> None: + path = _path(folder, key) + with _LOCK: + if not path.is_file(): + raise store.StoreError(404, f"없는 실무자료 「{key}」") + path.unlink() + + +# ── 맞춤 기억 ───────────────────────────────────────────────────────── +def memory(folder: str | Path) -> dict[str, dict[str, Any]]: + """층 맞춤 기억 `{줄 열쇠: 확정}` — 없으면 빈 것.""" + doc = _load(_dir(folder) / f"{MEMORY}.json") + got = doc.get("줄") if isinstance(doc, dict) else None + return got if isinstance(got, dict) else {} + + +def remember(folder: str | Path, entries: dict[str, dict[str, Any] | None]) -> dict[str, Any]: + """기억 더함 · 바꿈(값 None = 지움) → 새 기억 전부.""" + with _LOCK: + got = memory(folder) + for key, value in entries.items(): + if value is None: + got.pop(key, None) + else: + got[key] = value + atomic_write_json(_dir(folder) / f"{MEMORY}.json", {"줄": got}) + return got diff --git a/M02_MasterTemplete/M02_Labor_Router.py b/M02_MasterTemplete/M02_Labor_Router.py index 5ddb1792d..a6007ed72 100644 --- a/M02_MasterTemplete/M02_Labor_Router.py +++ b/M02_MasterTemplete/M02_Labor_Router.py @@ -3,19 +3,23 @@ 문서 읽기 · 쓰기는 양식 길(`GET/PUT /api/m02/layers/{층}/templates/labor/{열}`) 그대로 · 여기는 저장하지 않는 셈만 + 서고(단가승격 · 사전조합 · 50-3) 새로 · 복사(키는 서버가 줌 · 고침 · 지우기는 양식 길 그대로) + 개인 자재 목록 · 추가(148-2 · 개인 층만 · 품번은 서버) + -사전조합 나무 자유 폴더(158-1 · `M02_Precombo_Folders` · 읽기 = 층 읽기 · 고침 = 층 쓰기). +사전조합 나무 자유 폴더(158-1 · `M02_Precombo_Folders` · 읽기 = 층 읽기 · 고침 = 층 쓰기) + +일위대가 실무자료(324 · `M02_Hopyo_*` · 올리기 · 지우기 = 층 쓰기 · 목록 · 읽기 = 층 읽기). 층 권한 = `Router_Items._root`(`M02_Layer_Access` 한 벌). ⚠ 등록은 `main.py`(로그인만). """ from __future__ import annotations +from pathlib import Path from typing import Any, Literal -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, File, HTTPException, UploadFile from pydantic import BaseModel from common_util.common_util_auth import verify_session +from M02_MasterTemplete import M02_Hopyo_Excel as hopyo_excel +from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store from M02_MasterTemplete import M02_Labor_Doc as labor from M02_MasterTemplete import M02_Labor_Ref as refs from M02_MasterTemplete import M02_Layer_Access as access @@ -159,3 +163,61 @@ def new_material( raise HTTPException(status_code=400, detail="신규 자재는 개인 층에서만") folder = _root(session, layer, owner, write=True) return _call(folder, own.add, folder, body) + + +# ── 일위대가 실무자료(324) ───────────────────────────────────────────── +HOPYO_MAX_BYTES = 20 * 1024 * 1024 + + +def _upload(name: str, data: bytes, who: dict[str, Any]) -> dict[str, Any]: + try: + parsed = hopyo_excel.read_book(name, data) + except hopyo_excel.ExcelError as error: + raise store.StoreError(400, str(error)) from error + return hopyo_store.save(store.root(), name, parsed, who) + + +@router.get("/layers/{layer}/hopyo") +def hopyo_list( + layer: Layer, owner: str | None = None, session: Session = Depends(verify_session) +) -> dict: + """실무자료 목록 `{줄: [{키, 파일, 올린때, 올린이, 시트, 호표수, 어긋남}]}`.""" + folder = _root(session, layer, owner) + return {"줄": hopyo_store.rows(folder)} + + +@router.post("/layers/{layer}/hopyo") +def hopyo_upload( + layer: Layer, + file: UploadFile = File(...), + owner: str | None = None, + session: Session = Depends(verify_session), +) -> dict: + """엑셀(xls · xlsx) 올리기 — 「일위대가」 시트를 읽어 결과만 저장 · 20 MB · + 엉터리 · 시트 없음 400.""" + folder = _root(session, layer, owner, write=True) + data = file.file.read(HOPYO_MAX_BYTES + 1) + if len(data) > HOPYO_MAX_BYTES: + raise HTTPException(status_code=400, detail="파일이 너무 큼(20 MB 넘음)") + name = Path(file.filename or "실무자료").name[:120] + who = {"id": session.get("user_id"), "이름": session.get("name")} + return _call(folder, _upload, name, data, who) + + +@router.get("/layers/{layer}/hopyo/{key}") +def hopyo_read( + layer: Layer, key: str, owner: str | None = None, session: Session = Depends(verify_session) +) -> dict: + """실무자료 하나(호표 · 줄 · 검산).""" + folder = _root(session, layer, owner) + return _call(folder, hopyo_store.read, folder, key) + + +@router.delete("/layers/{layer}/hopyo/{key}") +def hopyo_delete( + layer: Layer, key: str, owner: str | None = None, session: Session = Depends(verify_session) +) -> dict: + """실무자료 지우기(만든 사전조합 · 맞춤 기억은 그대로).""" + folder = _root(session, layer, owner, write=True) + _call(folder, hopyo_store.delete, folder, key) + return {"지움": key} diff --git a/requirements.txt b/requirements.txt index 09d89ee34..befd6bc20 100644 --- a/requirements.txt +++ b/requirements.txt @@ -62,5 +62,6 @@ uvicorn==0.24.0 watchfiles==1.2.0 wheel==0.47.0 whitebox==2.3.6 +xlrd==2.0.2 pytest>=8 httpx<0.28 diff --git a/resources/tester/test_m02_hopyo_excel.py b/resources/tester/test_m02_hopyo_excel.py new file mode 100644 index 000000000..6e6b14fdf --- /dev/null +++ b/resources/tester/test_m02_hopyo_excel.py @@ -0,0 +1,209 @@ +"""일위대가 실무자료 읽기 · 저장 · 올리기 길(PLAN 324-1 · `M02_Hopyo_Excel` · `M02_Hopyo_Store`). + +견본 = `resources/knowledge/original/실무문서/2026_실무일위대가 모음/`(북부청 임도 · 사방) · +검산 어긋남 0 · 호표 값 · 시트 없음 400 · xlsx 길 · 역할 표 · 엉터리 · 20 MB · 지우기 · +양식 목록에 안 섞임 · +master_data 원본 그대로(sha256). 층은 임시 폴더. +""" + +import hashlib +import io +from pathlib import Path + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from common_util.common_util_auth import verify_session +from config import config_system +from M02_MasterTemplete import M02_Hopyo_Excel as hx +from M02_MasterTemplete import M02_Hopyo_Store as hs +from M02_MasterTemplete import M02_Labor_Router as labor_router +from M02_MasterTemplete import M02_MasterTemplete_Store as store +from M02_MasterTemplete import M02_Template_Layers as layers +from resources.tester.test_m02_structure import write_sample_table + +ROOT = Path(__file__).resolve().parents[2] +SAMPLES = ROOT / "resources" / "knowledge" / "original" / "실무문서" / "2026_실무일위대가 모음" +NORTH = SAMPLES / "2026년_북부지방산림청_임도사업_설계내역서(공통단가).xls" +SABANG = SAMPLES / "2026년_북부지방산림청_사방사업(계류+댐)_설계내역서.xls" +NO_SHEET = SAMPLES / "2016_상반_산림청_사방사업표준품셈(안).xls" +URL = "/api/m02/layers/{}/hopyo" +USER = {"user_id": 42, "company_id": 7, "role": "USER", "is_master": False, "name": "갑"} + + +def master_hash() -> dict[str, str]: + """master_data 원본(*.json · 일위대가조합.json) sha256 — 올리기 · 맞춤 · 만들기 뒤 그대로.""" + folder = ROOT / "resources" / "master_data" + return {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in folder.glob("*.json")} + + +@pytest.fixture(scope="module") +def north() -> dict: + return hx.read_book(NORTH.name, NORTH.read_bytes()) + + +def _one(book: dict, no: int) -> dict: + return next(h for h in book["호표"] if h["번"] == no) + + +# ── 읽기 ────────────────────────────────────────────────────────────── +def test_북부청_68_호표_검산_어긋남_0(north): + assert [h["번"] for h in north["호표"]] == list(range(1, 69)) + assert north["검산"] == {"어긋남": [], "총괄표": 67} # 제61호표 = 값 없음 + lines = [line for h in north["호표"] for line in h["줄"]] + assert len(lines) == 415 + kinds = {k: sum(1 for x in lines if x["갈래"] == k) for k in ("단가산출", "%", "호표", "기계")} + assert kinds == {"단가산출": 268, "%": 1, "호표": 1, "기계": 3} + + +def test_사방_27_호표_검산_어긋남_0(): + got = hx.read_book(SABANG.name, SABANG.read_bytes()) + assert len(got["호표"]) == 27 and got["검산"]["어긋남"] == [] + + +def test_호표_값과_줄(north): + one = _one(north, 1) + assert (one["이름"], one["규격"], one["단위"]) == ( + "배수관날개벽(유입구)", + "D=1,000mm,일반형", + "개소", + ) + assert one["금액"] == {"계": 1040052, "노무비": 777224, "재료비": 239284, "경비": 23544} + assert one["줄"][0]["갈래"] == "별도계상" and one["줄"][0]["규격"] == "별도계상" + three = _one(north, 3) + assert [(x["이름"], x["수량"], x["단위"], x["갈래"]) for x in three["줄"]] == [ + ("임도표지석(받침,글새김포함)", 1, "㎥", "자재"), + ("구조물터파기(인력100%)", 0.14, "㎥", "단가산출"), + ("구조물되메우기(인력100%)", 0.05, "㎥", "단가산출"), + ("잔토처리(인력100%)", 0.09, "㎥", "단가산출"), + ] + assert three["줄"][1]["참조"] == {"종류": "단산", "번": 93} + assert three["금액"]["계"] == 1415169 and three["금액"]["경비"] is None + assert _one(north, 45)["줄"][0]["참조"] == {"종류": "일위", "번": 46} + pct = [x for x in _one(north, 46)["줄"] if x["갈래"] == "%"] + assert [(x["규격"], x["수량"], x["단가"]["계"]) for x in pct] == [("재료비의", 3, 42390)] + empty = _one(north, 61) + assert empty["줄"] == [] and empty["금액"]["계"] is None + + +def test_일위대가_시트_없음(): + with pytest.raises(hx.ExcelError, match="일위대가 시트 없음"): + hx.read_book(NO_SHEET.name, NO_SHEET.read_bytes()) + with pytest.raises(hx.ExcelError, match="읽지 못함"): + hx.read_book("x.xls", b"not an excel file") + + +def _xlsx() -> bytes: + import openpyxl + + book = openpyxl.Workbook() + ws = book.active + ws.title = "일위대가" + ws.append([None, "일위대가"]) + ws.append([]) + ws.append([None, "호 표", "품 명", "규 격", "수 량", "단위", "합 계", None, "노무비", None]) + ws.append([None, None, None, None, None, None, "단 가", "금 액", "단 가", "금 액"]) + ws.append([None, "제 1 호표", "표지석　", "A×B", None, "개소", None, 1000, None, 1000]) + ws.append([None, None, "터파기", "토사", 0.5, "㎥", 2000, 1000, 2000, 1000, "비고"]) + out = io.BytesIO() + book.save(out) + return out.getvalue() + + +def test_xlsx_길(): + got = hx.read_book("a.xlsx", _xlsx()) + one = got["호표"][0] + assert (one["번"], one["이름"], one["금액"]["계"]) == (1, "표지석", 1000) + assert one["줄"][0]["수량"] == 0.5 and got["검산"]["어긋남"] == [] + + +# ── 저장 · 길 ────────────────────────────────────────────────────────── +@pytest.fixture() +def client(tmp_path, monkeypatch): + system = tmp_path / "system" + system.mkdir() + write_sample_table(system) + monkeypatch.setattr(store, "FOLDER", system) + monkeypatch.setattr(layers, "SYSTEM_ROOT", system) + (tmp_path / "storage").mkdir() + monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(tmp_path / "storage")) + app = FastAPI() + app.include_router(labor_router.router) + who = dict(USER) + app.dependency_overrides[verify_session] = lambda: who + got = TestClient(app) + got.who, got.system = who, system + return got + + +def _post(client, layer="personal", data=None, name=NORTH.name, **params): + files = {"file": (name, data if data is not None else NORTH.read_bytes())} + return client.post(URL.format(layer), files=files, params=params) + + +def test_올리기_목록_읽기_지우기(client): + before = master_hash() + res = _post(client) + assert res.status_code == 200, res.text + row = res.json() + assert (row["키"], row["파일"], row["호표수"], row["어긋남"]) == ( + "HP-000001", + NORTH.name, + 68, + 0, + ) + assert row["올린이"] == {"id": 42, "이름": "갑"} + assert client.get(URL.format("personal")).json()["줄"] == [row] + doc = client.get(URL.format("personal") + "/HP-000001").json() + assert len(doc["호표"]) == 68 and doc["시트"] == "일위대가" + folder = layers.personal_dir(7, 42) + assert (folder / hs.DIRNAME / "HP-000001.json").is_file() + assert all(r["종류"] in layers.KINDS for r in layers.list_templates(folder)) # 양식에 안 섞임 + assert not any("실무자료" in str(p) for _, p in layers.layer_files("precombo")) + assert _post(client).json()["키"] == "HP-000002" + assert client.delete(URL.format("personal") + "/HP-000001").status_code == 200 + assert [r["키"] for r in client.get(URL.format("personal")).json()["줄"]] == ["HP-000002"] + assert client.get(URL.format("personal") + "/HP-000001").status_code == 404 + assert client.get(URL.format("personal") + "/../x").status_code == 404 + assert master_hash() == before + + +def test_엉터리_시트없음_큰_파일_400(client, monkeypatch): + bad = _post(client, data=b"garbage", name="a.xls") + assert bad.status_code == 400 and "읽지 못함" in bad.json()["detail"] + none = _post(client, data=NO_SHEET.read_bytes(), name=NO_SHEET.name) + assert none.status_code == 400 and none.json()["detail"] == "일위대가 시트 없음" + monkeypatch.setattr(labor_router, "HOPYO_MAX_BYTES", 1000) + big = _post(client) + assert big.status_code == 400 and "20 MB" in big.json()["detail"] + assert client.get(URL.format("personal")).json()["줄"] == [] + + +@pytest.mark.parametrize( + ("role", "layer", "owner", "want"), + [ + ({"role": "SYSTEM_ADMIN"}, "system", None, 200), + ({"role": "USER"}, "system", None, 403), + ({"role": "ADMIN"}, "company", None, 200), + ({"role": "USER"}, "company", None, 403), + ({"role": "USER"}, "personal", None, 200), + ({"role": "USER"}, "personal", "7/43", 403), + ({"role": "USER"}, "personal", "8/50", 403), + ], +) +def test_역할_표(client, role, layer, owner, want): + client.who.update(role) + params = {"owner": owner} if owner else {} + assert _post(client, layer, **params).status_code == want + if want == 200: + assert client.get(URL.format(layer)).json()["줄"][0]["호표수"] == 68 + + +def test_마스터_층은_마스터_폴더(client): + client.who.update(role="SYSTEM_ADMIN") + assert _post(client, "system").status_code == 200 + assert (client.system / hs.DIRNAME / "HP-000001.json").is_file() + client.who.update(role="USER") + assert client.get(URL.format("system")).json()["줄"][0]["키"] == "HP-000001" # 읽기 = 누구나 + assert client.delete(URL.format("system") + "/HP-000001").status_code == 403