Merge remote-tracking branch 'origin/sub_desktop_1' into sub_laptop_2

2026-09-20 22:02:21 +09:00
parent d8fa316dd9 0b794cea7d
commit e54db25302
85 changed files with 3545 additions and 302 deletions
@@ -0,0 +1,214 @@
+# -*- 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 = {}
+
+
+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("상세구분")
+            공종 = 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()