feat(master_data): 자재품목 관급 열을 조달청 값으로 — 다산소프트 값 제거 · update_관급단가.py · 설계 문서
- 관급 열 4,552줄: 조달청 종합쇼핑몰(MAS · 일반 · 3자단가)에서 받은 값 2,981 · 못 채움·보류 null 1,571 · 비고 「관급: 조달청 … 받은 날」 - 짝짓기: 레미콘 2,095줄 중 2,028줄 다산 값과 같음 · 아스콘은 기준일(2026-08-01) 차이 · 제조사별 값이 갈리는 갈래는 보류 - 다시 돌리기: update_관급단가.py update (받기 → 원본 보관 → 관급·비고만 갈음) - ref/_설계_관급단가_갱신.md: API · 짝짓기 · 대조 결과 표 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PAdA5ThqmtVSsbzjusk1cJ
This commit is contained in:
Claude Sonnet 5
@@ -1,8 +1,11 @@
|
||||
"""관급 단가 갱신 — 조달청 종합쇼핑몰 API 에서 받아 자재품목(재료_시중물가) 「조달」 칸을 채움.
|
||||
"""관급 단가 갱신 — 조달청 종합쇼핑몰 API 에서 받아 자재품목(재료_자재품목) 「관급」 열을 채움.
|
||||
|
||||
python update_관급단가.py fetch # 받기 → 원본 보관 (resources/knowledge/original/원가계산/자재단가/관급/<날짜>/)
|
||||
python update_관급단가.py fill [날짜] # 원본 → 사본에서 조달 칸을 지우고 다시 채움 · 다산소프트 값과 맞대기
|
||||
마스터 정본은 덮지 않음 — 사본(tmp/관급_사본.json)과 대조표(tmp/관급_대조.json)만 씀.
|
||||
다시 돌리는 법 (한 줄) — 받기 + 정본 관급 열 갱신 + 바뀐 줄 목록:
|
||||
./venv/Scripts/python.exe resources/master_data/scripts/update_관급단가.py update
|
||||
|
||||
fetch 받기 → 원본 보관 (resources/knowledge/original/원가계산/자재단가/관급/<날짜>/ · gz + _meta.md)
|
||||
test [날짜] 관급 열을 지우고 다시 채워 옛 값(다산소프트)과 맞댐 — 정본 안 덮음 · tmp/관급_대조.json
|
||||
apply [날짜] 정본 관급 열(관급 · 비고)만 받은 값으로 바꿈 · tmp/관급_바뀐줄.json
|
||||
"""
|
||||
|
||||
import gzip
|
||||
@@ -40,9 +43,11 @@ JOBS = [ # 분류명은 부분 글자 조회 — 「철근」은 암거블록
|
||||
(UNIT, "콘크리트관"),
|
||||
]
|
||||
# 보관하는 칸 — 나머지(첨부·원산지·이미지·긴 설명)는 버림(원본이 수백 MB)
|
||||
KEEP = ("cntrctMthdNm cntrctCorpNm prdctSpecNm cntrctPrceAmt prdctUnit prdctSplyRgnNm splyJrsdctRgnNm "
|
||||
"prdctDlvryCndtnNm vatAplDivNm prdctClsfcNoNm prdctClsfcNo dtilPrdctClsfcNo prdctIdntNo prdctMakrNm "
|
||||
"fctryLocplc dscntAmt dscntBgnDate dscntEndDate cntrctBgnDate cntrctEndDate rgstDt chgDt").split()
|
||||
KEEP = (
|
||||
"cntrctMthdNm cntrctCorpNm prdctSpecNm cntrctPrceAmt prdctUnit prdctSplyRgnNm splyJrsdctRgnNm "
|
||||
"prdctDlvryCndtnNm vatAplDivNm prdctClsfcNoNm prdctClsfcNo dtilPrdctClsfcNo prdctIdntNo prdctMakrNm "
|
||||
"fctryLocplc dscntAmt dscntBgnDate dscntEndDate cntrctBgnDate cntrctEndDate rgstDt chgDt"
|
||||
).split()
|
||||
|
||||
|
||||
def call(op: str, key: str, **kw) -> dict:
|
||||
@@ -55,9 +60,14 @@ def fetch_all() -> Path:
|
||||
key = cc.read_g2b_key()
|
||||
out = RAW / date.today().isoformat()
|
||||
out.mkdir(parents=True, exist_ok=True)
|
||||
lines = ["# 관급 단가 원본 — 조달청 종합쇼핑몰 품목정보 서비스", "",
|
||||
f"받은 날 {date.today().isoformat()} · 수집 `scripts/update_관급단가.py fetch` · 조건 = inqryDiv=2 · prdctClsfcNoNm=<분류명> · 쪽당 999", "",
|
||||
"| 파일 | 오퍼레이션 | 분류명 | 줄 수 |", "|---|---|---|---|"]
|
||||
lines = [
|
||||
"# 관급 단가 원본 — 조달청 종합쇼핑몰 품목정보 서비스",
|
||||
"",
|
||||
f"받은 날 {date.today().isoformat()} · 수집 `scripts/update_관급단가.py fetch` · 조건 = inqryDiv=2 · prdctClsfcNoNm=<분류명> · 쪽당 999",
|
||||
"",
|
||||
"| 파일 | 오퍼레이션 | 분류명 | 줄 수 |",
|
||||
"|---|---|---|---|",
|
||||
]
|
||||
for op, name in JOBS:
|
||||
rows, page, total = [], 1, 1
|
||||
while len(rows) < total:
|
||||
@@ -79,7 +89,6 @@ def fetch_all() -> Path:
|
||||
return out
|
||||
|
||||
|
||||
|
||||
# ── 짝짓기 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@@ -95,10 +104,25 @@ def latest() -> str:
|
||||
return max(p.name for p in RAW.iterdir() if p.is_dir())
|
||||
|
||||
|
||||
ALIAS = {"경기": "경기도", "경남": "경상남도", "경북": "경상북도", "충남": "충청남도", "충북": "충청북도",
|
||||
"전남": "전라남도", "전북": "전라북도", "강원": "강원도", "제주": "제주특별자치도",
|
||||
"서울": "서울특별시", "부산": "부산광역시", "대구": "대구광역시", "인천": "인천광역시",
|
||||
"광주": "광주광역시", "대전": "대전광역시", "울산": "울산광역시", "세종": "세종특별자치시"}
|
||||
ALIAS = {
|
||||
"경기": "경기도",
|
||||
"경남": "경상남도",
|
||||
"경북": "경상북도",
|
||||
"충남": "충청남도",
|
||||
"충북": "충청북도",
|
||||
"전남": "전라남도",
|
||||
"전북": "전라북도",
|
||||
"강원": "강원도",
|
||||
"제주": "제주특별자치도",
|
||||
"서울": "서울특별시",
|
||||
"부산": "부산광역시",
|
||||
"대구": "대구광역시",
|
||||
"인천": "인천광역시",
|
||||
"광주": "광주광역시",
|
||||
"대전": "대전광역시",
|
||||
"울산": "울산광역시",
|
||||
"세종": "세종특별자치시",
|
||||
}
|
||||
WORDS = re.compile(r"[가-힣]+")
|
||||
|
||||
|
||||
@@ -107,7 +131,9 @@ def places(text: str) -> frozenset:
|
||||
out = set()
|
||||
text = re.sub(r"전지역\([^)]*\)", "전지역", text or "") # 「전지역(구 나열)」 = 전지역
|
||||
for w in WORDS.findall(text):
|
||||
if w in ("전지역", "지역", "협", "사업협동조합", "레미콘", "레미콘공업") or w.endswith(("조합", "레미콘")):
|
||||
if w in ("전지역", "지역", "협", "사업협동조합", "레미콘", "레미콘공업") or w.endswith(
|
||||
("조합", "레미콘")
|
||||
):
|
||||
continue
|
||||
w = ALIAS.get(w, w)
|
||||
if w.endswith(("광역시", "특별시", "특별자치시", "특별자치도")):
|
||||
@@ -134,7 +160,325 @@ def region_pick(cands: dict, want: frozenset):
|
||||
if best and best[0][0] >= 0.5 and (len(best) == 1 or best[0][0] > best[1][0]):
|
||||
return cands[best[0][1]], f"지역 비슷함 {best[0][0]:.2f}"
|
||||
return None, "지역 못 찾음"
|
||||
|
||||
|
||||
MASTER = ROOT / "resources/master_data/재료_자재품목.json"
|
||||
TMP = ROOT / "tmp"
|
||||
|
||||
|
||||
def newest(rows: list[dict]) -> dict:
|
||||
return max(rows, key=lambda r: (r["cntrctBgnDate"] or "", r["rgstDt"] or ""))
|
||||
|
||||
|
||||
def pick(rows: list[dict]):
|
||||
"""같은 짝 후보가 여럿이면 값이 하나일 때 그 값 · 값이 갈리면 가장 낮은 값(제조사별 단가) — 사유를 함께 돌려줌."""
|
||||
if not rows:
|
||||
return None, "후보 없음"
|
||||
vals = {int(r["cntrctPrceAmt"]) for r in rows}
|
||||
if len(vals) == 1:
|
||||
return newest(rows), "짝 하나" if len(rows) == 1 else f"같은 값 {len(rows)}건"
|
||||
low = min(vals)
|
||||
return newest(
|
||||
[r for r in rows if int(r["cntrctPrceAmt"]) == low]
|
||||
), f"제조사별 값 {len(vals)}가지 중 최저"
|
||||
|
||||
|
||||
def _dia(t):
|
||||
m = re.search(r"[ΦD](\d+)", t or "")
|
||||
return int(m.group(1)) if m else None
|
||||
|
||||
|
||||
def _thick(t):
|
||||
m = re.search(r"[×xX*](\d+(?:\.\d+)?)", t or "")
|
||||
return float(m.group(1)) if m else None
|
||||
|
||||
|
||||
def _size(t):
|
||||
m = re.search(
|
||||
r"(\d+(?:\.\d+)?)[×*x](\d+(?:\.\d+)?)[×*x](\d+(?:\.\d+)?)[×*x](\d+(?:\.\d+)?)", t or ""
|
||||
)
|
||||
return tuple(float(x) for x in m.groups()) if m else None
|
||||
|
||||
|
||||
class Matcher:
|
||||
"""자재품목 줄 → 조달청 줄. 짝이 서면 (줄, 방식), 못 서면 (None, 까닭)."""
|
||||
|
||||
def __init__(self, raw: dict):
|
||||
self.rmc = {}
|
||||
for r in raw.get("MASCntrct_레미콘", []):
|
||||
if r["prdctMakrNm"] == "조합공통품목":
|
||||
self.rmc.setdefault(strength(r["prdctSpecNm"]), {}).setdefault(
|
||||
places(r["prdctSplyRgnNm"]), []
|
||||
).append(r)
|
||||
self.asc = {}
|
||||
for f in (
|
||||
"MASCntrct_아스팔트콘크리트",
|
||||
"Ucntrct_아스팔트콘크리트",
|
||||
"ThptyUcntrct_아스팔트콘크리트",
|
||||
):
|
||||
for r in raw.get(f, []):
|
||||
if "조합공통품목" in r["prdctSpecNm"]:
|
||||
k = (
|
||||
r["prdctSpecNm"].startswith("순환"),
|
||||
re.sub(r"\s+", "", r["prdctSpecNm"].split("조합공통품목,")[1]),
|
||||
)
|
||||
self.asc.setdefault(k, {}).setdefault(places(r["prdctSplyRgnNm"]), []).append(r)
|
||||
self.bar = raw.get("MASCntrct_봉강", [])
|
||||
self.hbeam = raw.get("MASCntrct_H빔", [])
|
||||
self.corr = raw.get("MASCntrct_파형강관", []) + raw.get("ThptyUcntrct_파형강관", [])
|
||||
self.hume = raw.get("MASCntrct_콘크리트관", []) + raw.get("Ucntrct_콘크리트관", [])
|
||||
|
||||
@staticmethod
|
||||
def _region(table: dict, name: str):
|
||||
want = places(name.split(")-", 1)[1] if ")-" in name else name)
|
||||
cands, why = region_pick(table, want)
|
||||
if cands is None:
|
||||
return None, why
|
||||
corp = re.findall(
|
||||
r"([가-힣]+)레미콘\(협\)", name
|
||||
) # 이름에 조합이 붙은 줄(「제주…레미콘(협)」)은 그 조합 줄만
|
||||
if corp and len(cands) > 1:
|
||||
cands = [r for r in cands if r["cntrctCorpNm"].startswith(corp[0])] or cands
|
||||
return cands, why
|
||||
|
||||
def find(self, row: dict):
|
||||
n, spec = row["이름"], row["규격"]
|
||||
if n.startswith("레미콘(관급)-"):
|
||||
k = strength(spec)
|
||||
if k not in self.rmc:
|
||||
return None, "강도 없음(원문 규격이 조달청에 없음)"
|
||||
c, why = self._region(self.rmc[k], n)
|
||||
if c is None:
|
||||
return None, why
|
||||
r, how = pick(c)
|
||||
return r, f"{why} · {how}"
|
||||
if re.match(r"(순환)?아스콘\(관급\)-", n):
|
||||
k = (n.startswith("순환"), re.sub(r"\s+", "", spec.split("조합공통품목,")[-1]))
|
||||
if k not in self.asc:
|
||||
return None, "규격 없음(원문 규격이 조달청에 없음)"
|
||||
c, why = self._region(self.asc[k], n)
|
||||
if c is None:
|
||||
return None, why
|
||||
r, how = pick(c)
|
||||
return r, f"{why} · {how}"
|
||||
m = re.match(r"이형철근\((SD\d+)(,제주)?\)-관급", n)
|
||||
if m:
|
||||
d = re.search(r"HD=(\d+)", spec)
|
||||
zone = "제주" if m.group(2) else "전지역(제주제외)"
|
||||
cond = (
|
||||
"생산공장상차도"
|
||||
if "생산공장" in spec
|
||||
else ("하치장상차도" if "하치장" in spec else None)
|
||||
)
|
||||
c = (
|
||||
[
|
||||
r
|
||||
for r in self.bar
|
||||
if f"{m.group(1)}, D{d.group(1)}" in r["prdctSpecNm"]
|
||||
and r["prdctSplyRgnNm"].startswith(zone)
|
||||
and (cond is None or r["prdctDlvryCndtnNm"].replace(" ", "") == cond)
|
||||
]
|
||||
if d
|
||||
else []
|
||||
)
|
||||
return pick(c)
|
||||
if n.startswith("H 형강(관급)"):
|
||||
sz = _size(spec)
|
||||
g = re.match(r"(SS\d+|SM\d+|SHN\d+)", spec)
|
||||
c = [
|
||||
r
|
||||
for r in self.hbeam
|
||||
if sz
|
||||
and _size(r["prdctSpecNm"]) == sz
|
||||
and g
|
||||
and g.group(1) in r["prdctSpecNm"]
|
||||
and r["prdctSplyRgnNm"].startswith("전지역")
|
||||
]
|
||||
return pick(c)
|
||||
if re.match(r"파형강관\(관급\)", n):
|
||||
dia, th = _dia(spec), _thick(spec)
|
||||
rs = re.search(r"(\d)RS", spec)
|
||||
c = [
|
||||
r
|
||||
for r in self.corr
|
||||
if dia
|
||||
and th
|
||||
and rs
|
||||
and _dia(r["prdctSpecNm"]) == dia
|
||||
and re.search(rf"t{th:g}mm", r["prdctSpecNm"])
|
||||
and f"({rs.group(1)}RS)" in r["prdctSpecNm"]
|
||||
]
|
||||
r, how = pick(c)
|
||||
return r, how
|
||||
if re.match(r"흄\s*관", n):
|
||||
dia = _dia(spec)
|
||||
c = [
|
||||
r
|
||||
for r in self.hume
|
||||
if dia and _dia(r["prdctSpecNm"]) == dia and "보통관" in r["prdctSpecNm"]
|
||||
]
|
||||
return pick(c)
|
||||
return None, "이 갈래는 아직 짝짓기 없음"
|
||||
|
||||
|
||||
FAMILY = (
|
||||
("레미콘", r"^레미콘\(관급\)"),
|
||||
("아스콘", r"^(순환)?아스콘\(관급\)"),
|
||||
("철근", r"^이형철근"),
|
||||
("H형강", r"^H 형강"),
|
||||
("파형강관", r"^파형강관\(관급\)"),
|
||||
("흄관", r"^흄\s*관"),
|
||||
)
|
||||
|
||||
|
||||
def family(name: str) -> str:
|
||||
return next((k for k, pat in FAMILY if re.match(pat, name)), "그 밖")
|
||||
|
||||
|
||||
def refill(rows: list[dict], raw: dict) -> list[dict]:
|
||||
"""조달 값(관급 열)이 있는 줄마다 조달청 값으로 다시 채운 결과."""
|
||||
m, out = Matcher(raw), []
|
||||
for x in rows:
|
||||
if x["관급"] is None:
|
||||
continue
|
||||
r, why = m.find(x)
|
||||
out.append(
|
||||
{
|
||||
"키": x["키"],
|
||||
"이름": x["이름"],
|
||||
"규격": x["규격"],
|
||||
"옛값": x["관급"],
|
||||
"옛기간": x["비고"],
|
||||
"갈래": 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"],
|
||||
},
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def test(day: str | None = None) -> list[dict]:
|
||||
day = day or latest()
|
||||
res = refill(json.loads(MASTER.read_text(encoding="utf-8"))["줄"], load(day))
|
||||
TMP.mkdir(exist_ok=True)
|
||||
(TMP / "관급_대조.json").write_text(
|
||||
json.dumps(res, ensure_ascii=False, indent=1), encoding="utf-8"
|
||||
)
|
||||
tab: dict = {}
|
||||
for r in res:
|
||||
t = tab.setdefault(
|
||||
r["갈래"], {"전체": 0, "짝": 0, "같음": 0, "다름": 0, "못채움": 0, "까닭": {}}
|
||||
)
|
||||
t["전체"] += 1
|
||||
if r["새값"] is None:
|
||||
t["못채움"] += 1
|
||||
t["까닭"][r["까닭"]] = t["까닭"].get(r["까닭"], 0) + 1
|
||||
else:
|
||||
t["짝"] += 1
|
||||
t["같음" if r["새값"] == r["옛값"] else "다름"] += 1
|
||||
for k, v in tab.items():
|
||||
print(
|
||||
k,
|
||||
{a: b for a, b in v.items() if a != "까닭"},
|
||||
dict(sorted(v["까닭"].items(), key=lambda kv: -kv[1])[:4]),
|
||||
)
|
||||
return res
|
||||
|
||||
|
||||
METHOD = {
|
||||
"다수 공급자 계약": "다수공급자계약",
|
||||
"일반단가계약": "일반단가계약",
|
||||
"3자단가계약": "3자단가계약",
|
||||
}
|
||||
|
||||
|
||||
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["조달청"]
|
||||
return r["새값"], (
|
||||
f"관급: 조달청 {METHOD.get(c['방식'], c['방식'])} {day} · 계약 {c['계약시작']}~ · {c['지역']} · {c['인도']} · {c['부가세']}"
|
||||
)
|
||||
|
||||
|
||||
def apply(day: str | None = None) -> None:
|
||||
"""정본 관급 열 · 비고만 바꿈 — 다산소프트 값은 하나도 안 남김(못 채운 줄 = null)."""
|
||||
day = day or latest()
|
||||
text = MASTER.read_bytes().decode("utf-8")
|
||||
data = json.loads(text)
|
||||
found = {r["키"]: r for r in refill(data["줄"], load(day))}
|
||||
changed, edits = [], {}
|
||||
for x in data["줄"]:
|
||||
r = found.get(x["키"])
|
||||
if r is None:
|
||||
continue
|
||||
val, note = decide(x, r, day)
|
||||
changed.append(
|
||||
{
|
||||
"키": x["키"],
|
||||
"이름": x["이름"],
|
||||
"규격": x["규격"],
|
||||
"갈래": r["갈래"],
|
||||
"옛값": x["관급"],
|
||||
"새값": val,
|
||||
"비고": note,
|
||||
}
|
||||
)
|
||||
edits[x["키"]] = (val, note)
|
||||
# 글 자리에서 「관급」·「비고」 두 칸만 갈음 — 나머지 서식·칸은 그대로
|
||||
lines = text.split("\n")
|
||||
put = lambda v: json.dumps(v, ensure_ascii=False) # noqa: E731
|
||||
for i, line in enumerate(lines):
|
||||
m = re.match(r'\s*\{ "키": "([^"]+)"', line)
|
||||
if not m or m.group(1) not in edits:
|
||||
continue
|
||||
val, note = edits.pop(m.group(1))
|
||||
line, n1 = re.subn(
|
||||
r'"관급": (?:null|-?[\d.]+)', lambda _: f'"관급": {put(val)}', line, count=1
|
||||
)
|
||||
line, n2 = re.subn(
|
||||
r'"비고": (?:null|"(?:[^"\\]|\\.)*")', lambda _: f'"비고": {put(note)}', line, count=1
|
||||
)
|
||||
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"))
|
||||
TMP.mkdir(exist_ok=True)
|
||||
(TMP / "관급_바뀐줄.json").write_text(
|
||||
json.dumps(changed, ensure_ascii=False, indent=1), encoding="utf-8"
|
||||
)
|
||||
ok = sum(c["새값"] is not None for c in changed)
|
||||
print(
|
||||
f"관급 열 {len(changed)} 줄 중 채움 {ok} · null {len(changed) - ok} — 목록 tmp/관급_바뀐줄.json"
|
||||
)
|
||||
|
||||
|
||||
def update() -> None:
|
||||
day = fetch_all().name
|
||||
apply(day)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
mode = sys.argv[1] if len(sys.argv) > 1 else "fetch"
|
||||
if mode == "fetch":
|
||||
fetch_all()
|
||||
elif mode == "test":
|
||||
test(*sys.argv[2:3])
|
||||
elif mode == "apply":
|
||||
apply(*sys.argv[2:3])
|
||||
elif mode == "update":
|
||||
update()
|
||||
|
||||
Reference in New Issue
Block a user