eomsangdonandClaude Opus 5.5
40b80ff1b6
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
94 lines
3.4 KiB
Python
94 lines
3.4 KiB
Python
"""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)
|