feat(master_data): 관급 보강 후보 합침 — 다산 같은 품명 줄에 조달청 최저가 · 새 줄 6,239
- 다산에 같은 품명(그레이팅 · 맨홀뚜껑 · 보행매트 · 합성목재 · 용접철망 · 갈매기표지판 · 고무링 · 혼합골재 등): 같은 치수 후보의 최저가를 관급 열에 채움 · 안 맞으면 비움 - 다산에 없는 품명(관 · 안전시설 · 종자·비료·식생 · 골재·경계석 · 매트 · 목재): 새 줄 6,239 · 자재품목 24,353 → 30,592 - update_관급단가_합침.py 신설 · apply 가 부름 · 다시 돌려도 같은 결과 - 설계 문서에 합친 결과 · 시험 추가 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
@@ -24,6 +24,7 @@ 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 (「그 밖」 갈래 짝짓기)
|
||||
import update_관급단가_합침 as boost # noqa: E402 (보강 후보 합치기)
|
||||
|
||||
BASE = "https://apis.data.go.kr/1230000/at/ShoppingMallPrdctInfoService/"
|
||||
MAS, UNIT, THPTY = (
|
||||
@@ -596,6 +597,42 @@ def add_rows(lines: list[str], data: dict, raw: dict, day: str, changed: list) -
|
||||
return lines
|
||||
|
||||
|
||||
def add_boost(lines: list[str], data: dict, changed: list) -> list[str]:
|
||||
"""보강 후보 중 새 줄로 더하는 품명(boost.NEW) — 이미 든 줄은 값·비고만 갈음."""
|
||||
book = mk.load_book()
|
||||
at = {x["키"]: x for x in data["줄"]}
|
||||
dasan = {(x["이름"], x["규격"] or "") for x in data["줄"] if "조달청" not in x["출처"]}
|
||||
pos = {
|
||||
m.group(1): i for i, ln in enumerate(lines) if (m := re.match(r'\s*\{ "키": "([^"]+)"', ln))
|
||||
}
|
||||
new = []
|
||||
for row in boost.new_rows(book, mk.issue, dasan):
|
||||
if row["키"] in at:
|
||||
i = pos[row["키"]]
|
||||
lines[i] = re.sub(
|
||||
r'"관급": (?:null|-?[\d.]+)', lambda _: f'"관급": {row["관급"]}', lines[i], count=1
|
||||
)
|
||||
lines[i] = re.sub(
|
||||
r'"비고": (?:null|"(?:[^"\\]|\\.)*")',
|
||||
lambda _: f'"비고": {json.dumps(row["비고"], ensure_ascii=False)}',
|
||||
lines[i],
|
||||
count=1,
|
||||
)
|
||||
continue
|
||||
new.append(
|
||||
" { "
|
||||
+ ", ".join(f'"{k}": {json.dumps(v, ensure_ascii=False)}' for k, v in row.items())
|
||||
+ " }"
|
||||
)
|
||||
changed.append({"키": row["키"], "이름": row["이름"], "규격": row["규격"], "갈래": "보강 새 줄", "옛값": None, "새값": row["관급"], "비고": row["비고"]}) # fmt: skip
|
||||
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()
|
||||
@@ -622,6 +659,12 @@ def apply(day: str | None = None) -> None:
|
||||
}
|
||||
)
|
||||
edits[x["키"]] = (val, note)
|
||||
now = {x["키"]: edits[x["키"]][0] if x["키"] in edits else x["관급"] for x in data["줄"]}
|
||||
for k, (val, note) in boost.fills(data["줄"], now, places, PROVINCE).items():
|
||||
x = next(y for y in data["줄"] if y["키"] == k)
|
||||
changed[:] = [c for c in changed if c["키"] != k] # 앞서 못 채움으로 적힌 줄을 갈음
|
||||
changed.append({"키": k, "이름": x["이름"], "규격": x["규격"], "갈래": "보강 채움", "옛값": x["관급"], "새값": val, "비고": note}) # fmt: skip
|
||||
edits[k] = (val, note)
|
||||
# 글 자리에서 「관급」·「비고」 두 칸만 갈음 — 나머지 서식·칸은 그대로
|
||||
lines = text.split("\n")
|
||||
put = lambda v: json.dumps(v, ensure_ascii=False) # noqa: E731
|
||||
@@ -640,6 +683,7 @@ def apply(day: str | None = None) -> None:
|
||||
lines[i] = line
|
||||
assert not edits, f"정본에서 못 찾은 줄 {len(edits)}"
|
||||
lines = add_rows(lines, data, load(day), day, changed)
|
||||
lines = add_boost(lines, data, 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)
|
||||
|
||||
@@ -0,0 +1,170 @@
|
||||
"""관급 보강 합치기 — `ref/_관급_보강_후보.json`(조달청 후보) → 자재품목 (`update_관급단가.py apply` 가 부름).
|
||||
|
||||
(1) 다산소프트에 같은 품명이 있는 것 — 그 줄의 치수와 같은 조달청 후보의 최저가를 관급 열에 채움(치수가 안 맞으면 비움)
|
||||
(2) 다산소프트에 없는 품명 가운데 NEW 목록의 갈래만 — 새 줄(관급만 · 물가지 네 열 null)
|
||||
나머지 후보(그레이팅 · 맨홀 · 수로 · 거푸집 · 가설재 · 철선류 …)는 더하지 않음.
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
CAND = Path(__file__).resolve().parents[1] / "ref/_관급_보강_후보.json"
|
||||
|
||||
# (2) 새 줄로 더하는 조달청 품명 → 갈래
|
||||
NEW = {
|
||||
"일반용폴리에틸렌관": "관",
|
||||
"수도용폴리에틸렌관": "관",
|
||||
"철제가드레일": "안전시설",
|
||||
"도로안전표지판지주": "안전시설",
|
||||
"낙석방지책": "안전시설",
|
||||
"조경용수목": "종자·비료·식생",
|
||||
"수목보호용지지대": "종자·비료·식생",
|
||||
"부숙유기질비료": "종자·비료·식생",
|
||||
"잔디보호매트": "종자·비료·식생",
|
||||
"자연석경계석": "골재·경계석",
|
||||
"투수골재포장재": "골재·경계석",
|
||||
"식생매트": "매트·부직포",
|
||||
"목재덱": "목재",
|
||||
"목재판재": "목재",
|
||||
}
|
||||
# (1) 다산소프트에 같은 품명이 있어 그 줄에 값을 채우는 조달청 품명 (다산 이름 글자 → 조달청 품명)
|
||||
SAME = {
|
||||
"스틸그레이팅": "스틸그레이팅",
|
||||
"주철맨홀뚜껑": "주철맨홀뚜껑",
|
||||
"보행매트": "보행매트",
|
||||
"합성목재": "합성목재",
|
||||
"용접철망": "용접철망",
|
||||
"갈매기표지판": "갈매기표지판",
|
||||
"고무링": "고무링",
|
||||
"유기질비료": "유기질비료",
|
||||
"혼합골재": "혼합골재",
|
||||
"배수판": "배수판",
|
||||
}
|
||||
# 그레이팅 — 특수형(조달청 표기에 이 글자가 있으면 다산 규격과 다른 물건)
|
||||
GRATING_SPECIAL = (
|
||||
"디자인",
|
||||
"투수",
|
||||
"악취",
|
||||
"미끄럼",
|
||||
"날개",
|
||||
"압연",
|
||||
"오물",
|
||||
"토사",
|
||||
"맞춤",
|
||||
"중하중",
|
||||
)
|
||||
UNIT = {"ea": "개", "m": "m"}
|
||||
|
||||
|
||||
def load() -> list[dict]:
|
||||
return json.loads(CAND.read_text(encoding="utf-8"))["줄"] if CAND.exists() else []
|
||||
|
||||
|
||||
def _unit(u: str) -> str:
|
||||
u = (u or "").strip().lower()
|
||||
return UNIT.get(u, u)
|
||||
|
||||
|
||||
def _nums(t: str) -> tuple:
|
||||
t = re.sub(r"(?<=\d),(?=\d{3})", "", t or "")
|
||||
t = re.sub(
|
||||
r"W(\d)\.(\d)", lambda m: str(int(m[1]) * 1000 + int(m[2]) * 100), t
|
||||
) # W1.2 = 1200mm
|
||||
return tuple(sorted(float(x) for x in re.findall(r"\d+(?:\.\d+)?", t)))
|
||||
|
||||
|
||||
def _region(note: str) -> str:
|
||||
"""후보 비고 「… · 계약 …~ · <공급 지역> · <인도> · …」 의 공급 지역."""
|
||||
p = (note or "").split(" · ")
|
||||
return p[2] if len(p) > 2 else ""
|
||||
|
||||
|
||||
def _delivery(note: str) -> str:
|
||||
p = (note or "").split(" · ")
|
||||
return p[3] if len(p) > 3 else ""
|
||||
|
||||
|
||||
def _grating(row: dict, c: dict) -> bool:
|
||||
m = re.fullmatch(r"(\d+)×(\d+)×(\d+)(?:\(I-\d+×(\d+)×(\d+)\))?", row["규격"] or "")
|
||||
if not m or "앵글" in row["이름"]:
|
||||
return False
|
||||
s = c["규격"]
|
||||
d = re.match(r"(\d+)×(\d+)×(\d+)", s)
|
||||
if not d or d.groups() != m.groups()[:3] or "뚜껑" not in s:
|
||||
return False
|
||||
if m[4] and not re.search(rf"(?<![\d.]){m[4]}[×x]{m[5]}(?!\d)", s):
|
||||
return False
|
||||
return "중하중" in row["이름"] or not any(w in s for w in GRATING_SPECIAL)
|
||||
|
||||
|
||||
def _same(row: dict, c: dict, places, province) -> bool:
|
||||
"""다산 줄 하나와 조달청 후보 하나가 같은 규격인지."""
|
||||
n, s, cs = row["이름"], row["규격"] or "", c["규격"]
|
||||
if "(부품)" in cs or _unit(row["단위"]) != _unit(c["단위"]):
|
||||
return False
|
||||
if c["이름"] == "스틸그레이팅":
|
||||
return _grating(row, c)
|
||||
mine, theirs = _nums(s), _nums(cs)
|
||||
if c["이름"] == "갈매기표지판": # 앞 두 치수(가로×세로)만 · 두께는 조달청만 적음
|
||||
if (
|
||||
mine != _nums("×".join(re.findall(r"\d+(?:\.\d+)?", cs)[:2]))
|
||||
or re.search(r"태양광|LED|광섬유|솔라", cs)
|
||||
and "솔라" not in n
|
||||
):
|
||||
return False
|
||||
return ("양면" in n) == ("양면" in cs)
|
||||
if not mine or mine != theirs:
|
||||
return False
|
||||
if c["이름"] == "혼합골재": # 숫자만 적힌 줄 · 공급 지역 · 인도 조건이 같아야
|
||||
if re.search(r"[가-힣]", re.sub(r"mm", "", cs)):
|
||||
return False
|
||||
want, reg = places(n) & province, _region(c["비고"])
|
||||
if want and not (reg.startswith("전지역") or want <= places(reg)):
|
||||
return False
|
||||
return ("도착도" if "도착도" in n else "상차") in _delivery(c["비고"])
|
||||
if c["이름"] == "용접철망":
|
||||
want, reg = places(n) & province, _region(c["비고"])
|
||||
return not want or reg.startswith("전지역") or want <= places(reg)
|
||||
if c["이름"] == "주철맨홀뚜껑":
|
||||
return not ("상수도용" in s and "상수도" not in cs) and not (
|
||||
"하수도용" in s and "상하수도용" not in s and "하수도" not in cs
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def fills(rows: list[dict], now: dict, places, province) -> dict:
|
||||
"""{키: (관급, 비고)} — 다산 줄 중 아직 관급이 null 인 것에 조달청 후보 최저가. now = {키: 현재 관급}."""
|
||||
cands: dict = {}
|
||||
for c in load():
|
||||
if c["이름"] in SAME.values():
|
||||
cands.setdefault(c["이름"], []).append(c)
|
||||
out = {}
|
||||
for r in rows:
|
||||
if now.get(r["키"]) is not None or "조달청" in r["출처"]:
|
||||
continue
|
||||
for word, name in SAME.items():
|
||||
if word in r["이름"]:
|
||||
hit = [c for c in cands.get(name, []) if _same(r, c, places, province)]
|
||||
if hit:
|
||||
low = min(hit, key=lambda c: c["관급"])
|
||||
out[r["키"]] = (low["관급"], low["비고"])
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def new_rows(book, issue, existing: set) -> list[dict]:
|
||||
"""NEW 품명의 후보 → 새 줄(키는 원문번호 「조달청 품명 규격」 로 대장에서 — 다시 돌려도 같은 키)."""
|
||||
out, seen = [], set()
|
||||
for c in load():
|
||||
if c["이름"] not in NEW or (c["이름"], c["규격"]) in existing:
|
||||
continue
|
||||
num = f"조달청 {c['이름']} {c['규격']}"
|
||||
if num in seen:
|
||||
num += f" ({c['단위']})"
|
||||
seen.add(num)
|
||||
row = dict(c)
|
||||
row["출처"] = "자재품목 " + c["출처"] # 틀 검사: 출처는 책 이름으로 시작
|
||||
row["키"] = issue(book, "MT", num, "재료_자재품목.json")
|
||||
out.append(row)
|
||||
return out
|
||||
Reference in New Issue
Block a user