chore(resources): 옛 마스터 파일 master_data · tmp 두 폴더로 모음

- resources 아래 옛 폴더 18개 → master_data(40) · tmp(8) — 하위 폴더 없이 파일만
- 파일 이름 = 분류_대상_내용_출처_기준일 (분류: 공종 · 단가 · 품셈 · 화면)
- tmp = 실무문서 · 교본 값이 섞인 파일 — 확인 뒤 지울 것
- 코드 연결은 끊긴 채 둠(새 마스터로 다시 세움)

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QqkvdyxUtANWroVKpQUyxE
This commit is contained in:
2026-09-19 13:17:40 +09:00
co-authored by Claude Opus 5
parent 834eccec2f
commit f0e44ecb13
48 changed files with 0 additions and 0 deletions
@@ -1,455 +0,0 @@
# -*- coding: utf-8 -*-
"""기계 원천 벌 뽑기 — 건설 품셈 제8장 md → `mach_base_2026.json` (2026-09-18 브레인).
왜 — 코덱스가 8장 md 를 손질(뭉친 표 되살리기 · 8-4 운전경비 92종 전사)했는데 자료가 앞 판 그대로였음.
품셈 벌(`_build_pum.py`)과 같은 병 — 뽑는 길이 저장소에 없어 md 가 고쳐져도 자료로 못 내려옴.
무엇을 하나
- ⚠ **줄 번호로 안 잡음.** 자리는 **기계 코드**(`5401-0015`)로 잡고, 절은 글(`8-3 기계손료` …)로 찾음.
- 8-5 기계가격 → 취득가 · 8-3 기계손료 → 손료 계수 · 8-4 운전경비 → 주연료·잡재료·조종원.
- 한 칸에 코드가 여럿인 줄(`0101-0007 0010 0012`)은 **토막 수가 맞을 때만** 읽고, 안 맞으면
`parse_audit` 에 적음(지어내지 않음). 8-4 원표는 `source_tables` 로 통째로 실어 둠.
- 이름은 `(0101) 불도저(무한궤도)` 꼴 머리글 · 가격표 `기 종` 칸 · 연료표 `기계명` 칸에서.
- **사람 판단·원문에 없는 칸은 옛 벌에서 이어받음** — 조종원 배정(121) · 암석 보정 · 붙임글 ·
손으로 맞춘 연료 줄(`parse_method`)과 손질 사유(`note`). 이어받은 것은 알림에 수로 뜸.
돌리기: `./venv/Scripts/python.exe resources/data_cost_input_value/_build_mach.py [--check]`
`--check` — 쓰지 않고 달라질 것만 알림 · 달라지면 끝 코드 1.
⚠ 금액 불변 — 기계경비는 뿌리라 움직이면 크게 움직임. 뽑은 뒤 `test_work_item_key_gate.py`
원가계산서 시험(직접공사비 5,983,724)을 돌릴 것.
"""
from __future__ import annotations
import json
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
TARGET = ROOT / "resources/data_cost_input_value/mach_base_2026.json"
#: 절 — 글로 찾음(줄 번호 아님).
MARKS = ("8-3 기계손료", "8-4 운전경비", "8-5 기계가격")
_SEP = re.compile(r"\s*\|(\s*:?-+:?\s*\|)+\s*")
_CODE = re.compile(r"\d{4}-\d{4}")
_NAME_AT = re.compile(
r"\((\d{4})\)\s*([^()]*(?:\([^)]*\)[^()]*)*?)(?=\(\d{4}\)|\('\d\d|시\s*간\s*당|$)"
)
#: 손료표 칸 — 머리글(공백 지운 것)에 이 글이 들어 있으면 그 칸.
LOSS_COLS = {
"economic_life_hours": "내용",
"annual_standard_hours": "연간표준",
"depreciation_ratio": "상각비율",
"maintenance_ratio": "정비비율",
"annual_management_ratio": "관리비율",
}
LOSS_COEF = (
"depreciation_coefficient_1e_minus_7",
"maintenance_coefficient_1e_minus_7",
"management_coefficient_1e_minus_7",
"source_coefficient_1e_minus_7",
)
FUEL_COLS = {
"fuel_rate_l_per_hour": "주연료",
"misc_material_percent_of_fuel": "잡재료",
"operator_person_per_day": "조종원",
}
def _t(text: str) -> str:
return re.sub(r"\s+", "", str(text))
def _cells(line: str) -> list[str]:
s = line.strip()
s = s[1:-1] if s.endswith("|") else s[1:]
return [c.strip() for c in s.split("|")]
def md_tables(lines: list[str]) -> list[dict]:
tables, i = [], 0
while i < len(lines):
if (
lines[i].lstrip().startswith("|")
and i + 1 < len(lines)
and _SEP.fullmatch(lines[i + 1])
):
j = i + 2
while j < len(lines) and lines[j].lstrip().startswith("|"):
j += 1
tables.append(
{
"line": i + 1,
"headers": _cells(lines[i]),
"rows": [_cells(x) for x in lines[i + 2 : j]],
}
)
i = j
else:
i += 1
return tables
def _num(text: str) -> float | int | None:
t = _t(text).replace(",", "").rstrip("%")
t = re.sub(r"^[^\d.]*", "", t) # `휘발유0.7` · `약1.2`
if not re.fullmatch(r"\d+(\.\d+)?", t):
return None
return float(t) if "." in t else int(t)
def codes_in(cell: str, prefix: str | None) -> tuple[list[str], str | None]:
"""`0101-0007 0010` → `['0101-0007','0101-0010']`. 앞자리는 이어짐. 코드가 없으면 빈 목록."""
out = []
for token in cell.split():
if _CODE.fullmatch(token):
prefix = token[:4]
out.append(token)
elif re.fullmatch(r"\d{4}-", token): # `9020- 0010 0015` — 앞자리만 먼저 적은 줄
prefix = token[:4]
elif re.fullmatch(r"-?\d{4}", token) and prefix: # `0010` · `-0184`
out.append(f"{prefix}-{token.lstrip('-')}")
else:
return [], prefix
return out, prefix
def names_in(lines: list[str]) -> dict[str, str]:
"""`(0101) 불도저(무한궤도)` 머리글 → {앞자리: 이름}. 먼저 나온 것을 씀."""
found: dict[str, str] = {}
for line in lines:
if line.lstrip().startswith("|"):
continue
for m in _NAME_AT.finditer(line):
name = m.group(2).strip(" .·")
# ⚠ 숫자로 시작하는 이름도 있다 — `(8201) 3D GNSS 머신 가이던스(굴착기용)`.
# 첫 낱자가 숫자면 버리던 잣대가 그것을 쪽번호로 보고 내버려, 옛 벌의 뭉개진 글을
# 이어받고 있었다(2026-09-18). 낱자가 하나도 없는 토막만 버린다.
if name and re.search(r"[가-힣A-Za-z]", name):
found.setdefault(m.group(1), re.sub(r"\s+", " ", name))
return found
def split_values(cell: str, count: int) -> list[str] | None:
"""칸 하나에 값이 여럿(`9.0 12.5 14.6`) — 토막 수가 맞을 때만.
⚠ 코드가 하나인 줄은 **칸 통째로** 한 값(`320~ 400` 처럼 값 안에 공백이 있음).
"""
if count == 1:
return [cell.strip()]
parts = cell.split()
if len(parts) == count:
return parts
# 값 안에 공백이 있는 규격 칸 — `전기식 …ø45(0.75㎾) 엔진식 …ø45(2.6㎾)` 은 닫는 괄호로 가름
chunks = [x.strip() for x in re.split(r"(?<=\))\s+", cell.strip()) if x.strip()]
return chunks if len(chunks) == count else None
def parse_price(tables: list[dict], names: dict[str, str]) -> tuple[list[dict], list[dict]]:
records, bad, prefix = [], [], None
for t in tables:
if [_t(h) for h in t["headers"][:3]] != ["기종", "분류번호", "가격(₩)"]:
continue
for row in t["rows"]:
codes, prefix = codes_in(row[1], prefix)
prices = row[2].split()
if not codes or len(codes) != len(prices):
bad.append({"what": row[1][:60], "codes": len(codes), "values": len(prices)})
continue
for code, price in zip(codes, prices):
records.append({"machine_code": code, "machine_name": names.get(code[:4], ""), "price_thousand_krw": _num(price)}) # fmt: skip
return records, bad
def parse_loss(tables: list[dict], names: dict[str, str]) -> tuple[list[dict], list[dict]]:
records, bad, prefix = [], [], None
for t in tables:
head = [_t(h) for h in t["headers"]]
if not head or "분류번호" not in head[0]:
continue
cols = {field: next((i for i, h in enumerate(head) if word in h), None) for field, word in LOSS_COLS.items()} # fmt: skip
sub = next((r for r in t["rows"] if not _t(r[0])), [])
if "상각비계수" in [_t(c) for c in sub]:
first = [_t(c) for c in sub].index("상각비계수")
coef = dict(zip(LOSS_COEF, range(first, first + 4)))
elif (only := next((i for i, h in enumerate(head) if "시간당" in h), None)) is not None:
coef = {LOSS_COEF[3]: only}
else:
bad.append({"what": f"손료표 계수 칸 없음 — {t['headers']}"})
continue
# 규격 칸 — 분류번호와 첫 값 칸 사이. 둘·셋이면(5202·5203) 첫 칸만 규격으로 봄
after = min(
[i for i in list(cols.values()) + list(coef.values()) if i is not None], default=1
)
spec_col = 1 if after > 1 else None
for row in t["rows"]:
if not _t(row[0]):
continue
codes, prefix = codes_in(row[0], prefix)
values = {}
ok = bool(codes)
for field, index in {**cols, **coef}.items():
if index is None or index >= len(row):
continue
parts = split_values(row[index], len(codes)) if codes else None
if parts is None:
ok = False
break
values[field] = [_num(p) for p in parts]
specs = split_values(row[spec_col], len(codes)) if (ok and spec_col is not None and spec_col < len(row)) else None # fmt: skip
if not ok or LOSS_COEF[3] not in values:
bad.append({"what": row[0][:40], "codes": len(codes)})
continue
for n, code in enumerate(codes):
record = {"machine_code": code, "machine_name": names.get(code[:4], "")}
if specs and _t(specs[n]) not in ("", "-"):
record["specification"] = specs[n].strip()
total = values[LOSS_COEF[3]][n]
record["loss_coefficient_per_hour"] = None if total is None else total * 1e-7
for field in (*LOSS_COEF, *LOSS_COLS):
if values.get(field, [None])[n] is not None:
record[field] = values[field][n]
records.append(record)
return records, bad
def sections_by_line(lines: list[str], mark: str) -> dict[int, str]:
"""줄 번호 → 그 줄 위에서 마지막으로 본 절 제목(`8-4-3 [20]운반 및 하역기계`).
⚠ 제목이 쪽 번호와 붙어 있어(`278공통부문8-4-3 …`) **절 번호부터** 끊어 씀. 제목이 없는 동안은
앞 제목을 이어 씀 — 해상기계처럼 기종 머리글만 잇달아 나오는 자리가 있음.
"""
found, current = {}, ""
for index, text in enumerate(lines, 1):
if not text.lstrip().startswith("|"):
if m := re.search(rf"{mark}(?:-\d+)?(?=[\s\[(가-힣])", text):
current = text[m.start() :].strip()
found[index] = current
return found
def parse_fuel(
sections: dict[int, str], tables: list[dict], names: dict[str, str]
) -> tuple[list[dict], list[dict], list[dict], list[dict]]:
records, bad, source_tables, skipped, prefix = [], [], [], [], None
for t in tables:
head = [_t(h) for h in t["headers"]]
source_tables.append({"section": sections.get(t["line"], ""), "line": t["line"], "headers": t["headers"], "rows": t["rows"]}) # fmt: skip
if not head or "분류번호" not in head[0]:
continue
cols = {field: next((i for i, h in enumerate(head) if word in h), None) for field, word in FUEL_COLS.items()} # fmt: skip
spec_col = next((i for i, h in enumerate(head) if "규격" in h), None)
for row in t["rows"]:
codes, prefix = codes_in(row[0], prefix)
values, ok = {}, bool(codes)
for field, index in cols.items():
if index is None or index >= len(row):
continue
parts = split_values(row[index], len(codes))
if parts is None:
ok = False
break
values[field] = parts
if not ok or "fuel_rate_l_per_hour" not in values:
bad.append({"what": row[0][:40], "codes": len(codes)})
continue
specs = split_values(row[spec_col], len(codes)) if spec_col is not None and spec_col < len(row) else None # fmt: skip
for n, code in enumerate(codes):
raw = values["fuel_rate_l_per_hour"][n]
fuels, power = split_fuels(raw)
if not fuels:
# ⚠ 주연료가 수가 아닌 기종(전기 `㎾` · `-`)은 **연료 목록에 안 실음** — ℓ/hr 자료임.
skipped.append({"machine_code": code, "what": raw.strip()[:20]})
continue
record = {
"machine_code": code,
"machine_name": names.get(code[:4], ""),
"fuel_type": fuels[0][0],
"fuel_rate_l_per_hour": fuels[0][1],
}
# ⚠ **겸용 연료는 둘 다 싣는다** — 한쪽만 적으면 그만큼 경비가 빈다
# (`73.7+휘발유54.5` 현장가열표층재생기 · `1.87㎾ 휘발유1.3` 모르타르 믹서).
if len(fuels) > 1:
record["secondary_fuel_type"] = fuels[1][0]
record["secondary_fuel_rate_l_per_hour"] = fuels[1][1]
if power is not None:
record["electric_power_kw"] = power
for field in ("misc_material_percent_of_fuel", "operator_person_per_day"):
value = _num(values[field][n]) if field in values else None
if value is not None:
record[field] = value
if specs and _t(specs[n]) not in ("", "-"):
record["specification"] = specs[n].strip()
records.append(record)
return records, bad, source_tables, skipped
#: 겸용 연료 칸을 토막낸다 — `73.7+휘발유54.5` · `1.87㎾ 휘발유1.3` · `중유487.2`.
#: ⚠ 표시가 없으면 경유다([주]① ㉰).
_FUEL_PIECE_RE = re.compile(r"(휘발유|가솔린|중유|경유)?\s*([\d,]+(?:\.\d+)?)\s*(㎾|kW|kw)?")
def split_fuels(raw: str) -> tuple[list[tuple[str, float]], float | None]:
"""연료 칸 → (`[(연료갈래, ℓ/hr), …]`, 전력 ㎾).
한 칸에 연료가 둘인 기종이 있다 — 그 둘을 **다 싣는다**. 수가 없으면 빈 목록(전기·`-`).
"""
fuels: list[tuple[str, float]] = []
power: float | None = None
for mark, number, kw in _FUEL_PIECE_RE.findall(str(raw or "")):
value = _num(number)
if value is None:
continue
if kw: # 전력은 연료(ℓ/hr)가 아니라 곁값으로 적어 둔다
power = value
continue
kind = (
"gasoline"
if mark in ("휘발유", "가솔린")
else "heavy_oil"
if mark == "중유"
else "diesel"
)
fuels.append((kind, value))
return fuels, power
def extract(old_doc: dict, data: bytes) -> tuple[dict, dict]:
"""옛 벌 + 8장 md → (새 벌, 알림). 사람 판단 칸은 옛 벌에서 이어받음."""
lines = data.decode("utf-8").splitlines()
at = {mark: next(i for i, x in enumerate(lines) if mark in x) for mark in MARKS}
tables = md_tables(lines)
zone = lambda a, b: [t for t in tables if at[a] < t["line"] <= (at[b] if b else len(lines))] # noqa: E731
names = names_in(lines)
price, bad_price = parse_price(zone("8-5 기계가격", None), names)
loss, bad_loss = parse_loss(zone("8-3 기계손료", "8-4 운전경비"), names)
fuel, bad_fuel, source_tables, no_fuel = parse_fuel(
sections_by_line(lines, "8-4"), zone("8-4 운전경비", "8-5 기계가격"), names
)
old = old_doc["variables"]
carried = {"fuel": 0, "note": 0, "specification": 0, "machine_name": 0}
# 원문에서 못 읽은 줄은 옛 벌 값을 이어받음(손으로 맞춘 줄 · 손질 사유 · 규격)
old_fuel = {r["machine_code"]: r for r in old["mach_fuel_rate"]["parsed_records"]}
got = {r["machine_code"] for r in fuel}
# ⚠ **원문에 줄이 있는데 연료를 못 읽은 코드만** 이어받는다 —
# 못 읽은 줄(토막 수 안 맞음) · 주연료 칸이 수가 아닌 줄(`1.49㎾` 처럼 칸이 밀린 양수기).
# 원문에 아예 없는 기종은 안 이어받는다(원문에 없는 줄이 자료에 남는다).
unread = {c for row in bad_fuel for c in codes_in(row["what"], None)[0]}
unread |= {r["machine_code"] for r in no_fuel}
for code, record in old_fuel.items():
if code not in got and code in unread:
fuel.append(record)
carried["fuel"] += 1
fuel.sort(key=lambda r: r["machine_code"])
# 가격표엔 규격 칸이 없음 — 같은 코드의 손료표 규격을 씀. 연료표 규격도 손료표 것이 먼저
# (`7ton` · `0.12㎥` 처럼 단위가 붙어 있어 갈래가 갈림).
loss_spec = {r["machine_code"]: r.get("specification") for r in loss}
for record in price + fuel:
if loss_spec.get(record["machine_code"]):
record["specification"] = loss_spec[record["machine_code"]]
for records, key, field in ((loss, "mach_loss_coef", "records"), (price, "mach_price", "records"), (fuel, "mach_fuel_rate", "parsed_records")): # fmt: skip
old_by_code = {r["machine_code"]: r for r in old[key][field]}
for record in records:
was = old_by_code.get(record["machine_code"], {})
if not record.get("machine_name") and was.get("machine_name"):
record["machine_name"] = was["machine_name"] # 머리글이 없는 기종(8202~8204)
carried["machine_name"] = carried.get("machine_name", 0) + 1
if "note" in was:
record["note"] = was["note"]
carried["note"] += 1
if not record.get("specification") and was.get("specification"):
record["specification"] = was["specification"]
carried["specification"] += 1
doc = {
**old_doc,
"sources": [
{**s, "sha256": __import__("hashlib").sha256(data).hexdigest()}
for s in old_doc["sources"]
],
"variables": {
**old,
"mach_price": {**old["mach_price"], "records": price},
"mach_loss_coef": {**old["mach_loss_coef"], "records": loss},
"mach_fuel_rate": {
**old["mach_fuel_rate"],
"parsed_records": fuel,
"source_tables": source_tables,
},
},
# ⚠ 자세한 줄은 **알림에 찍음** — 자료엔 셈만 둠(이름표가 표로 잡지 않게).
"parse_audit": {
"unparsed_price_rows": len(bad_price),
"unparsed_loss_rows": len(bad_loss),
"unparsed_fuel_rows": len(bad_fuel),
"fuel_not_litre_rows": len(no_fuel),
},
}
report = {
"price": len(price), "loss": len(loss), "fuel": len(fuel),
"carried": carried,
"unparsed": {"price": len(bad_price), "loss": len(bad_loss), "fuel": len(bad_fuel)},
"unparsed_rows": {"price": bad_price, "loss": bad_loss, "fuel": bad_fuel},
"주연료가 수가 아님": len(no_fuel),
} # fmt: skip
if doc != old_doc:
doc["generated_at"] = datetime.now(timezone(timedelta(hours=9))).isoformat(
timespec="seconds"
)
return doc, report
def diff_report(old_doc: dict, doc: dict) -> dict:
"""옛 벌 ↔ 새 벌 — 갈래마다 새 코드 · 사라진 코드 · 값이 달라진 코드."""
out = {}
for key, field in (
("mach_price", "records"),
("mach_loss_coef", "records"),
("mach_fuel_rate", "parsed_records"),
):
old = {r["machine_code"]: r for r in old_doc["variables"][key][field]}
new = {r["machine_code"]: r for r in doc["variables"][key][field]}
changed = {
code: {
k: (old[code].get(k), new[code].get(k))
for k in set(old[code]) | set(new[code])
if old[code].get(k) != new[code].get(k)
}
for code in sorted(set(old) & set(new))
if old[code] != new[code]
}
out[key] = {
"새 코드": sorted(set(new) - set(old)),
"사라진 코드": sorted(set(old) - set(new)),
"값 바뀐 코드": changed,
}
return out
def main(check: bool) -> int:
old_doc = json.loads(TARGET.read_text(encoding="utf-8"))
data = (ROOT / old_doc["sources"][0]["path"]).read_bytes()
doc, report = extract(old_doc, data)
changed = doc != old_doc
diff = diff_report(old_doc, doc)
print(f"── mach_base: 가격 {report['price']} · 손료 {report['loss']} · 연료 {report['fuel']}")
print(
f" 못 읽은 줄 {report['unparsed']} · 주연료가 수가 아닌 기종 {report['주연료가 수가 아님']}"
)
print(f" 옛 벌에서 이어받음 {report['carried']}")
for kind, rows in report["unparsed_rows"].items():
for row in rows:
print(f" 못 읽음({kind}): {json.dumps(row, ensure_ascii=False)}")
for key, item in diff.items():
name_changed = sum(1 for v in item["값 바뀐 코드"].values() if set(v) == {"machine_name"})
print(
f" {key}: 새 {len(item['새 코드'])} · 사라짐 {len(item['사라진 코드'])} · "
f"값 바뀜 {len(item['값 바뀐 코드'])}(이름만 {name_changed})"
)
if changed and not check:
TARGET.write_text(json.dumps(doc, ensure_ascii=False, indent=1) + "\n", encoding="utf-8")
return 1 if (check and changed) else 0
if __name__ == "__main__":
sys.exit(main("--check" in sys.argv))
@@ -1,100 +0,0 @@
# -*- coding: utf-8 -*-
"""관급자재 단가 — **0 원 공시를 건너뛰고 앞 제값 공시를 채택**하는 규칙(2026-09-16 브레인).
왜 — 우리 고르기 규칙이 「물품코드마다 가장 나중 공시」라, 값이 안 실린 공시(`prce = 0`)가
옛 제값을 덮었다. 각형강관 25*25*1 이 `1,460 → 1,500 → 0` 으로 끝난 것이 그것이다.
「가격이 0」이 아니라 **그 공시에 값이 안 실린 것**이라, 0 을 값으로 받으면 안 된다.
무엇을 고치나
- 마스터의 0 원 줄만 고친다. **원천 스냅숏은 안 건드린다**(읽기만).
- 0 이 아닌 줄은 한 글자도 안 건드린다 — 규칙을 통째로 다시 돌리지 않는다.
- 고친 줄마다 **사유**(`price_adopted_reason`)를 남긴다. 조용히 바꾸지 않는다.
- 공시 날짜·번호도 **채택한 공시의 것**으로 바꾼다 — 어느 판 값인지 화면에서 보여야 한다.
- 앞 공시에도 값이 없어 0 이 남는 줄은 `zero_price_unrecovered` 로 낸다(규칙으로 못 고침).
⚠ 같은 날짜에 0 과 제값이 함께 실린다(2025-04-24) — 날짜만 보지 말고 **0 을 건너뛴 뒤**
가장 나중 공시를 고른다.
돌리기: `./venv/Scripts/python.exe resources/data_cost_input_value/_build_mat_price.py`
"""
from __future__ import annotations
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
RAW = ROOT / "resources/knowledge/original/원가계산/자재단가/나라장터_시설공통자재_2026-08-14.json"
MASTER = ROOT / "resources/data_cost_input_value/mat_price_public_2026-08-14.json"
#: 값이 안 실린 공시를 가리는 잣대 — 빈 칸·0 은 「값 없음」이다.
EMPTY_PRICES = {"", "0", "0.0", "0.00"}
SELECTION_POLICY = (
"latest notice per product identification number within the local snapshot, "
"skipping notices whose price is empty or zero "
"(a zero price means the notice carried no price, not a price of zero)"
)
UNRECOVERED_NOTE = (
"앞 공시에도 값이 없어 0 이 남은 줄. **규칙으로 못 고친다** — 0 을 값으로 쓰지 말고 "
"그 자재가 실제로 필요해지면 원천을 다시 받거나 견적을 받아야 한다."
)
def price_of(row: dict) -> str:
return str(row.get("prce", "")).strip()
def has_price(row: dict) -> bool:
return price_of(row) not in EMPTY_PRICES
def notice_day(row: dict) -> str:
return str(row.get("nticeDt", ""))[:10]
def build() -> dict:
raw = json.loads(RAW.read_text(encoding="utf-8"))
master = json.loads(MASTER.read_text(encoding="utf-8"))
records = master["variables"]["mat_price"]["records"]
history: dict[str, list[dict]] = {}
for row in raw:
history.setdefault(str(row.get("prdctIdntNo")), []).append(row)
for rows in history.values():
rows.sort(key=lambda r: str(r.get("nticeDt")), reverse=True)
repaired, unrecovered = 0, []
for record in records:
if str(record.get("price_krw")).strip() not in EMPTY_PRICES:
continue
was = notice_day(record | {"nticeDt": record.get("notice_datetime")})
adopted = next((r for r in history.get(str(record["item_code"]), ()) if has_price(r)), None)
if adopted is None:
unrecovered.append(
{
"item_code": record["item_code"],
"classification_name": record.get("classification_name", ""),
"specification": record.get("specification", ""),
"notice_datetime": record.get("notice_datetime", ""),
"reason": "이 물품코드의 모든 공시에 값이 없음",
}
)
continue
record["price_krw"] = int(price_of(adopted))
record["notice_datetime"] = str(adopted.get("nticeDt", ""))
record["notice_number"] = str(adopted.get("prceNticeNo", ""))
record["price_adopted_reason"] = f"{was} 공시에 값이 없어 {notice_day(adopted)} 값을 씀"
repaired += 1
master["selection_policy"] = SELECTION_POLICY
master["zero_price_unrecovered"] = {"note": UNRECOVERED_NOTE, "items": unrecovered}
return {"master": master, "repaired": repaired, "unrecovered": len(unrecovered)}
if __name__ == "__main__":
result = build()
MASTER.write_text(
json.dumps(result["master"], ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
print(f"고친 줄 {result['repaired']} · 못 고친 줄 {result['unrecovered']}")
@@ -1,198 +0,0 @@
# -*- coding: utf-8 -*-
"""원천 벌 뽑기 — 품셈 md → `pum_forest_2026.json` · `pum_const_2026.json` (2026-09-18 브레인).
왜 — 원문 md 가 고쳐지면(줄 보탬 · 표 머리 복원) 원천 벌의 **줄 번호가 밀려** 공종 마스터가 남의 줄을
읽었음(산림 밑수 238 → 105). 머리글로 다시 맞추니 12장에 같은 머리글이 많아 **이웃 표를 집었고**
직접공사비가 조용히 움직였음. 원천 벌을 뽑던 추출기가 저장소에 없었음.
무엇을 하나
- 표 = md 의 `|` 표(머리 줄 + `|---|` 줄 + 몸 줄). 칸 글자를 **그대로** 옮김 —
⚠ 2단 머리를 펼치지 않음(윗머리 「시 간 당(10-7)」 에 밑수·절 번호가 들어 있음).
- ⚠ **줄 번호로 옛 표를 찾지 않음.** 표 번호(`F0001`·`C0001`)와 절(`section`)은 옛 벌에서 이어받되,
짝은 이렇게 지음 — ① 글자(머리·몸)가 같은 표를 **차례를 지키며** 맞댐(앞뒤가 뒤바뀌지 않으니
머리글이 같은 이웃 표로 건너뛸 수 없음) ② 맞은 표 사이 빈 자리의 표 수가 같으면 **같은 차례끼리**
(글자만 고쳐진 표) ③ 수가 다르면 그 안에서 머리가 같은 표끼리 차례대로.
- 새 표 = 가장 큰 번호 다음 번호 · 절은 표 위로 올라가며 처음 만나는 `###` 제목이나
「숫자-숫자 글」 줄(옛 벌의 절 잡는 법 — 산림 476표가 이 규칙과 같음). ⚠ 알림에 뜸 — 사람이 볼 것.
- 사라진 표 = 번호를 버리고 알림에 적음(다른 표에 번호를 물려주지 않음).
- 줄 번호·sha256 은 새 md 에서. 표가 한 칸도 안 바뀌면 파일을 그대로 둠(`generated_at` 도).
돌리기: `./venv/Scripts/python.exe resources/data_cost_input_value/_build_pum.py [--check]`
`--check` — 쓰지 않고 달라질 것만 알림 · 달라지면 끝 코드 1.
금액 불변은 `resources/tester/test_work_item_key_gate.py` 원가계산서 시험(직접공사비 5,983,724)이 잼 —
다시 뽑은 뒤 공종 마스터를 다시 짓고 그 시험을 돌릴 것.
"""
from __future__ import annotations
import difflib
import hashlib
import json
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Callable
ROOT = Path(__file__).resolve().parents[2]
CATALOGS = {
"pum_forest": ROOT / "resources/data_cost_input_value/pum_forest_2026.json",
"pum_const": ROOT / "resources/data_cost_input_value/pum_const_2026.json",
}
PREFIX = {"pum_forest": "F", "pum_const": "C"}
_SEP = re.compile(r"\s*\|(\s*:?-+:?\s*\|)+\s*")
#: 옛 벌의 절 줄 — `12-2 표면 마무리를 따른다.` 처럼 본문 인용도 절로 잡혀 있음(마스터 빌더가 바로잡음).
_SECTION_LINE = re.compile(r"(\d+-\d+(?:-\d+)*\.?\s+\S.*)$")
def _cells(line: str) -> list[str]:
s = line.strip()
s = s[1:-1] if s.endswith("|") else s[1:]
return [c.strip() for c in s.split("|")]
def md_tables(lines: list[str]) -> list[dict]:
"""md `|` 표 목록 — `line` 은 머리 줄 번호(1부터)."""
tables, i = [], 0
while i < len(lines):
if (
lines[i].lstrip().startswith("|")
and i + 1 < len(lines)
and _SEP.fullmatch(lines[i + 1])
):
j = i + 2
while j < len(lines) and lines[j].lstrip().startswith("|"):
j += 1
tables.append(
{
"line": i + 1,
"headers": _cells(lines[i]),
"rows": [_cells(x) for x in lines[i + 2 : j]],
}
)
i = j
else:
i += 1
return tables
def section_above(lines: list[str], line: int) -> str | None:
"""새 표의 절 — 위로 올라가며 처음 만나는 `###` 제목 또는 「숫자-숫자 글」 줄(표 줄은 건너뜀)."""
for k in range(line - 2, -1, -1):
s = lines[k]
if s.startswith("### "):
return s[4:].strip()
if s.lstrip().startswith("|"):
continue
if m := _SECTION_LINE.search(s):
return m.group(1).strip()
return None
def _sig(table: dict) -> str:
return json.dumps([table["headers"], table["rows"]], ensure_ascii=False)
def align(old: list[dict], new: list[dict]) -> list[int | None]:
"""`new[j]` 의 짝 `old` 차례(없으면 None). 차례를 지킴 — 짝끼리 앞뒤가 뒤바뀌지 않음."""
pair: list[int | None] = [None] * len(new)
def pair_blocks(a: list[str], b: list[str], i0: int, j0: int, inner: Callable | None) -> None:
matcher = difflib.SequenceMatcher(None, a, b, autojunk=False)
for tag, i1, i2, j1, j2 in matcher.get_opcodes():
if tag == "equal" or (tag == "replace" and i2 - i1 == j2 - j1):
for k in range(j2 - j1):
pair[j0 + j1 + k] = i0 + i1 + k
elif tag == "replace" and inner:
inner(i0 + i1, i0 + i2, j0 + j1, j0 + j2)
def by_headers(i1: int, i2: int, j1: int, j2: int) -> None:
heads = lambda ts: [json.dumps(t["headers"], ensure_ascii=False) for t in ts] # noqa: E731
matcher = difflib.SequenceMatcher(
None, heads(old[i1:i2]), heads(new[j1:j2]), autojunk=False
)
for tag, a1, a2, b1, b2 in matcher.get_opcodes():
if tag == "equal":
for k in range(b2 - b1):
pair[j1 + b1 + k] = i1 + a1 + k
pair_blocks([_sig(t) for t in old], [_sig(t) for t in new], 0, 0, by_headers)
return pair
def extract(old_doc: dict, read: Callable[[str], bytes], dataset: str) -> tuple[dict, dict]:
"""옛 벌 + 원문 → (새 벌, 알림). `read(path)` 는 원문 파일 바이트."""
old_tables = old_doc["variables"]["pum"]["tables"]
per_file = "source_file" in old_tables[0]
sources = [s for s in old_doc["sources"] if s.get("role") == "primary"]
numbers = [int(t["table_id"][1:]) for t in old_tables]
next_number = max(numbers) + 1
report = {"same": 0, "moved": 0, "changed": [], "new": [], "removed": []}
tables, new_sources = [], []
for src in old_doc["sources"]:
data = read(src["path"])
new_sources.append({**src, "sha256": hashlib.sha256(data).hexdigest()})
for src in sources:
lines = read(src["path"]).decode("utf-8").splitlines()
old = [t for t in old_tables if not per_file or t["source_file"] == src["path"]]
new = md_tables(lines)
for j, k in enumerate(align(old, new)):
t = new[j]
if k is None:
table_id = f"{PREFIX[dataset]}{next_number:04d}"
next_number += 1
section = section_above(lines, t["line"])
report["new"].append({"table_id": table_id, "line": t["line"], "section": section})
else:
table_id, section = old[k]["table_id"], old[k]["section"]
if _sig(old[k]) != _sig(t):
report["changed"].append({"table_id": table_id, "line": t["line"]})
elif old[k]["line"] != t["line"]:
report["moved"] += 1
else:
report["same"] += 1
entry = {"table_id": table_id, "section": section, "line": t["line"], "headers": t["headers"], "rows": t["rows"]} # fmt: skip
if per_file:
entry["source_file"] = src["path"]
tables.append(entry)
kept = {old[k]["table_id"] for k in align(old, new) if k is not None}
report["removed"] += [{"table_id": t["table_id"], "section": t["section"]} for t in old if t["table_id"] not in kept] # fmt: skip
doc = {**old_doc, "sources": new_sources}
doc["variables"] = {
**old_doc["variables"],
"pum": {**old_doc["variables"]["pum"], "tables": tables},
}
if doc != old_doc:
doc["generated_at"] = datetime.now(timezone(timedelta(hours=9))).isoformat(
timespec="seconds"
)
return doc, report
def _read(path: str) -> bytes:
return (ROOT / path).read_bytes()
def main(check: bool) -> int:
dirty = False
for dataset, path in CATALOGS.items():
old_doc = json.loads(path.read_text(encoding="utf-8"))
doc, report = extract(old_doc, _read, dataset)
changed = doc != old_doc
dirty |= changed
print(
f"── {dataset}: 표 {len(doc['variables']['pum']['tables'])} · 그대로 {report['same']} · "
f"줄만 밀림 {report['moved']} · 글자 바뀜 {len(report['changed'])} · 새 표 {len(report['new'])} · "
f"사라진 표 {len(report['removed'])}" + ("" if changed else " — 바뀐 것 없음")
)
for key in ("changed", "new", "removed"):
for item in report[key]:
print(f" {key}: {json.dumps(item, ensure_ascii=False)}")
if changed and not check:
path.write_text(json.dumps(doc, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
return 1 if (check and dirty) else 0
if __name__ == "__main__":
sys.exit(main("--check" in sys.argv))
@@ -1,137 +0,0 @@
{
"schema_version": "1.0",
"dataset_id": "data_cost_input_value_manifest",
"generated_at": "2026-08-15T08:10:50+09:00",
"policy": {
"no_direct_download_performed": true,
"no_invented_values": true,
"no_null_placeholders": true,
"specification_catalogs_are_transferred_as_complete_local_source_tables": true
},
"files": [
{
"file": "coef_2026.json",
"dataset_id": "coef",
"effective_date": "2026-01-01",
"sha256": "daba170cd806374c92e086dec69f9559271c0317a7426e66e9dc841c2ca264bd",
"size_bytes": 31715
},
{
"file": "fx_2026-08-14.json",
"dataset_id": "fx",
"effective_date": "2026-08-14",
"sha256": "9b3325d2cc7f4508d57b677233cd9a2ffdf6d15ff9eabfc454190ee497f0337e",
"size_bytes": 1142
},
{
"file": "labor_const_2026-01-01.json",
"dataset_id": "labor_const",
"effective_date": "2026-01-01",
"sha256": "4a7bf8ed334103a7bf166e4499314061813caacff1a12adea7d595a508ca7e14",
"size_bytes": 32828
},
{
"file": "labor_mfg_2026-07-01.json",
"dataset_id": "labor_mfg",
"effective_date": "2026-07-01",
"sha256": "cef66bec83c1a07592ff26dbd557f9f0c46e1f7dfd2919447f3bec103562c73b",
"size_bytes": 38189
},
{
"file": "mach_base_2026.json",
"dataset_id": "mach_base",
"effective_date": "2026-01-01",
"sha256": "b66e9f658da2ed4a02931b3a220cb81f7bce0ba845b92d7562ae2c27b62dda1f",
"size_bytes": 457054
},
{
"file": "mat_price_public_2026-08-14.json",
"dataset_id": "mat_price_public",
"effective_date": "2026-08-14",
"sha256": "75ddfe85b0140d782cd00187ec459e995cd0db9982e7bd48f62f83cd65248acc",
"size_bytes": 4748976
},
{
"file": "oil_2026-08-14.json",
"dataset_id": "oil",
"effective_date": "2026-08-14",
"sha256": "c2d9f95ff5d6c8a6624298c2f2d43b581a7e5bbc3be64a5af15c763b804e0a7a",
"size_bytes": 879
},
{
"file": "oil_regional_2026-09-09.json",
"dataset_id": "oil_regional",
"effective_date": "2026-09-09",
"sha256": "f9c1ce162b326b324604a2681a6078cccbc11624bbdbab6f0791e4368140b659",
"size_bytes": 5249
},
{
"file": "pum_const_2026.json",
"dataset_id": "pum_const",
"effective_date": "2026-01-01",
"sha256": "dbc46ef1316b83ed710f25bc402dcecc4a979bbfc4a6641a0554252bca326ca7",
"size_bytes": 3095263
},
{
"file": "pum_forest_2026.json",
"dataset_id": "pum_forest",
"effective_date": "2026-01-01",
"sha256": "111208fd482dcfc3b8c8dd58004b153382fc46555d0d00985c19b45ebbc2e0dd",
"size_bytes": 916135
},
{
"file": "rates_2026.json",
"dataset_id": "rates",
"effective_date": "2026-04-13",
"sha256": "ac0328df68c70b7c0643bf1b9b1ad69e6125bf2ae3cbfda9825519600fde5a87",
"size_bytes": 39623
}
],
"deferred": [
{
"variable": "mat_rebar",
"reason": "official shopping mall snapshot not locally stored"
},
{
"variable": "mat_remicon",
"reason": "official shopping mall snapshot not locally stored"
},
{
"source_group": "asphalt_concrete",
"reason": "official shopping mall snapshot not locally stored"
},
{
"variable": "pum_std_drawing",
"reason": "구조물 표준도 원단위(옹벽·날개벽·집수정 규격별 콘크리트·거푸집·철근량)를 담을 자리. 품셈·교본에 표가 없어 실무 구조물 일람표에서 뽑아야 하는데 그 자료가 아직 안 들어옴 — 자료 확보 대기",
"source_needed": "실무 구조물 일람표(옹벽·날개벽·집수정 규격별 수량표)",
"knowledge_ref": "resources/knowledge/technical_info/01_임도/05_원가정보/원가_입력변수_사전.md G6 — 「현행 공백 E1(실무 일람표 추출 필요)」"
}
],
"program_entry": [
{
"source_group": "paid_price_guides",
"responsibility": "user_or_program_input"
},
{
"source_group": "vendor_quotes",
"responsibility": "user_or_program_input"
}
],
"selection_pending": [
{
"variable": "coef_soil_L/coef_soil_C — 원문 중복 2행 채택",
"detail": "품셈 원문이 '역이 섞인 점질토'(값 상이 2행)·'암괴나 호박돌이 섞인 점토'(동일값 2행)를 중복 수록 — 원문 그대로 수록(duplicate_group·selection:pending). 프로그램 기본 채택행은 개발 단계 사용자 선정",
"candidates": [
"duplicate_index_1",
"duplicate_index_2"
]
},
{
"variable": "rate_environment",
"candidates": [
"road_0.9_percent",
"other_civil_0.8_percent"
]
}
]
}
File diff suppressed because it is too large Load Diff
@@ -1,45 +0,0 @@
{
"schema_version": "1.0",
"dataset_id": "fx",
"effective_date": "2026-08-14",
"generated_at": "2026-08-14T19:54:28+09:00",
"sources": [
{
"path": "resources/knowledge/original/원가계산/환율_한국은행ECOS/환율_일별_20260101_2026-08-14.csv",
"sha256": "36a9684af9d6b4a59c3d9c96beec2d75d1d8ad3f01b6af21f078f39fa1d5ffbd",
"role": "primary"
}
],
"variables": {
"fx_usd": {
"value": 1414.9,
"unit": "KRW/USD",
"date": "2026-08-14",
"source_item_code": "0000001"
},
"fx_jpy100": {
"value": 887.33,
"unit": "KRW/JPY100",
"date": "2026-08-14",
"source_item_code": "0000002"
},
"fx_eur": {
"value": 1631.87,
"unit": "KRW/EUR",
"date": "2026-08-14",
"source_item_code": "0000003"
},
"fx_gbp": {
"value": 1908.7,
"unit": "KRW/GBP",
"date": "2026-08-14",
"source_item_code": "0000012"
},
"fx_cny": {
"value": 209.75,
"unit": "KRW/CNY",
"date": "2026-08-14",
"source_item_code": "0000053"
}
}
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -1,31 +0,0 @@
{
"schema_version": "1.0",
"dataset_id": "oil",
"effective_date": "2026-08-14",
"generated_at": "2026-08-14T19:54:28+09:00",
"sources": [
{
"path": "resources/knowledge/original/원가계산/유가_오피넷/유가_전국평균_2026-08-14.json",
"sha256": "3726397c89108ca5d98b65c0bc615443d02c6fe84794bd95937dabe46464a747",
"role": "primary"
}
],
"variables": {
"oil_gasoline": {
"value": 1863.25,
"unit": "KRW/L",
"scope": "national_average",
"date": "2026-08-14",
"source_product_code": "B027",
"source_product_name": "휘발유"
},
"oil_diesel": {
"value": 1846.39,
"unit": "KRW/L",
"scope": "national_average",
"date": "2026-08-14",
"source_product_code": "D047",
"source_product_name": "자동차용경유"
}
}
}
@@ -1,204 +0,0 @@
{
"schema_version": "1.0",
"dataset_id": "oil_regional",
"effective_date": "2026-09-09",
"generated_at": "2026-09-09T21:11:38+09:00",
"note": "시도별 유가 — 품셈 8-1-7 5호 「유류가격은 해당지역의 가격으로 한다」. 코드 00(전국)은 원문 그대로 두되 지역 선택지에서는 전국평균 판이 맡는다. ⚠ 원천이 시도를 17줄로 주는데 그 가름이 행정구역과 다르다 — 코드 20 이 「전남광주」 한 줄이고 07(전남)·16(광주)은 아예 없다. 원문 이름을 그대로 두고 우리가 쪼개지 않는다.",
"sources": [
{
"path": "resources/knowledge/original/원가계산/유가_오피넷/유가_시도별_2026-09-09.json",
"sha256": "6f1b58101fbe57be402d44582763ca1d54aea8d9bb5ee9b85b38bdeea2d4e439",
"role": "primary"
}
],
"variables": {
"oil_gasoline": {
"unit": "KRW/L",
"scope": "regional",
"date": "2026-09-09",
"source_product_code": "B027",
"source_product_name": "휘발유",
"records": [
{
"sido_code": "00",
"sido_name": "전국",
"value": 1858.93
},
{
"sido_code": "01",
"sido_name": "서울",
"value": 1905.02
},
{
"sido_code": "02",
"sido_name": "경기",
"value": 1855.78
},
{
"sido_code": "03",
"sido_name": "강원",
"value": 1870.23
},
{
"sido_code": "04",
"sido_name": "충북",
"value": 1866.12
},
{
"sido_code": "05",
"sido_name": "충남",
"value": 1864.78
},
{
"sido_code": "06",
"sido_name": "전북",
"value": 1859.7
},
{
"sido_code": "08",
"sido_name": "경북",
"value": 1852.71
},
{
"sido_code": "09",
"sido_name": "경남",
"value": 1854.71
},
{
"sido_code": "10",
"sido_name": "부산",
"value": 1842.66
},
{
"sido_code": "11",
"sido_name": "제주",
"value": 1889.27
},
{
"sido_code": "14",
"sido_name": "대구",
"value": 1831.78
},
{
"sido_code": "15",
"sido_name": "인천",
"value": 1845.64
},
{
"sido_code": "17",
"sido_name": "대전",
"value": 1836.79
},
{
"sido_code": "18",
"sido_name": "울산",
"value": 1838.49
},
{
"sido_code": "19",
"sido_name": "세종",
"value": 1855.8
},
{
"sido_code": "20",
"sido_name": "전남광주",
"value": 1862.28
}
]
},
"oil_diesel": {
"unit": "KRW/L",
"scope": "regional",
"date": "2026-09-09",
"source_product_code": "D047",
"source_product_name": "자동차용경유",
"records": [
{
"sido_code": "00",
"sido_name": "전국",
"value": 1843.79
},
{
"sido_code": "01",
"sido_name": "서울",
"value": 1886.15
},
{
"sido_code": "02",
"sido_name": "경기",
"value": 1838.22
},
{
"sido_code": "03",
"sido_name": "강원",
"value": 1859.9
},
{
"sido_code": "04",
"sido_name": "충북",
"value": 1851.55
},
{
"sido_code": "05",
"sido_name": "충남",
"value": 1850.08
},
{
"sido_code": "06",
"sido_name": "전북",
"value": 1846.86
},
{
"sido_code": "08",
"sido_name": "경북",
"value": 1838.1
},
{
"sido_code": "09",
"sido_name": "경남",
"value": 1837.71
},
{
"sido_code": "10",
"sido_name": "부산",
"value": 1825.55
},
{
"sido_code": "11",
"sido_name": "제주",
"value": 1869.16
},
{
"sido_code": "14",
"sido_name": "대구",
"value": 1815.72
},
{
"sido_code": "15",
"sido_name": "인천",
"value": 1829.14
},
{
"sido_code": "17",
"sido_name": "대전",
"value": 1824.58
},
{
"sido_code": "18",
"sido_name": "울산",
"value": 1826.63
},
{
"sido_code": "19",
"sido_name": "세종",
"value": 1841.72
},
{
"sido_code": "20",
"sido_name": "전남광주",
"value": 1850.47
}
]
}
}
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff