# -*- 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)