feat(일위대가실무): 324-1 실무자료 엑셀 읽기 · 저장 · 올리기 길 — 「일위대가」 시트 → 호표 · 줄 · 갈래 · 참조 · 검산

- 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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ji4nuyLVHcxrXWHTdb5nRD
This commit is contained in:
2026-10-14 22:03:59 +09:00
co-authored by Claude Opus 5.5
parent ecd4a95ab5
commit 63ba44a717
5 changed files with 659 additions and 2 deletions
+269
View File
@@ -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
+116
View File
@@ -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
+64 -2
View File
@@ -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}