merge(마스터): sub_laptop_4 관급 확대 합침 — 자재품목 열 정리 위에 조달 값·새 줄 얹기

- 자재품목 = 내 열 정리(유류 5줄 삭제 · 원문번호 칸 없음) 기준 · 저쪽 관급 값 1,012 줄 · 비고 1,507 줄 채택
- 조달청에만 있는 380 줄 더함 — 원문번호 칸 없이 · 키 MT023981~MT024360 은 _키대장.json 에 등록
- update_관급단가.py 는 procured() 로 대장에서 「조달청 …」 키를 가려 짝지음 — 줄의 원문번호 칸을 보지 않음
- _키대장.json 「다음」 MT 24361 · MO 73 · 「폐기」 는 양쪽 합집합

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JgUhN55Z3BCYhUJTjwTNfj
This commit is contained in:
2026-09-20 18:58:02 +09:00
co-authored by Claude Opus 5
40 changed files with 2872 additions and 1596 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,40 @@ 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, "부직포"),
(MAS, "경질폴리염화비닐관"),
(THPTY, "경질폴리염화비닐관"),
]
# 보관하는 칸 — 나머지(첨부·원산지·이미지·긴 설명)는 버림(원본이 수백 MB)
KEEP = (
@@ -53,10 +90,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 +112,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 +129,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)
@@ -171,6 +215,10 @@ def region_pick(cands: dict, want: frozenset):
hit = [k for k in cands if k <= want and k - PROVINCE]
if towns <= set().union(*hit) if hit else False:
return [r for k in hit for r in cands[k]], f"지역 묶음 {len(hit)}"
# 다산 묶음의 시군이 조달청 한 줄의 공급 지역 안에 모두 들어 있음 — 그 줄들의 최저가(값이 갈리면 pick 이 최저)
sup = [k for k in cands if towns and towns <= k]
if sup:
return [r for k in sup for r in cands[k]], f"지역 포함 {len(sup)}"
return None, "지역 못 찾음"
@@ -201,7 +249,7 @@ def _dia(t):
def _thick(t):
m = re.search(r"[×xX*](\d+(?:\.\d+)?)", t or "")
m = re.search(r"[×xX*]t?(\d+(?:\.\d+)?)", t or "")
return float(m.group(1)) if m else None
@@ -212,10 +260,15 @@ def _size(t):
return tuple(float(x) for x in m.groups()) if m else None
CORR_WORDS = ("PE양면피복", "PE내면피복", "아연도금", "유공관", "접속티", "커플링밴드", "직관")
CORR_EXTRA = ("내부평활", "플랜지", "세라믹", "복합", "PE+PP", "PE/PP", "내식")
class Matcher:
"""자재품목 줄 → 조달청 줄. 짝이 서면 (줄, 방식), 못 서면 (None, 까닭)."""
def __init__(self, raw: dict):
self.raw = raw
self.rmc = {}
for r in raw.get("MASCntrct_레미콘", []):
if r["prdctMakrNm"] == "조합공통품목":
@@ -244,8 +297,16 @@ class Matcher:
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: # 구분 없이 붙은 시군(「진주사천남해…」)은 두 글자씩 끊어 다시
cut = frozenset(
q
for w in want
for q in ([w[i : i + 2] for i in range(0, len(w), 2)] if len(w) >= 6 else [w])
)
if cut != want:
cands, why = region_pick(table, cut)
if cands is None or (
why.startswith("지역 묶음") and "제외" in name
why.startswith(("지역 묶음", "지역 포함")) and "제외" in name
): # 「~ 제외」 묶음은 짝짓지 않음
return None, why if cands is None else "지역 못 찾음"
corp = re.findall(
@@ -256,6 +317,12 @@ class Matcher:
return cands, why
def find(self, row: dict):
r, why = self._find(row)
if r is None:
why = misc.absent(row["이름"], why)
return r, why
def _find(self, row: dict):
n, spec = row["이름"], row["규격"]
if n.startswith("레미콘(관급)-"):
k = strength(spec)
@@ -269,7 +336,15 @@ class Matcher:
if re.match(r"(순환)?아스콘\(관급\)-", n):
k = (
n.startswith("순환"),
re.sub(r"\s+", "", re.split(r"조합공통품목(?:\([^)]*\))?,", spec)[-1]),
re.sub(
r"\s+",
"",
re.sub(
r"^\s*[A-Z]\d-H-[^,]*,",
"",
re.split(r"조합공통품목(?:\([^)]*\))?,", spec)[-1],
),
), # 「R3-H-BB2, BB-2, …」 = 앞의 순환 관리 번호를 떼고 「BB-2, …」
)
if k not in self.asc:
return None, "규격 없음(원문 규격이 조달청에 없음)"
@@ -315,18 +390,31 @@ class Matcher:
if re.match(r"파형강관\(관급\)", n):
dia, th = _dia(spec), _thick(spec)
rs = re.search(r"(\d)RS", spec)
c = [
if not (dia and th and rs):
return None, "후보 없음"
coat = [w for w in ("PE양면피복", "PE내면피복") if w in spec]
c = [ # 먼저 「(nRS)」 꼴 — 이미 채운 값이 흔들리지 않게 · PE 피복 줄은 피복 글자가 맞아야
r
for r in self.corr
if dia
and th
and rs
and _dia(r["prdctSpecNm"]) == dia
if _dia(r["prdctSpecNm"]) == dia
and re.search(rf"t{th:g}mm", r["prdctSpecNm"])
and f"({rs.group(1)}RS)" in r["prdctSpecNm"]
and all(re.search(rf"(?<!복합)(?<!\+)(?<!/){w}", r["prdctSpecNm"]) for w in coat)
]
r, how = pick(c)
return r, how
if c:
return pick(c)
kinds = [w for w in CORR_WORDS if w in spec] # 없을 때만 — 종류·피복 글자가 맞는 줄
for r in self.corr:
t = r["prdctSpecNm"]
if (
_dia(t) == dia
and re.search(rf"t{re.escape(f'{th:g}')}(\.0)?mm", t)
and re.search(rf"(?<!\d){rs.group(1)}RS", t)
and all(re.search(rf"(?<!복합)(?<!\+)(?<!/){w}", t) for w in kinds)
and all(w in spec or w not in t for w in CORR_EXTRA) # 다산에 없는 덧붙임은 뺌
):
c.append(r)
return pick(c)
if re.match(r"\s*관", n):
dia = _dia(spec)
c = [
@@ -335,7 +423,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 +437,38 @@ 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 procured(book: dict | None = None) -> dict[str, str]:
"""{자재품목 키: 조달청 원문번호} — 줄에 원문번호 칸이 없어 _키대장.json 으로 가름."""
book = book or mk.load_book()
return {
k: v["원문번호"]
for k, v in book[""].items()
if v["파일"] == "재료_자재품목.json" and str(v["원문번호"]).startswith("조달청 ")
}
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), []
m, out, mine = Matcher(raw), [], procured()
for x in rows:
if x[""] in mine: # 조달청에서 새로 들인 줄은 아래 `add_rows` 몫
continue
if x["관급"] is None and not (x["비고"] or "").startswith(
"관급:"
): # 지난번에 못 채운 줄도 다시 시도
@@ -371,17 +484,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 +500,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 +509,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 +531,76 @@ 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 · 같은 품명·규격은 최저가). 이미 든 줄은 값·비고만 갈음."""
book, new = mk.load_book(), []
mine = procured(book)
have = {mine[x[""]]: x for x in data[""] if x[""] in mine}
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, "구분": 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 +638,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,189 @@
"""관급 단가 — 「그 밖」 갈래 짝짓기(주철관 · 조경석 · 골재 · 시멘트 · 이음관 …) — `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_폴리에틸렌"],
),
)
# 조달청에 그 품목·규격이 없거나 다산 규격으로는 물건을 가릴 수 없는 것 — 「관급」 못 채움의 까닭 글
ABSENT = (
(
r"^벤치플륨",
"조달청 종합쇼핑몰에 없음 — 벤치플륨 품명 없음(수로관 유사품이 같은 물건인지 확인 못 함)",
),
(r"^무게수로관", "조달청 종합쇼핑몰에 없음 — 무게수로관 품명 없음"),
(r"^HI-VP", "조달청 종합쇼핑몰에 없음 — 삼중벽 품명 없음(단층 HIVG 만 있음)"),
(r"^KP 접합부속", "조달청 종합쇼핑몰에 없음 — 접합부속 세트 품명 없음"),
(
r"^소형고압블록",
"조달청 종합쇼핑몰에 없음 — 소형고압블록 품명 없음(보차도용콘크리트블록은 형·색 표기가 달라 안 지음)",
),
(
r"^(PE직관\(수도용\)|PE이중벽)",
"짝 못 지음 — 조달청은 압력·강성 등급별 값인데 다산 규격에 등급 없음",
),
(r"^VR관", "조달청 종합쇼핑몰에 없음 — VR관 품명 없음"),
(
r"^(순환)?아스콘.*제외",
"짝 못 지음 — 다산 묶음이 「~구 제외」 꼴이라 조달청 공급 지역과 안 맞음",
),
(r"^측구수로관", "짝 못 지음 — 조달청은 길이·형별 표기(가로×세로×길이)"),
)
GONE = ("후보 없음", "강도 없음(원문 규격이 조달청에 없음)", "규격 없음(원문 규격이 조달청에 없음)")
def absent(name: str, why: str) -> str:
"""못 채운 까닭 → 조달청에 없음 · 짝 못 지음 을 가려 적음."""
for pat, text in ABSENT:
if re.search(pat, name):
return text
if why == "지역 못 찾음":
return "조달청 종합쇼핑몰에 그 지역 묶음의 줄 없음 — 지역 못 찾음"
return f"조달청 종합쇼핑몰에 없음 — {why}" if why in GONE else why
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),(?=\d{3})", "", t) # 「1,000」
t = re.sub(r"(?<=[×x*\s,])t(?=\d)", "", t).replace("solid", "솔리드") # 「×t30」 = 「×30」
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 None if m.group(1) == "관급" else 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
if fam[0] == "골재": # 「토금산업(홍성), 75mm」 — 공장 이름 + 굵기 · 지역은 공장이 정함
spec, reg = re.sub(r"\([^)]*\)", "", spec), None
paren = re.search(r"\(([가-힣]+)", n) if fam[0] in ("교량받침", "합성목재") 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 paren and norm(paren.group(1)) not in norm(
r["prdctSpecNm"] + r["cntrctCorpNm"] + str(r["prdctMakrNm"])
):
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