Files
Aislo/resources/data_cost_input_value/_build_mach.py
T
eomsangdonandClaude Opus 5 1660ca48ff feat(tester): 건설 PDF ↔ md 전수 커버리지 검사 + 겸용 연료 둘 갈라 읽기
브레인 B08·B14.

- **PDF ↔ md 커버리지**(`test_const_pdf_coverage.py`) — PDF 글자층을 좌표로 줄로 다시 묶어
  값 수를 뽑고 md 전문과 양방향으로 맞댐. 지금 **결손 749**(PDF 에 있는데 md 에 없음) ·
  **허구 547**(md 에만 — 대부분 붙은 줄 잔재 `0008900.90` 꼴)
  · 쪽별 셈을 목록에 두어 **늘면 빨강**(새 구멍) · `--report` 로 구멍 큰 쪽 스물을 뽑음
  · ⚠ 줄 짝은 안 맞춤(PDF 는 칸마다 줄바꿈) · 값처럼 생긴 수만(소수점·세 자리 이상, 연도 뺌) ·
    차례 줄·쪽번호 줄은 뺌
- **겸용 연료 둘**(`_build_mach.py`) — 3450-0642 현장가열 표층재생기 `73.7+휘발유54.5` ·
  7992-0001 모르타르 믹서 `1.87㎾ 휘발유1.3`
  · 둘째 연료는 `secondary_fuel_*`, 전력은 `electric_power_kw`(연료가 아니라 곁값)
  · 연료 줄 274 → 276 · 대조 시험도 같은 꼴을 읽게 넓혀 **기계 못 보는 자리 38 → 27**
  · 이름표에 새 열 셋 + 줄 수 갱신
- 전체 시험 2,442 통과

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BW6Jdsh18WPtYUR6THZJqn
2026-09-18 14:35:24 +09:00

453 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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(" .·")
if name and not name[0].isdigit():
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))