knowledge(마스터): 관급 보강 — 조달청에서 자재품목에 없는 품목 후보 23,377줄 · 원본 보관
- 새 스크립트 update_관급단가_보강.py(받기 fetch · 후보 build) — 기존 update_관급단가.py 는 그대로 불러 씀 - 후보 = ref/_관급_보강_후보.json (자재품목 줄 모양 · 키 없음 · 같은 품명·규격은 최저가) — 정본·키대장은 건드리지 않음 - 원본 = 지식DB 관급/2026-09-20/보강/ - 없음(조달청 종합쇼핑몰) — 합판 · 각재 · 유로폼 · 강관동바리 · 비계 · 철선 · 못 · 폼타이 · 유공관 · 종자 · 방부목 · 잡석 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01V1MKKZKpUHTPKb513FneU8
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1,182 @@
|
||||
"""관급 보강 — 다산소프트에 없는 자재를 조달청 종합쇼핑몰에서 더 받아 후보 파일로 모음.
|
||||
|
||||
./venv/Scripts/python.exe resources/master_data/scripts/update_관급단가_보강.py fetch 받기 → 지식DB 관급/<날짜>/보강/
|
||||
./venv/Scripts/python.exe resources/master_data/scripts/update_관급단가_보강.py build 후보 → ref/_관급_보강_후보.json
|
||||
|
||||
정본(재료_자재품목.json · _키대장.json)은 건드리지 않음.
|
||||
"""
|
||||
|
||||
import gzip
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
import update_관급단가 as g # noqa: E402
|
||||
|
||||
MAS, UNIT, THPTY = g.MAS, g.UNIT, g.THPTY
|
||||
OPS = (MAS, THPTY, UNIT)
|
||||
# (갈래, 분류명 부분 글자들) — 품명 하나에 조회 두세 번 안쪽
|
||||
CATS = [
|
||||
("거푸집·가설재", ["합판", "각재", "유로폼", "거푸집", "동바리", "비계", "안전난간", "가설"]),
|
||||
("철선·못·폼타이", ["철선", "못", "폼타이"]),
|
||||
("관", ["유공관", "폴리에틸렌관", "고무링"]),
|
||||
("맨홀·그레이팅·수로", ["맨홀", "그레이팅", "측구", "수로관"]),
|
||||
("안전시설", ["가드레일", "표지판", "낙석방지", "방호"]),
|
||||
("종자·비료·식생", ["종자", "식생", "비료", "녹화"]),
|
||||
("목재", ["방부목", "제재목", "목재"]),
|
||||
("골재·경계석", ["잡석", "경계석", "골재", "석재"]),
|
||||
("매트·부직포", ["매트", "부직포", "지오텍스타일", "벤토나이트"]),
|
||||
# 두 번째 조회 — 첫 글자로 없던 것
|
||||
("거푸집·가설재", ["강관", "난간", "발판", "서포트"]),
|
||||
("철선·못·폼타이", ["와이어", "철망", "앵커", "결속"]),
|
||||
("관", ["배수관", "집수관", "배수판"]),
|
||||
("종자·비료·식생", ["씨앗", "잔디", "수목", "초화"]),
|
||||
("목재", ["방부", "판재"]),
|
||||
("골재·경계석", ["쇄석", "자갈", "모래"]),
|
||||
]
|
||||
DAY = date.today().isoformat()
|
||||
OUT = g.RAW / DAY / "보강"
|
||||
CAP = 25 # 분류 하나에 쪽 상한(쪽당 999)
|
||||
|
||||
|
||||
def fetch() -> None:
|
||||
key = g.cc.read_g2b_key()
|
||||
OUT.mkdir(parents=True, exist_ok=True)
|
||||
lines = [
|
||||
"# 관급 보강 원본",
|
||||
"",
|
||||
f"받은 날 {DAY} · `scripts/update_관급단가_보강.py fetch`",
|
||||
"",
|
||||
"| 파일 | 갈래 | 줄 수 |",
|
||||
"|---|---|---|",
|
||||
]
|
||||
t0 = time.time()
|
||||
for cat, words in CATS:
|
||||
for w in words:
|
||||
for op in OPS:
|
||||
f = f"{op.removeprefix('get').removesuffix('PrdctInfoList')}_{w}.json.gz"
|
||||
if (OUT / f).exists():
|
||||
continue
|
||||
if time.time() - t0 > 2100:
|
||||
print("시간 초과 — 중단", flush=True)
|
||||
break
|
||||
rows, page, total = [], 1, 1
|
||||
try:
|
||||
while len(rows) < total and page <= CAP:
|
||||
b = g.call(
|
||||
op, key, inqryDiv=2, prdctClsfcNoNm=w, numOfRows=999, pageNo=page
|
||||
)["body"]
|
||||
total = int(b.get("totalCount") or 0)
|
||||
items = b.get("items") or []
|
||||
if isinstance(items, dict):
|
||||
items = items.get("item", [])
|
||||
if not items:
|
||||
break
|
||||
rows += [{k: it.get(k) for k in g.KEEP} for it in items]
|
||||
page += 1
|
||||
except Exception as e:
|
||||
print(f, "실패", str(e)[:80], flush=True)
|
||||
continue
|
||||
(OUT / f).write_bytes(
|
||||
gzip.compress(json.dumps(rows, ensure_ascii=False).encode("utf-8"))
|
||||
)
|
||||
print(cat, f, len(rows), total, flush=True)
|
||||
for f in sorted(OUT.glob("*.json.gz")):
|
||||
n = len(json.loads(gzip.decompress(f.read_bytes())))
|
||||
lines.append(
|
||||
f"| {f.name} | {next((c for c, ws in CATS if any(f.name.endswith('_' + w + '.json.gz') for w in ws)), '')} | {n:,} |"
|
||||
)
|
||||
(OUT / "_meta.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
# 후보로 삼을 조달청 분류명 — 갈래별(나머지는 다른 물건이 섞여 온 것이라 뺌)
|
||||
KEEP_CLS = {
|
||||
"거푸집·가설재": ["건설용거푸집", "금속제울타리", "교량난간"],
|
||||
"철선·못·폼타이": ["용접철망"],
|
||||
"관": ["폴리에틸렌관", "오링", "배수판"],
|
||||
"맨홀·그레이팅·수로": [
|
||||
"맨홀박스",
|
||||
"맨홀뚜껑",
|
||||
"맨홀보조물",
|
||||
"스틸그레이팅",
|
||||
"그레이팅덮개",
|
||||
"플라스틱그레이팅",
|
||||
"배수로",
|
||||
],
|
||||
"안전시설": ["가드레일", "안전표지판", "도로안전표지판지주", "낙석방지책"],
|
||||
"종자·비료·식생": [
|
||||
"부숙유기질비료",
|
||||
"유기질비료",
|
||||
"조경용수목",
|
||||
"수목보호용지지대",
|
||||
"잔디보호매트",
|
||||
],
|
||||
"목재": ["덱(Deck)", "합성목재", "목재판재"],
|
||||
"골재·경계석": ["석재블록", "혼합골재", "투수골재포장재"],
|
||||
"매트·부직포": ["토목섬유"],
|
||||
}
|
||||
CLS_OF = {c: k for k, cs in KEEP_CLS.items() for c in cs}
|
||||
OUTFILE = g.ROOT / "resources/master_data/ref/_관급_보강_후보.json"
|
||||
|
||||
|
||||
def split_spec(t: str) -> tuple[str, str]:
|
||||
"""「품명, 제조사, 모델, 규격…」 → (품명, 규격) — 제조사·모델은 뺌(조달청 표기 그대로)."""
|
||||
p = [x.strip() for x in t.split(",")]
|
||||
return p[0], ", ".join(p[3:] if len(p) >= 4 else p[2:] if len(p) == 3 else p[1:])
|
||||
|
||||
|
||||
def build() -> None:
|
||||
have = {
|
||||
(x["이름"], x["규격"] or "") for x in json.loads(g.MASTER.read_text(encoding="utf-8"))["줄"]
|
||||
}
|
||||
grp: dict = {}
|
||||
for f in sorted(OUT.glob("*.json.gz")):
|
||||
op = f.name.split("_")[0]
|
||||
for r in json.loads(gzip.decompress(f.read_bytes())):
|
||||
cat = CLS_OF.get(r["prdctClsfcNoNm"])
|
||||
if cat and r["cntrctPrceAmt"]:
|
||||
name, spec = split_spec(r["prdctSpecNm"])
|
||||
grp.setdefault((cat, name, spec, r["prdctUnit"]), []).append((op, r))
|
||||
rows, per, dup = [], {}, {}
|
||||
for (cat, name, spec, unit), items in sorted(grp.items()):
|
||||
if (name, spec) in have: # 자재품목에 같은 품명·규격이 이미 있음
|
||||
dup[cat] = dup.get(cat, 0) + 1
|
||||
continue
|
||||
best, how = g.pick([r for _, r in items])
|
||||
ops = {o for o, r in items if r is best}
|
||||
_, note = g.decide(
|
||||
None, {"새값": int(best["cntrctPrceAmt"]), "까닭": how, "조달청": g.info(best)}, DAY
|
||||
)
|
||||
rows.append(
|
||||
{
|
||||
"키": None, "구분": None, "상세구분": None, "이름": name, "규격": spec, "단위": unit,
|
||||
"물가자료": None, "유통물가": None, "물가정보": None, "거래가격": None,
|
||||
"관급": int(best["cntrctPrceAmt"]),
|
||||
"출처": "조달청 " + ("/".join(sorted(ops)) or MAS),
|
||||
"비고": note, "면수": None,
|
||||
}
|
||||
) # fmt: skip
|
||||
per[cat] = per.get(cat, 0) + 1
|
||||
api = {"MASCntrct": MAS, "Ucntrct": UNIT, "ThptyUcntrct": THPTY}
|
||||
for r in rows:
|
||||
r["출처"] = "조달청 " + "/".join(api[o] for o in r["출처"].split(" ", 1)[1].split("/"))
|
||||
doc = {
|
||||
"받은날": DAY,
|
||||
"설명": "관급 보강 후보 — 재료_자재품목.json 에 아직 없는 조달청 품목(같은 품명·규격은 최저가). 키 없음 — 합칠 때 새로 부여.",
|
||||
"갈래별 줄 수": per,
|
||||
"이미 자재품목에 있어 뺀 수": dup,
|
||||
"줄": rows,
|
||||
}
|
||||
OUTFILE.write_text(json.dumps(doc, ensure_ascii=False, indent=1), encoding="utf-8")
|
||||
print("후보", len(rows), per, "이미 있음", dup)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
mode = (sys.argv[1:] or ["fetch"])[0]
|
||||
if mode == "fetch":
|
||||
fetch()
|
||||
elif mode == "build":
|
||||
build()
|
||||
Reference in New Issue
Block a user