Files
Aislo/tools/stmate-test/server.py
T
eomsangdonandClaude Opus 5.5 086750b6d4 feat(stmate-test): 2차 화면·코드관리 합침 · % 줄(조달수수료) 계산·저장 재계산 · 줄 고정 단가 보존 · 장부
- 코드 등록·삭제·변경·병합·데이타병합·일괄변경·검색·동일코드·풀어넣기 끝전 등(ui_code · ui_merge · api_ui_code)
- 코드 붙은 % 줄 = 같은 상위 공종 앞 말단 줄 비목 합 × 율(server pct_row_base · calc.js) · 저장 때 다시 셈(stc_save _pct_rows) — 실무 7건 저장값 그대로
- 내역 줄 단가 ≠ 단가목록이면 줄 고정 단가(fixedUnit)
- 메뉴 94개 눌러 오류 0 · 시험 502 통과

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-13 20:10:16 +09:00

619 lines
23 KiB
Python

# -*- coding: utf-8 -*-
"""STMATE 원본 DBF / SMT / STC / JB_R 100% 무손실 직접 읽기·쓰기 연동 서버 (server.py)."""
from __future__ import annotations
import importlib
import pathlib
import re
from bdqty import decode_bdqty
from excel_builder import build_stmate_full_xlsx
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import Response
from fastapi.staticfiles import StaticFiles
from kanjub_cfg import build_kanjub_config, stage_slots
from pydantic import BaseModel
from smt_parser import (
extract_costn_memos,
load_stmate_catalog,
parse_dbf_bytes,
parse_ini_bytes,
parse_prjlst_smt,
)
from stc_project import safe_name, write_output
from stc_read import Source, extra_tables
from stc_save import apply_save
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 open_source(source_path: str) -> Source:
try:
return Source(source_path)
except FileNotFoundError:
raise HTTPException(status_code=404, detail=f"경로를 찾을 수 없음: {source_path}")
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
def read_project_files(source_path: str) -> dict[str, bytes]:
"""STC(ZIP)·폴더 → 대문자 이름 → 바이트 (stc_read.Source)."""
return open_source(source_path).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 pct_row_base(boq_rows: list[dict], i: int) -> dict:
"""코드 붙은 % 줄(조달수수료 M%·T%) 밑수 — 같은 상위 공종 안 그 위 말단 줄들(닫힌 하위 공종 포함).
단가 칸 = 밑수 합 · 수량 칸 = 율(%) — 실무 봉화·울진·대흥 저장값 대조.
"""
ids, depth = [], []
for j in range(i - 1, -1, -1):
bd = (boq_rows[j].get("BDCODE1") or "").strip()
if re.match(r"^E\d$", bd):
depth.append(bd[1])
elif re.match(r"^T\d$", bd):
if depth and depth[-1] == bd[1]:
depth.pop()
else:
break
elif bd and not STRUCT_RE.match(bd):
ids.append(f"BOQ_{j + 1}")
unit = (boq_rows[i].get("UNIT1") or "").strip()
return {"pctComp": unit[0] if unit[:1] in ("M", "L", "S", "T") else "T", "pctBase": ids[::-1]}
def boq_fixed_unit(r: dict, crow: dict | None, is_pct: bool) -> dict:
"""내역 줄 단가가 단가목록(COSTN 1칸)과 다르면 줄 고정 단가(fixedUnit) — 단가 끝전 보정 등 줄 단위 수정 보존.
0원 줄(변경 차수에서만 쓰는 신규 공종의 당초 칸)은 제외.
"""
if is_pct or not crow:
return {}
unit = {c: fnum(r.get(f"{c}AMT1")) for c in "MLS"}
if not any(unit.values()) or all(
abs(unit[c] - fnum(crow.get(f"{c}AMT1"))) < 1e-6 for c in "MLS"
):
return {}
return {"fixedUnit": unit}
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 {}
costn_by_code = {(r.get("CODE") or "").strip(): r for r in costn_rows}
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"
# 비율 줄(PR)은 수량 칸 = 율 그대로(calc.js prRow · PRPT/PRIN/PROUT) — 코드 붙은 % 줄만 옛 꼴(1 × 저장 금액)
is_pct_fee = bd != "PR" and 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,
"qtys": [1.0] * 4 if is_pct_fee else [fnum(r.get(f"QTY{n}")) for n in range(1, 5)],
"units": {
n: [fnum(r.get(f"{c}_AMT{n}" if is_pct_fee else f"{c}AMT{n}")) for c in "MLS"]
for n in range(2, 5)
if any(
fnum(r.get(f"{c}_AMT{n}" if is_pct_fee else f"{c}AMT{n}")) for c in "MLS"
)
},
"_recno": r.get("_recno", i),
"PRPT": r.get("PRPT", ""),
"PRIN": r.get("PRIN", ""),
"PROUT": r.get("PROUT", ""),
"prQtys": [fnum(r.get(f"QTY{n}")) for n in range(1, 5)],
**(pct_row_base(boq_rows, i) if is_pct_fee else {}),
**boq_fixed_unit(r, costn_by_code.get(bd), is_pct_fee),
}
)
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": build_kanjub_config(smt_kv),
"kanjubStages": {n: build_kanjub_config(smt_kv, n) for n in stage_slots(smt_kv)},
}
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] = {}
boq_table: str = ""
boq_deleted: list[int] = []
start_ini: dict = {}
prjlst_rows: list = []
kanjub_stages: dict = {} # 제잡비 차수별 설정 → PRJLST.SMT(바뀐 키만)
bdqty_rows: list[dict] | None = None # 구성표 행 전체(decode_bdqty 꼴) → BDQTY.DBF(바뀐 호표만)
qcode_rows: list[dict] | None = None # 수량코드 행 전체 → QCODE.DBF·SMT(바뀌었을 때만)
@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""), (prj_rows[0] if prj_rows else {}).get("PRJ_MEMO_HEX", "")
)
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
)
# 노임 환산 직종·환산식 = Start8x.Ini CONLD · RXXNM
ini = parse_ini_bytes(files.get("START8X.INI") or files.get("STINI.MEM") or b"")
conld = ini.get("CONLD", "")
sim_state["sysConfig"]["condCodes"] = ", ".join(
c for c in (conld[i : i + 6].strip() for i in range(0, len(conld), 6)) if c
)
sim_state["sysConfig"]["rxFormula"] = ini.get("RXXNM") or sim_state["sysConfig"]["rxFormula"]
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,
"bdqty_rows": decode_bdqty(files.get("BDQTY.DBF", b"")),
"start_ini": ini,
"sim_state": sim_state,
**extra_tables(files),
}
@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):
"""바뀐 칸만 고쳐 work_dbf/<이름>/ 와 <이름>.STC 로 씀(stc_save · stc_project) — 무변경이면 원본과 같은 바이트."""
src = open_source(req.source_path)
try:
files, changes = apply_save(src.files, req.model_dump())
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
res = write_output(
WORK_DBF_ROOT, safe_name(req.target_folder_name), files, src.names, src.entries
)
res["changed_files"] = sorted(k for k in files if files[k] is not src.files.get(k))
res["changes"] = changes[:500]
res["change_count"] = len(changes)
return res
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))
# 끼워 넣기 — api_*.py 의 router 자동 연결 · scr_*.js 화면 파일 자동 불러오기
for _mod in sorted(HERE.glob("api_*.py")):
app.include_router(importlib.import_module(_mod.stem).router)
@app.get("/plugins_manifest.js")
def plugins_manifest():
tags = "".join(f'<script src="{f.name}"></script>' for f in sorted(HERE.glob("scr_*.js")))
return Response(content=f"document.write({tags!r});", media_type="application/javascript")
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)