feat(B09 설계내역서): 경쟁사 내역 엑셀 줄 불러오기 서버 — 읽기 · 짜기 · 고정 단가 칸 (PLAN 707-1)
- 읽기(Bill_Excel) — STmate 「설계내역서」 숨은 코드 · 부모 코드 열 · EST 「내역서」 공종 번호 깊이 · 비고 대가 · 산근 · 단산 · 일위 · 제외 묶음(관급 · 제외금액) · 조달수수료 = 율 · DRM = 「읽을 수 없음」 · 열 머리 = M02 실무자료 칸 읽기 재사용 - 짜기(Bill_Import) — 공종 = 새 묶음 · 항목 = 새 줄 + 고정 단가 칸 · 수량 갈래 take · price · 이름 같은 자동 줄 붙임 · 남은 자동 줄 지움 · 같은 파일 다시 = 갈아 끼움 · 어긋난 줄 목록 - 길(Router_Import) — import/bill(file · file_key) 미리 보기 · import/bill/apply(같은 파일 이름 실무자료 사전조합 → 서고 사본 연결 · 층 쓰기 0) · 저장 안 함 - 셈 짝(Calc_Bill py · ts) — 줄 price = 연결이 단가를 못 낼 때 금액 · fixed · src_qty · Store 다듬기 - 확인 = 기번3 710,359,136 · 영월 767,148,700 · 봉화 781,283,415 · 거창 384,916,227 직접비 합 = 엑셀 · 빈 줄 0 · 어긋남 0 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,238 @@
|
||||
"""B09 경쟁사 내역 엑셀 읽기(PLAN 707-1) — 파일 → 내역 줄(묶음 · 항목) · 엑셀 합계. 저장 안 함.
|
||||
|
||||
꼴 둘(`_분석_705-4_불러오기_범위.md` ②):
|
||||
STmate 「설계내역서」(없으면 「착공내역서」 · 「설계변경내역서」) —
|
||||
숨은 열 = 줄 코드 · 부모 코드(`B00194_1` · `E2_1` — 비고 열 뒤 첫 둘) ·
|
||||
묶음 = 코드 E · C(C = 제외금액 맨 위) ·
|
||||
변경 차수(`_2` 뒤) 줄 = 안 읽음 · 부모 = 열린 묶음 가운데 부모 코드가 같은 가장 안쪽 ·
|
||||
호표 = 코드 B 대가 · D 산근 · M 자재(번 = 비고 「대가 5호표」 · 「산근 13호표」 숫자)
|
||||
EST 「내역서」 — 숨은 열 없음 · 묶음 = 공종 칸 있음 + 수량 없음 · 깊이 = 공종 번호
|
||||
(Ⅰ · Ⅱ = 1 · 「4.1」 = 점 둘레 수 + 1) · 공종 · 수량 없이 금액만 = 합계 줄(안 읽음) ·
|
||||
호표 = 비고 「단산46참조」 · 「일위8참조」 · 「관급」
|
||||
열 = 머리 글(앞 10 줄 · 빈칸 뺌) 공종 · 명칭|품명 · 규격 · 수량 · 단위 · 합계 ·
|
||||
노무비 · 재료비 · 경비 · 비고 — 돈 머리 = 단가 열 · 다음 열 = 금액
|
||||
(`M02_Hopyo_Excel` 의 칸 읽기 재사용) · 빈 단가 = 0.
|
||||
제외 = 맨 위 묶음 가운데 STmate 코드 C · 비고 「제외」 · 이름 「관급」 — 직접비 합에서 뺌.
|
||||
조달수수료 = 제외 묶음 안 단위 % · 이름 「수수료」 줄 → `조달수수료율`
|
||||
(줄로 안 실음 · 우리 조달수수료 줄이 셈).
|
||||
그 밖 % 줄 = 1 식 · 단가 = 엑셀 금액(금액 그대로).
|
||||
감싼 묶음 = 직접 맨 위 묶음이 하나뿐(공사 이름)이면 벗김(깊이 5 한도 · 레벨 1 = 공종).
|
||||
DRM 파일 = 「읽을 수 없음」(`M02_Hopyo_Excel.DRM`).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import re
|
||||
import warnings
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
from M02_MasterTemplete import M02_Hopyo_Excel as hx
|
||||
|
||||
SHEETS = ("설계내역서", "내역서", "착공내역서", "설계변경내역서")
|
||||
HEADS = {
|
||||
"공종": "공종",
|
||||
"명칭": "품명",
|
||||
"품명": "품명",
|
||||
"규격": "규격",
|
||||
"수량": "수량",
|
||||
"단위": "단위",
|
||||
"합계": "합계",
|
||||
"노무비": "노무비",
|
||||
"재료비": "재료비",
|
||||
"경비": "경비",
|
||||
"비고": "비고",
|
||||
}
|
||||
PARTS = (("노무비", "labor"), ("재료비", "material"), ("경비", "expense"))
|
||||
_CODE = re.compile(r"^([A-Z]+\d*)_(\d+)$")
|
||||
_ROMAN = re.compile(r"^[ⅠⅡⅢⅣⅤⅥⅦⅧⅨⅩ]+$|^[IVX]+\.?$")
|
||||
_ST_NOTE = re.compile(r"^(대가|산근|자재)(\d+)")
|
||||
_EST_NOTE = re.compile(r"^(단산|일위)(\d+)")
|
||||
NO_SHEET = "내역서 시트(설계내역서 · 내역서) 없음"
|
||||
|
||||
|
||||
class BillExcelError(ValueError):
|
||||
"""못 읽는 파일 — 길이 400 으로 답함."""
|
||||
|
||||
|
||||
def _sheet(name: str, data: bytes) -> tuple[str, list[list[Any]]]:
|
||||
if not data:
|
||||
raise BillExcelError("빈 파일")
|
||||
if b"DRMONE" in data[:256] and data[:4] not in (b"PK\x03\x04", b"\xd0\xcf\x11\xe0"):
|
||||
raise BillExcelError(f"읽을 수 없음 — {hx.DRM}")
|
||||
try:
|
||||
if data[:4] == b"PK\x03\x04" or name.lower().endswith(".xlsx"):
|
||||
import openpyxl
|
||||
|
||||
with warnings.catch_warnings(): # 인쇄 영역 이름 경고(STmate) — 읽기와 무관
|
||||
warnings.simplefilter("ignore")
|
||||
book = openpyxl.load_workbook(io.BytesIO(data), read_only=True, data_only=True)
|
||||
titles = {ws.title.strip(): ws for ws in book.worksheets}
|
||||
pick = next((t for t in SHEETS if t in titles), None)
|
||||
rows = [list(r) for r in titles[pick].iter_rows(values_only=True)] if pick else []
|
||||
book.close()
|
||||
else:
|
||||
import xlrd
|
||||
|
||||
book = xlrd.open_workbook(file_contents=data, logfile=io.StringIO())
|
||||
titles = {sh.name.strip(): sh for sh in book.sheets()}
|
||||
pick = next((t for t in SHEETS if t in titles), None)
|
||||
sh = titles.get(pick or "")
|
||||
rows = [sh.row_values(r) for r in range(sh.nrows)] if sh else []
|
||||
except Exception as error: # 라이브러리마다 다른 예외 — 깨진 파일 · 엑셀 아님
|
||||
raise BillExcelError("엑셀 파일을 읽지 못함 — xls · xlsx 가 아니거나 깨진 파일") from error
|
||||
if pick is None:
|
||||
raise BillExcelError(NO_SHEET)
|
||||
return pick, rows
|
||||
|
||||
|
||||
def _columns(rows: list[list[Any]]) -> tuple[int, dict[str, int]]:
|
||||
for at, row in enumerate(rows[:10]):
|
||||
flat = [hx._flat(v) for v in row]
|
||||
if "공종" not in flat or not ({"명칭", "품명"} & set(flat)):
|
||||
continue
|
||||
cols: dict[str, int] = {}
|
||||
for col, head in enumerate(flat):
|
||||
if head in HEADS and HEADS[head] not in cols:
|
||||
cols[HEADS[head]] = col
|
||||
below = [hx._flat(v) for v in rows[at + 1]] if at + 1 < len(rows) else []
|
||||
return (at + 2 if "단가" in below or "금액" in below else at + 1), cols
|
||||
raise BillExcelError("내역서 머리 글(공종 · 명칭)을 못 찾음")
|
||||
|
||||
|
||||
def _codes(row: list[Any], after: int) -> tuple[str, str] | None:
|
||||
"""STmate 숨은 열 (줄 코드, 부모 코드) — 비고 뒤 「X_n」 꼴 첫 둘 ·
|
||||
변경 차수(_2 뒤)면 ("", "")."""
|
||||
got = [m for v in row[after + 1 :] if (m := _CODE.match(hx._text(v)))]
|
||||
if not got:
|
||||
return None
|
||||
if got[0].group(2) != "1":
|
||||
return "", ""
|
||||
return got[0].group(1), (got[1].group(1) if len(got) > 1 else "")
|
||||
|
||||
|
||||
def _depth(head: str) -> int:
|
||||
if _ROMAN.match(head):
|
||||
return 1
|
||||
parts = [p for p in head.split(".") if p.strip()]
|
||||
return len(parts) + 1 if parts and all(p.strip().isdigit() for p in parts) else 1
|
||||
|
||||
|
||||
def _money(row: list[Any], cols: dict[str, int], offset: int) -> dict[str, Decimal]:
|
||||
got = hx._money(row, cols, offset)
|
||||
return {key: got.get(head) or Decimal(0) for head, key in PARTS}
|
||||
|
||||
|
||||
def _hopyo(note: str, code: str) -> dict[str, Any] | None:
|
||||
flat = hx._flat(note)
|
||||
if hit := _ST_NOTE.match(flat):
|
||||
return {"종류": hit.group(1), "번": int(hit.group(2)), "코드": code}
|
||||
if hit := _EST_NOTE.match(flat):
|
||||
return {"종류": "산근" if hit.group(1) == "단산" else "대가", "번": int(hit.group(2))}
|
||||
if code[:1] in ("B", "D", "M"):
|
||||
return {"종류": {"B": "대가", "D": "산근", "M": "자재"}[code[0]], "코드": code}
|
||||
return None
|
||||
|
||||
|
||||
def read_bill(name: str, data: bytes) -> dict[str, Any]:
|
||||
"""엑셀 → `{꼴, 시트, 줄 [{행, 묶음, 위, 공종, 이름, 규격, 단위, 수량, 단가 {labor, material,
|
||||
expense}, 금액 {…}, 비고, 호표, 제외}], 합계 {직접, 제외}, 조달수수료율, 합계줄, 율줄}` —
|
||||
`위` = 부모 줄 행(None = 맨 위) · 수 = JSON 수."""
|
||||
title, rows = _sheet(name, data)
|
||||
start, cols = _columns(rows)
|
||||
lines: list[dict[str, Any]] = []
|
||||
stack: list[tuple[Any, dict[str, Any]]] = [] # (코드 | 깊이, 묶음 줄)
|
||||
stmate, skipped, fee, pct_lines = False, 0, None, 0
|
||||
for at, row in enumerate(rows[start:], start + 1):
|
||||
cell = lambda h: hx._cell(row, cols.get(h)) # noqa: E731
|
||||
head, item_name = hx._text(cell("공종")), hx._text(cell("품명"))
|
||||
if not item_name:
|
||||
continue
|
||||
qty = hx._num(cell("수량"))
|
||||
codes = _codes(row, cols.get("비고", 13))
|
||||
stmate = stmate or codes is not None
|
||||
if codes == ("", ""):
|
||||
continue
|
||||
if codes:
|
||||
own, parent = codes
|
||||
group = own[:1] in ("E", "C")
|
||||
at_up = next((n for n in range(len(stack) - 1, -1, -1) if stack[n][0] == parent), None)
|
||||
stack = stack[: at_up + 1] if at_up is not None else []
|
||||
else:
|
||||
own = ""
|
||||
group = bool(head) and qty is None
|
||||
if not group and qty is None:
|
||||
skipped += 1 # 「순 공 사 비」 같은 합계 줄
|
||||
continue
|
||||
if group:
|
||||
depth = _depth(head)
|
||||
while stack and stack[-1][0] >= depth:
|
||||
stack.pop()
|
||||
up = stack[-1][1] if stack else None
|
||||
note = hx._text(cell("비고"))
|
||||
line = {
|
||||
"행": at,
|
||||
"묶음": group,
|
||||
"위": up["행"] if up else None,
|
||||
"공종": head,
|
||||
"이름": item_name,
|
||||
"규격": hx._text(cell("규격")),
|
||||
"단위": hx._text(cell("단위")),
|
||||
"수량": qty,
|
||||
"단가": _money(row, cols, 0),
|
||||
"금액": _money(row, cols, 1),
|
||||
"비고": note,
|
||||
"호표": None if group else _hopyo(note, own),
|
||||
"제외": (up or {}).get("제외", False)
|
||||
or (up is None and (own[:1] == "C" or "제외" in note or "관급" in item_name)),
|
||||
}
|
||||
if group:
|
||||
stack.append((own if codes else _depth(head), line))
|
||||
elif hx._flat(line["단위"]) == "%":
|
||||
if line["제외"] and "수수료" in item_name:
|
||||
fee = fee if fee is not None else qty
|
||||
continue
|
||||
pct_lines += 1
|
||||
line.update({"단가": line["금액"], "수량": Decimal(1), "단위": "식"})
|
||||
line["규격"] = " ".join(t for t in (line["규격"], f"({hx._out(qty)} %)") if t)
|
||||
lines.append(line)
|
||||
_unwrap(lines)
|
||||
total = {"직접": Decimal(0), "제외": Decimal(0)}
|
||||
for line in lines:
|
||||
if not line["묶음"]:
|
||||
total["제외" if line["제외"] else "직접"] += sum(line["금액"].values())
|
||||
return _plain(
|
||||
{
|
||||
"꼴": "STmate" if stmate else "EST",
|
||||
"시트": title,
|
||||
"줄": lines,
|
||||
"합계": total,
|
||||
"조달수수료율": fee,
|
||||
"합계줄": skipped,
|
||||
"율줄": pct_lines,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _unwrap(lines: list[dict[str, Any]]) -> None:
|
||||
"""직접 맨 위 묶음이 하나뿐이면(공사 이름) 벗김 — 자식이 맨 위로(되풀이)."""
|
||||
while True:
|
||||
tops = [x for x in lines if x["위"] is None and not x["제외"]]
|
||||
if len(tops) != 1 or not tops[0]["묶음"]:
|
||||
return
|
||||
wrap = tops[0]["행"]
|
||||
lines.remove(tops[0])
|
||||
for line in lines:
|
||||
if line["위"] == wrap:
|
||||
line["위"] = None
|
||||
|
||||
|
||||
def _plain(value: Any) -> Any:
|
||||
if isinstance(value, Decimal):
|
||||
return hx._out(value)
|
||||
if isinstance(value, dict):
|
||||
return {k: _plain(v) for k, v in value.items()}
|
||||
if isinstance(value, list):
|
||||
return [_plain(v) for v in value]
|
||||
return value
|
||||
@@ -0,0 +1,221 @@
|
||||
"""B09 경쟁사 내역 줄 불러오기 짜기(PLAN 707-1) — 읽은 내역(`Bill_Excel.read_bill`) → 설계내역서
|
||||
덮개 v2(`user.bill` · 계약 = `B09_Estimation_Calc_Bill` 머리). 파일 · 저장 안 만짐(부르는 쪽 몫).
|
||||
|
||||
먼저 싣고 나중 잇기 — 빠짐 0:
|
||||
묶음 줄 → 새 묶음(`hand:x{표지}-{행}` · 제외 맨 위 = 관급자재대 `sec:G` 아래)
|
||||
항목 줄 → 새 줄 + 고정 단가 칸 `price`(엑셀 단가) + `src_qty`(엑셀 수량) + `note`(엑셀 비고) ·
|
||||
연결 = 같은 책 일위대가 사본(`links` {대가 번: ref}) — 없으면 고정 단가로 금액
|
||||
같은 파일을 다시 부르면(같은 표지) 앞서 실은 줄을 갈아 끼움.
|
||||
수량 갈래 — `take`(가 넘겨받기) 붙인 자동 줄 수량 = 엑셀(B08 = `auto_qty` 로 나란히) ·
|
||||
`price`(나 단가만) 붙인 자동 줄 수량 = B08 · 엑셀 수량 = 비교 칸만. 새 줄 수량 = 늘 엑셀.
|
||||
두 번 셈 막음 — `attach` = 이름(빈칸 뺌)이 같은 자동 항목 줄에 붙임(그 줄 = 불러온 자리로 옮기고
|
||||
단가 = 연결 · 고정 단가 칸 · 자재 줄 · 조달수수료 뺌) ·
|
||||
`drop` = 안 붙은 자동 항목 줄 지움(`gone`).
|
||||
조달수수료율 = 사용자 값이 비었을 때만 엑셀 율.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import hashlib
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
MODES = ("take", "price")
|
||||
FIELDS = ("name", "spec", "unit", "qty", "ref")
|
||||
PARTS = ("labor", "material", "expense")
|
||||
|
||||
|
||||
def tag_of(data: bytes) -> str:
|
||||
"""파일 표지 — 같은 내용 = 같은 표지(새 줄 키 머리)."""
|
||||
return hashlib.sha1(data).hexdigest()[:8]
|
||||
|
||||
|
||||
def _flat(text: Any) -> str:
|
||||
return re.sub(r"\s+", "", str(text or ""))
|
||||
|
||||
|
||||
def freeze(user: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
"""화면 굳힘(`UI_Tabs_Bill_Edit.freeze`)과 같은 일 — v2 = 그대로 + 덮개에 없는 보이는 줄을
|
||||
보이는 앞 줄 뒤에 · 옛 꼴 = 보이는 줄로(자동 줄 수량 덮개 · 새 줄 칸 옮김)."""
|
||||
bill = user.get("bill") or {}
|
||||
v2 = bill.get("v") == 2
|
||||
entries = copy.deepcopy(bill.get("rows") or []) if v2 else []
|
||||
gone = list(bill.get("gone") or []) if v2 else []
|
||||
prev = -1
|
||||
for row in rows:
|
||||
key = row["key"]
|
||||
if key == "fee":
|
||||
continue
|
||||
at = next((n for n, e in enumerate(entries) if e["key"] == key), -1)
|
||||
if at < 0 and key not in gone:
|
||||
entry = {"key": key, **({"parent": row["parent"]} if row.get("parent") else {})}
|
||||
if not v2 and row.get("auto"):
|
||||
if key in (bill.get("qty") or {}):
|
||||
entry["qty"] = bill["qty"][key]
|
||||
elif not v2:
|
||||
hand = next((h for h in bill.get("rows") or [] if h.get("key") == key), {})
|
||||
entry.update({f: hand[f] for f in FIELDS if f in hand})
|
||||
if row["kind"] != "item":
|
||||
entry["kind"] = "group"
|
||||
entries.insert(prev + 1, entry)
|
||||
at = prev + 1
|
||||
if at >= 0:
|
||||
prev = at
|
||||
return {"v": 2, "rows": entries, "gone": gone}
|
||||
|
||||
|
||||
def _amount(line: dict[str, Any]) -> float:
|
||||
return sum(float(line["금액"].get(p) or 0) for p in PARTS)
|
||||
|
||||
|
||||
def _price(line: dict[str, Any]) -> dict[str, float]:
|
||||
return {p: float(line["단가"].get(p) or 0) for p in PARTS}
|
||||
|
||||
|
||||
def _autos(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""붙일 수 있는 자동 항목 줄 — 자재 줄 · 조달수수료 뺌."""
|
||||
return [
|
||||
r
|
||||
for r in rows
|
||||
if r["kind"] == "item" and r.get("auto") and r["key"] != "fee" and r["key"][:4] != "mat:"
|
||||
]
|
||||
|
||||
|
||||
def matches(parsed: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, str]:
|
||||
"""{불러온 줄 행(글): 자동 줄 키} — 이름(빈칸 뺌)이 같은 첫 남은 자동 줄 · 제외 줄 뺌."""
|
||||
free: dict[str, list[str]] = {}
|
||||
for row in _autos(rows):
|
||||
free.setdefault(_flat(row["name"]), []).append(row["key"])
|
||||
out = {}
|
||||
for line in parsed["줄"]:
|
||||
if line["묶음"] or line["제외"]:
|
||||
continue
|
||||
keys = free.get(_flat(line["이름"])) or []
|
||||
if keys:
|
||||
out[str(line["행"])] = keys.pop(0)
|
||||
return out
|
||||
|
||||
|
||||
def preview(parsed: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
"""불러오기 창 수 — 묶음 · 항목 · 이름 같은 자동 줄 · 그 밖 자동 줄 · 엑셀 합계."""
|
||||
lines = parsed["줄"]
|
||||
pairs = matches(parsed, rows)
|
||||
return {
|
||||
"꼴": parsed["꼴"],
|
||||
"시트": parsed["시트"],
|
||||
"묶음": sum(1 for x in lines if x["묶음"]),
|
||||
"항목": sum(1 for x in lines if not x["묶음"]),
|
||||
"합계": parsed["합계"],
|
||||
"조달수수료율": parsed["조달수수료율"],
|
||||
"합계줄": parsed["합계줄"],
|
||||
"율줄": parsed["율줄"],
|
||||
"같은이름": len(pairs),
|
||||
"남는자동": len(_autos(rows)) - len(pairs),
|
||||
}
|
||||
|
||||
|
||||
def apply(
|
||||
parsed: dict[str, Any],
|
||||
user: dict[str, Any],
|
||||
rows: list[dict[str, Any]],
|
||||
tag: str,
|
||||
mode: str = "take",
|
||||
attach: bool = True,
|
||||
drop: bool = False,
|
||||
links: dict[int, dict[str, Any]] | None = None,
|
||||
) -> tuple[dict[str, Any], dict[str, Any]]:
|
||||
"""(새 사용자 값, 알림 {실음, 붙임, 지움, 이음, 줄 {행: 줄 키}}) — `rows` = 지금 셈 줄."""
|
||||
if mode not in MODES:
|
||||
raise ValueError(f"수량 갈래는 {', '.join(MODES)} 중 하나")
|
||||
links = links or {}
|
||||
head = f"hand:x{tag}-"
|
||||
bill = freeze(user, rows)
|
||||
kept = []
|
||||
for e in bill[
|
||||
"rows"
|
||||
]: # 앞서 실은 줄 갈아 끼움 — 그 묶음에 있던 자동 줄 = 제자리 · 새 줄 = 맨 위
|
||||
if e["key"].startswith(head):
|
||||
continue
|
||||
if str(e.get("parent") or "").startswith(head):
|
||||
if not e["key"].startswith("hand:"):
|
||||
continue
|
||||
e.pop("parent")
|
||||
kept.append(e)
|
||||
bill["rows"] = kept
|
||||
pairs = matches(parsed, rows) if attach else {}
|
||||
keys = {line["행"]: f"{head}{line['행']:05d}" for line in parsed["줄"]}
|
||||
put, report = [], {"실음": 0, "붙임": 0, "지움": 0, "이음": 0, "줄": {}}
|
||||
for line in parsed["줄"]:
|
||||
parent = keys.get(line["위"]) if line["위"] is not None else None
|
||||
if parent is None and line["제외"]:
|
||||
parent = "sec:G"
|
||||
hopyo = line.get("호표") or {}
|
||||
link = links.get(hopyo["번"]) if hopyo.get("종류") == "대가" and "번" in hopyo else None
|
||||
report["이음"] += bool(link)
|
||||
cells = {"price": _price(line), "src_qty": line["수량"]}
|
||||
auto = pairs.get(str(line["행"]))
|
||||
if auto:
|
||||
entry = next((e for e in bill["rows"] if e["key"] == auto), None)
|
||||
if entry is None:
|
||||
entry = {"key": auto}
|
||||
else:
|
||||
bill["rows"].remove(entry)
|
||||
entry.update({**cells, "ref": link})
|
||||
if mode == "take":
|
||||
entry["qty"] = line["수량"]
|
||||
entry.pop("parent", None)
|
||||
put.append({**entry, **({"parent": parent} if parent else {})})
|
||||
report["붙임"] += 1
|
||||
report["줄"][str(line["행"])] = auto
|
||||
continue
|
||||
entry = {"key": keys[line["행"]], **({"parent": parent} if parent else {})}
|
||||
entry.update({"name": line["이름"], "spec": line["규격"], "unit": line["단위"]})
|
||||
if line["묶음"]:
|
||||
entry.update({"kind": "group", "qty": None, "ref": None})
|
||||
else:
|
||||
entry.update({"qty": line["수량"], "ref": link, **cells, "note": line["비고"]})
|
||||
report["줄"][str(line["행"])] = entry["key"]
|
||||
put.append(entry)
|
||||
report["실음"] += 1
|
||||
if drop:
|
||||
used = set(pairs.values())
|
||||
for row in _autos(rows):
|
||||
if row["key"] not in used and row["key"] not in bill["gone"]:
|
||||
bill["gone"].append(row["key"])
|
||||
report["지움"] += 1
|
||||
bill["rows"] = [e for e in bill["rows"] if e["key"] not in bill["gone"]]
|
||||
bill["rows"] += put
|
||||
out = {**user, "bill": bill}
|
||||
estimate = dict(user.get("estimate") or {})
|
||||
if estimate.get("procurement_fee_pct") is None and parsed.get("조달수수료율") is not None:
|
||||
estimate["procurement_fee_pct"] = parsed["조달수수료율"]
|
||||
out["estimate"] = estimate
|
||||
return out, report
|
||||
|
||||
|
||||
def differences(
|
||||
parsed: dict[str, Any], report: dict[str, Any], rows: list[dict[str, Any]]
|
||||
) -> list[dict[str, Any]]:
|
||||
"""불러온 직후 엑셀 금액 ≠ 우리 금액인 줄 [{행, 이름, 엑셀, 우리, 까닭}]."""
|
||||
by_key = {r["key"]: r for r in rows}
|
||||
out = []
|
||||
for line in parsed["줄"]:
|
||||
row = by_key.get(report["줄"].get(str(line["행"]), ""))
|
||||
if row is None:
|
||||
continue
|
||||
excel, ours = _amount(line), row.get("sum")
|
||||
if ours is not None and abs(ours - excel) < 1:
|
||||
continue
|
||||
if ours is None:
|
||||
why = "빈칸 — 단가 · 수량 없음"
|
||||
elif row.get("qty") != line["수량"]:
|
||||
why = "수량 = B08(단가만)"
|
||||
elif not row.get("fixed"):
|
||||
why = "이음 단가 = 우리 셈"
|
||||
else:
|
||||
why = "끝수"
|
||||
out.append(
|
||||
{"행": line["행"], "이름": line["이름"], "엑셀": excel, "우리": ours, "까닭": why}
|
||||
)
|
||||
return out
|
||||
@@ -27,7 +27,10 @@
|
||||
(제목 `sec:*` = name 만 · `mat:` = ref 없음 ·
|
||||
`fee` = 덮개 없음 · 차례에 안 넣음 — `sec:G` 끝 고정)
|
||||
새 줄(`hand:*`) = 모든 칸 + "kind": "group"(묶음 · 수량 · 연결 없음) | 없음 = 항목 ·
|
||||
"cost": "direct"(맨 위 새 묶음 · 없음도 direct)}],
|
||||
"cost": "direct"(맨 위 새 묶음 · 없음도 direct) ·
|
||||
고정 단가 칸(707-1 · 새 줄 · 자동 줄 둘 다) = "price" {labor, material, expense}(제 단가 ·
|
||||
연결이 단가를 못 내면 이것으로 금액 · 줄 `fixed`) · "src_qty"(불러온 수량 = 비교 칸) ·
|
||||
"note"(새 줄 비고 · 연결 비고가 이김)}],
|
||||
"gone": [지운 자동 줄 키 — `sec:G` · `fee` · `mat:` 는 안 받음(자재 줄 주인 = 대비표 표시)]}
|
||||
덮개 없음 = 기본 표 그대로 · [초기화] = `bill` 지움 ·
|
||||
`rows` 에 없고 `gone` 에도 없는 자동 줄(B08 새 열 · 대비표 새 표시) =
|
||||
@@ -113,6 +116,8 @@ def _item_row(item: dict[str, Any], prices: dict[str, Any] | None) -> dict[str,
|
||||
"ref": item.get("ref"),
|
||||
**{k: v for k, v in item.items() if k.startswith("auto_")},
|
||||
**({"hand": True} if item.get("hand") else {"auto": True}),
|
||||
**({"fixed": True} if item.get("fixed") else {}),
|
||||
**({"src_qty": item["src_qty"]} if num(item.get("src_qty")) is not None else {}),
|
||||
**_amounts(qty, prices),
|
||||
}
|
||||
row["_gaps"] = []
|
||||
@@ -170,16 +175,27 @@ def ref_price(
|
||||
return hopyo.get(key) if key else None
|
||||
|
||||
|
||||
def fixed_price(price: Any) -> dict[str, Any] | None:
|
||||
"""고정 단가 칸(707-1) → 1 단위 단가 — 빈 비목 0 · 칸 없음 = None."""
|
||||
if not isinstance(price, dict):
|
||||
return None
|
||||
out = {p: num(price.get(p)) or 0.0 for p in PARTS}
|
||||
return {"no": "", **out, "sum": out["labor"] + out["material"] + out["expense"]}
|
||||
|
||||
|
||||
def hand_items(user: dict[str, Any], books: dict[str, Any], hopyo: dict) -> list[tuple]:
|
||||
"""새 줄(`hand:`) → (항목, 단가) — 연결 = 고른 호표(로직 = 새 산근 키로 풀어 둠)."""
|
||||
"""새 줄(`hand:`) → (항목, 단가) — 연결 = 고른 호표(로직 = 새 산근 키로 풀어 둠) ·
|
||||
연결이 단가를 못 내면 고정 단가 칸(707-1)."""
|
||||
out = []
|
||||
for hand in (user.get("bill") or {}).get("rows") or []:
|
||||
if not str(hand.get("key") or "").startswith("hand:"):
|
||||
continue
|
||||
price = ref_price(hand.get("ref"), books, hopyo, hand["key"])
|
||||
ref = {"tab": price["tab"], "key": price["key"]} if price and price.get("tab") else None
|
||||
fixed = price is None and (price := fixed_price(hand.get("price"))) is not None
|
||||
why = WHY_LOGIC if (hand.get("ref") or {}).get("tab") == "logic" else WHY_HAND
|
||||
item = {**hand, "hand": True, "note": (price or {}).get("no") or "", "ref": ref}
|
||||
note = (price or {}).get("no") or hand.get("note") or ""
|
||||
item = {**hand, "hand": True, "note": note, "ref": ref, "fixed": fixed}
|
||||
out.append((item, price, why))
|
||||
return out
|
||||
|
||||
@@ -209,6 +225,10 @@ def _covered(
|
||||
out["note"] = (prices or {}).get("no") or ""
|
||||
if (cover["ref"] or {}).get("tab") == "logic":
|
||||
out["why"] = WHY_LOGIC
|
||||
if prices is None and (prices := fixed_price(cover.get("price"))) is not None:
|
||||
out["fixed"] = True # 707-1 연결이 단가를 못 냄 = 고정 단가 칸
|
||||
if num(cover.get("src_qty")) is not None:
|
||||
out["src_qty"] = num(cover["src_qty"])
|
||||
return out, prices
|
||||
|
||||
|
||||
|
||||
@@ -13,6 +13,8 @@
|
||||
* 단가 = `refPrice(ref)` 하나(dg 대가 목록 · sg 산근 목록 · logic = 서버 sg_hand 산근(702-3 연결 즉시 뒤에서 셈) · 그 밖 hopyo)
|
||||
* 자유 편집(184) = `user.bill` v2 덮개 {v: 2, rows: [표 차례 · 고친 칸 · 새 줄], gone: [지운 자동 줄]} ·
|
||||
* 옛 꼴(qty · 손 줄 after · parent) 그대로 읽음 · 저장 꼴 · 출력 줄 칸(계약) = 짝 .py 머리
|
||||
* 고정 단가 칸(707-1) = 줄 price {labor, material, expense} — 연결이 단가를 못 내면 그것으로 금액(줄 fixed) ·
|
||||
* src_qty = 불러온 수량(비교 칸)
|
||||
* 출력 줄 칸 = kind · key · name · parent · level · no · cost · auto | hand · auto_{칸} · section
|
||||
* ========================================================================== */
|
||||
|
||||
@@ -79,6 +81,8 @@ function itemRow(item: Row, prices: Row | null): Row {
|
||||
ref: item.ref ?? null,
|
||||
...Object.fromEntries(Object.entries(item).filter(([k]) => k.startsWith("auto_"))),
|
||||
...(item.hand ? { hand: true } : { auto: true }),
|
||||
...(item.fixed ? { fixed: true } : {}),
|
||||
...(num(item.src_qty) !== null ? { src_qty: item.src_qty } : {}),
|
||||
...amounts(qty, prices),
|
||||
};
|
||||
row._gaps = [];
|
||||
@@ -134,15 +138,31 @@ export function refPrice(ref: Row | null, books: Row, hopyo: Row, rowKey = ""):
|
||||
return key ? (hopyo[key] ?? null) : null;
|
||||
}
|
||||
|
||||
/** 새 줄(hand:) → [항목, 단가, 까닭] — 연결 = 고른 호표(로직 = 새 산근 키로 풀어 둠) */
|
||||
/** 고정 단가 칸(707-1) → 1 단위 단가 — 빈 비목 0 · 칸 없음 = null */
|
||||
export function fixedPrice(price: unknown): Row | null {
|
||||
if (!price || typeof price !== "object" || Array.isArray(price)) return null;
|
||||
const out: Row = { no: "" };
|
||||
for (const p of PARTS) out[p] = num((price as Row)[p]) || 0;
|
||||
out.sum = out.labor + out.material + out.expense;
|
||||
return out;
|
||||
}
|
||||
|
||||
/** 새 줄(hand:) → [항목, 단가, 까닭] — 연결 = 고른 호표(로직 = 새 산근 키로 풀어 둠) ·
|
||||
* 연결이 단가를 못 내면 고정 단가 칸(707-1) */
|
||||
export function handItems(user: Row, books: Row, hopyo: Row): Array<[Row, Row | null, string]> {
|
||||
const out: Array<[Row, Row | null, string]> = [];
|
||||
for (const hand of (user.bill || {}).rows || []) {
|
||||
if (!String(hand.key || "").startsWith("hand:")) continue;
|
||||
const price = refPrice(hand.ref ?? null, books, hopyo, hand.key);
|
||||
let price = refPrice(hand.ref ?? null, books, hopyo, hand.key);
|
||||
const ref = price?.tab ? { tab: price.tab, key: price.key } : null;
|
||||
let fixed = false;
|
||||
if (price === null) {
|
||||
price = fixedPrice(hand.price);
|
||||
fixed = price !== null;
|
||||
}
|
||||
const why = hand.ref?.tab === "logic" ? WHY_LOGIC : WHY_HAND;
|
||||
out.push([{ ...hand, hand: true, note: price?.no || "", ref }, price, why]);
|
||||
const note = price?.no || hand.note || "";
|
||||
out.push([{ ...hand, hand: true, note, ref, fixed }, price, why]);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
@@ -171,6 +191,11 @@ function covered(
|
||||
out.note = prices?.no || "";
|
||||
if (cover.ref?.tab === "logic") out.why = WHY_LOGIC;
|
||||
}
|
||||
if (prices === null) {
|
||||
prices = fixedPrice(cover.price); // 707-1 연결이 단가를 못 냄 = 고정 단가 칸
|
||||
if (prices !== null) out.fixed = true;
|
||||
}
|
||||
if (num(cover.src_qty) !== null) out.src_qty = num(cover.src_qty);
|
||||
return [out, prices];
|
||||
}
|
||||
|
||||
|
||||
@@ -41,6 +41,7 @@
|
||||
지난 공사 일위대가 → 이 공사 서고 사본(덮개째) → {폴더, 만듦, key, cover, 덮개, payload} ·
|
||||
덮개 = 그 공사 저장 칸(화면 캐시 cover 가 비었을 때만 얹음 · 같은 꼴 블록이 먼저 있으면 빈 것) ·
|
||||
같은 회사만(남의 회사 403)
|
||||
경쟁사 내역 줄 불러오기(707-1) = `B09_Estimation_Router_Import`.
|
||||
운반 사본 연결 = 셈 때마다 거리로 사본 다시 고름(`Bill_Library.resolve_hauls` · `fix_keys`).
|
||||
|
||||
저장 자리 = 프로젝트 안 `B09_Estimation/v1/`(`B09_Estimation_Store`) ·
|
||||
@@ -85,6 +86,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_Import import router as import_router
|
||||
from B09_Estimation.B09_Estimation_Router_Quotes import router as quotes_router
|
||||
from B09_Estimation.B09_Estimation_Summary import compute_summary, hopyo_lines, used_materials
|
||||
from common_util.common_util_auth import verify_session
|
||||
@@ -554,8 +556,9 @@ async def past_import(
|
||||
return {**made, "key": key, "cover": cover, "payload": out}
|
||||
|
||||
|
||||
# 166-4 견적서 파일 길 — 같은 머리 · 같은 권한
|
||||
# 166-4 견적서 파일 길 · 707-1 경쟁사 내역 줄 불러오기 길 — 같은 머리 · 같은 권한
|
||||
router.include_router(quotes_router)
|
||||
router.include_router(import_router)
|
||||
|
||||
|
||||
# 302 초안 길(draft/*_Router.py 의 router · 폴더째 지우면 길도 없음) — 같은 머리 · 같은 권한
|
||||
|
||||
@@ -0,0 +1,190 @@
|
||||
"""B09 경쟁사 내역 줄 불러오기 길(PLAN 707-1) — 저장 안 함(화면 캐시로 → [저장] 때 서버 정본).
|
||||
|
||||
POST /api/projects/{id}/estimation/import/bill (multipart: file | file_key · user?)
|
||||
읽기 + 미리 보기 — 파일 = 프로젝트 `B09_Estimation/v1/import/{표지}/`
|
||||
(file_key = 표지 · 같은 파일 다시 씀) → {file_key, 파일, 꼴, 시트, 묶음, 항목, 합계,
|
||||
조달수수료율, 합계줄, 율줄, 같은이름, 남는자동, 이음} ·
|
||||
user = 화면 캐시(JSON 글 · 없으면 저장 값) · DRM = 400 「읽을 수 없음」
|
||||
POST /api/projects/{id}/estimation/import/bill/apply {file_key, mode, attach, drop, user?}
|
||||
짜기(`B09_Estimation_Bill_Import.apply`) →
|
||||
{user(새 캐시), 알림, 어긋남, 엑셀합계, payload(그 캐시로 셈)} ·
|
||||
이음 = 읽을 수 있는 층(개인 → 회사 → 시스템)에서 같은 파일 이름으로 들여온 실무자료 사전조합
|
||||
(본뜬 「HP-… 제N호표」) → 서고 사본(층 쓰기 0 · 프로젝트 사본만) · 대가 N 줄 연결.
|
||||
`B09_Estimation_Router` 가 끼워 넣음(같은 머리 · 같은 권한).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
|
||||
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_Store as store
|
||||
from B09_Estimation.B09_Estimation_Bill_Excel import BillExcelError, read_bill
|
||||
from B09_Estimation.B09_Estimation_Labor_Source import daega_inputs
|
||||
from common_util.common_util_auth import verify_session
|
||||
from M02_MasterTemplete import M02_Hopyo_Build as hopyo_build
|
||||
from M02_MasterTemplete import M02_Labor_Ref as refs
|
||||
from M02_MasterTemplete import M02_Layer_Access as access
|
||||
from M02_MasterTemplete.M02_Item_Snapshot import importer as who_of
|
||||
|
||||
router = APIRouter()
|
||||
PATH = "/{project_id}/estimation/import/bill"
|
||||
IMPORT = "import"
|
||||
MAX_BYTES = 50 * 1024 * 1024
|
||||
_TAG = re.compile(r"^[0-9a-f]{8}$")
|
||||
_MARK = re.compile(r"^HP-\d{6} 제(\d+)호표$")
|
||||
|
||||
|
||||
def _saved(root: Path, key: str) -> Path:
|
||||
"""file_key → 둔 파일 — 없으면 404."""
|
||||
folder = store.folder(root) / IMPORT / key
|
||||
files = (
|
||||
[f for f in folder.iterdir() if f.is_file()] if _TAG.match(key) and folder.is_dir() else []
|
||||
)
|
||||
if not files:
|
||||
raise HTTPException(status_code=404, detail="불러온 파일이 없음 — 다시 고름")
|
||||
return files[0]
|
||||
|
||||
|
||||
def _keep(root: Path, name: str, data: bytes) -> str:
|
||||
key = importer.tag_of(data)
|
||||
folder = store.folder(root) / IMPORT / key
|
||||
folder.mkdir(parents=True, exist_ok=True)
|
||||
for old in folder.iterdir():
|
||||
old.unlink()
|
||||
(folder / (store.safe_name(name) or "내역.xlsx")).write_bytes(data)
|
||||
return key
|
||||
|
||||
|
||||
def _parse(path: Path) -> dict[str, Any]:
|
||||
try:
|
||||
return read_bill(path.name, path.read_bytes())
|
||||
except BillExcelError as error:
|
||||
raise HTTPException(status_code=400, detail=str(error)) from error
|
||||
|
||||
|
||||
def _draft(raw: Any) -> dict[str, Any] | None:
|
||||
if raw in (None, ""):
|
||||
return None
|
||||
try:
|
||||
user = json.loads(raw) if isinstance(raw, str) else raw
|
||||
except ValueError as error:
|
||||
raise HTTPException(status_code=400, detail="user 가 JSON 이 아님") from error
|
||||
return store.clean_user(user) if isinstance(user, dict) else None
|
||||
|
||||
|
||||
def _found(session: dict[str, Any], name: str, numbers: set[int]) -> dict[int, tuple]:
|
||||
"""대가 번 → (층, 주인, 사전조합 키) — 같은 파일 이름 실무자료 본뜬 · 앞 층이 이김."""
|
||||
out: dict[int, tuple] = {}
|
||||
for layer in ("personal", "company", "system"):
|
||||
try:
|
||||
got = access.grant(session, layer)
|
||||
except HTTPException: # 회사 없음 등 — 그 층 건너뜀
|
||||
continue
|
||||
for key, origin, _raw in hopyo_build._origins(refs.folder_of(got.layer, got.owner)):
|
||||
hit = _MARK.match(str(origin.get("키") or ""))
|
||||
no = int(hit.group(1)) if hit else None
|
||||
if origin.get("파일") == name and no in numbers and no not in out:
|
||||
out[no] = (got.layer, got.owner, key)
|
||||
return out
|
||||
|
||||
|
||||
def _links(root: Path, found: dict[int, tuple], who: Any) -> dict[int, dict[str, Any]]:
|
||||
folders = {}
|
||||
for no, (layer, owner, key) in found.items():
|
||||
try:
|
||||
folders[no] = library.import_copy(root, "precombo", layer, owner, key, who)["폴더"]
|
||||
except (LookupError, ValueError): # 원본 없음 · 모양 틀림 = 고정 단가로
|
||||
continue
|
||||
libs = (daega_inputs(root).get("libs") or {}) if folders else {}
|
||||
return {
|
||||
no: {"tab": "dg", "key": libs[name], "lib": name}
|
||||
for no, name in folders.items()
|
||||
if libs.get(name)
|
||||
}
|
||||
|
||||
|
||||
def _numbers(parsed: dict[str, Any]) -> set[int]:
|
||||
return {
|
||||
int(h["번"])
|
||||
for line in parsed["줄"]
|
||||
if (h := line.get("호표") or {}).get("종류") == "대가" and "번" in h
|
||||
}
|
||||
|
||||
|
||||
@router.post(PATH)
|
||||
async def import_bill(
|
||||
project_id: UUID,
|
||||
file: UploadFile | None = File(None),
|
||||
file_key: str | None = Form(None),
|
||||
user: str | None = Form(None),
|
||||
session: dict[str, Any] = Depends(verify_session),
|
||||
) -> dict[str, Any]:
|
||||
from B09_Estimation.B09_Estimation_Router import payload # 서로 부름 — 부를 때만
|
||||
|
||||
root = await _project_root(project_id)
|
||||
if file is not None:
|
||||
data = await file.read(MAX_BYTES + 1)
|
||||
if len(data) > MAX_BYTES:
|
||||
raise HTTPException(status_code=400, detail="파일이 너무 큼(50MB 넘음)")
|
||||
file_key = await asyncio.to_thread(_keep, root, file.filename or "", data)
|
||||
if not file_key:
|
||||
raise HTTPException(status_code=400, detail="파일을 고름")
|
||||
path = _saved(root, file_key)
|
||||
parsed = await asyncio.to_thread(_parse, path)
|
||||
rows = (await payload(project_id, root, _draft(user), True))["sheets"]["bill"]["rows"]
|
||||
found = await asyncio.to_thread(_found, session, path.name, _numbers(parsed))
|
||||
return {
|
||||
"file_key": file_key,
|
||||
"파일": path.name,
|
||||
**importer.preview(parsed, rows),
|
||||
"이음": len(found),
|
||||
}
|
||||
|
||||
|
||||
class ApplyBody(BaseModel):
|
||||
file_key: str = Field(min_length=8, max_length=8)
|
||||
mode: str = "take"
|
||||
attach: bool = True
|
||||
drop: bool = False
|
||||
user: dict[str, Any] | None = None
|
||||
|
||||
|
||||
@router.post(PATH + "/apply")
|
||||
async def apply_bill(
|
||||
project_id: UUID, body: ApplyBody, session: dict[str, Any] = Depends(verify_session)
|
||||
) -> dict[str, Any]:
|
||||
from B09_Estimation.B09_Estimation_Router import payload # 서로 부름 — 부를 때만
|
||||
|
||||
if body.mode not in importer.MODES:
|
||||
raise HTTPException(status_code=400, detail="수량 갈래는 take · price 중 하나")
|
||||
root = await _project_root(project_id)
|
||||
path = _saved(root, body.file_key)
|
||||
parsed = await asyncio.to_thread(_parse, path)
|
||||
draft = _draft(body.user) or await asyncio.to_thread(store.read_user, root)
|
||||
rows = (await payload(project_id, root, draft, True))["sheets"]["bill"]["rows"]
|
||||
found = await asyncio.to_thread(_found, session, path.name, _numbers(parsed))
|
||||
links = await asyncio.to_thread(_links, root, found, who_of(session))
|
||||
made, report = importer.apply(
|
||||
parsed, draft, rows, body.file_key, body.mode, body.attach, body.drop, links
|
||||
)
|
||||
made = store.clean_user(made)
|
||||
out = await payload(project_id, root, made, True)
|
||||
lines = out["sheets"]["bill"]["rows"]
|
||||
return {
|
||||
"user": made,
|
||||
"알림": {k: v for k, v in report.items() if k != "줄"},
|
||||
"어긋남": importer.differences(parsed, report, lines),
|
||||
"엑셀합계": parsed["합계"],
|
||||
"payload": out,
|
||||
}
|
||||
@@ -152,9 +152,22 @@ def _clean_ref(ref: Any) -> dict[str, Any] | None:
|
||||
return None
|
||||
|
||||
|
||||
def _fixed_cells(row: dict[str, Any]) -> dict[str, Any]:
|
||||
"""707-1 고정 단가 칸 `price` {노 · 재 · 경 수(빈 = 0)} · 불러온 수량 `src_qty` — 있는 것만."""
|
||||
out: dict[str, Any] = {}
|
||||
if isinstance(row.get("price"), dict):
|
||||
out["price"] = {
|
||||
p: _num(row["price"].get(p)) or 0.0 for p in ("labor", "material", "expense")
|
||||
}
|
||||
if _num(row.get("src_qty")) is not None:
|
||||
out["src_qty"] = _num(row["src_qty"])
|
||||
return out
|
||||
|
||||
|
||||
def _clean_v2(raw: dict[str, Any]) -> dict[str, Any]:
|
||||
"""184 덮개 v2 — 표 차례(겹친 키 · `fee` 버림) · 새 줄(`hand:`) = 모든 칸 ·
|
||||
자동 줄 = 고친 칸만(값 · null = 일부러 빈칸 · 못 쓰는 값 = 칸 버림) · `gone` = 자동 줄 키."""
|
||||
자동 줄 = 고친 칸만(값 · null = 일부러 빈칸 · 못 쓰는 값 = 칸 버림) · `gone` = 자동 줄 키 ·
|
||||
고정 단가 칸 · 불러온 수량(707-1) = 항목 줄만 · 새 줄 비고 `note` 글."""
|
||||
rows, seen = [], set()
|
||||
for row in raw.get("rows") if isinstance(raw.get("rows"), list) else []:
|
||||
key = row.get("key") if isinstance(row, dict) else None
|
||||
@@ -174,6 +187,10 @@ def _clean_v2(raw: dict[str, Any]) -> dict[str, Any]:
|
||||
out.update({k: _text(row.get(k)) for k in ("name", "spec", "unit")})
|
||||
out["qty"] = None if group else _num(row.get("qty"))
|
||||
out["ref"] = None if group else _clean_ref(row.get("ref"))
|
||||
if not group:
|
||||
out.update(_fixed_cells(row))
|
||||
if _text(row.get("note")):
|
||||
out["note"] = _text(row["note"])
|
||||
else:
|
||||
for k in ("name", "spec", "unit"):
|
||||
if k in row:
|
||||
@@ -182,6 +199,7 @@ def _clean_v2(raw: dict[str, Any]) -> dict[str, Any]:
|
||||
out["qty"] = _num(row["qty"])
|
||||
if "ref" in row and (row["ref"] is None or _clean_ref(row["ref"]) is not None):
|
||||
out["ref"] = _clean_ref(row["ref"])
|
||||
out.update(_fixed_cells(row))
|
||||
rows.append(out)
|
||||
given = raw.get("gone") if isinstance(raw.get("gone"), list) else []
|
||||
gone = list(
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
"""707-1 경쟁사 내역 줄 불러오기 — 읽기(`Bill_Excel`) · 짜기(`Bill_Import`) · 고정 단가 칸 셈 거울.
|
||||
|
||||
- 실무 xlsx(STmate 기번3 · 영월 · 봉화 · EST 거창) = 빠짐 0 · 불러온 뒤 직접비 합 = 엑셀 합 ·
|
||||
어긋난 줄 0 · 조달수수료 = 율(줄로 안 실음) · DRM = 「읽을 수 없음」
|
||||
- 수량 갈래 (가) take = 붙인 자동 줄 수량 엑셀 · auto_qty = B08 / (나) price = B08 · src_qty = 엑셀
|
||||
- 두 번 셈 막음 = 이름 같은 자동 줄에 붙임(한 번만 셈) · drop = 남은 자동 줄 지움 ·
|
||||
같은 파일 다시 = 갈아 끼움 · 조달수수료율 = 빈 때만
|
||||
- 고정 단가 칸 = 연결이 단가를 못 낼 때만 · 다듬기 · TS 짝 같은 값
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
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_Bill_Import as imp # noqa: E402
|
||||
from B09_Estimation import B09_Estimation_Store as store # noqa: E402
|
||||
from B09_Estimation.B09_Estimation_Bill_Excel import BillExcelError, read_bill # noqa: E402
|
||||
from B09_Estimation.B09_Estimation_Calc_Bill import compute_estimation # noqa: E402
|
||||
|
||||
K = ROOT / "resources" / "knowledge" / "original" / "실무문서"
|
||||
FILES = {
|
||||
"기번3": K / "2024년 간선임도(기번3-울진.대흥)(공통)/1. 설계원본/01 내역자료/2024년 간선임도사업(기번3).xlsx",
|
||||
"영월": K / "05. 동부청 영월/내역서/2024년 임도시설사업2차(산불진화_기번6)-영월.xlsx",
|
||||
"봉화": K / "04.산불 봉화/01. 내역서/2024년 산불진화임도 실시설계(기번41_현동)_ L=3.46km.xlsx",
|
||||
"거창": K / "01.산불거창(거창북상병곡)/01.내역서/00.25년_산불_거창(거창,북상,병곡,산8외)내역+수량.xlsx",
|
||||
} # fmt: skip
|
||||
#: (꼴, 항목 수, 엑셀 직접비 합)
|
||||
WANT = {
|
||||
"기번3": ("STmate", 136, 710359136),
|
||||
"영월": ("STmate", 108, 767148700),
|
||||
"봉화": ("STmate", 77, 781283415),
|
||||
"거창": ("EST", 60, 384916227),
|
||||
}
|
||||
EMPTY = {"materials": [], "bill": {"items": []}, "hopyo": {}}
|
||||
ITEMS = [
|
||||
{"key": "earth:4", "section": "1.2", "name": "절토(토사)", "unit": "㎥", "qty": 10,
|
||||
"ref": {"tab": "sg_list", "key": "D01341"}},
|
||||
{"key": "earth:9", "section": "1.6", "name": "노면 정리X", "unit": "㎡", "qty": 3,
|
||||
"ref": {"tab": "sg_list", "key": "D01341"}},
|
||||
] # fmt: skip
|
||||
INPUTS = {"materials": [], "bill": {"items": ITEMS}, "hopyo": {"D01341": {
|
||||
"labor": 1, "material": 0, "expense": 0}}} # fmt: skip
|
||||
IMPORTS = re.compile(r"^import[^;]*;\n", re.MULTILINE)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def parsed():
|
||||
return {k: read_bill(p.name, p.read_bytes()) for k, p in FILES.items()}
|
||||
|
||||
|
||||
def _rows(inputs, user):
|
||||
return compute_estimation(inputs, user, None, {})["bill"]
|
||||
|
||||
|
||||
def _load(got, inputs=EMPTY, user=None, **kw):
|
||||
user = user or {}
|
||||
made, report = imp.apply(got, user, _rows(inputs, user)["rows"], "abcd1234", **kw)
|
||||
made = store.clean_user(made)
|
||||
return made, report, _rows(inputs, made)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(FILES))
|
||||
def test_실무_내역_빠짐_0_합계_엑셀과_같음(parsed, name):
|
||||
got = parsed[name]
|
||||
kind, items, total = WANT[name]
|
||||
assert (got["꼴"], sum(1 for x in got["줄"] if not x["묶음"])) == (kind, items)
|
||||
assert got["합계"]["직접"] == total and got["조달수수료율"] == 0.54
|
||||
made, report, bill = _load(got)
|
||||
assert report["실음"] == len(got["줄"]) and report["붙임"] == 0
|
||||
assert bill["totals"]["direct"]["sum"] == total and bill["totals"]["direct"]["blank"] == 0
|
||||
assert imp.differences(got, report, bill["rows"]) == []
|
||||
assert made["estimate"]["procurement_fee_pct"] == 0.54
|
||||
assert max(r["level"] for r in bill["rows"]) <= 5
|
||||
keys = {r["key"] for r in bill["rows"]}
|
||||
assert all(k in keys for k in report["줄"].values()) # 실은 줄 모두 보임
|
||||
# 관급 = 제외(관급자재대 아래 · 직접비 밖)
|
||||
assert bill["totals"]["owner"]["net_material"] > 0
|
||||
|
||||
|
||||
def test_EST_비고_참조_STmate_코드(parsed):
|
||||
est = [x["호표"] for x in parsed["거창"]["줄"] if not x["묶음"]]
|
||||
assert {"종류": "산근", "번": 46} in est and {"종류": "대가", "번": 8} in est
|
||||
st = next(x for x in parsed["기번3"]["줄"] if x["이름"] == "규준틀설치")
|
||||
assert st["호표"] == {"종류": "대가", "번": 5, "코드": "B00194"}
|
||||
|
||||
|
||||
def test_DRM_읽을_수_없음():
|
||||
with pytest.raises(BillExcelError, match="읽을 수 없음"):
|
||||
read_bill("a.xlsx", b"\x9b DRMONE" + b"\x00" * 300)
|
||||
with pytest.raises(BillExcelError, match="엑셀"):
|
||||
read_bill("a.xlsx", b"PK\x03\x04 broken")
|
||||
|
||||
|
||||
def test_가_넘겨받기_붙인_자동_줄_수량_엑셀(parsed):
|
||||
made, report, bill = _load(parsed["기번3"], INPUTS, mode="take")
|
||||
rows = {r["key"]: r for r in bill["rows"]}
|
||||
assert report["붙임"] == 1 # 절토(토사) 첫 줄만(1공구)
|
||||
cut = rows["earth:4"]
|
||||
assert (cut["qty"], cut["auto_qty"], cut["src_qty"]) == (5675, 10, 5675)
|
||||
assert cut["fixed"] and cut["sum"] == 11003825 # 엑셀 단가 · 연결 덮개 null
|
||||
assert cut["parent"].startswith("hand:xabcd1234-") # 불러온 자리로 옮김
|
||||
assert sum(1 for r in bill["rows"] if r["name"] == "절토(토사)") == 3 # 두 번 셈 없음
|
||||
# 붙인 줄 금액 = 엑셀 · 안 붙은 자동 줄(노면 정리X) 그대로 = 직접비 = 엑셀 + 3
|
||||
assert bill["totals"]["direct"]["sum"] == 710359136 + 3
|
||||
|
||||
|
||||
def test_나_단가만_B08_이김_비교_칸(parsed):
|
||||
made, report, bill = _load(parsed["기번3"], INPUTS, mode="price")
|
||||
cut = {r["key"]: r for r in bill["rows"]}["earth:4"]
|
||||
assert (cut["qty"], cut["src_qty"], "auto_qty" in cut) == (10, 5675, False)
|
||||
assert cut["sum"] == 1939 * 10 # 엑셀 단가 × B08 수량
|
||||
diff = imp.differences(parsed["기번3"], report, bill["rows"])
|
||||
assert [d["까닭"] for d in diff] == ["수량 = B08(단가만)"]
|
||||
|
||||
|
||||
def test_붙이지_않음_지우기_다시_부르기(parsed):
|
||||
_, report, bill = _load(parsed["기번3"], INPUTS, attach=False, drop=True)
|
||||
assert (report["붙임"], report["지움"]) == (0, 2)
|
||||
assert bill["totals"]["direct"]["sum"] == 710359136
|
||||
made, _, _ = _load(parsed["기번3"], INPUTS)
|
||||
made["bill"]["rows"].append({"key": "hand:mine", "parent": "hand:xabcd1234-00007",
|
||||
"name": "내 줄", "qty": 1, "ref": None}) # fmt: skip
|
||||
again, report, bill = _load(parsed["기번3"], INPUTS, user=made)
|
||||
keys = [e["key"] for e in again["bill"]["rows"]]
|
||||
assert len(keys) == len(set(keys)) and report["붙임"] == 1
|
||||
assert bill["totals"]["direct"]["sum"] == 710359136 + 3 # 갈아 끼움 · 두 번 안 셈
|
||||
mine = next(e for e in again["bill"]["rows"] if e["key"] == "hand:mine")
|
||||
assert "parent" not in mine # 내 줄 = 남음(맨 위)
|
||||
|
||||
|
||||
def test_조달수수료율_빈_때만(parsed):
|
||||
made, _, _ = _load(parsed["거창"], user={"estimate": {"procurement_fee_pct": 1.0}})
|
||||
assert made["estimate"]["procurement_fee_pct"] == 1.0
|
||||
|
||||
|
||||
def test_고정_단가_칸_다듬기_연결이_이김():
|
||||
raw = {"bill": {"v": 2, "rows": [
|
||||
{"key": "hand:a", "name": "a", "qty": 2, "ref": {"tab": "sg", "key": "SG:1"},
|
||||
"price": {"labor": 5, "material": "x"}, "src_qty": 3, "note": "대가 5호표"},
|
||||
{"key": "earth:4", "price": {"expense": 7}, "src_qty": "y"},
|
||||
]}} # fmt: skip
|
||||
user = store.clean_user(raw)
|
||||
hand, auto = user["bill"]["rows"]
|
||||
assert hand["price"] == {"labor": 5, "material": 0, "expense": 0} and hand["src_qty"] == 3
|
||||
assert hand["note"] == "대가 5호표" and auto == {"key": "earth:4", "price": {
|
||||
"labor": 0, "material": 0, "expense": 7}} # fmt: skip
|
||||
books = {"sg_list": [{"no": 1, "key": "SG:1", "labor": 100, "material": 0, "expense": 0,
|
||||
"sum": 100}]} # fmt: skip
|
||||
rows = {r["key"]: r for r in compute_estimation(INPUTS, user, None, books)["bill"]["rows"]}
|
||||
assert (rows["hand:a"]["sum"], rows["hand:a"].get("fixed")) == (200, None) # 연결 이김
|
||||
assert rows["earth:4"]["sum"] == 10 and "fixed" not in rows["earth:4"] # 자동 연결 이김
|
||||
rows = {r["key"]: r for r in compute_estimation(INPUTS, user, None, {})["bill"]["rows"]}
|
||||
assert (rows["hand:a"]["sum"], rows["hand:a"]["fixed"]) == (10, True)
|
||||
assert rows["hand:a"]["note"] == "대가 5호표"
|
||||
|
||||
|
||||
def test_거울_TS_가_같은_값(parsed):
|
||||
calc = "".join(
|
||||
IMPORTS.sub("", (ROOT / "B09_Estimation" / f"B09_Estimation_{n}.ts").read_text("utf-8"))
|
||||
for n in ("Calc_Base", "Calc_Bill")
|
||||
)
|
||||
users = [
|
||||
_load(parsed["거창"], INPUTS, mode="take")[0],
|
||||
_load(parsed["영월"], INPUTS, mode="price")[0],
|
||||
{"bill": {"v": 2, "rows": [{"key": "earth:4", "ref": None, "price": {"labor": 3}},
|
||||
{"key": "hand:a", "qty": 2, "price": {"material": 4}}]}},
|
||||
] # fmt: skip
|
||||
body = (
|
||||
f"\nconst i = {json.dumps(INPUTS, ensure_ascii=False)};"
|
||||
f"\nconst us = {json.dumps(users, ensure_ascii=False)};"
|
||||
"\nconsole.log(JSON.stringify(us.map((u) => computeEstimation(i, u, null, {}))));"
|
||||
)
|
||||
result = subprocess.run(
|
||||
["node", "--experimental-strip-types", "--no-warnings",
|
||||
"--input-type=module-typescript"],
|
||||
input=calc + body, capture_output=True, text=True, encoding="utf-8", timeout=60,
|
||||
) # fmt: skip
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert json.loads(result.stdout) == [compute_estimation(INPUTS, u, None, {}) for u in users]
|
||||
Reference in New Issue
Block a user