eomsangdonandClaude Opus 5
fd6865c19a
**① 기종 이름 — `(8201) 3D GNSS 머신 가이던스(굴착기용)`** 추출기가 **첫 낱자가 숫자면 쪽번호로 보고 버리던** 탓에 8201 이름을 못 찾고 **옛 벌의 뭉개진 글**을 이어받고 있었음(`배관파이프시 간 당(10-7)분류번호규 격7995-0050…`). 낱자가 하나도 없는 토막만 버리게 고침 — 옛 벌에서 이어받은 이름 7 → **0** · 이름 바로잡힌 줄 1(값은 안 움직임). ⚠ `_build_mach.py` 는 랩탑 메인이 지은 것 — 한 줄 잣대만 고쳤음. **② 8-1-7 손료보정 — 원문이 제 표로 고쳐졌음** 전엔 글로 뭉개져 `덤프트럭2510` 처럼 붙어 있어 「25·10 인지 2·510 인지 못 가름」으로 세고 있었음. 이제 `| 덤프트럭 | 25 | 10 |` 이라 **칸에서 바로** 읽음 — 기종 글은 원문이 낱자 사이에 빈칸을 두므로 (`불도저( 1 9 톤이상 제외)`) 빈칸을 지워 맞댐. 네 줄이 한꺼번에 풀림(보정 셋 + 수 자리 가름). **③ 이름 없는 기종을 세는 자리** — 8-3 에 `(코드) 이름` 머리글이 없어 **양쪽이 다 비어 조용히 지나가는** 자리를 세게 함. 지금은 **0** — 8202~8204 도 머리글이 [주] 줄에 섞여 있어 둘 다 읽고 있음. ⇒ 기초단가 어긋남 23 → **17**(잣대 못 알아봄 16 → **10** · 원문 부재 7). 남은 잣대 10 은 다 랩탑 서브 PDF 몫 — 규격 빈칸 일곱(원문이 기종 이름을 여러 줄로 쪼갬) · 양수기 7993-0020 셋(규격이 주연료 칸에 밀림). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QsGw1Dz9HmhuAxGisxmDPF
773 lines
47 KiB
Python
773 lines
47 KiB
Python
"""원문 ↔ 기초단가 다섯 갈래(노임·기계·자재·유가·제비율) 맞대기 — 2026-09-18 랩탑 메인.
|
||
|
||
사람이 손으로 하던 값 대조를 시험으로 옮김. 원문 자리는 자료 `sources[].path` 를 그대로 씀.
|
||
|
||
- 어긋남·셈 틀어짐은 `fixtures/base_prices_source_known.json` 에 적어 두고 **그 목록 밖**이면 빨강.
|
||
⚠ 목록에 있는데 고쳐졌거나 모습이 바뀐 것도 빨강 — 목록이 낡지 않게.
|
||
고친 뒤 목록 지우기: `python resources/tester/test_base_prices_source_match.py --prune`
|
||
(지우기만 함 · 새 어긋남은 안 보탬).
|
||
- `*_blind` 갈래 = **시험이 못 보는 자리**. 못 보는 것을 본 척하지 않고 목록에 까닭과 함께 둠.
|
||
자료에 안 재는 칸이 새로 생겨도 목록 밖이라 빨강.
|
||
|
||
⚠ 빌더 코드를 안 씀 — 원문 읽기를 여기서 따로 함. 표 읽기(`md_tables`)만 옆 시험 한 벌.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import openpyxl
|
||
import pytest
|
||
from openpyxl.utils import get_column_letter
|
||
from test_work_item_master_source_match import ROOT, _without_why, md_tables
|
||
|
||
DATA = ROOT / "resources/data_cost_input_value"
|
||
KNOWN = Path(__file__).resolve().parent / "fixtures/base_prices_source_known.json"
|
||
FILES = {
|
||
"labor_const": "labor_const_2026-01-01.json",
|
||
"labor_mfg": "labor_mfg_2026-07-01.json",
|
||
"mach": "mach_base_2026.json",
|
||
"mat": "mat_price_public_2026-08-14.json",
|
||
"oil": "oil_2026-08-14.json",
|
||
"oil_regional": "oil_regional_2026-09-09.json",
|
||
"rates": "rates_2026.json",
|
||
}
|
||
KINDS = ["labor", "machine", "material", "oil", "rate"]
|
||
KIND_NAMES = {
|
||
"labor": "노임", "machine": "기계", "material": "자재", "oil": "유가", "rate": "제비율",
|
||
**{f"{k}_blind": f"{n} — 못 보는 자리" for k, n in
|
||
{"labor": "노임", "machine": "기계", "material": "자재", "oil": "유가", "rate": "제비율"}.items()},
|
||
} # fmt: skip
|
||
|
||
|
||
# ── 공통 ────────────────────────────────────────────────────────────────
|
||
def _num(text):
|
||
"""`12,000` · `0.9` · `(1.8)` · `16%` → 수. `-` · `*` · 빈 칸 → None. 수가 아니면 ValueError."""
|
||
t = str(text).strip().strip("()%").replace(",", "")
|
||
if t in ("", "-", "*", "None"):
|
||
return None
|
||
return float(t) if "." in t else int(t)
|
||
|
||
|
||
def _t(value) -> str:
|
||
return re.sub(r"\s+", "", str(value))
|
||
|
||
|
||
def _compare(out: dict, key: str, source: dict | None, data: dict | None, fields) -> None:
|
||
"""한 줄 맞대기 — 한쪽에만 있으면 줄 통째로, 둘 다 있으면 칸마다 `{source, data}`."""
|
||
if source is None or data is None:
|
||
if source != data:
|
||
out[key] = {"source": source, "data": data}
|
||
return
|
||
for f in fields:
|
||
if source.get(f) != data.get(f):
|
||
out[f"{key} {f}"] = {"source": source.get(f), "data": data.get(f)}
|
||
|
||
|
||
def _unchecked(blind: dict, where: str, records, checked) -> None:
|
||
"""자료에 있는데 원문과 안 맞대는 칸 — 못 본다고 적음."""
|
||
for field in sorted({k for r in records for k in r} - set(checked)):
|
||
blind[f"안 잰 칸 {where}.{field}"] = {"what": "원문과 안 맞댐"}
|
||
|
||
|
||
def _lines(path: Path) -> list[str]:
|
||
return path.read_text(encoding="utf-8").splitlines()
|
||
|
||
|
||
# ── 노임 ────────────────────────────────────────────────────────────────
|
||
def read_labor_const(path: Path) -> dict:
|
||
"""건설업 md — `| 직종코드 | 직종명 | 신뢰도 | 2026.1.1 | …` 표 한 벌 · 맨 앞 열이 적용일 노임."""
|
||
lines = _lines(path)
|
||
table = next(t for t in md_tables(lines) if t["headers"][:2] == ["직종코드", "직종명"])
|
||
head = "\n".join(lines[: table["line"]])
|
||
hours = int(re.search(r"1일 (\d+)시간 기준", head).group(1))
|
||
rows = {
|
||
code: {
|
||
"occupation_name": name,
|
||
"reliability": mark or None,
|
||
"hours_per_day": hours,
|
||
"daily_wage_krw": _num(wage),
|
||
"status": "not_published" if wage == "-" else "published",
|
||
}
|
||
for code, name, mark, wage, *_ in table["rows"]
|
||
}
|
||
y, m, d = table["headers"][3].split(".")
|
||
published, effective = re.search(r"공표 ([\d-]+), 적용 ([\d-]+)", head).groups()
|
||
dates = {"publication_date": published, "effective_date": effective}
|
||
return {"rows": rows, "wage_column": f"{y}-{int(m):02d}-{int(d):02d}", "dates": dates}
|
||
|
||
|
||
_JOB = re.compile(r"^(\d+)\.(.+)$")
|
||
|
||
|
||
def read_labor_mfg(path: Path) -> dict:
|
||
"""제조업 md — `업 종` 표들. 업종 칸은 빈 줄이 위 칸을 이어받음 · 하위 업종(`화학 공학`)은 `상위 / 하위`."""
|
||
lines = _lines(path)
|
||
rows, main, sub = {}, "", ""
|
||
for table in md_tables(lines):
|
||
if table["headers"][0] != "업 종":
|
||
continue
|
||
for r in table["rows"]:
|
||
i = next((k for k, c in enumerate(r) if _JOB.match(c)), None)
|
||
if i is None:
|
||
continue
|
||
if r[0]:
|
||
main, sub = r[0], ""
|
||
if i == 2 and r[1]:
|
||
sub = r[1]
|
||
code, name = _JOB.match(r[i]).groups()
|
||
wage = r[i + 1]
|
||
rows[code] = {
|
||
"occupation_name": name,
|
||
"industry": f"{main} / {sub}" if sub else main,
|
||
"daily_wage_krw": _num(wage.lstrip("*")),
|
||
"variation_coefficient": _num(r[i + 2]),
|
||
"status": "not_published" if wage == "*" else "published",
|
||
"source_flag": re.match(r"\**", wage).group() or None,
|
||
}
|
||
text = "\n".join(lines)
|
||
y, m, d = re.search(r"적용시점 : (\d{4})\. (\d+)\. (\d+)\.", text).groups()
|
||
sy, sm = re.search(r"조사 기 준 일 : (\d{4})\. (\d+)\.", text).groups()
|
||
return {
|
||
"rows": rows,
|
||
"dates": {"effective_date": f"{y}-{int(m):02d}-{int(d):02d}", "survey_month": f"{sy}-{int(sm):02d}"},
|
||
} # fmt: skip
|
||
|
||
|
||
LABOR = {
|
||
"labor_const": ("건설", "labor_rate", ("occupation_name", "reliability", "hours_per_day", "daily_wage_krw", "status")),
|
||
"labor_mfg": ("제조", "labor_mfg", ("occupation_name", "industry", "daily_wage_krw", "variation_coefficient", "status", "source_flag")),
|
||
} # fmt: skip
|
||
|
||
|
||
def check_labor(src: dict, bad: dict, blind: dict) -> dict:
|
||
counts = {}
|
||
for name, (label, var, fields) in LABOR.items():
|
||
source, data = src[name]["source"], src[name]["data"]
|
||
rows = source["rows"]
|
||
records = {r["occupation_code"]: r for r in data["variables"][var]["records"]}
|
||
for code in rows | records:
|
||
_compare(bad, f"{label} {code}", rows.get(code), records.get(code), fields)
|
||
_compare(bad, f"{label} 날짜", source["dates"], data, source["dates"])
|
||
if source.get("wage_column", data["effective_date"]) != data["effective_date"]:
|
||
bad[f"{label} 노임 열"] = {
|
||
"source": source["wage_column"],
|
||
"data": data["effective_date"],
|
||
}
|
||
for alias, code in data["variables"].get("aliases", {}).items():
|
||
if code not in rows:
|
||
bad[f"{label} 별칭 {alias}"] = {"source": None, "data": code}
|
||
counts[f"{label} 직종 수"] = (len(rows), len(records))
|
||
counts[f"{label} 원문에 없는 자료 줄"] = (0, len(set(records) - set(rows)))
|
||
blind[f"{label} pdf"] = {"what": src[name]["paths"][0].name}
|
||
_unchecked(blind, var, records.values(), {"occupation_code", *fields})
|
||
blind["건설 별칭 이름"] = {"what": sorted(src["labor_const"]["data"]["variables"]["aliases"])}
|
||
return counts
|
||
|
||
|
||
# ── 기계 ────────────────────────────────────────────────────────────────
|
||
def _codes(cell: str, prefix: str | None) -> tuple[list[str] | None, str | None]:
|
||
"""`0101-0007 0010 0012` → 코드 목록 · 머리(`0101`)는 앞 코드를 이어받음(앞 표에서도).
|
||
|
||
`9020- 0010` 처럼 머리만 떨어진 토막도 받음. 코드 아닌 토막이 끼면 `(None, None)`.
|
||
"""
|
||
codes = []
|
||
for tok in cell.split():
|
||
m = re.fullmatch(r"-?(?:(\d{4})-)?(\d{4})?", tok)
|
||
prefix = (m.group(1) or prefix) if m and tok.strip("-") else None
|
||
if prefix is None:
|
||
return None, None
|
||
if m.group(2):
|
||
codes.append(f"{prefix}-{m.group(2)}")
|
||
return codes, prefix
|
||
|
||
|
||
LOSS_COLS = {"economic_life_hours": "내용시간", "annual_standard_hours": "가동시간", "depreciation_ratio": "상각비율", "maintenance_ratio": "정비비율", "annual_management_ratio": "관리비율"} # fmt: skip
|
||
LOSS_COEF = ("depreciation_coefficient_1e_minus_7", "maintenance_coefficient_1e_minus_7", "management_coefficient_1e_minus_7", "source_coefficient_1e_minus_7") # fmt: skip
|
||
FUEL_FIELDS = ("fuel_type", "fuel_rate_l_per_hour", "misc_material_percent_of_fuel", "operator_person_per_day", "secondary_fuel_type", "secondary_fuel_rate_l_per_hour", "electric_power_kw") # fmt: skip
|
||
|
||
|
||
#: 연료 토막 — `9.0` 경유([주]① ㉰) · `휘발유0.7` · `중유487.2` · 겸용 `73.7+휘발유54.5` ·
|
||
#: 전력과 겸용 `1.87㎾ 휘발유1.3`. ⚠ 꼴이 다르면 ValueError — 지어 읽지 않는다.
|
||
_FUEL_TOKEN_RE = re.compile(r"(휘발유|중유|경유)?(\d+(?:\.\d+)?)(㎾)?")
|
||
|
||
|
||
def _fuel(tok: str) -> tuple:
|
||
"""`(갈래, ℓ/hr, 겸용 갈래, 겸용 ℓ/hr, 전력㎾)`. 값이 없으면 자리마다 `None`."""
|
||
if tok == "-":
|
||
return None, None, None, None, None
|
||
rest, fuels, power = tok, [], None
|
||
for m in _FUEL_TOKEN_RE.finditer(tok):
|
||
rest = rest.replace(m.group(0), "", 1)
|
||
if m.group(3): # 전력은 ℓ/hr 가 아니다 — 곁값
|
||
power = float(m.group(2))
|
||
continue
|
||
kind = {"휘발유": "gasoline", "중유": "heavy_oil", "경유": "diesel", None: "diesel"}[m.group(1)] # fmt: skip
|
||
fuels.append((kind, float(m.group(2))))
|
||
if rest.replace("+", "").strip() or not (fuels or power):
|
||
raise ValueError(tok)
|
||
first = fuels[0] if fuels else (None, None)
|
||
second = fuels[1] if len(fuels) > 1 else (None, None)
|
||
return first[0], first[1], second[0], second[1], power
|
||
|
||
|
||
def _cells(cell: str, n: int) -> list[str]:
|
||
"""한 칸을 토막 냄 — **빈 칸은 「없음」** 이고 `동력` 낱말은 뗀다.
|
||
|
||
원문 8-4 는 없음을 「-」 로도 쓰고 빈 칸으로도 두며, 전력 기종은 `동력 15.1㎾` 처럼
|
||
낱말을 붙인다. 빈 칸을 토막 없음으로 두면 줄을 통째로 못 읽어 **원문에 있는 값까지
|
||
놓친다**(7360-0090 차선 제거기 — 휘발유 5.53 이 원문에 적혀 있음).
|
||
"""
|
||
toks = cell.replace("동력", " ").split()
|
||
return toks or ["-"] * n
|
||
|
||
|
||
def read_machine(path: Path) -> dict:
|
||
"""제8장 md — 8-3 손료 · 8-4 운전경비 · 8-5 가격. 여러 코드가 한 칸에 뭉친 줄은 토막 수가 맞을 때만 읽음."""
|
||
lines = _lines(path)
|
||
tables = md_tables(lines)
|
||
at = {w: next(i for i, x in enumerate(lines, 1) if w in x) for w in ("8-3 기계손료", "8-4 운전경비", "8-4-9 [90]", "8-5 기계가격")} # fmt: skip
|
||
out = {k: {} for k in ("price", "loss", "fuel", "fuel_tables", "skipped", "names")}
|
||
out["spec_blank"] = set() # 원문이 규격 칸을 비워 둔 줄 — 자료가 채웠는지는 맞대기에서 봄
|
||
out["lines"] = lines
|
||
name_at = re.compile(r"\((\d{4})\) ?(.+?)(?=시 간 당\(10-7\)|\((?:'\d\d|\d{4}\))|$)")
|
||
for x in lines[
|
||
at["8-3 기계손료"] - 1 : at["8-4 운전경비"]
|
||
]: # 손료 표 머리 `(0101) 불도저(무한궤도)`
|
||
for m in name_at.finditer("" if x.lstrip().startswith("|") else x):
|
||
out["names"].setdefault(m.group(1), m.group(2).strip())
|
||
prefix = None
|
||
for t in tables:
|
||
if t["line"] > at["8-5 기계가격"] and t["headers"] == ["기 종", "분류번호", "가격(￦)"]:
|
||
for n, r in enumerate(t["rows"]):
|
||
codes, prefix = _codes(r[1], prefix)
|
||
prices = r[2].split()
|
||
if codes is None or len(codes) != len(prices):
|
||
out["skipped"][f"가격 줄 {_t(r[1])[:24]}"] = {"what": r[1][:40]}
|
||
continue
|
||
out["price"].update(
|
||
(c, {"price_thousand_krw": _num(p)}) for c, p in zip(codes, prices)
|
||
)
|
||
prefix = None
|
||
for t in tables:
|
||
head = [_t(h) for h in t["headers"]]
|
||
if not (at["8-3 기계손료"] < t["line"] < at["8-4 운전경비"] and head[0] == "분류번호"):
|
||
continue
|
||
cols = {f: next(i for i, h in enumerate(head) if w in h) for f, w in LOSS_COLS.items() if any(w in h for h in head)} # fmt: skip
|
||
sub = next((r for r in t["rows"] if not r[0]), [])
|
||
if "상각비 계수" in sub:
|
||
cols.update(zip(LOSS_COEF, range(sub.index("상각비 계수"), 99)))
|
||
elif "시간당(10-7)" in head:
|
||
cols[LOSS_COEF[3]] = head.index("시간당(10-7)")
|
||
else: # 원문 표에 계수 칸이 없음(5220·7995 — 머리 「시 간 당(10-7)」 은 표 위 글줄로만 남음)
|
||
out["skipped"][f"손료 계수 칸 없는 표 {t['rows'][0][0][:4] if t['rows'] else '?'}"] = {"what": t["headers"]} # fmt: skip
|
||
continue
|
||
for n, r in enumerate(t["rows"]):
|
||
if not r[0]:
|
||
continue
|
||
codes, prefix = _codes(r[0], prefix)
|
||
# `| 9020- | 비항SD | | … |` 꼴 — **앞자리만 알리는 머리줄**이라 값이 애초에 없다.
|
||
# 못 읽은 자리가 아니므로 「못 보는 자리」로 세지 않고 앞자리만 이어받고 지난다.
|
||
if codes == [] and not any(re.search(r"\d", c) for c in r[1:]):
|
||
continue
|
||
try:
|
||
values = {f: [_num(x) for x in r[i].split()] for f, i in cols.items()}
|
||
if not codes or any(len(v) != len(codes) for v in values.values()):
|
||
raise ValueError(r[0])
|
||
except ValueError:
|
||
out["skipped"][f"손료 줄 {_t(r[0])[:24]}"] = {"what": r[0][:40]}
|
||
continue
|
||
out["loss"].update(
|
||
(c, {f: v[k] for f, v in values.items()}) for k, c in enumerate(codes)
|
||
)
|
||
# 규격 — 원문 손료표의 「규 격」 칸(분류번호 바로 뒤). 자료는 이 값을 가격·연료 벌에도 함께 씀.
|
||
# 토막이 코드 수와 안 맞는 줄(`2,200×5,150 ×1,000` 처럼 규격 안에 빈칸이 든 것)은 안 담는다 —
|
||
# 맞대기에서 자료 값으로 갈음되므로 지어 채우는 일이 없다.
|
||
specs = r[1].split()
|
||
if len(codes) == 1 and specs: # 한 코드 줄은 규격 안에 빈칸이 들어도 칸을 통째로 씀
|
||
specs = [" ".join(specs)]
|
||
if len(specs) == len(codes):
|
||
for k, c in enumerate(codes):
|
||
out["loss"][c]["specification"] = specs[k]
|
||
elif len(codes) == 1 and not specs:
|
||
# 원문이 그 칸을 **비워 둔** 줄 — 표에서 「위와 같음」으로 읽히는 자리다.
|
||
# 자료가 앞 줄 값을 이어받아 채운 줄만 못 보는 자리로 센다(맞대기에서 판다).
|
||
out["spec_blank"].add(codes[0])
|
||
for k in range(at["8-3 기계손료"], at["8-4 운전경비"]):
|
||
# 머리 없는 **이어 표** — 책장 넘김에 밀린 줄을 빈 줄 뒤에 따로 적어 둔 자리(9020-0022~0200).
|
||
# `|---|` 가 없어 표로 잡히지 않아 **추출기도 이 읽기도 못 봄** — 원문에 있는 값이 조용히 빠진다.
|
||
cells = [c.strip() for c in lines[k - 1].strip().strip("|").split("|")]
|
||
if len(cells) >= 10 and re.fullmatch(r"\d{4}-\d{4}", cells[0] or ""):
|
||
# 칸이 열 이상인 것만 — 축약 손료표(5220·7995)는 「계수 칸 없는 표」 로 이미 세고 있다.
|
||
if cells[0] not in out["loss"]:
|
||
out["skipped"][f"손료 머리 없는 이어 표 {cells[0]}"] = {"what": lines[k - 1][:48]}
|
||
if not lines[k - 1].lstrip().startswith("|") and "시 간 당(10-7)" in lines[k - 1]:
|
||
heads = re.findall(r"\((\d{4})\) ?(?:[^|()]{0,30}\([^)]*\)){0,2}[^|()]{0,30}?시 간 당\(10-7\)", lines[k - 1]) # fmt: skip
|
||
if heads: # 표 위 머리 글줄(`시 간 당(10-7)` 한 줄)은 표로 읽힘 — 뭉개진 글만
|
||
out["skipped"][f"손료 뭉개진 글 {'·'.join(heads)}"] = {"what": heads}
|
||
prefix = None
|
||
for t in tables:
|
||
if not at["8-4 운전경비"] < t["line"] < at["8-5 기계가격"]:
|
||
continue
|
||
out["fuel_tables"][len(out["fuel_tables"])] = {"headers": t["headers"], "rows": t["rows"]}
|
||
if t["line"] > at["8-4-9 [90]"] or t["headers"][0] != "분류번호":
|
||
continue
|
||
for r in t["rows"]:
|
||
codes, prefix = _codes(r[0], prefix)
|
||
# 서브의 칸 토막내기(빈 칸 = 없음 · `동력` 뗌)를 쓰되, ⚠ **코드가 하나면 칸 통째로** —
|
||
# `1.87㎾ 휘발유1.3` 처럼 한 값 안에 공백이 온다(2026-09-18 겸용 연료).
|
||
cols = [
|
||
[r[i].replace("동력", " ").strip() or "-"]
|
||
if len(codes or []) == 1
|
||
else _cells(r[i], len(codes or []))
|
||
for i in (3, 4, 5)
|
||
]
|
||
try:
|
||
if not codes or any(len(c) != len(codes) for c in cols):
|
||
raise ValueError(r[0])
|
||
parsed = [
|
||
(ft[0], ft[1], _num(m), _num(o), ft[2], ft[3], ft[4])
|
||
for f, m, o in zip(*cols)
|
||
for ft in (_fuel(f),)
|
||
]
|
||
except ValueError:
|
||
out["skipped"][f"연료 표 {codes[0] if codes else _t(r[0])[:24]}"] = {
|
||
"codes": codes and [codes[0], codes[-1]]
|
||
}
|
||
out["fuel_unread"] = out.get("fuel_unread", set()) | set(codes or ())
|
||
continue
|
||
out["fuel"].update((c, dict(zip(FUEL_FIELDS, p))) for c, p in zip(codes, parsed) if p[1] is not None) # fmt: skip
|
||
return out
|
||
|
||
|
||
#: 8-1-7 손료보정 표의 기종 글 — 원문이 낱자 사이에 빈칸을 두므로(`불도저( 1 9 톤이상 제외)`) **빈칸을 지워** 맞댄다.
|
||
#: 2026-09-18 md 가 제 표로 고쳐져 두 수(암석작업·전석섞인토사)를 **칸에서 바로** 읽는다(전엔 `덤프트럭2510` 처럼 글로 뭉개져 있었다).
|
||
ROCK = {"bulldozer_under_19_ton": "불도저(19톤이상제외)", "crawler_excavator_or_loader": "굴착기(무한궤도)및로더(무한궤도)", "dump_truck": "덤프트럭"} # fmt: skip
|
||
|
||
|
||
def check_machine(src: dict, bad: dict, blind: dict) -> dict:
|
||
source, v = src["mach"]["source"], src["mach"]["data"]["variables"]
|
||
counts = {}
|
||
for label, var, rows, fields in (
|
||
("가격", "mach_price", source["price"], ("price_thousand_krw", "specification")),
|
||
("손료", "mach_loss_coef", source["loss"], (*LOSS_COLS, *LOSS_COEF, "specification")),
|
||
("연료", "mach_fuel_rate", source["fuel"], (*FUEL_FIELDS, "specification")),
|
||
):
|
||
records = v[var].get("records") or v[var]["parsed_records"]
|
||
by_code = {r["machine_code"]: r for r in records}
|
||
unread = source.get("fuel_unread", set()) if label == "연료" else set()
|
||
missing: dict[str, list] = {}
|
||
for code in rows | by_code:
|
||
if code in unread and code not in rows:
|
||
blind[f"연료 못 읽은 줄 {code}"] = {"what": "원문 표 토막 수가 안 맞음"}
|
||
continue
|
||
if code not in by_code:
|
||
missing.setdefault(code[:4], []).append(code)
|
||
continue
|
||
if code in rows: # 원문에 칸이 없는 표(축약 손료표)는 있는 칸만 맞댐
|
||
rows_code = {f: rows[code].get(f, by_code[code].get(f)) for f in fields}
|
||
_compare(bad, f"{label} {code}", rows_code, by_code[code], fields)
|
||
else:
|
||
_compare(bad, f"{label} {code}", None, by_code[code], fields)
|
||
heading, name = source["names"].get(code[:4]), by_code[code].get("machine_name")
|
||
if _t(heading) != _t(name):
|
||
bad[f"{label} {code} machine_name"] = {"source": heading, "data": name}
|
||
for head, codes in missing.items():
|
||
bad[f"{label} {head} 자료에 없음"] = {"source": codes, "data": None}
|
||
counts[f"{label} 코드 수"] = (len(rows), len(by_code))
|
||
counts[f"{label} 원문에 없는 자료 줄"] = (0, len(set(by_code) - set(rows) - unread))
|
||
_unchecked(blind, var, records, {"machine_code", "machine_name", *fields})
|
||
# 원문 8-3 에 `(코드) 이름` 머리글이 아예 없어 **이름이 빈** 기종 — 양쪽이 다 비어 조용히 지나간다.
|
||
# 표는 있는데 이름만 없는 자리라 세어 둔다(md 에 머리글이 붙으면 저절로 빠짐).
|
||
for code, rec in sorted({r["machine_code"]: r for r in v["mach_loss_coef"]["records"]}.items()):
|
||
if not rec.get("machine_name") and source["names"].get(code[:4]) is None:
|
||
blind[f"기종 이름 없는 머리글 {code[:4]}"] = {"what": code}
|
||
# 원문이 규격 칸을 비워 둔 자리에 **자료가 앞 줄 값을 이어받아 채운** 줄 — 원문에 없는 글이라 사람이 볼 자리.
|
||
loss_by_code = {r["machine_code"]: r for r in v["mach_loss_coef"]["records"]}
|
||
for code in sorted(source.get("spec_blank", ())):
|
||
spec = (loss_by_code.get(code) or {}).get("specification")
|
||
if spec is not None:
|
||
blind[f"규격 빈칸 {code}"] = {"what": spec}
|
||
blind.update(source["skipped"])
|
||
tables = v["mach_fuel_rate"]["source_tables"]
|
||
for n, st in enumerate(
|
||
tables
|
||
): # ⚠ 줄 번호로 안 맞댐 — md 가 고쳐지면 통째로 밀림. 8-4 안 표 차례로
|
||
t = source["fuel_tables"].get(n)
|
||
if t != {"headers": st["headers"], "rows": st["rows"]}:
|
||
bad[f"운전경비 원표 {n + 1}번째"] = {
|
||
"source": t,
|
||
"data": {k: st[k] for k in ("headers", "rows")},
|
||
}
|
||
counts["운전경비 원표 수"] = (len(source["fuel_tables"]), len(tables))
|
||
_unchecked(blind, "mach_fuel_rate.source_tables", tables, {"line", "headers", "rows"})
|
||
|
||
op = next(t for t in md_tables(source["lines"]) if t["headers"] == ["구 분", "해 당 기 계"])
|
||
labels = [_t(r[0]) for r in op["rows"]]
|
||
names = src["labor_const"]["source"]["rows"]
|
||
for rule in v["mach_operator_map"]["rules"]:
|
||
name = names.get(rule["occupation_code"], {}).get("occupation_name")
|
||
if name not in labels:
|
||
bad[f"운전사 {rule['rule_id']}"] = {"source": labels, "data": [rule["occupation_code"], name]} # fmt: skip
|
||
blind["운전사 기종별 판단"] = {"what": len(v["mach_operator_map"]["explicit_mappings"])}
|
||
_unchecked(
|
||
blind, "mach_operator_map.rules", v["mach_operator_map"]["rules"], {"occupation_code"}
|
||
)
|
||
|
||
text = _t("".join(source["lines"]))
|
||
rock_rows = {} # 기종 글(빈칸 지움) → (암석작업, 전석섞인토사)
|
||
for t in md_tables(source["lines"]):
|
||
if [_t(h) for h in t["headers"]][:2] != ["기종", "가산비율"]:
|
||
continue
|
||
for r in t["rows"]:
|
||
try:
|
||
rock_rows[_t(r[0])] = (_num(r[1]), _num(r[2]))
|
||
except ValueError: # 머리 곁줄(`암석작업(연암…)` · `전석섞인토사`)
|
||
continue
|
||
for rule in v["mach_rock_adj"]["rules"]:
|
||
group = rule["machine_group"]
|
||
pair = (rule["rock_work"], rule["boulder_mixed_soil"])
|
||
if group in ROCK:
|
||
got = rock_rows.get(ROCK[group])
|
||
else: # 「불도저(19톤이상)은 보정하지 않는다」 — 표에 줄이 없고 [주]①이 0 을 말한다
|
||
got = (0, 0) if "불도저(19톤이상)의경우는보정하지않는다" in text else None
|
||
if got != pair:
|
||
bad[f"손료보정 {group}"] = {"source": got, "data": list(pair)}
|
||
_unchecked(blind, "mach_rock_adj.rules", v["mach_rock_adj"]["rules"], {"machine_group", "rock_work", "boulder_mixed_soil"}) # fmt: skip
|
||
return counts
|
||
|
||
|
||
# ── 자재 ────────────────────────────────────────────────────────────────
|
||
MAT_FIELDS = {
|
||
"classification_code": "prdctClsfcNo", "classification_name": "prdctClsfcNoNm",
|
||
"specification": "krnPrdctNm", "unit": "unit", "price_krw": "prce", "notice_datetime": "nticeDt",
|
||
"notice_number": "prceNticeNo", "business_division_code": "bsnsDivCd",
|
||
"business_division_name": "bsnsDivNm", "vat_basis": "vatYnNm", "price_type": "prceDiv",
|
||
"delivery_condition": "dlvryCndtnNm", "field": "분야",
|
||
} # fmt: skip
|
||
|
||
|
||
def read_material(path: Path) -> dict:
|
||
"""나라장터 API 스냅숏 — 물품식별번호마다 **값이 실린 공시 중 가장 나중** 것(같은 때가 둘이면 둘 다).
|
||
|
||
값 없는 공시(`0`·빈 칸)를 건너뜀은 자료 `selection_policy` 가 밝힌 규칙.
|
||
"""
|
||
raw = json.loads(path.read_text(encoding="utf-8"))
|
||
latest: dict[str, list[dict]] = {}
|
||
for row in raw:
|
||
if not _num(row["prce"]):
|
||
continue
|
||
mapped = {k: (_num(row[f]) if k == "price_krw" else row[f]) for k, f in MAT_FIELDS.items()}
|
||
best = latest.setdefault(row["prdctIdntNo"], [])
|
||
if best and best[0]["notice_datetime"] > mapped["notice_datetime"]:
|
||
continue
|
||
if best and best[0]["notice_datetime"] < mapped["notice_datetime"]:
|
||
best.clear()
|
||
if mapped not in best:
|
||
best.append(mapped)
|
||
return {"rows": len(raw), "ids": len({r["prdctIdntNo"] for r in raw}), "latest": latest}
|
||
|
||
|
||
def check_material(src: dict, bad: dict, blind: dict) -> dict:
|
||
source, data = src["mat"]["source"], src["mat"]["data"]
|
||
records = {r["item_code"]: r for r in data["variables"]["mat_price"]["records"]}
|
||
latest = source["latest"]
|
||
for code in latest | records:
|
||
cands, rec = latest.get(code, []), records.get(code)
|
||
if len(cands) > 1:
|
||
if rec and any(all(rec.get(k) == c[k] for k in MAT_FIELDS) for c in cands):
|
||
blind[f"같은 때 공시 {code}"] = {"what": len(cands)}
|
||
else:
|
||
bad[code] = {"source": cands, "data": rec}
|
||
continue
|
||
_compare(bad, code, cands[0] if cands else None, rec, MAT_FIELDS)
|
||
blind["API 스냅숏"] = {"what": src["mat"]["paths"][0].name}
|
||
blind["원천 없는 묶음"] = {"what": [g["group"] for g in data.get("excluded_named_groups", [])]}
|
||
_unchecked(blind, "mat_price", records.values(), {"item_code", *MAT_FIELDS})
|
||
return {
|
||
"원천 줄 수": (source["rows"], data["source_row_count"]),
|
||
"물품코드 수": (source["ids"], len(records)),
|
||
"원문에 없는 자료 줄": (0, len(set(records) - set(latest))),
|
||
}
|
||
|
||
|
||
# ── 유가 ────────────────────────────────────────────────────────────────
|
||
def check_oil(src: dict, bad: dict, blind: dict) -> dict:
|
||
counts = {}
|
||
raw, data = src["oil"]["source"], src["oil"]["data"]
|
||
prices = {r["PRODCD"]: r for r in raw["전국평균"]}
|
||
for var, v in data["variables"].items():
|
||
row = prices.get(v["source_product_code"])
|
||
d = row and row["TRADE_DT"]
|
||
source = row and {"value": _num(row["PRICE"]), "source_product_name": row["PRODNM"], "date": f"{d[:4]}-{d[4:6]}-{d[6:]}"} # fmt: skip
|
||
_compare(bad, f"전국 {var}", source, v, ("value", "source_product_name", "date"))
|
||
_compare(bad, "전국 적용일", {"effective_date": raw["수집일"]}, data, ("effective_date",))
|
||
_unchecked(blind, "oil", data["variables"].values(), {"value", "source_product_name", "date", "source_product_code"}) # fmt: skip
|
||
|
||
raw, data = src["oil_regional"]["source"], src["oil_regional"]["data"]
|
||
for var, v in data["variables"].items():
|
||
rows = {r["SIDOCD"]: {"sido_name": r["SIDONM"], "value": r["PRICE"]} for r in raw["시도별"].get(v["source_product_code"], [])} # fmt: skip
|
||
records = {r["sido_code"]: r for r in v["records"]}
|
||
for code in rows | records:
|
||
_compare(bad, f"시도 {var} {code}", rows.get(code), records.get(code), ("sido_name", "value")) # fmt: skip
|
||
_compare(bad, f"시도 {var} 날짜", {"date": raw["수집일"]}, v, ("date",))
|
||
counts[f"시도 {var} 줄 수"] = (len(rows), len(records))
|
||
_unchecked(blind, f"oil_regional.{var}.records", v["records"], {"sido_code", "sido_name", "value"}) # fmt: skip
|
||
_unchecked(blind, "oil_regional", data["variables"].values(), {"records", "date", "source_product_code"}) # fmt: skip
|
||
blind["API 스냅숏"] = {"what": [src[n]["paths"][0].name for n in ("oil", "oil_regional")]}
|
||
blind["시도 거래일"] = {"what": "시도별 원천에 거래일 칸이 없음 — 수집일과만 맞댐"}
|
||
return counts
|
||
|
||
|
||
# ── 제비율 ──────────────────────────────────────────────────────────────
|
||
def read_xlsx(path: Path) -> dict:
|
||
"""첫 시트 칸 — `raw` 는 적힌 칸만, `filled` 는 병합 칸 전체에 왼쪽 위 값을 퍼뜨림."""
|
||
ws = openpyxl.load_workbook(path, data_only=True).worksheets[0]
|
||
raw = {c.coordinate: c.value for row in ws.iter_rows() for c in row if c.value not in (None, "")} # fmt: skip
|
||
filled = dict(raw)
|
||
for m in ws.merged_cells.ranges:
|
||
top = ws.cell(m.min_row, m.min_col).value
|
||
for row in range(m.min_row, m.max_row + 1):
|
||
for col in range(m.min_col, m.max_col + 1):
|
||
if top not in (None, ""):
|
||
filled[f"{get_column_letter(col)}{row}"] = top
|
||
return {"raw": raw, "filled": filled}
|
||
|
||
|
||
def read_pension(path: Path) -> dict:
|
||
"""국민연금법 — 부칙 제4조① 해마다 사용자 부담률(1만분의) · 제88조③ 그 뒤 부담률(1천분의)."""
|
||
text = path.read_text(encoding="utf-8")
|
||
part = re.search(r"제4조\(연금보험료에 관한 특례\) ①(.*?)②", text, re.S).group(1)
|
||
years = {int(y): int(n) / 100 for y, n in re.findall(r"(\d{4})년은 1만분의 (\d+)", part)}
|
||
after = re.search(r"사용자가 각각 부담하되, 그 금액은 각각 기준소득월액의 1천분의 (\d+)", text)
|
||
return {"years": years, "after": int(after.group(1)) / 10}
|
||
|
||
|
||
BASE = {"노": "total_labor_cost", "직노": "direct_labor_cost", "건강보험료": "health_insurance_amount", "재+노": "material_cost_plus_total_labor_cost", "재+노+경": "material_plus_labor_plus_expense", "노+경+일": "labor_plus_expense_plus_overhead", "직접공사비": "direct_construction_cost", "재+직노": "material_cost_plus_direct_labor_cost"} # fmt: skip
|
||
BASE_CELLS = {"rate_goyong.base": "B63", "rate_indirect_labor.base": "AC7", "rate_other_expense.base": "AL7", "rate_overhead.base": "BD7", "rate_profit.base": "BS7", "rate_equipment_payment_guarantee.base": "BZ7", "rate_subcontract_payment_guarantee.base": "BZ63", "rate_environment.base": "AU63", "rate_performance_guarantee_fee.base": "B95", "rate_safety_pct.base_without_owner_supplied_material": "EB7"} # fmt: skip
|
||
SINGLE = {"rate_sanjae": "BZ51", "rate_health": "AU51", "rate_care": "BD51", "rate_retirement_mutual_aid": "AU110", "rate_wage_claim_contribution": "CS99", "rate_asbestos_contribution": "BZ99", "rate_pension": "BR51"} # fmt: skip
|
||
SCALE = {"10억미만": "lt_1_billion", "10억-50억미만": "1_to_5_billion", "50억-300억미만": "5_to_30_billion", "300억-1000억미만": "30_to_100_billion", "1000억이상": "gte_100_billion"} # fmt: skip
|
||
TERM = {"6개월이하(183일)": "lte_183_days", "7~12개월(365일)": "184_to_365_days", "13~36개월(1095일)": "366_to_1095_days", "36개월초과(1096일)": "gte_1096_days"} # fmt: skip
|
||
WORK_TYPE = {"토목": "civil", "조경": "landscape", "산업설비(토목)": "industrial_facilities_civil"}
|
||
SAFETY_SCALE = {"5억미만": "lt_500_million", "5억~50억미만": "500_million_to_5_billion", "50억이상": "gte_5_billion", "추정금액800억미만건설공사(주공종이토목인경우1,000억)": "gte_5_billion_below_manager_threshold", "추정금액800억이상건설공사(주공종이토목인경우1,000억)": "gte_5_billion_at_or_above_manager_threshold"} # fmt: skip
|
||
SAFETY_WORK = {"건축공사": "building", "토목공사": "civil", "중건설공사": "heavy_construction", "특수건설공사": "special_construction"} # fmt: skip
|
||
CIVIL_PRICE = {"5억미만": "lt_5_billion", "5억-30억미만": "lt_5_billion", "30억-50억미만": "lt_5_billion", "50억-100억미만": "5_to_30_billion", "100억-300억미만": "5_to_30_billion", "300억-1000억미만": "30_to_100_billion", "1000억이상": "gte_100_billion"} # fmt: skip
|
||
SPECIAL_PRICE = {"5억미만": "lt_500_million", "5억-30억미만": "500_million_to_3_billion", "30억-50억미만": "3_to_10_billion", "50억-100억미만": "3_to_10_billion", "100억-300억미만": "10_to_30_billion", "300억-1000억미만": "30_to_100_billion", "1000억이상": "gte_100_billion"} # fmt: skip
|
||
#: (라벨 열, 첫 줄, 끝 줄, 값 열, 라벨→자료 구간, 자료 열쇠 틀)
|
||
BANDS = [
|
||
("B", 67, 82, "AH", {"[1등급]1400억이상": "1|gte_140_billion", "[2등급]900억~1400억미만": "2|90_to_140_billion", "[3등급]570억~900억미만": "3|57_to_90_billion", "[4등급]370억~570억미만": "4|37_to_57_billion", "[5등급]220억~370억미만": "5|22_to_37_billion", "[6등급]140억~220억미만": "6|14_to_22_billion", "[7등급]고시금액~140억미만": "7|official_threshold_to_14_billion", "7등급미만": "below_7|below_official_threshold"}, "rate_goyong.brackets[{}].rate_percent"),
|
||
("AZ", 68, 94, "BU", {"도로(예시:교량,터널,활주로등)": "civil_road", "플랜트(예시:발전소,쓰레기소각장등)": "civil_plant", "지하철": "civil_subway", "철도": "civil_railway", "상하수도(예시:폐수,하수처리장,정수장등)": "civil_water_and_sewer", "항만": "civil_port", "(오탁방지막또는준설토방지막을설치하는경우)": "civil_port_with_silt_screen", "댐": "civil_dam", "택지개발": "civil_land_development", "그밖의토목공사(하천등)": "civil_other", "조경": "landscape", "주택(재개발및재건축)": "building_housing_redevelopment", "주택(신축)": "building_new_housing", "그밖의건축공사": "building_other"}, "rate_environment.all_work_types[{}].rate_percent"),
|
||
("BZ", 68, 83, "DC", {"50억미만": "lt_5_billion", "50억이상-100억미만": "5_to_10_billion", "100억이상-300억미만": "10_to_30_billion", "300억이상종심∙종평제(토목및산업설비)": "gte_30_billion_integrated_civil_or_industrial", "300억이상종심∙종평제(건축)": "gte_30_billion_integrated_building", "턴키∙대안공사": "turnkey_or_alternative"}, "rate_subcontract_payment_guarantee.brackets[{}].rate_percent"),
|
||
("BZ", 19, 30, "CI", {"토목공사(토건)": "civil_general", "산업∙설비공사": "industrial_facilities", "조경공사": "landscape"}, "rate_equipment_payment_guarantee.general_construction[{}].rate_percent"),
|
||
("CN", 19, 38, "DE", {"준설공사,포장공사": "dredging_or_paving_or_earthwork_or_scaffolding_dismantling", "토공사,비계구조물해제공사": "dredging_or_paving_or_earthwork_or_scaffolding_dismantling", "상하수도설비공사,수중공사": "water_sewer_or_underwater_or_boring_grouting", "보링그라우팅공사": "water_sewer_or_underwater_or_boring_grouting", "석공사,철근콘크리트공사": "stone_or_reinforced_concrete", "조경시설물설치,조경식재공사": "landscape_facility_or_planting_or_painting_or_rail_track_or_steel_installation", "도장공사,철도궤도공사": "landscape_facility_or_planting_or_painting_or_rail_track_or_steel_installation", "철강재설치공사": "landscape_facility_or_planting_or_painting_or_rail_track_or_steel_installation", "그외": "other"}, "rate_equipment_payment_guarantee.specialty_construction[{}].rate_percent"),
|
||
("B", 98, 112, "P", {"70억미만": "lt_7_billion", "70억이상-120억미만": "7_to_12_billion", "120억이상-250억미만": "12_to_25_billion", "250억이상-500억미만": "25_to_50_billion", "500억이상": "gte_50_billion"}, "rate_performance_guarantee_fee.brackets[{}].formula"),
|
||
("AU", 15, 46, "BD", CIVIL_PRICE, "rate_overhead.civil_landscape_industrial[{}].rate_percent"),
|
||
("AU", 15, 46, "BL", SPECIAL_PRICE, "rate_overhead.specialty_electric_communication_fire_other[{}].rate_percent"),
|
||
("AU", 15, 46, "BS", CIVIL_PRICE, "rate_profit.brackets[{}].rate_percent"),
|
||
] # fmt: skip
|
||
|
||
|
||
def _formula(text) -> tuple[float, ...]:
|
||
"""수수료 식 속 수 — `79만원`·`75억원` 은 원으로. 원문 `[79만원+(직공비-75억원)x0.0070%]` ↔ 자료 식 글."""
|
||
unit = {"만원": 10**4, "억원": 10**8, "": 1}
|
||
return tuple(float(n.replace(",", "")) * unit[u] for n, u in re.findall(r"(\d[\d,.]*)(만원|억원)?", str(text))) # fmt: skip
|
||
|
||
|
||
def rates_expected(xlsx: dict, pension: dict) -> dict:
|
||
"""조달청 제비율 xlsx · 국민연금법 → 자료 `variables` 를 편 열쇠(`flatten`)와 같은 꼴의 값."""
|
||
raw, filled = xlsx["raw"], xlsx["filled"]
|
||
exp = {key: BASE[re.search(r"\(([^)]*)\)", _t(raw[cell])).group(1)] for key, cell in BASE_CELLS.items()} # fmt: skip
|
||
for var, cell in SINGLE.items():
|
||
base, rate = re.fullmatch(r"\(([^)]*)\)x([\d.]+)", _t(raw[cell])).groups()
|
||
exp[f"{var}.base"] = BASE[base]
|
||
exp["연금 2026 xlsx" if var == "rate_pension" else f"{var}.rate_percent"] = float(rate)
|
||
for year, rate in (pension["years"] | {2033: pension["after"]}).items():
|
||
exp[f"rate_pension.annual_rates[{year}].rate_percent"] = rate
|
||
exp["rate_pension.rate_from_2033_percent"] = pension["after"]
|
||
for r in range(15, 55):
|
||
if (term := raw.get(f"M{r}")) is None:
|
||
continue
|
||
for var, cols in (("rate_indirect_labor", "AC AF AI"), ("rate_other_expense", "AL AO AR")):
|
||
for col in cols.split():
|
||
key = f"{SCALE[_t(filled[f'B{r}'])]}|{TERM[_t(term)]}|{WORK_TYPE[_t(raw[f'{col}9'])]}" # fmt: skip
|
||
exp[f"{var}.brackets[{key}].rate_percent"] = _num(filled[f"{col}{r}"])
|
||
unit = 1000 if "천원" in raw["FA15"] else 1
|
||
for r in range(19, 51):
|
||
if (label := raw.get(f"EB{r}")) is None:
|
||
continue
|
||
scale = SAFETY_SCALE[_t(filled.get(f"DR{r}", filled[f"DK{r}"]))]
|
||
key = f"rate_safety_pct.brackets[{scale}|{SAFETY_WORK[_t(label)]}]"
|
||
exp[f"{key}.rate_percent"] = _num(filled[f"ES{r}"])
|
||
if f"FA{r}" in filled:
|
||
exp[f"{key}.base_amount_krw"] = _num(filled[f"FA{r}"]) * unit
|
||
default, civil = re.findall(r"([\d,]+)억", _t(raw["DR35"]))
|
||
exp["rate_safety_pct.manager_thresholds.default_estimated_amount_krw"] = _num(default) * 10**8
|
||
exp["rate_safety_pct.manager_thresholds.civil_main_work_estimated_amount_krw"] = _num(civil) * 10**8 # fmt: skip
|
||
exp["rate_safety_pct.minimum_total_construction_amount_krw"] = int(re.search(r"(\d+)천만원", raw["DK51"]).group(1)) * 10**7 # fmt: skip
|
||
exp["rate_retirement_mutual_aid.minimum_estimated_amount_krw"] = int(re.search(r"(\d+)억", raw["AU113"]).group(1)) * 10**8 # fmt: skip
|
||
for col, first, last, value_col, groups, template in BANDS:
|
||
got: dict[str, set] = {}
|
||
for r in range(first, last + 1):
|
||
if (label := raw.get(f"{col}{r}")) is None:
|
||
continue
|
||
value = filled.get(f"{value_col}{r}")
|
||
value = _formula(value) if template.endswith("formula") else _num(value)
|
||
got.setdefault(groups.get(_t(label), f"({_t(label)})"), set()).add(value)
|
||
for group, values in got.items():
|
||
if not (group.startswith("(") and values == {None}):
|
||
value = values.pop() if len(values) == 1 else sorted(values, key=str)
|
||
exp[template.format(group)] = list(value) if isinstance(value, tuple) else value
|
||
for candidate, work_type in (("road", "civil_road"), ("other_civil_work", "civil_other")):
|
||
exp[f"rate_environment.forest_road_candidates[{candidate}].rate_percent"] = exp[f"rate_environment.all_work_types[{work_type}].rate_percent"] # fmt: skip
|
||
return exp
|
||
|
||
|
||
def flatten(obj, path: str = "", out: dict | None = None) -> dict:
|
||
"""자료 `variables` 를 `이름.칸[구간|종류].rate_percent` 꼴로 폄 — 목록 줄은 값 아닌 칸을 이어 열쇠로."""
|
||
out = {} if out is None else out
|
||
if isinstance(obj, dict):
|
||
for k, v in obj.items():
|
||
flatten(v, f"{path}.{k}" if path else k, out)
|
||
elif isinstance(obj, list) and obj and isinstance(obj[0], dict):
|
||
values = {"rate_percent", "base_amount_krw", "formula"}
|
||
for item in obj:
|
||
label = "|".join(str(v) for k, v in item.items() if k not in values)
|
||
for k in values & item.keys():
|
||
out[f"{path}[{label}].{k}"] = item[k]
|
||
else:
|
||
out[path] = obj
|
||
return out
|
||
|
||
|
||
def check_rate(src: dict, bad: dict, blind: dict) -> dict:
|
||
exp = dict(src["rates"]["source"])
|
||
data = flatten(src["rates"]["data"]["variables"])
|
||
xlsx_2026 = exp.pop("연금 2026 xlsx")
|
||
law_2026 = exp.get("rate_pension.annual_rates[2026].rate_percent")
|
||
if xlsx_2026 != law_2026:
|
||
bad["연금 2026 xlsx↔법"] = {"source": xlsx_2026, "data": law_2026}
|
||
for key in exp | data:
|
||
if key not in exp:
|
||
blind[f"안 잰 칸 {key}"] = {"what": data[key]}
|
||
continue
|
||
value = list(_formula(data[key])) if key.endswith("formula") and key in data else data.get(key) # fmt: skip
|
||
if exp[key] != value:
|
||
bad[key] = {"source": exp[key], "data": value}
|
||
blind["processing_rules"] = {"what": sorted(src["rates"]["data"].get("processing_rules", {}))}
|
||
blind["조달청 요약표"] = {"what": src["rates"]["paths"][0].name}
|
||
return {"원문 값 수": (len(exp), len(set(exp) & set(data)))}
|
||
|
||
|
||
# ── 맞대기 ──────────────────────────────────────────────────────────────
|
||
READERS = {
|
||
"labor_const": read_labor_const, "labor_mfg": read_labor_mfg, "mach": read_machine,
|
||
"mat": read_material, "oil": lambda p: json.loads(p.read_text(encoding="utf-8")),
|
||
"oil_regional": lambda p: json.loads(p.read_text(encoding="utf-8")),
|
||
} # fmt: skip
|
||
CHECKS = {"labor": check_labor, "machine": check_machine, "material": check_material, "oil": check_oil, "rate": check_rate} # fmt: skip
|
||
|
||
|
||
def load_sources() -> dict:
|
||
src = {}
|
||
for name, file in FILES.items():
|
||
data = json.loads((DATA / file).read_text(encoding="utf-8"))
|
||
paths = [ROOT / s["path"] for s in data["sources"]]
|
||
src[name] = {"data": data, "paths": paths}
|
||
if name == "rates":
|
||
src[name]["source"] = rates_expected(read_xlsx(paths[0]), read_pension(paths[1]))
|
||
else:
|
||
src[name]["source"] = READERS[name](paths[0])
|
||
return src
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def src() -> dict:
|
||
return load_sources()
|
||
|
||
|
||
def run_checks(src: dict) -> tuple[dict[str, dict], dict[str, dict]]:
|
||
"""(갈래별 어긋남·못 보는 자리, 갈래별 셈 `(원문, 자료)`). 틀어진 셈은 어긋남에 `셈 …` 으로도 들어감."""
|
||
out: dict[str, dict] = {kind: {} for kind in KIND_NAMES}
|
||
counts = {}
|
||
for kind, check in CHECKS.items():
|
||
counts[kind] = check(src, out[kind], out[f"{kind}_blind"])
|
||
for name, (s, d) in counts[kind].items():
|
||
if s != d:
|
||
out[kind][f"셈 {name}"] = {"source": s, "data": d}
|
||
return out, counts
|
||
|
||
|
||
def current_mismatches(src: dict) -> dict[str, dict]:
|
||
return run_checks(src)[0]
|
||
|
||
|
||
def load_known() -> dict[str, dict]:
|
||
"""알려진 어긋남 목록. **「할 일 아님」 칸(`_할일_아님`)도 함께 읽는다.**
|
||
|
||
⚠ 왜 합쳐 읽나 — ㉮(원문에 없음·거절 자리·판정으로 닫힘)는 **바퀴마다 다시 세어지지 않게**
|
||
갈래에서 빼 별도 칸에 둔다(2026-09-18 브레인). 그렇다고 비교에서까지 빠지면 그 줄이 다시
|
||
「새로 어긋남」으로 빨개진다 — 그래서 **세는 곳에서만 빼고 비교에는 넣는다.**
|
||
"""
|
||
data = json.loads(KNOWN.read_text(encoding="utf-8"))
|
||
for kind, rows in ((data.get("_할일_아님") or {}).get("줄") or {}).items():
|
||
data.setdefault(kind, {}).update(rows)
|
||
return data
|
||
|
||
|
||
# ── 시험 ────────────────────────────────────────────────────────────────
|
||
@pytest.mark.parametrize("kind", list(KIND_NAMES))
|
||
def test_알려진_어긋남_밖은_없음(src: dict, kind: str) -> None:
|
||
"""알려진 목록과 똑같아야 함 — 새로 어긋남 · 모습 바뀜 · 고쳐짐 셋 다 빨강. `why` 는 맞대기에서 뺌."""
|
||
now = current_mismatches(src)[kind]
|
||
known = {k: _without_why(v) for k, v in load_known()[kind].items()}
|
||
lines: list[str] = []
|
||
for key in sorted(set(now) - set(known)):
|
||
lines.append(f" 새로 어긋남 {key}: {json.dumps(now[key], ensure_ascii=False)}")
|
||
for key in sorted(set(now) & set(known)):
|
||
if now[key] != known[key]:
|
||
lines.append(
|
||
f" 모습 바뀜 {key}: 목록 {json.dumps(known[key], ensure_ascii=False)}"
|
||
f"\n 지금 {json.dumps(now[key], ensure_ascii=False)}"
|
||
)
|
||
for key in sorted(set(known) - set(now)):
|
||
lines.append(f" 고쳐짐 {key} — 목록에서 지울 것(--prune)")
|
||
assert not lines, f"{KIND_NAMES[kind]} {len(lines)}건 — 원문 ↔ 자료:\n" + "\n".join(lines)
|
||
|
||
|
||
@pytest.mark.parametrize("kind", KINDS)
|
||
def test_셈이_맞음(src: dict, kind: str) -> None:
|
||
"""원문 줄 수 = 자료 줄 수 · 자료 줄은 모두 원문에서 찾음 — 틀어진 셈은 목록 `셈 …` 과 똑같을 때만 통과."""
|
||
known = load_known()[kind]
|
||
wrong = [
|
||
f" {name}: 원문 {s} · 자료 {d}"
|
||
for name, (s, d) in run_checks(src)[1][kind].items()
|
||
if s != d and _without_why(known.get(f"셈 {name}", {})) != {"source": s, "data": d}
|
||
]
|
||
assert not wrong, f"{KIND_NAMES[kind]} 셈 틀어짐(목록 밖):\n" + "\n".join(wrong)
|
||
|
||
|
||
if __name__ == "__main__" and "--prune" in sys.argv:
|
||
# 고쳐졌거나 모습이 바뀐 줄만 지움 — 새 어긋남은 안 보탬(보태면 빨강이 조용히 묻힘).
|
||
# ⚠ 「할 일 아님」 칸(`_할일_아님`)은 **제자리에서** 지운다. `load_known()` 이 합쳐 읽은 것을
|
||
# 그대로 되쓰면 2026-09-18 에 가른 칸이 다시 갈래로 섞여 「없는 할 일」이 되살아난다.
|
||
current = current_mismatches(load_sources())
|
||
data = json.loads(KNOWN.read_text(encoding="utf-8"))
|
||
parked = (data.get("_할일_아님") or {}).get("줄") or {}
|
||
|
||
def _keep(kind: str, rows: dict) -> dict:
|
||
return {k: v for k, v in rows.items() if current[kind].get(k) == _without_why(v)}
|
||
|
||
for kind in KIND_NAMES:
|
||
before = len(data.get(kind) or {})
|
||
data[kind] = _keep(kind, data.get(kind) or {})
|
||
print(f"{KIND_NAMES[kind]}: {before} → {len(data[kind])}")
|
||
if kind in parked:
|
||
was = len(parked[kind])
|
||
parked[kind] = _keep(kind, parked[kind])
|
||
if len(parked[kind]) != was:
|
||
print(f" 할 일 아님 {KIND_NAMES[kind]}: {was} → {len(parked[kind])}")
|
||
KNOWN.write_text(json.dumps(data, ensure_ascii=False, indent=1) + "\n", encoding="utf-8")
|