"""M02 일위대가 계산 · 알약 길(PLAN 39-1a · 설계_정함 1장) — `/api/m02/layers/{층}/labor/{열}/…`. 문서 읽기 · 쓰기는 양식 길(`GET/PUT /api/m02/layers/{층}/templates/labor/{열}`) 그대로 · 여기는 저장하지 않는 셈만 + 서고(단가승격 · 사전조합 · 50-3) 새로 · 복사(키는 서버가 줌 · 고침 · 지우기는 양식 길 그대로) + 개인 자재 목록 · 추가(148-2 · 개인 층만 · 품번은 서버) + 사전조합 나무 자유 폴더(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, File, HTTPException, UploadFile from pydantic import BaseModel from common_util.common_util_auth import verify_session from M02_MasterTemplete import M02_Hopyo_Build as hopyo_build from M02_MasterTemplete import M02_Hopyo_Excel as hopyo_excel from M02_MasterTemplete import M02_Hopyo_Match as hopyo_match 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 from M02_MasterTemplete import M02_MasterTemplete_Store as store from M02_MasterTemplete import M02_Material_Own as own from M02_MasterTemplete import M02_Precombo_Folders as pre_folders from M02_MasterTemplete.M02_MasterTemplete_Router_Items import ( Layer, Session, TreeBody, _call, _root, ) router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Labor"]) class CalcBody(BaseModel): """`doc` = 저장 전 화면 문서(없으면 저장본) · `수량` = {알약 키: 값}(없으면 저장된 면적산출).""" doc: dict[str, Any] | None = None 수량: dict[str, Any] | None = None def _doc(column: str, sent: dict[str, Any] | None) -> dict[str, Any]: got = sent if sent is not None else store.read("labor", column)["문서"] return labor.clean({**got, "열": column}) def _calc(column: str, body: CalcBody) -> dict[str, Any]: doc = _doc(column, body.doc) return {**labor.calc(doc, body.수량), "알약": labor.pills(doc)} def _pills(column: str) -> dict[str, Any]: return {"열": column, "알약": labor.pills(_doc(column, None))} @router.post("/layers/{layer}/labor/{column}/calc") def calc_labor( layer: Layer, column: str, body: CalcBody, owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: """줄마다 기준 · 반영 · 합 + 알약(저장 안 함 · 화면은 최신 응답만).""" return _call(_root(session, layer, owner), _calc, column, body) @router.get("/layers/{layer}/labor/{column}/pills") def labor_pills( layer: Layer, column: str, owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: """면적산출 알약 — 로직 물량 알약 「이름|단위」 + 자재 알약 「자재|MT키」(저장본).""" return _call(_root(session, layer, owner), _pills, column) class LibraryBody(BaseModel): """새로 = `문서` · 복사 = `본뜬` {층, 주인, 키}(원본 읽기 권한 · 못 읽는 참조는 같이 복사).""" 문서: dict[str, Any] | None = None 본뜬: dict[str, Any] | None = None def _copy(kind: str, doc: dict[str, Any], source: refs.Here, ref: dict) -> dict[str, Any]: return refs.create(kind, refs.carry({**doc, "본뜬": ref}, source)) @router.post("/layers/{layer}/library/{kind}") def new_library( layer: Layer, kind: Literal["precombo"], body: LibraryBody, owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: """서고 새로 · 복사 — 키는 서버가 그 층 다음 번호(SP · SC) · 답 `이름` = 새 키.""" folder = _root(session, layer, owner, write=True) if body.본뜬 is None: return _call(folder, refs.create, kind, body.문서 or {}) src = body.본뜬 if src.get("층") not in ("system", "company", "personal"): raise HTTPException(status_code=400, detail="본뜬 층이 틀림") got = access.grant(session, src["층"], src.get("주인")) # 원본 읽기 권한 ref = {"층": got.layer, "주인": got.owner, "키": str(src.get("키") or "")} if refs.kind_of(ref["키"]) != kind: raise HTTPException(status_code=400, detail="본뜬 종류가 틀림") doc = refs.read_ref(ref) if refs._KEY.match(ref["키"]) else None if doc is None: raise HTTPException(status_code=404, detail=f"없는 {refs.TITLE[kind]} 「{ref['키']}」") return _call(folder, _copy, kind, doc, (got.layer, got.owner), ref) @router.get("/layers/{layer}/precombo-tree") def precombo_tree( layer: Layer, owner: str | None = None, session: Session = Depends(verify_session) ) -> dict: """사전조합 나무 폴더 — `{판, 폴더: [경로 …], 자리: {키: 경로}}`(자리 없는 문서 = 뿌리).""" return _call(_root(session, layer, owner), pre_folders.view) @router.post("/layers/{layer}/precombo-tree") def edit_precombo_tree( layer: Layer, body: TreeBody, owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: """새 폴더 · 이름 바꾸기 · 옮기기(폴더 · 문서 `키`) · 지우기(빈 폴더만) → 새 나무 · 판 다르면 409.""" folder = _root(session, layer, owner, write=True) return _call(folder, pre_folders.edit, body.model_dump()) @router.get("/layers/{layer}/materials") def own_materials( layer: Layer, owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: """개인 자재 목록(148-2) — 개인 층만 · 회사 · 마스터 = 빈 목록.""" if layer != "personal": return {"줄": []} return {"줄": own.rows(_root(session, layer, owner))} @router.post("/layers/{layer}/materials") def new_material( layer: Layer, body: dict[str, Any], owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: """신규 자재(148-2) — `{이름, 규격, 단위, 값}` → 새 줄(품번 `PM` + 6자리 서버 일련) · 개인 층만.""" if layer != "personal": 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.get("/layers/{layer}/hopyo-picks") def hopyo_picks( layer: Layer, owner: str | None = None, session: Session = Depends(verify_session) ) -> dict: """[로직 추가] 「실무 호표」 줄 — 이 층 · 위 층 파일의 호표마다 `{층, 주인, 키, 파일, 번, 이름, 규격, 단위, 맞춤}`(맞춤 n/m = 이 층 기억).""" folder = _root(session, layer, owner) return {"줄": _call(folder, hopyo_match.picks, 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_match.view, folder, key) @router.get("/layers/{layer}/hopyo/{key}/{no}") def hopyo_one( layer: Layer, key: str, no: int, owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: """호표 하나 — 줄마다 맞춤(상태 · 대상 · 입력값 · 환산 · 들어감 · 후보 5).""" folder = _root(session, layer, owner) return _call(folder, hopyo_match.view, folder, key, no) class GuessBody(BaseModel): """`차례` = 호표 안 줄(0부터) · `대상` = {로직} | {자재}.""" 차례: int 대상: dict[str, Any] @router.post("/layers/{layer}/hopyo/{key}/{no}/guess") def hopyo_guess( layer: Layer, key: str, no: int, body: GuessBody, owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: """고른 대상 미리 채움(저장 안 함) — 규격 → 입력값 · 환산.""" folder = _root(session, layer, owner) return _call(folder, hopyo_match.guess, folder, key, no, body.차례, body.대상) class MemoryBody(BaseModel): """`줄` = {줄 열쇠: {로직|자재, 입력값, 환산} | {빼기: true} | null(지움)}.""" 줄: dict[str, Any] @router.post("/layers/{layer}/hopyo-memory") def hopyo_memory( layer: Layer, body: MemoryBody, owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: """맞춤 확정 · 빼기 → 층 기억(층 쓰기) — 다음부터 같은 줄은 자동.""" folder = _root(session, layer, owner, write=True) return _call(folder, hopyo_match.confirm, folder, body.줄) class BuildBody(BaseModel): """실무 호표 → 이 층 사전조합 — 파일 층 `층` · `주인` · `키` + 호표 `번` · `새로` = 있던 것을 덮음.""" 층: Layer 주인: str | None = None 키: str 번: int 새로: bool = False @router.post("/layers/{layer}/hopyo-build") def hopyo_make( layer: Layer, body: BuildBody, owner: str | None = None, session: Session = Depends(verify_session), ) -> dict: """실무 호표 → 우리 일위대가 사전조합(이 층 쓰기 · 파일 층 읽기) — `{참조, 이름, 다시씀, 빠진줄, 비교}`.""" folder = _root(session, layer, owner, write=True) source = _root(session, body.층, body.주인) return _call(folder, hopyo_build.build, folder, source, body.키, body.번, body.새로) @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}