feat(tools): STMATE 원본 DBF/SMT/STC/JB_R 직접 읽기·쓰기 브리지 서버 및 UI 연동 추가 (tools/stmate-test)

This commit is contained in:
2026-11-04 22:23:34 +09:00
parent a56da7bb2b
commit 1cc9dca9fc
5 changed files with 1060 additions and 4 deletions
+11 -3
View File
@@ -4,10 +4,18 @@
경쟁사 설계예산 프로그램 **STMATE**(경영정보(주))에 대한 **REA 6.7.0 / Ghidra 12.1.4 정적 분석** 및 지식DB(`resources/knowledge/original/경쟁사 프로그램/STmate 분석/`) 결과를 바탕으로, 주요 기능과 계산 사슬·제어 옵션을 브라우저에서 독립적으로 확인하고 조작할 수 있도록 만든 테스트 웹앱입니다.
- **완전 분리 동작**: Aislo 본체 서버·DB·로그인 세션과 무관하게 `index.html` 단독으로 동작합니다.
- **상태 보관 및 입출력**: 현재 브라우저 탭(`sessionStorage`)에 자동 보관되며, 상단 버튼으로 **JSON 저장 · 파일 열기 · 예제 초기화**를 지원합니다.
- **완전 분리 동작**: Aislo 본체 서버·공용 MariaDB·로그인 세션과 무관하게 동작합니다.
- **PC 내 STMATE 원본 DBF · STC · JB_R 직접 읽기/쓰기 연동 (`server.py`)**:
- 실행 명령: `.\venv\Scripts\python.exe tools\stmate-test\server.py` (포트 `8095`, `http://127.0.0.1:8095`)
- PC의 `C:\경영정보(주)\STmate 자료\설계자료\실습(맛보기)` 원본 DBF 폴더, `C:\경영정보(주)\System\ST_BSC\*.JB_R` 제비율 기준 DB(2015~2026년 41종), 및 `resources/knowledge/original/실무문서/*.STC` 실무 프로젝트 DB를 직접 읽어 시뮬레이터에 동기화하고, 수정된 단가·수량·제비율 결과를 CP949 고정폭 바이너리 DBF(`COSTN.DBF`, `BOQ11/12.DBF`, `PRJLST.DBF/.SMT`) 및 `.STC` 패키지로 직접 기록합니다.
- **상태 보관 및 입출력**: 현재 브라우저 탭(`sessionStorage`)에 자동 보관되며, 상단 버튼으로 **JSON 저장 · 파일 열기 · 예제 초기화**도 함께 지원합니다.
## 주요 확인 및 제어 기능 (6개 탭)
## 주요 확인 및 제어 기능 (7개 탭)
0. **00 원본 DBF · STC 연동 (`server.py`, `db_bridge.js`)**
- PC 내 STMATE 원본 DBF 폴더 및 `.STC` 패키지 선택·불러오기 (`COSTN.DBF`, `BOQ11/12.DBF`, `PRJLST.DBF`, `PRJLST.SMT`)
- `C:\경영정보(주)\System\ST_BSC\*.JB_R` 제비율 기준 DBF(행20·45·46 좌표) 로드 및 원가계산 요율 즉시 적용
- `COSTN.DBF` 단가 직접 편집 및 시뮬레이터 계산 결과의 **원본 호환 DBF / `.STC` 파일 직접 쓰기(`tools/stmate-test/work_dbf/`)**
1. **01 기초자원 · 노임환산 (`ST_Sys.CPM`, `COSTN.DBF`)**
- 자재·노무·경비 기초자원의 5개 가격지 후보(`AMT_11..51`: 조달·물자·물정·거래·견적) 입력
+15 -1
View File
@@ -256,7 +256,8 @@ function renderAll() {
});
const panel = $("tabPanel");
if (activeTab === "resources") panel.innerHTML = renderResourcesTab(res);
if (activeTab === "db") panel.innerHTML = StmateDbBridge.renderDbTab(state, res);
else if (activeTab === "resources") panel.innerHTML = renderResourcesTab(res);
else if (activeTab === "machines") panel.innerHTML = renderMachinesTab(res);
else if (activeTab === "ilwi") panel.innerHTML = renderIlwiTab(res);
else if (activeTab === "boq") panel.innerHTML = StmateCostViews.renderBoqTab(state, res);
@@ -681,4 +682,17 @@ $("fileOpen").addEventListener("change", async (e) => {
}
});
if (typeof StmateDbBridge !== "undefined") {
StmateDbBridge.initEvents(
() => state,
(newSt) => {
state = newSt;
selectedIlwiId = state.ilwiItems[0]?.id || "";
},
() => {
saveState();
renderAll();
},
);
}
renderAll();
+345
View File
@@ -0,0 +1,345 @@
/* -*- coding: utf-8 -*- */
/**
* STMATE 원본 DBF / SMT / STC / JB_R 직접 읽기·쓰기 연동 컨트롤러 (db_bridge.js)
*/
const StmateDbBridge = (() => {
const API_BASE = "http://127.0.0.1:8095";
const esc = (s) =>
String(s ?? "").replace(
/[&<>"']/g,
(c) => ({ "&": "&amp;", "<": "&lt;", ">": "&gt;", '"': "&quot;", "'": "&#39;" })[c],
);
const fmt = (n, maxDec = 2) =>
new Intl.NumberFormat("ko-KR", { maximumFractionDigits: maxDec }).format(Number(n) || 0);
const dbCtx = {
connected: false,
sources: { projects: [], jbr_files: [], work_root: "" },
selectedProjectPath: "",
selectedJbrPath: "",
targetFolderName: "stmate_edited",
loadedDbf: null,
loadedJbr: null,
lastSaveResult: null,
statusMsg: "로컬 DB 브리지(127.0.0.1:8095) 연결 확인 중...",
};
async function checkAndLoadSources(onUpdate) {
try {
const resp = await fetch(`${API_BASE}/api/sources`);
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
const data = await resp.json();
dbCtx.connected = true;
dbCtx.sources = data;
if (!dbCtx.selectedProjectPath && data.projects.length > 0) {
dbCtx.selectedProjectPath = data.projects[0].path;
}
const onlyJbr = data.jbr_files.filter((f) => f.ext === ".JB_R");
if (!dbCtx.selectedJbrPath && onlyJbr.length > 0) {
dbCtx.selectedJbrPath = onlyJbr[onlyJbr.length - 1].path;
}
dbCtx.statusMsg = `연결됨 — PC 프로젝트 DB ${data.projects.length}건 · ST_BSC 제비율 DB ${data.jbr_files.length}건 감지`;
} catch (_) {
dbCtx.connected = false;
dbCtx.statusMsg =
"오프라인 — 터미널에서 .\\venv\\Scripts\\python.exe tools\\stmate-test\\server.py 실행 시 DBF 직접 읽기/쓰기 활성화";
}
if (onUpdate) onUpdate();
}
async function loadProjectDbf(applyStateCallback, onUpdate) {
if (!dbCtx.selectedProjectPath) return;
try {
dbCtx.statusMsg = `DBF/STC 읽는 중: ${dbCtx.selectedProjectPath} ...`;
if (onUpdate) onUpdate();
const resp = await fetch(`${API_BASE}/api/load`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ source_path: dbCtx.selectedProjectPath }),
});
if (!resp.ok) throw new Error(await resp.text());
const data = await resp.json();
dbCtx.loadedDbf = data;
dbCtx.lastSaveResult = null;
dbCtx.statusMsg = `로드 완료: ${data.prjlst?.PNAME || data.source_path} (COSTN ${data.costn_count}건, ${data.boq_table_name} ${data.boq_count}건)`;
if (data.sim_state && applyStateCallback) {
applyStateCallback(data.sim_state);
}
} catch (err) {
dbCtx.statusMsg = `DBF 로드 실패: ${err.message || err}`;
if (onUpdate) onUpdate();
}
}
async function loadAndApplyJbr(state, saveAndRender) {
if (!dbCtx.selectedJbrPath) return;
try {
const resp = await fetch(`${API_BASE}/api/load-jbr`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ jbr_path: dbCtx.selectedJbrPath }),
});
if (!resp.ok) throw new Error(await resp.text());
const data = await resp.json();
dbCtx.loadedJbr = data;
const r = data.extracted_rates;
Object.assign(state.kanjubConfig.rates, {
KANNO: r.KANNO || state.kanjubConfig.rates.KANNO,
KITA: r.KITA || state.kanjubConfig.rates.KITA,
ILBAN: r.ILBAN || state.kanjubConfig.rates.ILBAN,
SANBO: r.SANBO,
KOBO: r.KOBO,
KUNBO: r.KUNBO,
YUNBO: r.YUNBO,
SLVBO: r.SLVBO,
IYUN: r.IYUN,
ANJUN: r.ANJUN,
});
state.kanjubConfig.anjunBaseAdd = r.anjunBaseAdd;
dbCtx.statusMsg = `ST_BSC 제비율 DB(${data.name}) 요율 적용 완료`;
if (saveAndRender) saveAndRender();
} catch (err) {
dbCtx.statusMsg = `JB_R 로드 실패: ${err.message || err}`;
if (saveAndRender) saveAndRender();
}
}
async function saveToDbf(state, calcRes, onUpdate) {
const srcPath = dbCtx.loadedDbf?.source_path || dbCtx.selectedProjectPath;
if (!srcPath) return;
try {
const k = calcRes.kanjubOrig;
const costnUpdates = (dbCtx.loadedDbf?.costn_rows || []).slice(0, 80).map((r) => ({
CODE: r.CODE,
NAME: r.NAME,
TAMT1: r.TAMT1,
LAMT1: r.LAMT1,
MAMT1: r.MAMT1,
SAMT1: r.SAMT1,
}));
// 시뮬레이터 기초자원 단가 변경분도 COSTN 업데이트에 병합
for (const resItem of calcRes.resources.list) {
costnUpdates.push({
CODE: resItem.code,
NAME: resItem.name,
TAMT1: resItem.adoptedPrice,
LAMT1: resItem.kind === "노무비" ? resItem.adoptedPrice : 0,
MAMT1: resItem.kind === "재료비" ? resItem.adoptedPrice : 0,
SAMT1: resItem.kind === "경비" ? resItem.adoptedPrice : 0,
});
}
const boqUpdates = calcRes.boqOrig.rows
.filter((r) => r._recno !== undefined)
.map((r) => ({
_recno: r._recno,
NAME1: r.name,
QTY1: r.qty,
MAMT1: r.unitInfo.M,
LAMT1: r.unitInfo.L,
SAMT1: r.unitInfo.S,
TAMT1: r.unitInfo.T,
}));
const prjlstUpdates = {
MCOST1: k.dM,
LCOST1: k.dL,
SCOST1: k.dS,
TCOST1: k.dT,
COSTW1: k.grandTotal,
};
const rates = state.kanjubConfig.rates;
const smtUpdates = {
"1_R_KANNO": rates.KANNO,
"1_R_SANBO": rates.SANBO,
"1_R_KOBO": rates.KOBO,
"1_R_KUNBO": rates.KUNBO,
"1_R_SLVBO": rates.SLVBO,
"1_R_YUNBO": rates.YUNBO,
"1_R_ANJUN": rates.ANJUN,
"1_R_T_KITA": rates.KITA,
"1_R_HWAN": rates.HWAN,
"1_R_EQIPSU": rates.EQIP,
"1_R_ILBAN": rates.ILBAN,
"1_R_IYUN": rates.IYUN,
"1_C_KANNO": k.kanno,
"1_C_ILBAN": k.ilbanAmt,
"1_C_IYUN_M": k.pretaxProfitAdj,
"1_C_KONGSA_J": k.contractBeforeAdj,
"1_C_UPDNKWN": k.govFinal,
};
const resp = await fetch(`${API_BASE}/api/save`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
source_path: srcPath,
target_folder_name: dbCtx.targetFolderName || "stmate_edited",
costn_updates: costnUpdates,
boq_updates: boqUpdates,
prjlst_updates: prjlstUpdates,
smt_updates: smtUpdates,
}),
});
if (!resp.ok) throw new Error(await resp.text());
const saved = await resp.json();
dbCtx.lastSaveResult = saved;
dbCtx.statusMsg = `DBF·STC 저장 완료: ${saved.saved_stc} (${fmt(saved.stc_size, 0)} bytes)`;
await checkAndLoadSources();
} catch (err) {
dbCtx.statusMsg = `DBF 저장 실패: ${err.message || err}`;
}
if (onUpdate) onUpdate();
}
function renderDbTab(state, calcRes) {
const projOpts = dbCtx.sources.projects
.map(
(p) =>
`<option value="${esc(p.path)}" ${dbCtx.selectedProjectPath === p.path ? "selected" : ""}>${esc(p.name)} — ${esc(p.path)}</option>`,
)
.join("");
const jbrOpts = dbCtx.sources.jbr_files
.map(
(f) =>
`<option value="${esc(f.path)}" ${dbCtx.selectedJbrPath === f.path ? "selected" : ""}>${esc(f.name)} (${fmt(f.size, 0)} B)</option>`,
)
.join("");
let dbfPreviewHtml =
'<div class="empty-box">상단에서 PC 내 STMATE 폴더 또는 <code>.STC</code> 파일을 선택하고 <strong>[원본 DBF/STC 불러오기]</strong>를 실행하세요.</div>';
if (dbCtx.loadedDbf) {
const d = dbCtx.loadedDbf;
const costnRows = (d.costn_rows || [])
.slice(0, 25)
.map(
(r, i) => `
<tr>
<td><code>${esc(r.CODE)}</code></td>
<td><input type="text" data-dbf="costn" data-idx="${i}" data-field="NAME" value="${esc(r.NAME)}"></td>
<td>${esc(r.SIZE)}</td>
<td class="center-col">${esc(r.UNIT)}</td>
<td class="num-col"><input type="number" data-dbf="costn" data-idx="${i}" data-field="MAMT1" value="${Number(r.MAMT1) || 0}" step="any"></td>
<td class="num-col"><input type="number" data-dbf="costn" data-idx="${i}" data-field="LAMT1" value="${Number(r.LAMT1) || 0}" step="any"></td>
<td class="num-col"><input type="number" data-dbf="costn" data-idx="${i}" data-field="SAMT1" value="${Number(r.SAMT1) || 0}" step="any"></td>
<td class="num-col"><input type="number" data-dbf="costn" data-idx="${i}" data-field="TAMT1" value="${Number(r.TAMT1) || 0}" step="any"></td>
<td class="money-col">${fmt(r.AMT_61, 0)}</td>
</tr>`,
)
.join("");
dbfPreviewHtml = `
<div class="sub-toolbar">
<span>현재 연결 원본: <strong>${esc(d.prjlst?.PNAME || d.source_path)}</strong> · 수록 테이블: <code>${esc(d.file_list.join(", "))}</code></span>
<span>원장 <code>COSTN.DBF</code> <strong>${d.costn_count}건</strong> · 내역 <code>${esc(d.boq_table_name)}</code> <strong>${d.boq_count}건</strong> · <code>PRJLST.SMT</code> 설정키 <strong>${Object.keys(d.prjlst_smt || {}).length}개</strong></span>
</div>
<div class="table-scroll">
<table class="data-table">
<thead><tr><th>CODE</th><th>품명(<code>NAME</code>)</th><th>규격(<code>SIZE</code>)</th><th>단위</th><th>재료단가(<code>MAMT1</code>)</th><th>노무단가(<code>LAMT1</code>)</th><th>경비단가(<code>SAMT1</code>)</th><th>합계단가(<code>TAMT1</code>)</th><th>채택6칸(<code>AMT_61</code>)</th></tr></thead>
<tbody>${costnRows}</tbody>
</table>
</div>`;
}
const saveAlert = dbCtx.lastSaveResult
? `<div class="sub-toolbar"><span>최근 쓰기 완료 폴더: <code>${esc(dbCtx.lastSaveResult.saved_dir)}</code></span><span>생성된 패키지: <code>${esc(dbCtx.lastSaveResult.saved_stc)}</code> (${fmt(dbCtx.lastSaveResult.stc_size, 0)} bytes, 파일 ${dbCtx.lastSaveResult.written_count}종)</span></div>`
: "";
return `
<div class="panel-card">
<div class="card-header">
<div>
<h2>PC 내 STMATE 원본 DBF · SMT · STC · JB_R 직접 읽기/쓰기 연동</h2>
<p><code>C:\\경영정보(주)</code> 설치 폴더의 프로젝트 DBF(<code>COSTN.DBF</code>, <code>BOQ11.DBF</code>, <code>PRJLST.DBF/.SMT</code>), <code>System\\ST_BSC\\*.JB_R</code> 제비율 기준 DB, 그리고 실무 <code>.STC</code> 패키지를 직접 읽고 CP949 고정폭 바이너리 DBF로 다시 기록합니다.</p>
</div>
<div>
<button type="button" class="btn secondary" data-db-action="refresh-sources">연결/목록 새로고침</button>
</div>
</div>
<div class="sub-toolbar">
<span>상태: <strong>${esc(dbCtx.statusMsg)}</strong></span>
<span>브리지 주소: <code>${API_BASE}</code></span>
</div>
<div class="control-grid">
<label class="ctrl-item">
<span>1. 프로젝트 원본 DB 폴더 또는 <code>.STC</code> 선택</span>
<select id="selDbProject">${projOpts || '<option value="">감지된 프로젝트 없음</option>'}</select>
</label>
<div class="ctrl-item" style="justify-content:flex-end;">
<button type="button" class="btn primary" data-db-action="load-project" ${!dbCtx.connected ? "disabled" : ""}>원본 DBF/STC 불러오기 (시뮬레이터 동기화)</button>
</div>
<label class="ctrl-item">
<span>2. 시스템 제비율 DB (<code>C:\\경영정보(주)\\System\\ST_BSC\\*.JB_R</code>) 선택</span>
<select id="selDbJbr">${jbrOpts || '<option value="">감지된 JB_R 없음</option>'}</select>
</label>
<div class="ctrl-item" style="justify-content:flex-end;">
<button type="button" class="btn secondary" data-db-action="load-jbr" ${!dbCtx.connected ? "disabled" : ""}>선택한 .JB_R 제비율 읽어서 원가계산 적용</button>
</div>
<label class="ctrl-item">
<span>3. DBF/STC 쓰기 대상 폴더명 (<code>tools/stmate-test/work_dbf/</code> 하위)</span>
<input type="text" id="inpTargetFolder" value="${esc(dbCtx.targetFolderName)}">
</label>
<div class="ctrl-item" style="justify-content:flex-end;">
<button type="button" class="btn primary" data-db-action="save-dbf" ${!dbCtx.connected ? "disabled" : ""}>현재 단가·내역·원가 결과를 DBF 및 .STC로 직접 쓰기</button>
</div>
</div>
${saveAlert}
</div>
<div class="panel-card">
<div class="card-header">
<div>
<h2>원본 <code>COSTN.DBF</code> 레코드 직접 조회 및 편집 (상위 25건 미리보기)</h2>
<p>여기서 수정한 단가와 01~06 탭에서 조작한 수량·요율·원가 결과가 <strong>[DBF 및 .STC로 직접 쓰기]</strong> 실행 시 <code>COSTN.DBF</code>, <code>BOQ11/12.DBF</code>, <code>PRJLST.DBF/.SMT</code>에 기록됩니다.</p>
</div>
</div>
${dbfPreviewHtml}
</div>`;
}
function initEvents(getState, setState, saveAndRender) {
document.addEventListener("click", (e) => {
const btn = e.target.closest("[data-db-action]");
if (!btn) return;
const act = btn.dataset.dbAction;
if (act === "refresh-sources") {
checkAndLoadSources(saveAndRender);
} else if (act === "load-project") {
loadProjectDbf((newState) => {
setState(newState);
saveAndRender();
}, saveAndRender);
} else if (act === "load-jbr") {
loadAndApplyJbr(getState(), saveAndRender);
} else if (act === "save-dbf") {
const st = getState();
saveToDbf(st, StmateCalc.calculateAll(st), saveAndRender);
}
});
document.addEventListener("change", (e) => {
const el = e.target;
if (el.id === "selDbProject") dbCtx.selectedProjectPath = el.value;
else if (el.id === "selDbJbr") dbCtx.selectedJbrPath = el.value;
else if (el.id === "inpTargetFolder") dbCtx.targetFolderName = el.value;
else if (el.dataset.dbf === "costn" && dbCtx.loadedDbf?.costn_rows) {
const idx = Number(el.dataset.idx);
const fld = el.dataset.field;
if (dbCtx.loadedDbf.costn_rows[idx]) {
dbCtx.loadedDbf.costn_rows[idx][fld] = el.value;
}
}
});
checkAndLoadSources(saveAndRender);
}
return {
dbCtx,
initEvents,
checkAndLoadSources,
loadProjectDbf,
loadAndApplyJbr,
saveToDbf,
renderDbTab,
};
})();
+2
View File
@@ -7,6 +7,7 @@
<link rel="stylesheet" href="style.css" />
<script defer src="calc.js"></script>
<script defer src="app_views_cost.js"></script>
<script defer src="db_bridge.js"></script>
<script defer src="app.js"></script>
</head>
<body>
@@ -47,6 +48,7 @@
<section class="kpi-grid" id="kpiSummary" aria-label="핵심 계산 요약"></section>
<nav class="tab-nav" id="mainTabs" aria-label="STMATE 기능 단계">
<button type="button" data-tab="db">00 원본 DBF · STC 연동</button>
<button type="button" data-tab="resources" class="active">01 기초자원 · 노임환산</button>
<button type="button" data-tab="machines">02 중기 · 단가산출(Q식)</button>
<button type="button" data-tab="ilwi">03 일위대가 조립</button>
+687
View File
@@ -0,0 +1,687 @@
# -*- 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)