feat(b08): STmate 엑셀 뽑기 창구 두 걸음 — [읽기]는 호표 수·구성 줄 수·못 읽은 사유만(안 씀) · [넣기]는 서버가 파일을 다시 읽어 고른 호표를 개인 단에 고정형으로(회사 없음 403 · 못 읽은 호표 404 · 모르는 종류 400) · main.py 에 창구 한 줄

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016VBGFXB9AbJBwXP19z75Qq
This commit is contained in:
2026-09-14 15:59:11 +09:00
co-authored by Claude Opus 5
parent d6fd3c1796
commit 38a5e785f2
5 changed files with 238 additions and 5 deletions
@@ -14,7 +14,7 @@ from __future__ import annotations
import re
from pathlib import Path
from typing import Any
from typing import Any, BinaryIO
SHEET = "일위대가표"
HEADER_ROW = 3
@@ -37,7 +37,7 @@ def _text(value: Any) -> str:
return "" if value is None else str(value).strip()
def read_recipes(path: str | Path) -> dict[str, Any]:
def read_recipes(path: str | Path | BinaryIO) -> dict[str, Any]:
"""`{"project": 공사명, "hopyo": [호표…], "problems": [사유…]}` — 못 읽은 호표는 목록에 없고 사유만."""
import openpyxl
@@ -127,6 +127,19 @@ def _write(folder: Path, item: dict[str, Any]) -> None:
(folder / f"{item['code']}.json").write_text(text, encoding="utf-8")
def _personal_code(folder: Path, type_id: Any) -> str:
"""개인 단 코드 — 같은 종류가 있으면 그 코드(덮어씀 · 판정 Ⓐ 종류당 하나), 없으면 새로."""
same = [item for item in _items(folder) if item.get("type_id") == type_id]
return str(same[0]["code"]) if same else f"AX-ST-{secrets.token_hex(4)}"
def save_item_personal(folder: Path, item: dict[str, Any]) -> str:
"""이미 만든 항목(STmate 에서 뽑은 고정형 등)을 개인 단에 씀. 코드."""
code = _personal_code(folder, item.get("type_id"))
_write(folder, {**item, "code": code, "library_tier": "personal"})
return code
def save_personal(
folder: Path,
template: dict[str, Any],
@@ -141,9 +154,7 @@ def save_personal(
"""
from B08_Quantity.B08_Quantity_Engine_StructureTemplate import overridden_rows
type_id = template.get("type_id")
same = [item for item in _items(folder) if item.get("type_id") == type_id]
code = str(same[0]["code"]) if same else f"AX-ST-{secrets.token_hex(4)}"
code = _personal_code(folder, template.get("type_id"))
# 고친 식·반올림이 곧 이 항목의 값 — 「사용자」 표시와 되돌릴 자리는 떼어 냄.
dropped = {"default_formula", "default_rounding"}
rows = [
@@ -0,0 +1,112 @@
"""B08 라이브러리 — STmate 출력 엑셀에서 **고정형 항목 뽑아 넣기** (PLAN 4장 · 2026-09-14 브레인 판정).
두 걸음:
① [읽기] 파일을 올리면 읽은 호표 수·구성 줄 수·못 읽은 사유를 돌려줌 — **아무것도 안 씀**.
사용자가 수를 보고 넣을지 정함(판정 「읽은 뒤 수를 보이고 묻는 걸음」).
② [넣기] 같은 파일 + 고른 호표 차례 + 우리 구조물 종류 → 서버가 **파일을 다시 읽어** 개인 단에 씀.
⚠ 브라우저가 보낸 줄을 받아 적지 않음(CLAUDE.md 5장) · 개인 단만(판정 Ⓗ) · 종류당 하나라 같은 종류 내 것은
덮어씀(판정 Ⓐ) · 종류는 사용자가 고름 — 이름으로 자동으로 안 붙임(판정 Ⓒ).
"""
import asyncio
import io
from typing import Any
from uuid import UUID
from fastapi import APIRouter, Depends, File, Form, UploadFile
from fastapi.responses import JSONResponse
from common_util.common_util_auth import verify_session
router = APIRouter(prefix="/api/projects", tags=["B08 Quantity"])
BASE = "/{project_id}/quantity/structure-sheets/library/stmate"
#: 출력 엑셀은 수백 KB — 넉넉히 두되 끝없이 받지 않음.
MAX_BYTES = 20 * 1024 * 1024
def _error(status: int, message: str, **extra: Any) -> JSONResponse:
return JSONResponse(
status_code=status, content={"status": "error", "message": message, **extra}
)
async def _read(file: UploadFile) -> dict[str, Any] | JSONResponse:
from B08_Quantity.B08_Quantity_Engine_StmateRecipe import read_recipes
data = await file.read(MAX_BYTES + 1)
if len(data) > MAX_BYTES:
return _error(413, "파일이 너무 큼(20MB 넘음)")
return await asyncio.to_thread(read_recipes, io.BytesIO(data))
@router.post(BASE + "/read")
async def read_stmate_recipes(
project_id: UUID,
file: UploadFile = File(...),
session: dict[str, Any] = Depends(verify_session),
) -> JSONResponse:
"""① 읽기만 — 호표 목록(차례·명칭·규격·단위·구성 줄 수)과 못 읽은 사유."""
read = await _read(file)
if isinstance(read, JSONResponse):
return read
hopyo = [
{
"no": h["no"],
"name": h["name"],
"spec": h["spec"],
"unit": h["unit"],
"rows": len(h["rows"]),
"percent_rows": sum(1 for row in h["rows"] if row["unit"] == "%"),
"contract_rows": h["contract_rows"],
}
for h in read["hopyo"]
]
return JSONResponse(
content={
"status": "success",
"file_name": file.filename or "",
"project": read["project"],
"hopyo": hopyo,
"counts": {"hopyo": len(hopyo), "rows": sum(h["rows"] for h in hopyo)},
"problems": read["problems"],
}
)
@router.post(BASE + "/save")
async def save_stmate_recipe(
project_id: UUID,
file: UploadFile = File(...),
hopyo_no: int = Form(...),
type_id: str = Form(..., min_length=1, max_length=100),
session: dict[str, Any] = Depends(verify_session),
) -> JSONResponse:
"""② 넣기 — 서버가 다시 읽은 호표 하나를 고정형 항목으로 **로그인한 사람 개인 단**에 씀."""
from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
from B08_Quantity.B08_Quantity_Engine_StmateRecipe import recipe_item
from B08_Quantity.B08_Quantity_Engine_StructureLibrary import save_item_personal, tier_dirs
folder = tier_dirs(session.get("company_id"), session.get("user_id")).get("personal")
if folder is None:
return _error(403, "개인 라이브러리는 회사에 속한 사용자만 씁니다.")
if type_id not in structure_type_map():
return _error(400, f"모르는 구조물 종류: {type_id}")
read = await _read(file)
if isinstance(read, JSONResponse):
return read
hopyo = next((h for h in read["hopyo"] if h["no"] == hopyo_no), None)
if hopyo is None:
return _error(404, f"{hopyo_no}호표를 읽지 못함", problems=read["problems"])
item = recipe_item(
hopyo, type_id=type_id, file_name=file.filename or "", project=read["project"]
)
code = await asyncio.to_thread(save_item_personal, folder, item)
return JSONResponse(
content={
"status": "success",
"code": code,
"name": item["name"],
"rows": len(item["rows"]),
"note": item["note"],
}
)
+2
View File
@@ -64,6 +64,7 @@ from B08_Quantity.B08_Quantity_Router import router as b08_quantity_router
from B08_Quantity.B08_Quantity_Router_Earthwork import router as b08_earthwork_router
from B08_Quantity.B08_Quantity_Router_Material import router as b08_material_router
from B08_Quantity.B08_Quantity_Router_StructureSheet import router as b08_structure_sheet_router
from B08_Quantity.B08_Quantity_Router_StmateLibrary import router as b08_stmate_library_router
from B09_Estimation.B09_Estimation_Router import router as b09_estimation_router
from B09_Estimation.B09_Estimation_Router_Contract import router as b09_contract_router
from B09_Estimation.B09_Estimation_Router_Execution import router as b09_execution_router
@@ -644,6 +645,7 @@ app.include_router(b08_quantity_router, dependencies=protected_with_company)
app.include_router(b08_earthwork_router, dependencies=protected_with_company)
app.include_router(b08_material_router, dependencies=protected_with_company)
app.include_router(b08_structure_sheet_router, dependencies=protected_with_company)
app.include_router(b08_stmate_library_router, dependencies=protected_with_company)
app.include_router(b09_estimation_router, dependencies=protected_with_company)
app.include_router(b09_cost_sheet_router, dependencies=protected_with_company)
app.include_router(b09_contract_router, dependencies=protected_with_company)
@@ -0,0 +1,108 @@
"""라이브러리 — STmate 출력 엑셀에서 고정형 항목 뽑아 넣기 창구(PLAN 4장 · 2026-09-14 브레인 판정).
두 걸음: [읽기]는 호표 수·구성 줄 수·못 읽은 사유만 돌려주고 아무것도 안 씀 → [넣기]는 같은 파일을
서버가 **다시 읽어** 고른 호표를 개인 단에 씀(브라우저가 보낸 줄을 받아 적지 않음 · 개인 단만 · 판정 Ⓗ).
"""
from __future__ import annotations
import io
import json
import sys
from pathlib import Path
import openpyxl
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
import B08_Quantity.B08_Quantity_Engine_StructureLibrary as library_module # noqa: E402
import B08_Quantity.B08_Quantity_Router_StmateLibrary as router_module # noqa: E402
from common_util.common_util_auth import verify_session # noqa: E402
PROJECT_ID = "33333333-3333-3333-3333-333333333333"
BASE = f"/api/projects/{PROJECT_ID}/quantity/structure-sheets/library/stmate"
def _xlsx() -> bytes:
book = openpyxl.Workbook()
ws = book.active
ws.title = "일위대가표"
for row in [
["일 위 대 가 표"],
["공사명 : 시험 공사"],
["명 칭", "규 격", "수 량", "단위"],
[None],
[" 제 1 호표"],
["돌기슭막이(메쌓기)", "H=2.0", None, "m"],
["메쌓기", "L3=55cm이하", 2.09, "m2"],
["고임돌채집", "기계", 0.31, "m3"],
["합 계"],
[" 제 2 호표"],
["규준틀설치", "종단", None, "개소"],
["각재", "외송", "약간", "M3"],
["합 계"],
]:
ws.append(row)
buffer = io.BytesIO()
book.save(buffer)
return buffer.getvalue()
@pytest.fixture()
def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> TestClient:
monkeypatch.setattr(library_module, "STORAGE_BASE_DIR", str(tmp_path / "storage"))
app = FastAPI()
app.include_router(router_module.router)
app.dependency_overrides[verify_session] = lambda: {"company_id": 7, "user_id": 42}
return TestClient(app)
def _files() -> dict:
return {"file": ("견본.xlsx", _xlsx(), "application/octet-stream")}
def test_읽기는_수와_사유만_주고_안_쓴다(client: TestClient, tmp_path: Path) -> None:
body = client.post(f"{BASE}/read", files=_files()).json()
assert body["counts"] == {"hopyo": 1, "rows": 2}
assert body["hopyo"][0]["name"] == "돌기슭막이(메쌓기)" and body["hopyo"][0]["rows"] == 2
assert any("C12" in p for p in body["problems"]) # 수량이 수가 아닌 2호표는 사유로
assert not (tmp_path / "storage").exists()
def test_넣기는_서버가_다시_읽어_개인_단에_고정형으로(client: TestClient, tmp_path: Path) -> None:
response = client.post(
f"{BASE}/save", files=_files(), data={"hopyo_no": "1", "type_id": "masonry_dry"}
)
assert response.status_code == 200, response.text
code = response.json()["code"]
saved = json.loads(
(tmp_path / "storage" / "7" / "42" / "library" / f"{code}.json").read_text(encoding="utf-8")
)
assert saved["library_tier"] == "personal" and saved["origin"]["project"] == "시험 공사"
assert [r["amount"] for r in saved["rows"]] == ["2.09", "0.31"]
dirs = library_module.tier_dirs(7, 42)
assert [i["kind"] for i in library_module.list_items(dirs, "masonry_dry")] == ["fixed"]
def test_회사_없으면_403_없는_호표는_404_모르는_종류는_400(client: TestClient) -> None:
bad = client.post(
f"{BASE}/save", files=_files(), data={"hopyo_no": "2", "type_id": "masonry_dry"}
)
assert bad.status_code == 404 # 못 읽은 호표는 넣을 수 없음
unknown = client.post(f"{BASE}/save", files=_files(), data={"hopyo_no": "1", "type_id": "없음"})
assert unknown.status_code == 400
client.app.dependency_overrides[verify_session] = lambda: {"company_id": None, "user_id": 1}
denied = client.post(
f"{BASE}/save", files=_files(), data={"hopyo_no": "1", "type_id": "masonry_dry"}
)
assert denied.status_code == 403
def test_앱에_창구가_걸린다() -> None:
main = (ROOT / "main.py").read_text(encoding="utf-8")
assert "B08_Quantity_Router_StmateLibrary" in main and "b08_stmate_library_router" in main