"""프로젝트 구조물 집계 항목 복사본 길 — `/api/projects/{id}/items…` (PLAN 25-1b · 26-1 · 26-2b). 폴더 = 구조물 id(`structure_id` · 관 지점은 `pipe@측점`) — 같은 항목을 고른 구조물도 복사본은 따로. GET …/items 목록(원본 · 가져온날 · 마스터바뀜 · 원본없음) POST …/items {structure, layer, owner, key} 미리 가져오기 → {폴더, 만듦, 가져온날} POST …/items/new {structure, 경로, 이름, 단위} 새 항목(개인 층에도) · 회사 없음 409 GET …/items/{폴더}/{종류} {종류, 판, 내용} PUT …/items/{폴더}/{종류} {판, 내용} — 판이 다르면 409 · 면적 · 재료는 서버 계산값 POST …/items/{폴더}/reapply 재적용(입력값 · 고른자재는 같은 키면 살림 · `덮임` 알림) POST …/items/{폴더}/save-as {to, 경로, 이름} 개인 층에 새 항목으로(단가 묶음은 안 가져감) POST …/items/{폴더}/calc 복사본만으로 계산(세 비목 · 계 · 빠진 입력) 종류 = 도면 · 면적 · 재료 · 자재조합 · 입력연결 · 입력값 · 고른자재 · `refresh` = 옛 이름(재적용). ⚠ 등록은 `main.py`(로그인만) — 프로젝트 권한(같은 회사)은 여기서 `_project` 로 봄(403 · 404). """ from __future__ import annotations import asyncio from typing import Any, Literal from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from common_util import common_util_spreadsheet as spreadsheet from common_util.common_util_auth import verify_session from common_util.common_util_summary_item import parse_summary_item from M02_MasterTemplete import M02_Item_Copy as copies from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layer_routes from M02_MasterTemplete import M02_MasterTemplete_Store as store from M02_MasterTemplete.M02_Item_Source import item_name router = APIRouter(prefix="/api/projects", tags=["M02 Project Items"]) Kind = Literal["도면", "면적", "재료", "자재조합", "입력연결", "입력값", "고른자재"] class WriteBody(BaseModel): 판: str | None = None 내용: Any class PrefetchBody(BaseModel): """`{structure, layer, owner, key}`(이름은 층 SS 줄에서) — 옛 모양 `{summary_item}` 도 받음.""" structure: str | None = None layer: str | None = None owner: str | None = None key: str | None = None name: str | None = None summary_item: dict[str, Any] | None = None class SaveAsBody(BaseModel): to: Literal["personal"] = "personal" 경로: list[str] = [] 이름: str | None = None class NewBody(BaseModel): structure: str 경로: list[str] = [] 이름: str | None = None 단위: str | None = None def _me(session: dict[str, Any]) -> tuple[int, int]: """개인 층 주인(회사 · 사용자) — 회사 없는 사용자는 409.""" if session.get("company_id") is None: raise HTTPException( status_code=409, detail="회사부터 등록 — 개인 층은 회사 연결이 필요합니다." ) return int(session["company_id"]), int(session["user_id"]) async def _root(session: dict[str, Any], project_id: str): return (await layer_routes._project(session, project_id))["root"] async def _run(fn, *args): try: return await asyncio.to_thread(fn, *args) except store.StoreError as error: raise HTTPException(status_code=error.status, detail=error.detail) from error except copies.Stale as error: raise HTTPException( status_code=409, detail={"message": str(error), "판": error.current} ) from error except LookupError as error: raise HTTPException(status_code=404, detail=str(error)) from error except ValueError as error: raise HTTPException(status_code=400, detail=str(error)) from error @router.get("/{project_id}/items") async def list_items( project_id: str, session: dict[str, Any] = Depends(verify_session) ) -> dict[str, Any]: return {"항목": await _run(copies.list_items, await _root(session, project_id))} @router.post("/{project_id}/items") async def prefetch_item( project_id: str, body: PrefetchBody, session: dict[str, Any] = Depends(verify_session) ) -> dict[str, Any]: if not body.structure: raise HTTPException(status_code=400, detail="미리 가져올 구조물(structure)이 필요합니다.") raw = body.summary_item or {"layer": body.layer, "owner": body.owner, "key": body.key} try: # 모양 먼저(이름은 자리만) — 그다음 층 SS 줄에서 이름 item = parse_summary_item({**raw, "name": raw.get("name") or body.name or "-"}) except ValueError as error: raise HTTPException(status_code=400, detail=str(error)) from error if not (raw.get("name") or body.name): try: item["name"] = item_name(item) except LookupError as error: raise HTTPException(status_code=404, detail=str(error)) from error root = await _root(session, project_id) return await _run(copies.prefetch, root, body.structure, item) @router.post("/{project_id}/items/new") async def new_item( project_id: str, body: NewBody, session: dict[str, Any] = Depends(verify_session) ) -> dict[str, Any]: root = await _root(session, project_id) company, user = _me(session) return await _run( copies.new_item, root, body.structure, company, user, body.경로, body.이름, body.단위 ) @router.get("/{project_id}/items/{folder}/{kind}") async def read_item_file( project_id: str, folder: str, kind: Kind, session: dict[str, Any] = Depends(verify_session) ) -> dict[str, Any]: return await _run(copies.read_file, await _root(session, project_id), folder, kind) def _recalc(document: dict) -> dict: try: return spreadsheet.with_server_values(document) except spreadsheet.RecalcError as error: raise HTTPException(status_code=503, detail=str(error)) from error @router.put("/{project_id}/items/{folder}/{kind}") async def write_item_file( project_id: str, folder: str, kind: Kind, body: WriteBody, session: dict[str, Any] = Depends(verify_session), ) -> dict[str, Any]: root = await _root(session, project_id) return await _run(copies.write_file, root, folder, kind, body.판, body.내용, _recalc) @router.post("/{project_id}/items/{folder}/reapply") @router.post("/{project_id}/items/{folder}/refresh") async def reapply_item( project_id: str, folder: str, session: dict[str, Any] = Depends(verify_session) ) -> dict[str, Any]: return await _run(copies.reapply, await _root(session, project_id), folder) @router.post("/{project_id}/items/{folder}/save-as") async def save_item_as( project_id: str, folder: str, body: SaveAsBody, session: dict[str, Any] = Depends(verify_session), ) -> dict[str, Any]: root = await _root(session, project_id) company, user = _me(session) return await _run(copies.save_as, root, folder, company, user, body.경로, body.이름) @router.post("/{project_id}/items/{folder}/calc") async def calc_item( project_id: str, folder: str, session: dict[str, Any] = Depends(verify_session) ) -> dict[str, Any]: return await _run(copies.calc, await _root(session, project_id), folder)