Files
Aislo/resources/master_data/scripts/build_용도.py
T

224 lines
7.2 KiB
Python

# -*- coding: utf-8 -*-
"""소요량 · 계수 표마다 「용도」(공종 · 대상 · 로직키) 채우기.
담당: 소요량_*.json · 계수_*.json (읽기+쓰기) · 로직_*.json · 원문 md (읽기만).
다시 돌리면 같은 결과 — 순수 함수 + 정렬.
"""
import glob
import json
import re
from pathlib import Path
ROOT = Path(__file__).resolve().parents[3]
MASTER = ROOT / "resources" / "master_data"
ORIG_KS = ROOT / "resources" / "knowledge" / "original" / "원가계산" / "건설공사_표준품셈" / "본문"
ORIG_SR = ROOT / "resources" / "knowledge" / "original" / "원가계산" / "산림_표준품셈" / "본문"
CODE_RE = re.compile(r"\b([A-Z]{2}\d{6})\b")
HEADING_RE = re.compile(r"^#{2,4}\s+([0-9]+(?:-[0-9]+)*)\.?\s+(.+?)\s*$")
TRAIL_PAREN_RE = re.compile(r"(\s*\([^()]*\))+\s*$")
_heading_cache = {}
# 원문번호 절 이름이 표 내용과 안 맞는 예외 — 검증(ref/_검증_기계_용도.md) 반영.
OVERRIDE_공종 = {
"CC000074": "불도저·굴착기", # 8-4-8 은 「[70]기타기계」 절이나 내용은 불도저·굴착기 잡재료 비율
}
def _find_부문_dir(부문):
if not ORIG_KS.exists():
return None
for d in sorted(ORIG_KS.iterdir()):
if d.is_dir() and d.name.endswith(부문 + "부문"):
return d
return None
def _find_장_dir(base, ):
m = re.match(r"(\d+)", or "")
if not m or not base or not base.exists():
return None
num = m.group(1).zfill(2)
for d in sorted(base.iterdir()):
if d.is_dir() and d.name.startswith(f"제{num}장"):
return d
return None
def _heading_index(folder):
key = str(folder)
if key in _heading_cache:
return _heading_cache[key]
idx = {}
if folder and folder.exists():
for md in sorted(folder.glob("*.md")):
for line in md.read_text(encoding="utf-8").splitlines():
m = HEADING_RE.match(line)
if not m:
continue
num, text = m.group(1), m.group(2)
text = TRAIL_PAREN_RE.sub("", text).strip()
idx.setdefault(num, text)
_heading_cache[key] = idx
return idx
def resolve_공종(원문, 부문, , 원문번호):
if 원문 == "건설품셈":
base = _find_부문_dir(부문 or "")
folder = _find_장_dir(base, or "")
elif 원문 == "산림품셈":
folder = _find_장_dir(ORIG_SR, or "")
else:
folder = None
idx = _heading_index(folder)
num = 원문번호 or ""
parts = num.split("-")
while parts:
cand = "-".join(parts)
if cand in idx:
return idx[cand]
parts = parts[:-1]
return None
def _load_names(path, key="이름"):
d = json.load(open(path, encoding="utf-8"))
return set(r[key] for r in d["줄"] if r.get(key))
def load_masters():
인력_이름 = _load_names(MASTER / "인력.json")
기계_d = json.load(open(MASTER / "기계.json", encoding="utf-8"))
기계_이름 = set()
for r in 기계_d["줄"]:
기계_이름.add(r["이름"])
if r.get("상세구분"):
기계_이름.add(r["상세구분"])
재료_이름 = _load_names(MASTER / "재료_품셈재료.json")
return 인력_이름, 기계_이름, 재료_이름
def _strip_spec(key):
return key.split("(", 1)[0].strip()
def classify_대상(item, 인력_이름, 기계_이름, 재료_이름):
대상 = set()
이름 = item.get("이름", "")
if "할증" in 이름:
대상.add("할증")
if "환산" in 이름:
대상.add("환산")
조건 = item.get("조건") or {}
if 조건:
대상.add("작업량 조건")
값칸 = item.get("값칸") or {}
keys = list(값칸.keys())
row_vals = set()
for row in item.get("줄") or []:
if isinstance(row, dict):
for v in row.values():
if isinstance(v, str):
row_vals.add(v)
candidates = set(keys) | row_vals
for c in candidates:
base = _strip_spec(c)
if c in 인력_이름 or base in 인력_이름:
대상.add("인력")
elif c in 기계_이름 or base in 기계_이름:
대상.add("기계")
elif c in 재료_이름 or base in 재료_이름:
대상.add("재료")
for k in keys:
if "시공량" in k:
대상.add("작업량 조건")
if not 대상:
대상.add("작업량 조건")
return sorted(대상)
def build_로직_index():
idx = {}
for fp in sorted(glob.glob(str(MASTER / "로직_*.json"))):
d = json.load(open(fp, encoding="utf-8"))
for row in d.get("줄", []):
own = row.get("키")
text = json.dumps(row, ensure_ascii=False)
for code in set(CODE_RE.findall(text)):
if code == own:
continue
idx.setdefault(code, set()).add(own)
return idx
def main():
인력_이름, 기계_이름, 재료_이름 = load_masters()
로직_idx = build_로직_index()
files = sorted(glob.glob(str(MASTER / "소요량_*.json"))) + sorted(
glob.glob(str(MASTER / "계수_*.json"))
)
총_항목 = 0
공종_미확인 = []
for fp in files:
d = json.load(open(fp, encoding="utf-8"))
원문 = d.get("원문")
부문 = d.get("부문")
용도_map = {}
for item in d.get("표", []):
총_항목 += 1
= item["키"]
= item.get("상세구분")
if in OVERRIDE_공종:
공종 = OVERRIDE_공종[]
else:
공종 = resolve_공종(원문, 부문, , item.get("원문번호"))
if 공종 is None:
공종_미확인.append((, fp, item.get("원문번호")))
공종 = item.get("이름", "").split("(", 1)[0].strip()
대상 = classify_대상(item, 인력_이름, 기계_이름, 재료_이름)
로직키 = sorted(로직_idx.get(, []))
용도_map[] = {"공종": 공종, "대상": 대상, "로직키": 로직키}
text = open(fp, encoding="utf-8").read()
lines = text.split("\n")
out = []
cur_키 = None
key_re = re.compile(r'"키":\s*"([A-Z]{2}\d{6})"')
src_re = re.compile(r'^(\s*)"출처":\s*".*",\s*$')
old_용도_re = re.compile(r'^\s*"용도":\s*\{.*\},\s*$')
for line in lines:
if old_용도_re.match(line):
continue
m = key_re.search(line)
if m:
cur_키 = m.group(1)
out.append(line)
sm = src_re.match(line)
if sm and cur_키 in 용도_map:
indent = sm.group(1)
payload = json.dumps(용도_map[cur_키], ensure_ascii=False)
out.append(f'{indent}"용도": {payload},')
new_text = "\n".join(out)
with open(fp, "w", encoding="utf-8") as f:
f.write(new_text)
print("총 항목:", 총_항목, "/ 공종 미확인:", len(공종_미확인))
if 공종_미확인:
for , fp, 번호 in 공종_미확인[:30]:
print(" 미확인:", , 번호, fp)
if __name__ == "__main__":
main()