chore(resources): 옛 생성 스크립트 2 master_data/old 로 옮김
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QqkvdyxUtANWroVKpQUyxE
This commit is contained in:
Claude Fable 5.1
@@ -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,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))
|
||||
Reference in New Issue
Block a user