126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""수량산출서·수량코드·표준중기공식 끼워 넣기 API.
|
|
|
|
근거: 지식DB STmate/old/20_분석 28 §1.1(중기공식 등록·수정·삭제) · §2(수량산출서 · 기본/확장 수량변수) ·
|
|
02(QCODE 15필드) · REA 28 TWQTYLIST · REA 30 TWM_EDIT_EQ(명칭 · 적용 공식/공식+코드 · EQSYNC).
|
|
쓰기는 tools/stmate-test/work_dbf/ 아래만.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pathlib
|
|
import re
|
|
import zipfile
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from qty_qcode import build_qcode, read_qcode
|
|
|
|
router = APIRouter()
|
|
HERE = pathlib.Path(__file__).resolve().parent
|
|
|
|
|
|
def _work_root() -> pathlib.Path:
|
|
import sys
|
|
|
|
srv = sys.modules.get("server") # 실행 중 서버가 있으면 그 작업 폴더(시험은 임시 폴더로 바꿈)
|
|
return pathlib.Path(getattr(srv, "WORK_DBF_ROOT", HERE / "work_dbf"))
|
|
|
|
|
|
def _read_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}")
|
|
if p.is_file() and p.suffix.upper() == ".STC":
|
|
with zipfile.ZipFile(p) as zf:
|
|
return {pathlib.Path(n).name.upper(): zf.read(n) for n in zf.namelist()}
|
|
if p.is_dir():
|
|
return {fp.name.upper(): fp.read_bytes() for fp in p.iterdir() if fp.is_file()}
|
|
raise HTTPException(status_code=400, detail="지원하지 않는 경로 형식")
|
|
|
|
|
|
class QcodeReq(BaseModel):
|
|
source_path: str
|
|
|
|
|
|
class QcodeSaveReq(BaseModel):
|
|
source_path: str
|
|
target_folder_name: str = "stmate_edited"
|
|
rows: list[dict] = []
|
|
|
|
|
|
@router.post("/api/qty/qcode")
|
|
def api_qty_qcode(req: QcodeReq):
|
|
"""프로젝트의 수량코드(QCODE) 행 — 메모(IFORM · XCODE)는 글자로."""
|
|
return read_qcode(_read_files(req.source_path))
|
|
|
|
|
|
@router.post("/api/qty/qcode-save")
|
|
def api_qty_qcode_save(req: QcodeSaveReq):
|
|
"""수량코드 저장 — work_dbf/<이름>/QCODE.DBF·SMT 와 같은 이름 STC 안의 두 파일을 바꿈."""
|
|
bad = [
|
|
r.get("CODE")
|
|
for r in req.rows
|
|
if not re.fullmatch(r"Q\d{5}", str(r.get("CODE", "")).strip())
|
|
]
|
|
if bad:
|
|
raise HTTPException(status_code=400, detail=f"수량코드 형식 오류: {bad[:5]}")
|
|
files = _read_files(req.source_path)
|
|
dbf, smt = build_qcode(req.rows, files)
|
|
safe = (
|
|
re.sub(r"[^\w\-가-힣]", "_", req.target_folder_name or "stmate_edited").strip("_")
|
|
or "stmate_edited"
|
|
)
|
|
root = _work_root()
|
|
out_dir = root / safe
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
(out_dir / "QCODE.DBF").write_bytes(dbf)
|
|
(out_dir / "QCODE.SMT").write_bytes(smt)
|
|
stc = root / f"{safe}.STC"
|
|
base = _read_files(str(stc)) if stc.exists() else files
|
|
base.update({"QCODE.DBF": dbf, "QCODE.SMT": smt})
|
|
with zipfile.ZipFile(stc, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
|
for name, content in base.items():
|
|
zf.writestr(name, content)
|
|
return {"ok": True, "saved_dir": str(out_dir), "saved_stc": str(stc), "count": len(req.rows)}
|
|
|
|
|
|
# 표준중기공식(EQSYNC) — 설치 폴더 공용 등록부 · 여기서는 work_dbf/EQSYNC.json
|
|
class EqItem(BaseModel):
|
|
name: str
|
|
text: str = ""
|
|
ins_x: int = 0 # 0 공식 · 1 공식+코드
|
|
|
|
|
|
class EqSaveReq(BaseModel):
|
|
items: list[EqItem] = []
|
|
|
|
|
|
def _eq_path() -> pathlib.Path:
|
|
return _work_root() / "EQSYNC.json"
|
|
|
|
|
|
@router.get("/api/qty/eq")
|
|
def api_qty_eq():
|
|
p = _eq_path()
|
|
items = json.loads(p.read_text(encoding="utf-8")) if p.exists() else []
|
|
return {"items": items}
|
|
|
|
|
|
@router.post("/api/qty/eq")
|
|
def api_qty_eq_save(req: EqSaveReq):
|
|
names = [i.name.strip() for i in req.items]
|
|
if any(not n for n in names):
|
|
raise HTTPException(status_code=400, detail="중기공식 명칭이 비어 있음")
|
|
if len(set(names)) != len(names):
|
|
raise HTTPException(status_code=400, detail="같은 명칭의 중기공식이 둘 이상")
|
|
p = _eq_path()
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(
|
|
json.dumps([i.model_dump() for i in req.items], ensure_ascii=False, indent=1),
|
|
encoding="utf-8",
|
|
)
|
|
return {"ok": True, "count": len(req.items)}
|