feat(master_data): 관급 단가 받기 — 조달청 종합쇼핑몰 API 원본 보관 · update_관급단가.py 받기·짝짓기 시작
- 받기: 레미콘 · 아스팔트콘크리트 · 봉강(철근) · H빔 · 파형강관 · 콘크리트관 — MAS · 일반단가 · 3자단가 (gz 원본 + _meta.md) - 짝짓기 도우미: 지명 묶음 · 강도 · 지역 고르기 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
@@ -0,0 +1,140 @@
|
||||
"""관급 단가 갱신 — 조달청 종합쇼핑몰 API 에서 받아 자재품목(재료_시중물가) 「조달」 칸을 채움.
|
||||
|
||||
python update_관급단가.py fetch # 받기 → 원본 보관 (resources/knowledge/original/원가계산/자재단가/관급/<날짜>/)
|
||||
python update_관급단가.py fill [날짜] # 원본 → 사본에서 조달 칸을 지우고 다시 채움 · 다산소프트 값과 맞대기
|
||||
마스터 정본은 덮지 않음 — 사본(tmp/관급_사본.json)과 대조표(tmp/관급_대조.json)만 씀.
|
||||
"""
|
||||
|
||||
import gzip
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import urllib.parse
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
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)
|
||||
|
||||
BASE = "https://apis.data.go.kr/1230000/at/ShoppingMallPrdctInfoService/"
|
||||
MAS, UNIT, THPTY = (
|
||||
"getMASCntrctPrdctInfoList",
|
||||
"getUcntrctPrdctInfoList",
|
||||
"getThptyUcntrctPrdctInfoList",
|
||||
)
|
||||
# (오퍼레이션, 분류명) — 분류명은 조회 조건(inqryDiv=2 · prdctClsfcNoNm)
|
||||
JOBS = [ # 분류명은 부분 글자 조회 — 「철근」은 암거블록까지 섞여 오므로 「봉강」(철근또는강철봉) · H형강은 「H빔」
|
||||
(MAS, "레미콘"),
|
||||
(THPTY, "레미콘"),
|
||||
(MAS, "아스팔트콘크리트"),
|
||||
(UNIT, "아스팔트콘크리트"),
|
||||
(THPTY, "아스팔트콘크리트"),
|
||||
(MAS, "봉강"),
|
||||
(MAS, "H빔"),
|
||||
(MAS, "파형강관"),
|
||||
(THPTY, "파형강관"),
|
||||
(MAS, "콘크리트관"),
|
||||
(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()
|
||||
|
||||
|
||||
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}"
|
||||
return json.loads(cc.fetch(url, timeout=90).decode("utf-8", "replace"))["response"]
|
||||
|
||||
|
||||
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", "",
|
||||
"| 파일 | 오퍼레이션 | 분류명 | 줄 수 |", "|---|---|---|---|"]
|
||||
for op, name in JOBS:
|
||||
rows, page, total = [], 1, 1
|
||||
while len(rows) < total:
|
||||
r = call(op, key, inqryDiv=2, prdctClsfcNoNm=name, numOfRows=999, pageNo=page)
|
||||
b = r["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 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)
|
||||
(out / "_meta.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
return out
|
||||
|
||||
|
||||
|
||||
# ── 짝짓기 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def load(day: str) -> dict[str, list[dict]]:
|
||||
"""받아 둔 원본 → {파일이름(.json.gz 뺀): 줄들}"""
|
||||
return {
|
||||
f.name.removesuffix(".json.gz"): json.loads(gzip.decompress(f.read_bytes()))
|
||||
for f in sorted((RAW / day).glob("*.json.gz"))
|
||||
}
|
||||
|
||||
|
||||
def latest() -> str:
|
||||
return max(p.name for p in RAW.iterdir() if p.is_dir())
|
||||
|
||||
|
||||
ALIAS = {"경기": "경기도", "경남": "경상남도", "경북": "경상북도", "충남": "충청남도", "충북": "충청북도",
|
||||
"전남": "전라남도", "전북": "전라북도", "강원": "강원도", "제주": "제주특별자치도",
|
||||
"서울": "서울특별시", "부산": "부산광역시", "대구": "대구광역시", "인천": "인천광역시",
|
||||
"광주": "광주광역시", "대전": "대전광역시", "울산": "울산광역시", "세종": "세종특별자치시"}
|
||||
WORDS = re.compile(r"[가-힣]+")
|
||||
|
||||
|
||||
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(("조합", "레미콘")):
|
||||
continue
|
||||
w = ALIAS.get(w, w)
|
||||
if w.endswith(("광역시", "특별시", "특별자치시", "특별자치도")):
|
||||
w = w[:2]
|
||||
elif len(w) > 2 and w.endswith(("시", "군", "구", "도")) and w not in ("경기도", "강원도"):
|
||||
w = w[:-1]
|
||||
elif w in ("경기도", "강원도"):
|
||||
w = w[:2]
|
||||
out.add(w)
|
||||
return frozenset(out)
|
||||
|
||||
|
||||
def strength(text: str):
|
||||
"""「25-18-080」·「25-18-80」 → (25, 18, 80) · 휨 강도 등은 글 그대로"""
|
||||
m = re.search(r"(\d+)-(\d+)-(\d+)", text or "")
|
||||
return tuple(int(x) for x in m.groups()) if m else None
|
||||
|
||||
|
||||
def region_pick(cands: dict, want: frozenset):
|
||||
"""cands = {지명묶음: 줄} — 같은 묶음 → 겹침 가장 큰 하나(동점이면 None)."""
|
||||
if want in cands:
|
||||
return cands[want], "지역 같음"
|
||||
best = sorted(((len(want & k) / len(want | k), k) for k in cands if want & k), reverse=True)
|
||||
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, "지역 못 찾음"
|
||||
if __name__ == "__main__":
|
||||
mode = sys.argv[1] if len(sys.argv) > 1 else "fetch"
|
||||
if mode == "fetch":
|
||||
fetch_all()
|
||||
Reference in New Issue
Block a user