feat(B09 불러오기): .estx 고리 — 내역 줄 · 단가표 같은 길 · 요율 = 엑셀 안내 · 호표 목록 = 이음 후보 수 (PLAN 707-5)

- 길: 이름 끝 .estx = `B09_Estimation_Estx` read_bill · read_prices(같은 그릇) · 요율 = 400 「같은 공사 엑셀에서」
- 미리 보기 `호표` = 대가 · 산근 목록 수 + 내역 줄이 코드로 가리킨 수(구성 줄 귀속은 안 함)
- 창: 파일 고르기 *.estx · [단가표도 같이](받은 patch 를 캐시에 함께) · 지난 파일 요율 실패 = 까닭 알림 뒤 다시 고름
- 시험 test_707_5 8(길 갈림 · 호표 수 · 깨진 파일 400)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-31 16:59:53 +09:00
co-authored by Claude Opus 5.5
parent 6b6b1575eb
commit 56ebe76e7f
3 changed files with 82 additions and 14 deletions
+29 -4
View File
@@ -18,6 +18,9 @@
{file_key, name, sheets, counts, linked, missing,
patch{labor_cover, machine_cover, materials}}
file_key = 세 길 같은 표지(`_keep` · `_saved`) — 같은 파일을 다시 올리지 않음 ·
.estx(EST 원본 · 707-5 `B09_Estimation_Estx`) = 이름 끝으로 갈림 — 내역 줄 · 단가표 같은 그릇 ·
요율 = 400 「같은 공사 엑셀에서」 ·
미리 보기 `호표` = 대가 · 산근 목록 수 + 코드로 가리킨 줄(이음 후보만)
화면은 `patch` 를 캐시에 얹고 [저장] 때 서버가 정본으로 다시 셈.
`B09_Estimation_Router` 가 끼워 넣음(같은 머리 · 같은 권한).
"""
@@ -37,6 +40,7 @@ from pydantic import BaseModel, Field
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
from B09_Estimation import B09_Estimation_Bill_Import as importer
from B09_Estimation import B09_Estimation_Bill_Library as library
from B09_Estimation import B09_Estimation_Estx as estx
from B09_Estimation import B09_Estimation_Import_Prices as import_prices
from B09_Estimation import B09_Estimation_Import_Rates as import_rates
from B09_Estimation import B09_Estimation_Prices as prices
@@ -82,13 +86,28 @@ def _keep(root: Path, name: str, data: bytes) -> str:
return key
def _is_estx(name: str) -> bool:
return name.lower().endswith(".estx")
def _parse(name: str, data: bytes) -> dict[str, Any]:
try:
return read_bill(name, data)
return (estx.read_bill if _is_estx(name) else read_bill)(name, data)
except BillExcelError as error:
raise HTTPException(status_code=400, detail=str(error)) from error
def _hopyo(name: str, data: bytes, parsed: dict[str, Any]) -> dict[str, int] | None:
""".estx 호표 목록 수 · 내역 줄이 코드로 가리킨 수(이음 후보 — 구성 줄은 `estx.UNKNOWN`)."""
if not _is_estx(name):
return None
got = estx.read_hopyo(name, data)
codes = {(k, x["코드"]) for k, v in got.items() for x in v}
hits = [h for line in parsed["줄"] if (h := line.get("호표"))]
return {**{k: len(v) for k, v in got.items()},
"가리킴": sum((h["종류"], h["코드"]) in codes for h in hits)} # fmt: skip
def _draft(raw: Any) -> dict[str, Any] | None:
if raw in (None, ""):
return None
@@ -158,8 +177,10 @@ async def import_bill(
if not file_key:
raise HTTPException(status_code=400, detail="파일을 고름")
path = _saved(root, file_key)
data = path.read_bytes()
if file is None:
parsed = await asyncio.to_thread(_parse, path.name, path.read_bytes())
parsed = await asyncio.to_thread(_parse, path.name, data)
hopyo = await asyncio.to_thread(_hopyo, path.name, data, parsed)
rows = (await payload(project_id, root, _draft(user), True))["sheets"]["bill"]["rows"]
found = await asyncio.to_thread(_found, session, path.name, _numbers(parsed))
return {
@@ -167,6 +188,7 @@ async def import_bill(
"파일": path.name,
**importer.preview(parsed, rows),
"이음": len(found),
"호표": hopyo,
}
@@ -234,6 +256,8 @@ async def import_rate_sheet(
root = await _project_root(project_id)
key, name, data = await _file(file, file_key, root)
draft = _draft(user)
if _is_estx(name):
raise HTTPException(400, "원가계산서 요율은 .estx 에서 못 읽음 — 같은 공사 엑셀에서 불러옴")
mine = (draft if draft is not None else await asyncio.to_thread(store.read_user, root))[
"estimate"
]
@@ -262,8 +286,9 @@ async def import_price_sheets(
draft = _draft(user)
mine = draft if draft is not None else await asyncio.to_thread(store.read_user, root)
try:
read = await asyncio.to_thread(import_prices.read_prices, name, data)
except ExcelError as error:
reader = estx.read_prices if _is_estx(name) else import_prices.read_prices
read = await asyncio.to_thread(reader, name, data)
except (ExcelError, BillExcelError) as error:
raise HTTPException(400, str(error)) from error
keys = await main._demand(root) # 화면 단가 목록(GET)과 같은 수요 — 사본 전부
b08 = await main._b08_for(project_id, [], True)