feat(b09): 자재 견적서 여러 파일 첨부 길 · 견적 · 고른 출처 저장(166-4)

Router_Quotes 새 길(목록 · 올리기 · 내려받기 · 지우기 · v1/quotes/<자재 키>/) · Store quote · quote_page · pick 칸 · 읽을 때 files 목록

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015sDZSjFvoiDjdGpgj3muc3

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)

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)

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("."))

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"]}}