688 lines
25 KiB
Python
688 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""STMATE 원본 DBF / SMT / STC / JB_R 직접 읽기·쓰기 연동 로컬 서버 (server.py)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import io
|
|
import pathlib
|
|
import re
|
|
import shutil
|
|
import struct
|
|
import zipfile
|
|
from decimal import Decimal, ROUND_DOWN
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel
|
|
|
|
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|:)$")
|
|
SMT_KV_RE = re.compile(r"([A-Za-z0-9_]+)=([^\r\n\x00]+)")
|
|
|
|
app = FastAPI(title="STMATE DBF Direct Bridge")
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
def parse_dbf_bytes(b: bytes) -> tuple[list[dict], list[dict], dict]:
|
|
"""dBASE III/IV 바이트를 파싱하여 (필드정의, 레코드목록, 헤더메타) 반환."""
|
|
if len(b) < 32:
|
|
raise ValueError("DBF 헤더 길이 부족")
|
|
ver = b[0]
|
|
nrec, hlen, rlen = struct.unpack_from("<IHH", b, 4)
|
|
fields: list[dict] = []
|
|
off = 32
|
|
while off < len(b) and b[off] != 0x0D:
|
|
raw_name = b[off : off + 11].split(b"\x00", 1)[0]
|
|
name = raw_name.decode("cp949", "replace").strip()
|
|
ftype = chr(b[off + 11])
|
|
flen = b[off + 16]
|
|
fdec = b[off + 17]
|
|
fields.append(
|
|
{
|
|
"name": name,
|
|
"type": ftype,
|
|
"len": flen,
|
|
"dec": fdec,
|
|
"raw_desc": b[off : off + 32].hex(),
|
|
}
|
|
)
|
|
off += 32
|
|
|
|
rows: list[dict] = []
|
|
for i in range(nrec):
|
|
pos = hlen + i * rlen
|
|
if pos + rlen > len(b):
|
|
break
|
|
rec = b[pos : pos + rlen]
|
|
deleted = rec[0:1] == b"*"
|
|
row: dict = {"_recno": i, "_deleted": deleted}
|
|
cur = 1
|
|
for f in fields:
|
|
val_bytes = rec[cur : cur + f["len"]]
|
|
cur += f["len"]
|
|
row[f["name"]] = val_bytes.decode("cp949", "replace").strip()
|
|
if not deleted:
|
|
rows.append(row)
|
|
|
|
meta = {"version": ver, "hlen": hlen, "rlen": rlen, "header_tail_hex": b[off:hlen].hex()}
|
|
return fields, rows, meta
|
|
|
|
|
|
def build_dbf_bytes(fields: list[dict], rows: list[dict], meta: dict | None = None) -> bytes:
|
|
"""(필드정의, 레코드목록)을 STMATE 호환 CP949 고정폭 DBF 바이트로 직렬화."""
|
|
ver = (meta or {}).get("version", 0x03)
|
|
today = datetime.date.today()
|
|
nrec = len(rows)
|
|
rlen = 1 + sum(int(f["len"]) for f in fields)
|
|
tail_bytes = bytes.fromhex((meta or {}).get("header_tail_hex", "0d"))
|
|
if not tail_bytes or tail_bytes[0] != 0x0D:
|
|
tail_bytes = b"\x0d"
|
|
hlen = 32 + 32 * len(fields) + len(tail_bytes)
|
|
|
|
hdr = bytearray(32)
|
|
hdr[0] = ver
|
|
hdr[1] = today.year - 1900
|
|
hdr[2] = today.month
|
|
hdr[3] = today.day
|
|
struct.pack_into("<IHH", hdr, 4, nrec, hlen, rlen)
|
|
|
|
out = io.BytesIO()
|
|
out.write(hdr)
|
|
for f in fields:
|
|
if f.get("raw_desc"):
|
|
desc = bytearray(bytes.fromhex(f["raw_desc"]))
|
|
desc[16] = int(f["len"])
|
|
desc[17] = int(f.get("dec", 0))
|
|
out.write(desc)
|
|
else:
|
|
desc = bytearray(32)
|
|
nm_b = f["name"].encode("cp949", "replace")[:10]
|
|
desc[: len(nm_b)] = nm_b
|
|
desc[11] = ord((f.get("type") or "C")[0])
|
|
desc[16] = int(f["len"])
|
|
desc[17] = int(f.get("dec", 0))
|
|
out.write(desc)
|
|
out.write(tail_bytes)
|
|
|
|
for row in rows:
|
|
out.write(b"*" if row.get("_deleted") else b" ")
|
|
for f in fields:
|
|
flen = int(f["len"])
|
|
fdec = int(f.get("dec", 0))
|
|
ftype = f.get("type", "C")
|
|
raw_val = row.get(f["name"], "")
|
|
if ftype in ("N", "F"):
|
|
sval = str(raw_val).strip()
|
|
if sval == "" or sval == "None":
|
|
sval = ""
|
|
else:
|
|
try:
|
|
num = float(sval)
|
|
sval = f"{num:.{fdec}f}" if fdec > 0 else f"{int(round(num))}"
|
|
except Exception:
|
|
pass
|
|
enc = sval.encode("ascii", "replace")[:flen]
|
|
out.write(enc.rjust(flen, b" "))
|
|
else:
|
|
sval = "" if raw_val is None else str(raw_val)
|
|
enc = sval.encode("cp949", "replace")[:flen]
|
|
out.write(enc.ljust(flen, b" "))
|
|
|
|
out.write(b"\x1a")
|
|
return out.getvalue()
|
|
|
|
|
|
def parse_prjlst_smt(b: bytes) -> dict[str, str]:
|
|
if len(b) <= 0x200:
|
|
return {}
|
|
txt = b[0x200:].decode("cp949", "replace")
|
|
return {k: v.strip() for k, v in SMT_KV_RE.findall(txt)}
|
|
|
|
|
|
def update_prjlst_smt_bytes(orig_b: bytes, updates: dict[str, str | int | float]) -> bytes:
|
|
"""PRJLST.SMT의 SIxMemo 헤더(0x200)를 보존하며 키=값 본문을 갱신."""
|
|
header = bytearray(orig_b[:0x200]) if len(orig_b) >= 0x200 else bytearray(b"\x00" * 0x200)
|
|
txt = orig_b[0x200:].decode("cp949", "replace") if len(orig_b) > 0x200 else ""
|
|
for k, v in updates.items():
|
|
pat = re.compile(rf"(^|\r?\n)({re.escape(k)}=)([^\r\n\x00]*)", re.M)
|
|
if pat.search(txt):
|
|
txt = pat.sub(rf"\g<1>\g<2>{v}", txt)
|
|
else:
|
|
txt += f"\r\n{k}={v}"
|
|
body_b = txt.encode("cp949", "replace")
|
|
return bytes(header) + body_b
|
|
|
|
|
|
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: list[dict] = []
|
|
if STMATE_SYS_BSC.exists():
|
|
for p in sorted(STMATE_SYS_BSC.glob("*.JB_R*")):
|
|
jbr_files.append(
|
|
{
|
|
"name": p.name,
|
|
"path": str(p),
|
|
"ext": p.suffix.upper(),
|
|
"size": p.stat().st_size,
|
|
}
|
|
)
|
|
|
|
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 convert_dbf_to_sim_state(
|
|
costn_rows: list[dict], boq_rows: list[dict], smt_kv: dict[str, str], prj_row: dict
|
|
) -> dict:
|
|
"""원본 COSTN.DBF / BOQ11~12.DBF / PRJLST.SMT 레코드를 시뮬레이터 state로 변환."""
|
|
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 = r.get("NAME") or code
|
|
spec = r.get("SIZE") or ""
|
|
unit = 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 = fnum(r.get("AMT_61")) or t1
|
|
|
|
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]
|
|
is_min = bool(non_zero and abs(min(non_zero) - amt61) < 1e-4)
|
|
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 is_min else "manual",
|
|
"manualPrice": amt61,
|
|
"convertHourly": False,
|
|
"isUsd": (r.get("DOLLOR") or "").upper() == "T",
|
|
}
|
|
)
|
|
else:
|
|
# B(일위대가), D(단가산출), X(중기)도 내역서 직접 참조가 가능하도록 복합단가 보존
|
|
ilwi_items.append(
|
|
{
|
|
"id": code,
|
|
"code": code,
|
|
"name": f"[{prefix}] {name}",
|
|
"spec": spec,
|
|
"unit": unit,
|
|
"fixedM": m1,
|
|
"fixedL": l1,
|
|
"fixedS": s1,
|
|
"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(
|
|
{
|
|
"id": res_id,
|
|
"code": res_id,
|
|
"name": f"{name} ({comp_kind})",
|
|
"spec": spec,
|
|
"unit": unit,
|
|
"kind": comp_kind,
|
|
"candidates": [val, 0, 0, 0, 0],
|
|
"adoptMode": "manual",
|
|
"manualPrice": val,
|
|
"convertHourly": False,
|
|
"isUsd": False,
|
|
}
|
|
)
|
|
ilwi_items[-1]["lines"].append(
|
|
{"refId": res_id, "qty": 1.0, "halPercent": 0, "devd": False}
|
|
)
|
|
|
|
# BOQ 제외금액 구역(C*) 판별 (16번 문서 §4.2)
|
|
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 = r.get("NAME1") or bd
|
|
if bd not in known_ids:
|
|
# BOQ 자체 단가를 가진 말단행이면 호표 자동 생성
|
|
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(
|
|
{
|
|
"id": rid,
|
|
"code": rid,
|
|
"name": f"{name1} ({ckind})",
|
|
"spec": r.get("SIZE1") or "",
|
|
"unit": r.get("UNIT1") or "",
|
|
"kind": ckind,
|
|
"candidates": [cval, 0, 0, 0, 0],
|
|
"adoptMode": "manual",
|
|
"manualPrice": cval,
|
|
"convertHourly": False,
|
|
"isUsd": False,
|
|
}
|
|
)
|
|
lines.append({"refId": rid, "qty": 1.0, "halPercent": 0, "devd": False})
|
|
ilwi_items.append(
|
|
{
|
|
"id": bd,
|
|
"code": bd,
|
|
"name": name1,
|
|
"spec": r.get("SIZE1") or "",
|
|
"unit": r.get("UNIT1") or "",
|
|
"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"
|
|
|
|
boq_items.append(
|
|
{
|
|
"id": f"BOQ_{i + 1}",
|
|
"code": bd,
|
|
"name": name1,
|
|
"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")
|
|
tot_jeol = -int(fnum(smt_kv.get("1_I_TOT_JEOL"), 3))
|
|
|
|
return {
|
|
"projectMeta": {
|
|
"pname": prj_row.get("PNAME", ""),
|
|
"pnum": prj_row.get("PNUM", ""),
|
|
"kOrd": int(fnum(prj_row.get("K_ORD"), 0)),
|
|
},
|
|
"sysConfig": {
|
|
"rxFormula": "1/8*16/12*25/20",
|
|
"condCodes": "L00008, L00009, L00010",
|
|
"usdRate1": 1350,
|
|
"usdRate2": 1400,
|
|
"useChangedUsd": False,
|
|
"ecdDigits": 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[:120],
|
|
"machines": machines,
|
|
"sanItems": san_items,
|
|
"ilwiItems": ilwi_items[:120],
|
|
"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": tot_jeol,
|
|
"enableProfitEndAdj": True,
|
|
"govRoundMode": "up",
|
|
"govRoundDigits": -int(fnum(smt_kv.get("1_I_KWN_JEOLSA"), 3)),
|
|
"rates": rates,
|
|
"revModes": {
|
|
"KANNO": "recalculated",
|
|
"SANBO": "recalculated",
|
|
"KOBO": "recalculated",
|
|
"KUNBO": "carry_original",
|
|
"SLVBO": "carry_original",
|
|
"YUNBO": "carry_original",
|
|
"TBUGUM": "recalculated",
|
|
"ANJUN": "recalculated",
|
|
"KITA": "recalculated",
|
|
"HWAN": "recalculated",
|
|
"EQIP": "recalculated",
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
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 = {}
|
|
|
|
|
|
@app.get("/api/sources")
|
|
def api_sources():
|
|
return discover_sources()
|
|
|
|
|
|
@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_flds, costn_rows, _ = parse_dbf_bytes(files["COSTN.DBF"])
|
|
boq_name = "BOQ12.DBF" if "BOQ12.DBF" in files else "BOQ11.DBF"
|
|
boq_flds, boq_rows, _ = parse_dbf_bytes(files[boq_name]) if boq_name in files else ([], [], {})
|
|
prj_flds, 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_r_rows = []
|
|
if "RATE_R.DBF" in files:
|
|
_, rate_r_rows, _ = parse_dbf_bytes(files["RATE_R.DBF"])
|
|
|
|
sim_state = convert_dbf_to_sim_state(
|
|
costn_rows, boq_rows, smt_kv, prj_rows[0] if prj_rows else {}
|
|
)
|
|
return {
|
|
"source_path": req.source_path,
|
|
"boq_table_name": boq_name,
|
|
"file_list": sorted(files.keys()),
|
|
"prjlst": prj_rows[0] if prj_rows else {},
|
|
"prjlst_smt": smt_kv,
|
|
"costn_count": len(costn_rows),
|
|
"costn_rows": costn_rows[:200],
|
|
"boq_count": len(boq_rows),
|
|
"boq_rows": boq_rows[:200],
|
|
"rate_r_count": len(rate_r_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: list[list[float]] = []
|
|
for r in rows:
|
|
matrix.append([fnum(r.get(f["name"])) for f in flds])
|
|
|
|
def cell(row_idx: int, col_1based: int, default: float = 0.0) -> float:
|
|
if 0 <= row_idx < len(matrix) and 0 <= col_1based - 1 < len(matrix[row_idx]):
|
|
return matrix[row_idx][col_1based - 1]
|
|
return default
|
|
|
|
raw_base = cell(20, 18, 5349.0)
|
|
anjun_base = raw_base * 1000 if 0 < raw_base < 100000 else raw_base
|
|
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": anjun_base,
|
|
}
|
|
return {
|
|
"name": p.name,
|
|
"row_count": len(matrix),
|
|
"col_count": len(flds),
|
|
"extracted_rates": extracted_rates,
|
|
"preview_rows": {
|
|
"row6_kanno_kita": matrix[6][:12] if len(matrix) > 6 else [],
|
|
"row20_iyun_anjun": matrix[20][:20] if len(matrix) > 20 else [],
|
|
"row45_sanbo_kobo": matrix[45][:20] if len(matrix) > 45 else [],
|
|
"row46_kunbo_yunbo_slvbo": matrix[46][:15] if len(matrix) > 46 else [],
|
|
},
|
|
}
|
|
|
|
|
|
@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)
|
|
|
|
# 1. COSTN.DBF 수정 반영 및 직렬화
|
|
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)
|
|
files["COSTN.DBF"] = build_dbf_bytes(flds, rows, meta)
|
|
|
|
# 2. BOQ11/12.DBF 수량·단가·비목금액 재계산 및 직렬화
|
|
boq_name = "BOQ12.DBF" if "BOQ12.DBF" in files else "BOQ11.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 ("NAME1", "QTY1", "MAMT1", "LAMT1", "SAMT1", "TAMT1"):
|
|
if k in u and k in r:
|
|
r[k] = str(u[k])
|
|
# 비목별 금액 ROUNDDOWN(QTY1 * 단가) 동기화
|
|
q = Decimal(str(fnum(r.get("QTY1"))))
|
|
m_u = Decimal(str(fnum(r.get("MAMT1"))))
|
|
l_u = Decimal(str(fnum(r.get("LAMT1"))))
|
|
s_u = Decimal(str(fnum(r.get("SAMT1"))))
|
|
m_a = int((q * m_u).quantize(Decimal(1), rounding=ROUND_DOWN))
|
|
l_a = int((q * l_u).quantize(Decimal(1), rounding=ROUND_DOWN))
|
|
s_a = int((q * s_u).quantize(Decimal(1), rounding=ROUND_DOWN))
|
|
r["M_AMT1"] = str(m_a)
|
|
r["L_AMT1"] = str(l_a)
|
|
r["S_AMT1"] = str(s_a)
|
|
r["T_AMT1"] = str(m_a + l_a + s_a)
|
|
files[boq_name] = build_dbf_bytes(flds, rows, meta)
|
|
|
|
# 3. PRJLST.DBF 및 PRJLST.SMT 갱신
|
|
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)
|
|
|
|
# 4. 폴더 내 개별 DBF/SMT 저장 및 .STC 패키지 생성
|
|
written_files = []
|
|
for fname, content in files.items():
|
|
fp = out_dir / fname
|
|
fp.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,
|
|
}
|
|
|
|
|
|
app.mount("/", StaticFiles(directory=str(HERE), html=True), name="static")
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
uvicorn.run(app, host="127.0.0.1", port=8095)
|