feat(master_data): 관급 제조사별 최저가 채움 · 조달청 2차 받기 · 짝짓기 갈래 · 새 줄 380

- 제조사별로 값이 갈리면 같은 규격 최저가 + 비고에 제조사 (보류 394 채움) - 조달청 분류 23개 더 받음 (원본 gz 보관) · 시멘트 · 조경석 · 주철관 · 이음관 · 골재 · 측구수로관 · 볼라드 · 맨홀 · 용접철망 · 돌망태 · 매트 짝짓기 - 조달청에만 있는 금속돌망태 372 · 토목용부직포 8 을 자재품목에 새 줄로 들임 (키대장 갱신) - 다산 값 대조 기준 ±5% · 짝짓기 시험 11건 · 설계 문서 갱신 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PAdA5ThqmtVSsbzjusk1cJ
2026-09-20 18:06:15 +09:00
parent ce38094e91
commit e44365365b
38 changed files with 2061 additions and 946 deletions
@@ -4,7 +4,7 @@
  ./venv/Scripts/python.exe resources/master_data/scripts/update_관급단가.py update

  fetch         받기 → 원본 보관 (resources/knowledge/original/원가계산/자재단가/관급/<날짜>/ · gz + _meta.md)
-  test [날짜]   관급 열을 지우고 다시 채워 옛 값(다산소프트)과 맞댐 — 정본 안 덮음 · tmp/관급_대조.json
+  test [날짜]   관급 열을 지우고 다시 채워 옛 값(다산소프트)과 맞댐(±5% 안 = 맞음) — 정본 안 덮음 · tmp/관급_대조.json
  apply [날짜]  정본 관급 열(관급 · 비고)만 받은 값으로 바꿈 · tmp/관급_바뀐줄.json
 """

@@ -12,6 +12,7 @@ import gzip
 import json
 import re
 import sys
+import time
 import urllib.parse
 from datetime import date
 from pathlib import Path
@@ -21,6 +22,8 @@ PIPE = ROOT / "resources/knowledge/original/_pipeline"
 RAW = ROOT / "resources/knowledge/original/원가계산/자재단가/관급"
 sys.path.insert(0, str(PIPE))
 import collect_cost_sources as cc  # noqa: E402  (키 읽기 · fetch)
+import master_keys as mk  # noqa: E402  (새 줄 키)
+import update_관급단가_기타 as misc  # noqa: E402  (「그 밖」 갈래 짝짓기)

 BASE = "https://apis.data.go.kr/1230000/at/ShoppingMallPrdctInfoService/"
 MAS, UNIT, THPTY = (
@@ -41,6 +44,38 @@ JOBS = [  # 분류명은 부분 글자 조회 — 「철근」은 암거블록
    (THPTY, "파형강관"),
    (MAS, "콘크리트관"),
    (UNIT, "콘크리트관"),
+    # 2차 — 다산소프트 「그 밖」 관급 · 품셈재료 중 조달청에 있는 것
+    (MAS, "시멘트"),
+    (MAS, "골재"),
+    (MAS, "조경석"),
+    (UNIT, "주철관"),
+    (MAS, "파형강관이음관"),
+    (THPTY, "파형강관이음관"),
+    (MAS, "배수로"),
+    (THPTY, "배수로"),
+    (MAS, "블록"),
+    (UNIT, "블록"),
+    (THPTY, "블록"),
+    (MAS, "돌망태"),
+    (THPTY, "돌망태"),
+    (MAS, "합성목재"),
+    (MAS, "볼라드"),
+    (MAS, "교량받침"),
+    (THPTY, "교량받침"),
+    (MAS, "용접철망"),
+    (THPTY, "섬유"),
+    (MAS, "폴리에틸렌"),
+    (THPTY, "폴리에틸렌"),
+    (MAS, "맨홀"),
+    (MAS, "매트"),
+    (THPTY, "매트"),
+    (MAS, "낙석"),
+    (THPTY, "그레이팅"),
+    (MAS, "벽돌"),
+    (THPTY, "시트"),
+    (MAS, "비료"),
+    (MAS, "압륜"),
+    (THPTY, "부직포"),
 ]
 # 보관하는 칸 — 나머지(첨부·원산지·이미지·긴 설명)는 버림(원본이 수백 MB)
 KEEP = (
@@ -53,10 +88,13 @@ KEEP = (
 def call(op: str, key: str, **kw) -> dict:
    q = "&".join(f"{k}={urllib.parse.quote(str(v))}" for k, v in kw.items())
    url = f"{BASE}{op}?serviceKey={key}&type=json&{q}"
-    try:
-        return json.loads(cc.fetch(url, timeout=90).decode("utf-8", "replace"))["response"]
-    except Exception as e:  # 오류 글에 받은 주소(키 포함)가 실려도 키는 가림
-        raise RuntimeError(f"{op} 호출 실패: {str(e).replace(key, '***')}") from None
+    for tries in range(5):  # 큰 분류는 시간 초과가 잦음 — 다섯 번까지
+        try:
+            return json.loads(cc.fetch(url, timeout=90).decode("utf-8", "replace"))["response"]
+        except Exception as e:  # 오류 글에 받은 주소(키 포함)가 실려도 키는 가림
+            if tries == 4:
+                raise RuntimeError(f"{op} 호출 실패: {str(e).replace(key, '***')}") from None
+            time.sleep(5)


 def fetch_all() -> Path:
@@ -72,6 +110,11 @@ def fetch_all() -> Path:
        "|---|---|---|---|",
    ]
    for op, name in JOBS:
+        f = f"{op.removeprefix('get').removesuffix('PrdctInfoList')}_{name}.json.gz"
+        if (out / f).exists():  # 같은 날 이미 받은 것은 건너뜀(이어받기)
+            n = len(json.loads(gzip.decompress((out / f).read_bytes())))
+            lines.append(f"| {f} | {op} | {name} | {n:,} |")
+            continue
        rows, page, total = [], 1, 1
        while len(rows) < total:
            r = call(op, key, inqryDiv=2, prdctClsfcNoNm=name, numOfRows=999, pageNo=page)
@@ -84,7 +127,6 @@ def fetch_all() -> Path:
                break
            rows += [{k: it.get(k) for k in KEEP} for it in items]
            page += 1
-        f = f"{op.removeprefix('get').removesuffix('PrdctInfoList')}_{name}.json.gz"
        (out / f).write_bytes(gzip.compress(json.dumps(rows, ensure_ascii=False).encode("utf-8")))
        lines.append(f"| {f} | {op} | {name} | {len(rows):,} (totalCount {total:,}) |")
        print(f, len(rows), total, flush=True)
@@ -216,6 +258,7 @@ class Matcher:
    """자재품목 줄 → 조달청 줄. 짝이 서면 (줄, 방식), 못 서면 (None, 까닭)."""

    def __init__(self, raw: dict):
+        self.raw = raw
        self.rmc = {}
        for r in raw.get("MASCntrct_레미콘", []):
            if r["prdctMakrNm"] == "조합공통품목":
@@ -335,7 +378,7 @@ class Matcher:
                if dia and _dia(r["prdctSpecNm"]) == dia and "보통관" in r["prdctSpecNm"]
            ]
            return pick(c)
-        return None, "이 갈래는 아직 짝짓기 없음"
+        return misc.find(row, self.raw, places, pick, PROVINCE)


 FAMILY = (
@@ -349,13 +392,30 @@ FAMILY = (


 def family(name: str) -> str:
-    return next((k for k, pat in FAMILY if re.match(pat, name)), "그 밖")
+    return next((k for k, pat in FAMILY if re.match(pat, name)), misc.label(name) or "그 밖")
+
+
+def info(r: dict) -> dict:
+    return {
+        "방식": r["cntrctMthdNm"],
+        "품명": r["prdctSpecNm"],
+        "지역": r["prdctSplyRgnNm"],
+        "계약시작": r["cntrctBgnDate"],
+        "부가세": r["vatAplDivNm"],
+        "인도": r["prdctDlvryCndtnNm"],
+        "단위": r["prdctUnit"],
+        "제조사": r["cntrctCorpNm"] if r["prdctMakrNm"] == "조합공통품목" else r["prdctMakrNm"],
+    }


 def refill(rows: list[dict], raw: dict) -> list[dict]:
    """조달 값(관급 열)이 있는 줄마다 조달청 값으로 다시 채운 결과."""
    m, out = Matcher(raw), []
    for x in rows:
+        if str(x.get("원문번호")).startswith(
+            "조달청 "
+        ):  # 조달청에서 새로 들인 줄은 아래 `add_rows` 몫
+            continue
        if x["관급"] is None and not (x["비고"] or "").startswith(
            "관급:"
        ):  # 지난번에 못 채운 줄도 다시 시도
@@ -371,17 +431,7 @@ def refill(rows: list[dict], raw: dict) -> list[dict]:
                "갈래": family(x["이름"]),
                "새값": int(r["cntrctPrceAmt"]) if r else None,
                "까닭": why,
-                "조달청": None
-                if not r
-                else {
-                    "방식": r["cntrctMthdNm"],
-                    "품명": r["prdctSpecNm"],
-                    "지역": r["prdctSplyRgnNm"],
-                    "계약시작": r["cntrctBgnDate"],
-                    "부가세": r["vatAplDivNm"],
-                    "인도": r["prdctDlvryCndtnNm"],
-                    "단위": r["prdctUnit"],
-                },
+                "조달청": info(r) if r else None,
            }
        )
    return out
@@ -397,7 +447,8 @@ def test(day: str | None = None) -> list[dict]:
    tab: dict = {}
    for r in res:
        t = tab.setdefault(
-            r["갈래"], {"전체": 0, "짝": 0, "같음": 0, "다름": 0, "못채움": 0, "까닭": {}}
+            r["갈래"],
+            {"전체": 0, "짝": 0, "맞음": 0, "다름": 0, "최저채움": 0, "못채움": 0, "까닭": {}},
        )
        t["전체"] += 1
        if r["새값"] is None:
@@ -405,7 +456,11 @@ def test(day: str | None = None) -> list[dict]:
            t["까닭"][r["까닭"]] = t["까닭"].get(r["까닭"], 0) + 1
        else:
            t["짝"] += 1
-            t["같음" if r["새값"] == r["옛값"] else "다름"] += 1
+            ok = (
+                r["옛값"] and abs(r["새값"] / r["옛값"] - 1) <= 0.05
+            )  # 받는 시점 차이는 ±5% 안이면 맞음
+            t["맞음" if ok else "다름"] += 1
+            t["최저채움"] += "최저" in r["까닭"]
    for k, v in tab.items():
        print(
            k,
@@ -423,21 +478,75 @@ METHOD = {


 def decide(x: dict, r: dict, day: str) -> tuple[int | None, str]:
-    """(새 관급 값, 새 비고) — 값은 짝이 서고 후보 값이 하나일 때만(제조사별로 값이 갈리면 null · 고르는 기준 미정)."""
+    """(새 관급 값, 새 비고) — 제조사별로 값이 갈리면 같은 규격의 가장 낮은 값 · 비고에 그 제조사."""
    if r["새값"] is None:
        return None, f"관급: 조달청 값 못 채움 — {r['까닭']} ({day})"
-    if "최저" in r["까닭"]:
-        return None, f"관급: 조달청 값 못 채움 — 제조사별로 값이 갈림 · 고르는 기준 미정 ({day})"
    c = r["조달청"]
+    who = f" {c['제조사']} 최저" if "최저" in r["까닭"] else ""
    return r["새값"], (
-        f"관급: 조달청 {METHOD.get(c['방식'], c['방식'])} {day} · 계약 {c['계약시작']}~ · {c['지역']} · {c['인도']} · {c['부가세']}"
+        f"관급: 조달청 {METHOD.get(c['방식'], c['방식'])}{who} {day} · 계약 {c['계약시작']}~ · {c['지역']} · {c['인도']} · {c['부가세']}"
    )


+def add_rows(lines: list[str], data: dict, raw: dict, day: str, changed: list) -> list[str]:
+    """조달청에만 있는 품목 — 자재품목에 새 줄(관급만 · 물가지 네 열 null · 같은 품명·규격은 최저가). 이미 든 줄은 값·비고만 갈음."""
+    have = {x["원문번호"]: x for x in data["줄"] if str(x["원문번호"]).startswith("조달청 ")}
+    book, new = mk.load_book(), []
+    for num, (name, spec, rows) in misc.groups(raw).items():
+        r, how = pick(rows)
+        val, note = decide(
+            None, {"새값": int(r["cntrctPrceAmt"]), "까닭": how, "조달청": info(r)}, day
+        )
+        if num in have:
+            for i, line in enumerate(lines):
+                if f'"키": "{have[num]["키"]}"' in line:
+                    line = re.sub(
+                        r'"관급": (?:null|-?[\d.]+)', lambda _: f'"관급": {val}', line, count=1
+                    )
+                    lines[i] = re.sub(
+                        r'"비고": (?:null|"(?:[^"\\]|\\.)*")',
+                        lambda _: f'"비고": {json.dumps(note, ensure_ascii=False)}',
+                        line,
+                        count=1,
+                    )
+                    break
+            continue
+        key = mk.issue(book, "MT", num, "재료_자재품목.json")
+        row = {
+            "키": key, "원문번호": num, "구분": None, "상세구분": None, "이름": name, "규격": spec,
+            "단위": r["prdctUnit"], "물가자료": None, "유통물가": None, "물가정보": None,
+            "거래가격": None, "관급": val, "출처": f"자재품목 조달청 {r['cntrctMthdNm']}", "비고": note, "면수": None,
+        }  # fmt: skip
+        new.append(
+            "    { "
+            + ", ".join(f'"{k}": {json.dumps(v, ensure_ascii=False)}' for k, v in row.items())
+            + " }"
+        )
+        changed.append(
+            {
+                "키": key,
+                "이름": name,
+                "규격": spec,
+                "갈래": "새 줄",
+                "옛값": None,
+                "새값": val,
+                "비고": note,
+            }
+        )
+    if new:
+        end = max(i for i, line in enumerate(lines) if line.strip() == "]")
+        lines[end - 1] += ","
+        lines[end:end] = [ln + "," for ln in new[:-1]] + new[-1:]
+        mk.save_book(mk.finish(book))
+    return lines
+
+
 def apply(day: str | None = None) -> None:
    """정본 관급 열 · 비고만 바꿈 — 다산소프트 값은 하나도 안 남김(못 채운 줄 = null)."""
    day = day or latest()
    text = MASTER.read_bytes().decode("utf-8")
+    crlf = "\r\n" in text  # 작업 폴더 줄바꿈(CRLF)을 그대로 지킴
+    text = text.replace("\r\n", "\n")
    data = json.loads(text)
    found = {r["키"]: r for r in refill(data["줄"], load(day))}
    changed, edits = [], {}
@@ -475,7 +584,9 @@ def apply(day: str | None = None) -> None:
        assert n1 == n2 == 1, f"{m.group(1)} 줄 서식이 달라 멈춤"
        lines[i] = line
    assert not edits, f"정본에서 못 찾은 줄 {len(edits)}"
-    MASTER.write_bytes("\n".join(lines).encode("utf-8"))
+    lines = add_rows(lines, data, load(day), day, changed)
+    out = "\n".join(lines)
+    MASTER.write_bytes((out.replace("\n", "\r\n") if crlf else out).encode("utf-8"))
    TMP.mkdir(exist_ok=True)
    (TMP / "관급_바뀐줄.json").write_text(
        json.dumps(changed, ensure_ascii=False, indent=1), encoding="utf-8"
@@ -0,0 +1,143 @@
+"""관급 단가 — 「그 밖」 갈래 짝짓기(주철관 · 조경석 · 골재 · 시멘트 · 이음관 …) — `update_관급단가.py` 가 부름.
+
+짝 = 자재품목 규격의 숫자 조각(치수·종별)이 조달청 품명·규격에 모두 있고, 글자 조각(제품·산지)도 모두 있을 때.
+이름 쪽 글자는 점수만(같은 값이면 더 많이 맞는 줄 먼저). 지역이 이름에 있으면 그 지역 줄만. 여럿이면 가장 낮은 값(호출한 쪽 `pick`).
+"""
+
+import re
+
+# (갈래, 이름 패턴, 받은 파일들) — 앞에서부터 처음 맞는 갈래
+FAMILIES = (
+    ("주철관", r"^(KP|타이튼)\s*주철직관", ["Ucntrct_주철관"]),
+    ("주철관부속", r"^(이탈방지압륜|고무링,타이튼)", ["MASCntrct_압륜"]),
+    (
+        "파형강관이음관",
+        r"^파형강관(이음관|\s*커플링)",
+        ["MASCntrct_파형강관이음관", "ThptyUcntrct_파형강관이음관"],
+    ),
+    ("조경석", r"^조경석", ["MASCntrct_조경석"]),
+    ("골재", r"^도로용혼합골재", ["MASCntrct_골재"]),
+    ("시멘트", r"^시멘트", ["MASCntrct_시멘트"]),
+    ("수로관", r"^측구수로관", ["MASCntrct_배수로", "ThptyUcntrct_배수로"]),
+    (
+        "블록",
+        r"^(소형고압블록|생태옹벽블록|생태어소블록)",
+        ["MASCntrct_블록", "Ucntrct_블록", "ThptyUcntrct_블록"],
+    ),
+    ("교량받침", r"^교량받침", ["MASCntrct_교량받침", "ThptyUcntrct_교량받침"]),
+    ("합성목재", r"^합성목재", ["MASCntrct_합성목재"]),
+    ("돌망태", r"(돌망태|개비온)", ["MASCntrct_돌망태", "ThptyUcntrct_돌망태"]),
+    ("볼라드", r"^볼라드", ["MASCntrct_볼라드"]),
+    ("매트", r"^(보행매트|고무매트)", ["MASCntrct_매트", "ThptyUcntrct_매트"]),
+    ("용접철망", r"^와이어메쉬", ["MASCntrct_용접철망"]),
+    ("맨홀", r"^주철맨홀", ["MASCntrct_맨홀"]),
+    ("보강섬유", r"^콘크리트보강섬유", ["ThptyUcntrct_섬유"]),
+    (
+        "폴리에틸렌관",
+        r"^(PE직관|PE이중벽|VR관)",
+        ["MASCntrct_폴리에틸렌", "ThptyUcntrct_폴리에틸렌"],
+    ),
+)
+SKIP_WORDS = {"관급", "포장품", "", "각종"}
+
+
+def label(name: str) -> str | None:
+    return next((k for k, pat, _ in FAMILIES if re.search(pat, name)), None)
+
+
+def norm(t: str) -> str:
+    t = (t or "").lower().replace("㎜", "mm").replace("㎥", "m3").replace("㎏", "kg")
+    t = re.sub(r"[φ∅øᴓф]", "d", t)
+    t = re.sub(r"(?<=\d)mm", "", t)
+    t = re.sub(r"(?<=[\d.])[×*x](?=[\d.d])", "x", t)
+    return re.sub(r"\s+", "", t).replace("入", "")
+
+
+def pieces(text: str) -> list[str]:
+    return [
+        p for p in (norm(x) for x in re.split(r"[,()/\-·\s]+", text or "")) if p not in SKIP_WORDS
+    ]
+
+
+def has(s: str, p: str) -> bool:
+    return re.search(rf"(?<![\d.]){re.escape(p)}(?![\d.])", s) is not None
+
+
+def tail_of(name: str) -> str | None:
+    """이름 끝의 「-지역」 또는 「-제조사」 · 「(충남지역」 · 「시멘트, 제주」."""
+    m = re.search(r"[\)\s]-([가-힣,·]+)$", name) or re.search(r"\(([가-힣]+)지역", name)
+    if m:
+        return m.group(1)
+    return "제주" if ", 제주" in name else None
+
+
+def find(row: dict, raw: dict, places, pick, PROVINCE=frozenset()):
+    """(줄, 방식) 또는 (None, 까닭) — `places` · `pick` 은 본 스크립트 것을 받음."""
+    n, spec = row["이름"], row["규격"]
+    fam = next(((k, f) for k, pat, f in FAMILIES if re.search(pat, n)), None)
+    if not fam:
+        return None, "이 갈래는 아직 짝짓기 없음"
+    rows = [r for f in fam[1] for r in raw.get(f, [])]
+    if not rows:
+        return None, "받은 원본 없음"
+    tail = tail_of(n)
+    reg = tail if tail and places(tail) & PROVINCE else None  # 지역 이름이 아니면 제조사
+    maker = norm(tail) if tail and not reg else None
+    need = pieces(spec)
+    if fam[0] == "주철관":  # 이름의 종별(2종 · 3종)도 맞아야
+        need += [p for p in pieces(n) if re.search(r"\d", p)]
+    if not pieces(spec):
+        return None, "규격 없음"
+    hits, seen_reg = [], False
+    for r in rows:
+        s = norm(r["prdctSpecNm"])
+        if not all(has(s, p) if re.search(r"\d", p) else p in s for p in need):
+            continue
+        if maker and maker not in norm(
+            r["prdctSpecNm"] + r["cntrctCorpNm"] + str(r["prdctMakrNm"])
+        ):
+            continue
+        if reg:
+            want, have = places(reg), places(r["prdctSplyRgnNm"])
+            if "제주" in reg:
+                ok = "제주" in r["prdctSplyRgnNm"].replace("제주제외", "")
+            elif want:
+                ok = (
+                    want <= have
+                    or r["prdctSplyRgnNm"].startswith("전지역")
+                    and "제주" not in r["prdctSplyRgnNm"]
+                )
+            else:
+                ok = True
+            if not ok:
+                seen_reg = True
+                continue
+        hits.append(r)
+    if not hits:
+        return None, "지역 못 찾음" if seen_reg else "규격 없음(원문 규격이 조달청에 없음)"
+    words = [p for p in pieces(n.replace(tail or "\0", "")) if not re.search(r"\d", p)]
+    score = lambda r: sum(w in norm(r["prdctSpecNm"]) for w in words)  # noqa: E731
+    top = max(map(score, hits))
+    r, how = pick([r for r in hits if score(r) == top])
+    return r, how
+
+
+# 조달청에만 있는 품목 — 자재품목에 새 줄로 들임 (갈래, 받은 파일들, 분류명)
+NEW = (
+    ("부직포", ["ThptyUcntrct_부직포"], None),
+    ("돌망태", ["MASCntrct_돌망태", "ThptyUcntrct_돌망태"], {"금속돌망태"}),
+)
+
+
+def groups(raw: dict) -> dict:
+    """{원문번호: (품명, 규격, 줄들)} — 「품명, 제조사, 모델, 규격…」 에서 제조사·모델을 뺀 같은 품명·규격끼리."""
+    out: dict = {}
+    for _, files, klass in NEW:
+        g: dict = {}
+        for f in files:
+            for r in raw.get(f, []):
+                p = [x.strip() for x in r["prdctSpecNm"].split(",")]
+                if len(p) >= 4 and (klass is None or r["prdctClsfcNoNm"] in klass):
+                    g.setdefault((p[0], ", ".join(p[3:])), []).append(r)
+        out.update({f"조달청 {n} {s}": (n, s, rows) for (n, s), rows in g.items()})
+    return out