From 40b80ff1b698468786bc37b1b3cb2a0118f24f0a Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sat, 10 Oct 2026 00:09:37 +0900 Subject: [PATCH] =?UTF-8?q?feat(b09):=20=EC=9E=90=EC=9E=AC=20=EA=B2=AC?= =?UTF-8?q?=EC=A0=81=EC=84=9C=20=EC=97=AC=EB=9F=AC=20=ED=8C=8C=EC=9D=BC=20?= =?UTF-8?q?=EC=B2=A8=EB=B6=80=20=EA=B8=B8=20=C2=B7=20=EA=B2=AC=EC=A0=81=20?= =?UTF-8?q?=C2=B7=20=EA=B3=A0=EB=A5=B8=20=EC=B6=9C=EC=B2=98=20=EC=A0=80?= =?UTF-8?q?=EC=9E=A5(166-4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Router_Quotes 새 길(목록 · 올리기 · 내려받기 · 지우기 · v1/quotes/<자재 키>/) · Store quote · quote_page · pick 칸 · 읽을 때 files 목록 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015sDZSjFvoiDjdGpgj3muc3 --- B09_Estimation/B09_Estimation_Router.py | 5 + .../B09_Estimation_Router_Quotes.py | 93 ++++++++++++++++ B09_Estimation/B09_Estimation_Store.py | 47 +++++++- resources/tester/test_166_4_b09_quote.py | 105 ++++++++++++++++++ 4 files changed, 247 insertions(+), 3 deletions(-) create mode 100644 B09_Estimation/B09_Estimation_Router_Quotes.py create mode 100644 resources/tester/test_166_4_b09_quote.py diff --git a/B09_Estimation/B09_Estimation_Router.py b/B09_Estimation/B09_Estimation_Router.py index fe9c66f5b..b973d2329 100644 --- a/B09_Estimation/B09_Estimation_Router.py +++ b/B09_Estimation/B09_Estimation_Router.py @@ -36,6 +36,7 @@ from B09_Estimation.B09_Estimation_Calc_Total import compute_total from B09_Estimation.B09_Estimation_Labor_Source import daega_inputs from B09_Estimation.B09_Estimation_Linked import linked_resource_keys from B09_Estimation.B09_Estimation_Rates import rate_candidates, rate_table +from B09_Estimation.B09_Estimation_Router_Quotes import router as quotes_router from B09_Estimation.B09_Estimation_Summary import compute_summary, hopyo_lines router = APIRouter(prefix="/api/projects", tags=["B09 Estimation"]) @@ -129,3 +130,7 @@ async def refresh_prices(project_id: UUID) -> dict[str, Any]: book = await asyncio.to_thread(prices.refresh_book, keys) await asyncio.to_thread(store.write_book, root, book) return await payload(project_id, root) + + +# 166-4 견적서 파일 길 — 같은 머리 · 같은 권한 +router.include_router(quotes_router) diff --git a/B09_Estimation/B09_Estimation_Router_Quotes.py b/B09_Estimation/B09_Estimation_Router_Quotes.py new file mode 100644 index 000000000..adb60a382 --- /dev/null +++ b/B09_Estimation/B09_Estimation_Router_Quotes.py @@ -0,0 +1,93 @@ +"""B09 자재 견적서 파일 길(PLAN 166-4) — 자재 줄마다 여러 파일 · DB 표 없음(폴더가 목록). + + GET /api/projects/{id}/estimation/quotes?key= 목록 [{name, size, at}] + POST /api/projects/{id}/estimation/quotes?key= 올리기(`files` 여러 개 · 같은 이름 = 교체) + GET /api/projects/{id}/estimation/quotes/file?key=&name= 내려받기 + DELETE /api/projects/{id}/estimation/quotes?key=&name= 지우기 → 목록 + +자재 키 · 파일 이름은 쿼리로(키에 `/` · `|` 가 섞임) · 저장 자리 = `B09_Estimation_Store.quote_dir`. +`B09_Estimation_Router` 가 끼워 넣음(같은 머리 · 같은 권한). +""" + +from __future__ import annotations + +import asyncio +from datetime import datetime +from pathlib import Path +from typing import Any +from uuid import UUID + +from fastapi import APIRouter, File, HTTPException, Query, UploadFile +from fastapi.responses import FileResponse + +from B03_FileInput.B03_FileInput_Engine import save_upload_stream +from B08_Quantity.B08_Quantity_Router_Hydro import _project_root +from B09_Estimation import B09_Estimation_Store as store + +router = APIRouter() +PATH = "/{project_id}/estimation/quotes" + + +def _listing(folder: Path) -> list[dict[str, Any]]: + return [ + { + "name": f.name, + "size": f.stat().st_size, + "at": datetime.fromtimestamp(f.stat().st_mtime).isoformat(timespec="seconds"), + } + for f in store.quote_files(folder) + ] + + +async def _folder(project_id: UUID, key: str) -> Path: + root = await _project_root(project_id) + try: + return store.quote_dir(root, key) + except ValueError as error: + raise HTTPException(status_code=400, detail=str(error)) from error + + +async def _file(project_id: UUID, key: str, name: str) -> Path: + folder = await _folder(project_id, key) + path = folder / name + if store.safe_name(name) != name or not path.is_file(): + raise HTTPException(status_code=404, detail="견적서 파일이 없습니다.") + return path + + +@router.get(PATH) +async def list_quotes(project_id: UUID, key: str = Query(...)) -> list[dict[str, Any]]: + folder = await _folder(project_id, key) + return await asyncio.to_thread(_listing, folder) + + +@router.post(PATH) +async def upload_quotes( + project_id: UUID, key: str = Query(...), files: list[UploadFile] = File(...) +) -> list[dict[str, Any]]: + folder = await _folder(project_id, key) + folder.mkdir(parents=True, exist_ok=True) + for upload in files: + name = store.safe_name(upload.filename or "") + if not name: + raise HTTPException(status_code=400, detail="파일 이름이 없습니다.") + try: + await save_upload_stream(upload, folder / name) + except ValueError as error: + raise HTTPException(status_code=400, detail=str(error)) from error + return await asyncio.to_thread(_listing, folder) + + +@router.get(PATH + "/file") +async def download_quote(project_id: UUID, key: str = Query(...), name: str = Query(...)): + path = await _file(project_id, key, name) + return FileResponse(path, filename=name) + + +@router.delete(PATH) +async def delete_quote( + project_id: UUID, key: str = Query(...), name: str = Query(...) +) -> list[dict[str, Any]]: + path = await _file(project_id, key, name) + await asyncio.to_thread(path.unlink) + return await asyncio.to_thread(_listing, path.parent) diff --git a/B09_Estimation/B09_Estimation_Store.py b/B09_Estimation/B09_Estimation_Store.py index 27ae5bf29..e32557ef1 100644 --- a/B09_Estimation/B09_Estimation_Store.py +++ b/B09_Estimation/B09_Estimation_Store.py @@ -4,9 +4,13 @@ "driver_factor": …, "fx": […]} — [단가 갱신] 때만 M01 로 새로 · [저장] 때 책에 없는 새 줄만 더함(있는 줄 안 바꿈) - user_values.json 사용자 값 {"materials": {자재 키: {applied?, supply?}}, "estimate": {…}} + user_values.json 사용자 값 {"materials": {자재 키: {applied?, supply?, quote?, quote_page?, + pick?}}, "estimate": {…}} — 고친 칸만(키가 있으면 그 값 · null = 빈칸 · 없으면 기본) + quotes/<자재 키>/ 견적서 파일(166-4 · 여러 개) — 폴더가 목록(키의 못 쓰는 글자는 %XX) · + 읽을 때 materials[키].files = 이름들 · 사용자 값으로는 저장 안 함 + 결과(표)는 저장하지 않음 — GET 때 서버가 셈(B08 꼴). """ @@ -15,14 +19,19 @@ from __future__ import annotations import json import math import os +import re from pathlib import Path from typing import Any +from urllib.parse import unquote -from B09_Estimation.B09_Estimation_Calc_Base import SUPPLY +from B09_Estimation.B09_Estimation_Calc_Base import SOURCES, SUPPLY FOLDER = ("B09_Estimation", "v1") BOOK = "price_book.json" USER = "user_values.json" +QUOTES = "quotes" +#: 폴더 · 파일 이름에 못 쓰는 글자(Windows) +BAD_NAME = re.compile(r'[<>:"/\\|?*\x00-\x1f]') #: 원가계산 사용자 입력 칸 — 요율 고르는 값 · 손입력(엑셀 총괄 · 원가계산서 · 내역 손입력 칸) ESTIMATE_FIELDS = ( @@ -89,6 +98,12 @@ def clean_user(raw: Any) -> dict[str, Any]: out["applied"] = _num(cell["applied"]) if cell.get("supply") in SUPPLY: out["supply"] = cell["supply"] + if "quote" in cell: + out["quote"] = _num(cell["quote"]) + if isinstance(cell.get("quote_page"), str) and cell["quote_page"].strip(): + out["quote_page"] = cell["quote_page"].strip() + if cell.get("pick") in SOURCES: + out["pick"] = cell["pick"] if out: materials[str(key)] = out estimate: dict[str, Any] = {} @@ -105,10 +120,36 @@ def clean_user(raw: Any) -> dict[str, Any]: def read_user(project_root: str | Path) -> dict[str, Any]: - return clean_user(_read(folder(project_root) / USER)) + user = clean_user(_read(folder(project_root) / USER)) + base = folder(project_root) / QUOTES + for sub in sorted(base.iterdir()) if base.is_dir() else []: + names = [f.name for f in quote_files(sub)] + if names: + user["materials"].setdefault(unquote(sub.name), {})["files"] = names + return user def write_user(project_root: str | Path, user: dict[str, Any]) -> dict[str, Any]: clean = clean_user(user) _write(folder(project_root) / USER, clean) return clean + + +def quote_dir(project_root: str | Path, key: str) -> Path: + """자재 키 → 견적서 폴더 — 못 쓰는 글자 · % 는 %XX(되풀면 키).""" + part = re.sub(r"[%]|" + BAD_NAME.pattern, lambda m: f"%{ord(m.group()):02X}", key) + if not part.strip(" ."): + raise ValueError("자재 키가 비었습니다.") + return folder(project_root) / QUOTES / part + + +def safe_name(name: str) -> str: + """올린 파일 이름 → 저장 이름(경로 떼고 · 못 쓰는 글자 _ · 앞뒤 점 · 빈칸 뺌).""" + return BAD_NAME.sub("_", re.split(r"[\\/]", name)[-1]).strip(" .") + + +def quote_files(path: Path) -> list[Path]: + """견적서 폴더 파일(올리는 중 임시 `.…upload` 뺌) · 이름 차례.""" + if not path.is_dir(): + return [] + return sorted(f for f in path.iterdir() if f.is_file() and not f.name.startswith(".")) diff --git a/resources/tester/test_166_4_b09_quote.py b/resources/tester/test_166_4_b09_quote.py new file mode 100644 index 000000000..9198884ba --- /dev/null +++ b/resources/tester/test_166_4_b09_quote.py @@ -0,0 +1,105 @@ +"""166-4 자재단가대비표 견적 열 · 적용 출처 고르기 · 견적서 여러 파일(임시 storage) · 거울.""" + +import sys +from pathlib import Path +from uuid import uuid4 + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from B09_Estimation import B09_Estimation_Store as store # noqa: E402 +from B09_Estimation.B09_Estimation_Calc_Base import compute_base # noqa: E402 +from resources.tester.test_144_b09_base_calc import _run_ts # noqa: E402 + +MATERIALS = [ + {"key": "MT1", "name": "자갈", "prices": {"물가정보": 23000, "물가자료": 22000}, "qty": 2}, + {"key": "MT2", "name": "폭약", "prices": {"물가자료": 12000}}, + {"key": "b08:번호판", "name": "국가지점번호판", "prices": {}}, + {"key": "MT4", "name": "모래", "prices": {"물가자료": 30000}}, +] +INPUTS = {"labor": [], "machines": [], "materials": MATERIALS} +USER = { + "materials": { + "MT1": {"quote": 21000, "quote_page": "견적 A"}, # 견적이 최저 → 자동 + "MT2": {"quote": 12712, "pick": "견적"}, # 견적이 높아도 고름 + "b08:번호판": {"quote": 520000}, # 견적만 있는 줄 = 견적 자동 + "MT4": {"quote": 29000, "applied": 31000, "pick": "물가자료"}, # 고름 › 손 수 + } +} + + +def test_견적_최저_고름_손수(): + s = compute_base(INPUTS, USER) + rows = {r["key"]: r for r in s["compare"]} + got = { + k: (r["lowest"], r["lowest_from"], r["applied"], r["applied_from"]) for k, r in rows.items() + } + assert got == { + "MT1": (21000, "견적", 21000, "견적"), + "MT2": (12000, "물가자료", 12712, "견적"), + "b08:번호판": (520000, "견적", 520000, "견적"), + "MT4": (29000, "견적", 30000, "물가자료"), + } + assert rows["MT1"]["pages"]["견적"] == "견적 A" and rows["MT2"]["pick"] == "견적" + assert {r["key"]: r["price"] for r in s["material"]}["MT2"] == 12712 # 목록표 단가 따라감 + assert s["material"][0]["amount"] == 42000 + # 손 수만 있으면 「사용자」 · 고름 없으면 최저(견적 포함) + s2 = compute_base(INPUTS, {"materials": {"MT4": {"quote": 29000, "applied": 31000}}}) + assert s2["compare"][3]["applied_from"] == "사용자" + + +def test_거울_TS(): + assert _run_ts(INPUTS, USER) == compute_base(INPUTS, USER) + + +def test_저장_다듬기_파일_목록(tmp_path): + raw = {"materials": {"MT2": {"quote": 12712, "quote_page": " 견적 B ", "pick": "없는 출처", + "files": ["x"]}, "MT1": {"pick": "견적"}}} # fmt: skip + clean = store.write_user(tmp_path, raw) + assert clean["materials"] == { + "MT2": {"quote": 12712, "quote_page": "견적 B"}, + "MT1": {"pick": "견적"}, + } # files 는 저장 안 함(폴더가 목록) + folder = store.quote_dir(tmp_path, "b08:이름|㎥/hr") + folder.mkdir(parents=True) + (folder / "a.pdf").write_bytes(b"1") + (folder / ".a.pdf.123.upload").write_bytes(b"temp") + assert folder.parent == store.folder(tmp_path) / "quotes" + user = store.read_user(tmp_path) + assert user["materials"]["b08:이름|㎥/hr"] == {"files": ["a.pdf"]} + assert store.safe_name("../../x\\y:z.pdf") == "y_z.pdf" and store.safe_name("..") == "" + with pytest.raises(ValueError): + store.quote_dir(tmp_path, "..") + + +def test_올리기_목록_내려받기_지우기(tmp_path, monkeypatch): + from fastapi import FastAPI + from fastapi.testclient import TestClient + + from B09_Estimation import B09_Estimation_Router_Quotes as quotes + + async def root(_project_id): + return tmp_path + + monkeypatch.setattr(quotes, "_project_root", root) + app = FastAPI() + app.include_router(quotes.router, prefix="/api/projects") + client = TestClient(app) + base = f"/api/projects/{uuid4()}/estimation/quotes" + key = {"key": "b08:pipe_600"} + files = [("files", ("견적1.pdf", b"one", "application/pdf")), + ("files", ("sub/견적2.xlsx", b"two!", "application/octet-stream"))] # fmt: skip + listed = client.post(base, params=key, files=files).json() + assert [(f["name"], f["size"]) for f in listed] == [("견적1.pdf", 3), ("견적2.xlsx", 4)] + assert (store.quote_dir(tmp_path, "b08:pipe_600") / "견적1.pdf").read_bytes() == b"one" + assert len(client.get(base, params=key).json()) == 2 + got = client.get(base + "/file", params={**key, "name": "견적2.xlsx"}) + assert got.status_code == 200 and got.content == b"two!" + assert client.get(base + "/file", params={**key, "name": "../x"}).status_code == 404 + assert client.get(base, params={"key": ".."}).status_code == 400 + left = client.delete(base, params={**key, "name": "견적1.pdf"}).json() + assert [f["name"] for f in left] == ["견적2.xlsx"] + assert store.read_user(tmp_path)["materials"] == {"b08:pipe_600": {"files": ["견적2.xlsx"]}}