693 lines
25 KiB
Python
693 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""STMATE 원본 DBF / SMT / STC / JB_R 100% 무손실 직접 읽기·쓰기 연동 서버 (server.py)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pathlib
|
|
import re
|
|
import zipfile
|
|
from decimal import Decimal, ROUND_DOWN
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel
|
|
|
|
from excel_builder import build_stmate_full_xlsx
|
|
from smt_parser import (
|
|
build_dbf_bytes,
|
|
extract_costn_memos,
|
|
load_stmate_catalog,
|
|
parse_dbf_bytes,
|
|
parse_prjlst_smt,
|
|
rebuild_costn_with_smt,
|
|
update_prjlst_smt_bytes,
|
|
)
|
|
|
|
HERE = pathlib.Path(__file__).resolve().parent
|
|
REPO_ROOT = HERE.parent.parent
|
|
WORK_DBF_ROOT = HERE / "work_dbf"
|
|
STMATE_ROOT = pathlib.Path(r"C:\경영정보(주)")
|
|
STMATE_SYS_BSC = STMATE_ROOT / "System" / "ST_BSC"
|
|
PRAC_DOCS_ROOT = REPO_ROOT / "resources" / "knowledge" / "original" / "실무문서"
|
|
|
|
STRUCT_RE = re.compile(r"^(T\d|E\d|C\d|:)$")
|
|
|
|
app = FastAPI(title="STMATE w 8.0 Full DBF/SMT Engine")
|
|
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
|
|
|
|
|
|
def fnum(val: object, default: float = 0.0) -> float:
|
|
try:
|
|
s = str(val or "").strip().replace(",", "")
|
|
return float(s) if s else default
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def discover_sources() -> dict:
|
|
projects: list[dict] = []
|
|
if STMATE_ROOT.exists():
|
|
for costn in sorted(STMATE_ROOT.rglob("COSTN.DBF")):
|
|
pdir = costn.parent
|
|
projects.append(
|
|
{
|
|
"id": str(pdir),
|
|
"name": f"[PC 폴더] {pdir.name}",
|
|
"path": str(pdir),
|
|
"kind": "folder",
|
|
}
|
|
)
|
|
for stc in sorted(STMATE_ROOT.rglob("*.STC")):
|
|
projects.append(
|
|
{"id": str(stc), "name": f"[PC STC] {stc.name}", "path": str(stc), "kind": "stc"}
|
|
)
|
|
if PRAC_DOCS_ROOT.exists():
|
|
for stc in sorted(PRAC_DOCS_ROOT.rglob("*.STC")):
|
|
projects.append(
|
|
{"id": str(stc), "name": f"[실무 STC] {stc.stem}", "path": str(stc), "kind": "stc"}
|
|
)
|
|
if WORK_DBF_ROOT.exists():
|
|
for costn in sorted(WORK_DBF_ROOT.rglob("COSTN.DBF")):
|
|
pdir = costn.parent
|
|
projects.append(
|
|
{
|
|
"id": str(pdir),
|
|
"name": f"[작업본 DBF] {pdir.name}",
|
|
"path": str(pdir),
|
|
"kind": "work_folder",
|
|
}
|
|
)
|
|
jbr_files = (
|
|
[
|
|
{"name": p.name, "path": str(p), "ext": p.suffix.upper(), "size": p.stat().st_size}
|
|
for p in sorted(STMATE_SYS_BSC.glob("*.JB_R*"))
|
|
]
|
|
if STMATE_SYS_BSC.exists()
|
|
else []
|
|
)
|
|
return {"projects": projects, "jbr_files": jbr_files, "work_root": str(WORK_DBF_ROOT)}
|
|
|
|
|
|
def read_project_files(source_path: str) -> dict[str, bytes]:
|
|
p = pathlib.Path(source_path)
|
|
if not p.exists():
|
|
raise HTTPException(status_code=404, detail=f"경로를 찾을 수 없음: {source_path}")
|
|
files: dict[str, bytes] = {}
|
|
if p.is_file() and p.suffix.upper() == ".STC":
|
|
with zipfile.ZipFile(p, "r") as zf:
|
|
for nm in zf.namelist():
|
|
files[pathlib.Path(nm).name.upper()] = zf.read(nm)
|
|
elif p.is_dir():
|
|
for fp in p.iterdir():
|
|
if fp.is_file():
|
|
files[fp.name.upper()] = fp.read_bytes()
|
|
else:
|
|
raise HTTPException(status_code=400, detail="지원하지 않는 경로 형식")
|
|
return files
|
|
|
|
|
|
def _make_syn_resource(res_id: str, name: str, spec: str, unit: str, kind: str, val: float) -> dict:
|
|
return {
|
|
"id": res_id,
|
|
"code": res_id,
|
|
"name": f"{name} ({kind})",
|
|
"spec": spec,
|
|
"unit": unit,
|
|
"kind": kind,
|
|
"candidates": [val, 0, 0, 0, 0],
|
|
"adoptMode": "manual",
|
|
"manualPrice": val,
|
|
"convertHourly": False,
|
|
"isUsd": False,
|
|
"isSynthetic": True,
|
|
}
|
|
|
|
|
|
def convert_dbf_to_sim_state(
|
|
costn_rows: list[dict],
|
|
boq_rows: list[dict],
|
|
smt_kv: dict[str, str],
|
|
prj_row: dict,
|
|
costn_memos: dict[str, dict] | None = None,
|
|
) -> dict:
|
|
"""원본 COSTN.DBF(전수) / COSTN.SMT / BOQ11~14.DBF(전수) / PRJLST.SMT를 무손실 변환."""
|
|
costn_memos = costn_memos or {}
|
|
resources: list[dict] = []
|
|
machines: list[dict] = []
|
|
san_items: list[dict] = []
|
|
ilwi_items: list[dict] = []
|
|
|
|
for r in costn_rows:
|
|
code = (r.get("CODE") or "").strip()
|
|
if not code:
|
|
continue
|
|
prefix = code[0].upper()
|
|
name, spec, unit = r.get("NAME") or code, r.get("SIZE") or "", r.get("UNIT") or ""
|
|
t1, l1, m1, s1 = (
|
|
fnum(r.get("TAMT1")),
|
|
fnum(r.get("LAMT1")),
|
|
fnum(r.get("MAMT1")),
|
|
fnum(r.get("SAMT1")),
|
|
)
|
|
cands = [fnum(r.get(f"AMT_{i}1")) for i in range(1, 6)]
|
|
amt61 = t1 or fnum(r.get("AMT_61"))
|
|
memo_obj = costn_memos.get(code)
|
|
|
|
if prefix in ("M", "L", "S"):
|
|
kind = "노무비" if prefix == "L" else ("경비" if prefix == "S" else "재료비")
|
|
non_zero = [x for x in cands if x > 0]
|
|
resources.append(
|
|
{
|
|
"id": code,
|
|
"code": code,
|
|
"name": name,
|
|
"spec": spec,
|
|
"unit": unit,
|
|
"kind": kind,
|
|
"candidates": cands if any(cands) else [amt61, 0, 0, 0, 0],
|
|
"adoptMode": "min"
|
|
if (non_zero and abs(min(non_zero) - amt61) < 1e-4)
|
|
else "manual",
|
|
"manualPrice": amt61,
|
|
"convertHourly": False,
|
|
"isUsd": (r.get("DOLLOR") or "").upper() == "T",
|
|
"jdCode": r.get("JDCODE") or "",
|
|
"uc": r.get("UC") or "F",
|
|
}
|
|
)
|
|
else:
|
|
if prefix == "X":
|
|
machines.append(
|
|
{
|
|
"id": code,
|
|
"code": code,
|
|
"name": name,
|
|
"spec": spec,
|
|
"unit": unit or "Hr",
|
|
"price": t1,
|
|
"hourlyL": l1,
|
|
"hourlyM": m1,
|
|
"hourlyS": s1,
|
|
"hourlyT": t1,
|
|
"smtText": memo_obj["text"] if memo_obj else "",
|
|
}
|
|
)
|
|
elif prefix == "D" or memo_obj:
|
|
vmap = memo_obj["vars"] if memo_obj else {}
|
|
ref_x = next(
|
|
(
|
|
rf["code"]
|
|
for rf in (memo_obj["refs"] if memo_obj else [])
|
|
if rf["code"].startswith("X")
|
|
),
|
|
"",
|
|
)
|
|
san_items.append(
|
|
{
|
|
"id": code,
|
|
"code": code,
|
|
"name": name,
|
|
"spec": spec,
|
|
"unit": unit,
|
|
"machineId": ref_x,
|
|
"q": fnum(vmap.get("q"), 0.7),
|
|
"K": fnum(vmap.get("K") or vmap.get("k"), 0.9),
|
|
"fExpr": str(vmap.get("f") or vmap.get("F") or "1.0"),
|
|
"EExpr": str(vmap.get("E") or vmap.get("E1") or "0.55"),
|
|
"Cm": fnum(vmap.get("Cm"), 20.0),
|
|
"cmFormula": memo_obj["cmFormula"] if memo_obj else "",
|
|
"qFormula": (memo_obj["qFormula"] if memo_obj else "") or "3600*q*K*f*E/Cm",
|
|
"smtText": memo_obj["text"] if memo_obj else "",
|
|
"smtRefs": memo_obj["refs"] if memo_obj else [],
|
|
"fixedM": m1,
|
|
"fixedL": l1,
|
|
"fixedS": s1,
|
|
"fixedT": t1,
|
|
}
|
|
)
|
|
|
|
lines = []
|
|
for comp_tag, comp_kind, val in (
|
|
("M", "재료비", m1),
|
|
("L", "노무비", l1),
|
|
("S", "경비", s1),
|
|
):
|
|
if val > 0:
|
|
res_id = f"{code}_{comp_tag}"
|
|
resources.append(_make_syn_resource(res_id, name, spec, unit, comp_kind, val))
|
|
lines.append({"refId": res_id, "qty": 1.0, "halPercent": 0, "devd": False})
|
|
ilwi_items.append(
|
|
{
|
|
"id": code,
|
|
"code": code,
|
|
"name": f"[{prefix}] {name}",
|
|
"rawName": name,
|
|
"prefix": prefix,
|
|
"spec": spec,
|
|
"unit": unit,
|
|
"fixedM": m1,
|
|
"fixedL": l1,
|
|
"fixedS": s1,
|
|
"fixedT": t1,
|
|
"smtText": memo_obj["text"] if memo_obj else "",
|
|
"smtRefs": memo_obj["refs"] if memo_obj else [],
|
|
"lines": lines,
|
|
}
|
|
)
|
|
|
|
excl_span: set[int] = set()
|
|
for i, r in enumerate(boq_rows):
|
|
bd = (r.get("BDCODE1") or "").strip()
|
|
if bd.startswith("C"):
|
|
end_code = r.get("ENDCODE")
|
|
start_idx = next(
|
|
(j for j, q in enumerate(boq_rows) if q.get("SORTCODE") == end_code), None
|
|
)
|
|
if start_idx is not None:
|
|
for k in range(min(start_idx, i), max(start_idx, i) + 1):
|
|
excl_span.add(k)
|
|
|
|
boq_items: list[dict] = []
|
|
known_ids = {x["id"] for x in resources} | {x["id"] for x in ilwi_items}
|
|
for i, r in enumerate(boq_rows):
|
|
bd = (r.get("BDCODE1") or "").strip()
|
|
if not bd or STRUCT_RE.match(bd):
|
|
continue
|
|
q1 = fnum(r.get("QTY1"))
|
|
q2 = fnum(r.get("QTY2")) if r.get("QTY2") else q1
|
|
name1, spec1, unit1 = r.get("NAME1") or bd, r.get("SIZE1") or "", r.get("UNIT1") or ""
|
|
if bd not in known_ids:
|
|
m1, l1, s1 = fnum(r.get("MAMT1")), fnum(r.get("LAMT1")), fnum(r.get("SAMT1"))
|
|
lines = []
|
|
for ctag, ckind, cval in (("M", "재료비", m1), ("L", "노무비", l1), ("S", "경비", s1)):
|
|
if cval > 0:
|
|
rid = f"{bd}_{ctag}"
|
|
resources.append(_make_syn_resource(rid, name1, spec1, unit1, ckind, cval))
|
|
lines.append({"refId": rid, "qty": 1.0, "halPercent": 0, "devd": False})
|
|
ilwi_items.append(
|
|
{
|
|
"id": bd,
|
|
"code": bd,
|
|
"name": name1,
|
|
"rawName": name1,
|
|
"prefix": bd[0].upper(),
|
|
"spec": spec1,
|
|
"unit": unit1,
|
|
"fixedM": m1,
|
|
"fixedL": l1,
|
|
"fixedS": s1,
|
|
"fixedT": fnum(r.get("TAMT1")),
|
|
"lines": lines,
|
|
}
|
|
)
|
|
known_ids.add(bd)
|
|
|
|
row_type = "normal"
|
|
if i in excl_span:
|
|
row_type = "gov_fee" if ("수수료" in name1 or fnum(r.get("S_AMT1")) > 0) else "gov_mat"
|
|
is_pct_fee = unit1.endswith("%") and fnum(r.get("T_AMT1")) > 0
|
|
if is_pct_fee:
|
|
fee_amt = fnum(r.get("T_AMT1"))
|
|
for res in resources:
|
|
if res["id"] == bd and res.get("manualPrice", 0) == 0:
|
|
res["manualPrice"] = fee_amt
|
|
res["candidates"][0] = fee_amt
|
|
q1, q2 = 1.0, 1.0
|
|
boq_items.append(
|
|
{
|
|
"id": f"BOQ_{i + 1}",
|
|
"code": bd,
|
|
"itnum": r.get("ITNUM") or "",
|
|
"name": name1,
|
|
"spec": spec1,
|
|
"unit": unit1,
|
|
"refId": bd,
|
|
"rowType": row_type,
|
|
"qty": q1,
|
|
"changedQty": q2 if q2 > 0 else q1,
|
|
"_recno": r.get("_recno", i),
|
|
}
|
|
)
|
|
|
|
rates = {
|
|
"KANNO": fnum(smt_kv.get("1_R_KANNO"), 14.5),
|
|
"SANBO": fnum(smt_kv.get("1_R_SANBO"), 3.56),
|
|
"KOBO": fnum(smt_kv.get("1_R_KOBO"), 1.01),
|
|
"KUNBO": fnum(smt_kv.get("1_R_KUNBO"), 3.545),
|
|
"SLVBO": fnum(smt_kv.get("1_R_SLVBO"), 12.95),
|
|
"YUNBO": fnum(smt_kv.get("1_R_YUNBO"), 4.5),
|
|
"TBUGUM": fnum(smt_kv.get("1_R_TBUGUM"), 0.0),
|
|
"ANJUN": fnum(smt_kv.get("1_R_ANJUN"), 1.86),
|
|
"KITA": fnum(smt_kv.get("1_R_T_KITA"), 6.0),
|
|
"HWAN": fnum(smt_kv.get("1_R_HWAN"), 0.8),
|
|
"EQIP": fnum(smt_kv.get("1_R_EQIPSU"), 0.4),
|
|
"ILBAN": fnum(smt_kv.get("1_R_ILBAN"), 6.0),
|
|
"IYUN": fnum(smt_kv.get("1_R_IYUN"), 15.0),
|
|
"S_SUDO": fnum(smt_kv.get("1_R_S_SUDO"), 0.0),
|
|
"S_BOKRI": fnum(smt_kv.get("1_R_S_BOKRI"), 0.0),
|
|
"S_SOMO": fnum(smt_kv.get("1_R_S_SOMO"), 0.0),
|
|
"S_TONGSIN": fnum(smt_kv.get("1_R_S_TONGSIN"), 0.0),
|
|
"S_SEGUM": fnum(smt_kv.get("1_R_S_SEGUM"), 0.0),
|
|
"S_PEGI": fnum(smt_kv.get("1_R_S_PEGI"), 0.0),
|
|
"S_DOSEO": fnum(smt_kv.get("1_R_S_DOSEO"), 0.0),
|
|
"S_JIGUP": fnum(smt_kv.get("1_R_S_JIGUP"), 0.0),
|
|
}
|
|
jab_type = int(fnum(smt_kv.get("JAB_TYPE"), 0))
|
|
cal_actl = int(fnum(smt_kv.get("CAL_ACTL"), 0))
|
|
scheme = ("SA" if jab_type == 1 else "TA") if cal_actl == 1 else ("S" if jab_type == 1 else "T")
|
|
|
|
return {
|
|
"projectMeta": {
|
|
"pname": prj_row.get("PNAME", ""),
|
|
"pnum": prj_row.get("PNUM", ""),
|
|
"pcode": prj_row.get("PCODE", "11"),
|
|
"pclass": int(fnum(prj_row.get("PCLASS"), 0)),
|
|
"kOrd": int(fnum(prj_row.get("K_ORD"), 0)),
|
|
"tcost1": fnum(prj_row.get("TCOST1"), 0),
|
|
"costw1": fnum(prj_row.get("COSTW1"), 0),
|
|
},
|
|
"sysConfig": {
|
|
"rxFormula": prj_row.get("RXNM_") or "1/8*16/12*25/20",
|
|
"condCodes": "L00008, L00009, L00010",
|
|
"usdRate1": fnum(prj_row.get("EXLNM1"), 1350),
|
|
"usdRate2": fnum(prj_row.get("EXLNM2"), 1400),
|
|
"useChangedUsd": False,
|
|
"ecdDigits": int(fnum(prj_row.get("ECD"), 0)),
|
|
"chnMode": "total" if prj_row.get("CHAMT") == "T" else "bimok",
|
|
"optOrder": "SML",
|
|
"kOrdSim": {
|
|
"pclass": int(fnum(prj_row.get("PCLASS"), 7)),
|
|
"regMode": 7,
|
|
"flag": 0,
|
|
"currentK": int(fnum(prj_row.get("K_ORD"), 2)) or 2,
|
|
},
|
|
},
|
|
"resources": resources,
|
|
"machines": machines,
|
|
"sanItems": san_items,
|
|
"ilwiItems": ilwi_items,
|
|
"boqItems": boq_items,
|
|
"kanjubConfig": {
|
|
"scheme": scheme,
|
|
"indirectTruncDigits": 0,
|
|
"anjunBaseAdd": fnum(smt_kv.get("1_C_ANJUN_G"), 5349000),
|
|
"anjunGovVatExcluded": True,
|
|
"anjunCap12": True,
|
|
"vatMode": "supply_10",
|
|
"totJeolDigits": -int(fnum(smt_kv.get("1_I_TOT_JEOL"), 3)),
|
|
"enableProfitEndAdj": True,
|
|
"govRoundMode": "up",
|
|
"govRoundDigits": -int(fnum(smt_kv.get("1_I_KWN_JEOLSA"), 3)),
|
|
"rates": rates,
|
|
"revModes": {
|
|
k: ("carry_original" if k in ("KUNBO", "SLVBO", "YUNBO") else "recalculated")
|
|
for k in (
|
|
"KANNO",
|
|
"SANBO",
|
|
"KOBO",
|
|
"KUNBO",
|
|
"SLVBO",
|
|
"YUNBO",
|
|
"TBUGUM",
|
|
"ANJUN",
|
|
"KITA",
|
|
"HWAN",
|
|
"EQIP",
|
|
)
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
class LoadReq(BaseModel):
|
|
source_path: str
|
|
|
|
|
|
class LoadJbrReq(BaseModel):
|
|
jbr_path: str
|
|
|
|
|
|
class SaveDbfReq(BaseModel):
|
|
source_path: str
|
|
target_folder_name: str = "stmate_edited"
|
|
costn_updates: list[dict] = []
|
|
boq_updates: list[dict] = []
|
|
prjlst_updates: dict = {}
|
|
smt_updates: dict = {}
|
|
costn_memo_updates: dict[str, str] = {}
|
|
|
|
|
|
@app.get("/api/sources")
|
|
def api_sources():
|
|
return discover_sources()
|
|
|
|
|
|
@app.get("/api/catalog")
|
|
def api_catalog():
|
|
return load_stmate_catalog()
|
|
|
|
|
|
@app.post("/api/load")
|
|
def api_load(req: LoadReq):
|
|
files = read_project_files(req.source_path)
|
|
if "COSTN.DBF" not in files:
|
|
raise HTTPException(status_code=400, detail="COSTN.DBF 파일이 포함되지 않은 경로임")
|
|
|
|
_, costn_rows, _ = parse_dbf_bytes(files["COSTN.DBF"])
|
|
costn_memos = extract_costn_memos(files["COSTN.DBF"], files.get("COSTN.SMT", b""))
|
|
|
|
boq_tables: dict[str, list[dict]] = {}
|
|
for bname in sorted(k for k in files if k.startswith("BOQ") and k.endswith(".DBF")):
|
|
_, brows, _ = parse_dbf_bytes(files[bname])
|
|
boq_tables[bname] = brows
|
|
|
|
boq_name = "BOQ11.DBF" if "BOQ11.DBF" in boq_tables else (next(iter(boq_tables), "BOQ11.DBF"))
|
|
boq_rows = boq_tables.get(boq_name, [])
|
|
_, prj_rows, _ = (
|
|
parse_dbf_bytes(files["PRJLST.DBF"]) if "PRJLST.DBF" in files else ([], [{}], {})
|
|
)
|
|
smt_kv = parse_prjlst_smt(files.get("PRJLST.SMT", b""))
|
|
|
|
rate_tables: dict[str, list[dict]] = {}
|
|
for rname in ("RATE_R.DBF", "RATE_RA.DBF", "RATE_M.DBF", "RATE_B.DBF", "RATE.DBF"):
|
|
if rname in files:
|
|
_, rrows, _ = parse_dbf_bytes(files[rname])
|
|
rate_tables[rname] = rrows
|
|
|
|
set_rep_rows = parse_dbf_bytes(files["SET_REP.DBF"])[1] if "SET_REP.DBF" in files else []
|
|
sim_state = convert_dbf_to_sim_state(
|
|
costn_rows, boq_rows, smt_kv, prj_rows[0] if prj_rows else {}, costn_memos
|
|
)
|
|
return {
|
|
"source_path": req.source_path,
|
|
"boq_table_name": boq_name,
|
|
"boq_tables": boq_tables,
|
|
"file_list": sorted(files.keys()),
|
|
"prjlst_all": prj_rows,
|
|
"prjlst": prj_rows[0] if prj_rows else {},
|
|
"prjlst_smt": smt_kv,
|
|
"costn_count": len(costn_rows),
|
|
"costn_rows": costn_rows,
|
|
"costn_memos": costn_memos,
|
|
"boq_count": len(boq_rows),
|
|
"boq_rows": boq_rows,
|
|
"rate_tables": rate_tables,
|
|
"set_rep_rows": set_rep_rows,
|
|
"sim_state": sim_state,
|
|
}
|
|
|
|
|
|
@app.post("/api/load-jbr")
|
|
def api_load_jbr(req: LoadJbrReq):
|
|
p = pathlib.Path(req.jbr_path)
|
|
if not p.exists():
|
|
raise HTTPException(status_code=404, detail=f"JB_R 파일 없음: {req.jbr_path}")
|
|
flds, rows, _ = parse_dbf_bytes(p.read_bytes())
|
|
matrix = [[fnum(r.get(f["name"])) for f in flds] for r in rows]
|
|
|
|
def cell(r_i: int, c_1: int, d: float = 0.0) -> float:
|
|
return (
|
|
matrix[r_i][c_1 - 1]
|
|
if 0 <= r_i < len(matrix) and 0 <= c_1 - 1 < len(matrix[r_i])
|
|
else d
|
|
)
|
|
|
|
raw_base = cell(20, 18, 5349.0)
|
|
extracted_rates = {
|
|
"KANNO": cell(20, 1, 14.5),
|
|
"KITA": cell(20, 6, 6.0),
|
|
"ILBAN": cell(20, 11, 6.0),
|
|
"SANBO": cell(45, 1, 3.56),
|
|
"KOBO": cell(45, 17, 1.01),
|
|
"KUNBO": cell(46, 1, 3.545),
|
|
"YUNBO": cell(46, 6, 4.5),
|
|
"SLVBO": cell(46, 11, 12.95),
|
|
"IYUN": cell(20, 16, 15.0),
|
|
"ANJUN": cell(20, 17, 1.86),
|
|
"anjunBaseAdd": raw_base * 1000 if 0 < raw_base < 100000 else raw_base,
|
|
}
|
|
return {
|
|
"name": p.name,
|
|
"row_count": len(matrix),
|
|
"col_count": len(flds),
|
|
"fields": [f["name"] for f in flds],
|
|
"rows": rows,
|
|
"matrix": matrix,
|
|
"extracted_rates": extracted_rates,
|
|
}
|
|
|
|
|
|
@app.post("/api/save")
|
|
def api_save(req: SaveDbfReq):
|
|
files = read_project_files(req.source_path)
|
|
safe_name = re.sub(r"[^\w\-가-힣]", "_", req.target_folder_name or "stmate_edited").strip("_")
|
|
out_dir = WORK_DBF_ROOT / (safe_name or "stmate_edited")
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
if "COSTN.DBF" in files:
|
|
flds, rows, meta = parse_dbf_bytes(files["COSTN.DBF"])
|
|
upd_map = {str(u.get("CODE", "")).strip(): u for u in req.costn_updates if u.get("CODE")}
|
|
for r in rows:
|
|
cd = (r.get("CODE") or "").strip()
|
|
if cd in upd_map:
|
|
for k, v in upd_map[cd].items():
|
|
if k in r:
|
|
r[k] = str(v)
|
|
dbf_b = build_dbf_bytes(flds, rows, meta)
|
|
if req.costn_memo_updates:
|
|
dbf_b, smt_b = rebuild_costn_with_smt(
|
|
dbf_b, files.get("COSTN.SMT", b""), req.costn_memo_updates
|
|
)
|
|
files["COSTN.SMT"] = smt_b
|
|
files["COSTN.DBF"] = dbf_b
|
|
|
|
boq_name = "BOQ11.DBF" if "BOQ11.DBF" in files else "BOQ12.DBF"
|
|
if boq_name in files:
|
|
flds, rows, meta = parse_dbf_bytes(files[boq_name])
|
|
upd_by_rec = {int(u["_recno"]): u for u in req.boq_updates if "_recno" in u}
|
|
for r in rows:
|
|
rno = int(r.get("_recno", -1))
|
|
if rno in upd_by_rec:
|
|
u = upd_by_rec[rno]
|
|
for k in (
|
|
"ITNUM",
|
|
"NAME1",
|
|
"SIZE1",
|
|
"UNIT1",
|
|
"QTY1",
|
|
"QTY2",
|
|
"MAMT1",
|
|
"LAMT1",
|
|
"SAMT1",
|
|
"TAMT1",
|
|
"MAMT2",
|
|
"LAMT2",
|
|
"SAMT2",
|
|
"TAMT2",
|
|
"MAMT3",
|
|
"LAMT3",
|
|
"SAMT3",
|
|
"TAMT3",
|
|
):
|
|
if k in u and k in r:
|
|
r[k] = str(u[k])
|
|
bd = (r.get("BDCODE1") or "").strip()
|
|
if bd and not STRUCT_RE.match(bd):
|
|
for s_idx in ("1", "2", "3"):
|
|
q_k = (
|
|
f"QTY{s_idx}" if f"QTY{s_idx}" in r and r.get(f"QTY{s_idx}") else "QTY1"
|
|
)
|
|
q = Decimal(str(fnum(r.get(q_k))))
|
|
m_a = int(
|
|
(q * Decimal(str(fnum(r.get(f"MAMT{s_idx}"))))).quantize(
|
|
Decimal(1), rounding=ROUND_DOWN
|
|
)
|
|
)
|
|
l_a = int(
|
|
(q * Decimal(str(fnum(r.get(f"LAMT{s_idx}"))))).quantize(
|
|
Decimal(1), rounding=ROUND_DOWN
|
|
)
|
|
)
|
|
s_a = int(
|
|
(q * Decimal(str(fnum(r.get(f"SAMT{s_idx}"))))).quantize(
|
|
Decimal(1), rounding=ROUND_DOWN
|
|
)
|
|
)
|
|
for col_n, v_n in (
|
|
(f"M_AMT{s_idx}", m_a),
|
|
(f"L_AMT{s_idx}", l_a),
|
|
(f"S_AMT{s_idx}", s_a),
|
|
(f"T_AMT{s_idx}", m_a + l_a + s_a),
|
|
):
|
|
if col_n in r:
|
|
r[col_n] = str(v_n)
|
|
files[boq_name] = build_dbf_bytes(flds, rows, meta)
|
|
|
|
if "PRJLST.DBF" in files and req.prjlst_updates:
|
|
flds, rows, meta = parse_dbf_bytes(files["PRJLST.DBF"])
|
|
if rows:
|
|
for k, v in req.prjlst_updates.items():
|
|
if k in rows[0]:
|
|
rows[0][k] = str(v)
|
|
files["PRJLST.DBF"] = build_dbf_bytes(flds, rows, meta)
|
|
|
|
if "PRJLST.SMT" in files and req.smt_updates:
|
|
files["PRJLST.SMT"] = update_prjlst_smt_bytes(files["PRJLST.SMT"], req.smt_updates)
|
|
|
|
written_files = []
|
|
for fname, content in files.items():
|
|
(out_dir / fname).write_bytes(content)
|
|
written_files.append({"name": fname, "size": len(content)})
|
|
|
|
stc_path = WORK_DBF_ROOT / f"{out_dir.name}.STC"
|
|
with zipfile.ZipFile(stc_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
|
for fname, content in files.items():
|
|
zf.writestr(fname, content)
|
|
|
|
return {
|
|
"ok": True,
|
|
"saved_dir": str(out_dir),
|
|
"saved_stc": str(stc_path),
|
|
"stc_size": stc_path.stat().st_size,
|
|
"written_count": len(written_files),
|
|
"written_files": written_files,
|
|
}
|
|
|
|
|
|
class ExportXlsxReq(BaseModel):
|
|
prj_meta: dict = {}
|
|
costn_rows: list = []
|
|
boq_rows: list = []
|
|
kanjub_data: dict = {}
|
|
costn_memos: dict = {}
|
|
recipes: list = []
|
|
|
|
|
|
@app.post("/api/export-xlsx")
|
|
def api_export_xlsx(req: ExportXlsxReq):
|
|
try:
|
|
xlsx_bytes = build_stmate_full_xlsx(
|
|
req.prj_meta,
|
|
req.costn_rows,
|
|
req.boq_rows,
|
|
req.kanjub_data,
|
|
req.costn_memos,
|
|
req.recipes,
|
|
)
|
|
return Response(
|
|
content=xlsx_bytes,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": 'attachment; filename="STMATE_Workbook.xlsx"'},
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
app.mount("/", StaticFiles(directory=str(HERE), html=True), name="static")
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
uvicorn.run(app, host="0.0.0.0", port=8095)
|