Aislo/resources/knowledge/original/_pipeline/kbiz_verify.py

# -*- coding: utf-8 -*-
"""중소제조업 임금조사 회차 md 표 검증 — 「3. 직종별 조사노임.md」 값을 PDF 글자층과 따로 읽어 대조 + 회차 사이 대조.

사용: python kbiz_verify.py            (원가계산/노임단가_제조업_중소기업중앙회/<적용일>/ 전부)
- 표 값: md 의 직종 줄(코드 · 이름 · 값 칸들) = PDF 단어 좌표로 다시 읽은 줄(코드 · 이름 · 값들) 인지.
- 회차 사이: 같은 조사월 값(직종별)이 다른 회차 보고서 칸에도 실려 있으면 같은지.
"""

import re
import sys
from collections import defaultdict
from pathlib import Path

import pymupdf as fitz

sys.path.insert(0, str(Path(__file__).resolve().parent))
from kbiz_report_md import pages_of  # noqa: E402

BASE = Path(__file__).resolve().parent.parent / "원가계산" / "노임단가_제조업_중소기업중앙회"
CODE = re.compile(r"^(\d{1,3})\.")
VAL = re.compile(r"^[　]*\*{0,2}[\d,]+(?:\.\d+)?\**[　]*$|^[　]*\*{1,2}[　]*$|^[　]*-[　]*$|^　$")


def md_rows(md_path):
    """조사노임표(머리에 「업 종」 · 「조사노임」) 의 직종 줄 [(코드, 이름, [값 칸들])]."""
    rows, blocks, cur = [], [], []
    for ln in Path(md_path).read_text(encoding="utf-8").split(chr(10)):
        if ln.startswith("|"):
            cur.append(ln)
        elif cur:
            blocks.append(cur)
            cur = []
    if cur:
        blocks.append(cur)
    for blk in blocks:
        head = " ".join(blk[:2])
        if not ("업 종" in head and "조사노임" in head):
            continue
        for ln in blk[2:]:
            cells = [c.strip() for c in ln.strip().strip("|").split("|")]
            for i, c in enumerate(cells):
                m = CODE.match(c)
                if m and i <= 2:
                    rows.append((int(m.group(1)), c, cells[i + 1 :]))
                    break
    return rows


def pdf_rows(pdf):
    doc, _, _ = pages_of(pdf)
    out = []
    for p in doc:
        ws = [(w[0], w[1], w[2], w[3], w[4]) for w in p.get_text("words")]
        for x0, y0, x1, y1, w in ws:
            if not (CODE.match(w) and x0 < 300):
                continue
            yc = (y0 + y1) / 2
            toks = sorted(
                (t for t in ws if abs((t[1] + t[3]) / 2 - yc) < 5 and t[0] >= x0),
                key=lambda t: t[0],
            )
            name, j = toks[0][4], 1
            while j < len(toks) and not VAL.match(toks[j][4]):
                name += toks[j][4]
                j += 1
            vals = [t[4] for t in toks[j:]]
            if vals and all(VAL.match(v) for v in vals):
                out.append((int(CODE.match(name).group(1)), name, vals))
    return out


def wage_pages_rows(rows):
    """값이 3~5칸인 줄만(=조사노임표 · 다른 표의 코드 줄 제외)."""
    return [(c, n, v) for c, n, v in rows if all(VAL.match(x) or x == "" for x in v)]


def check_round(folder):
    pdf = next(folder.glob("*.pdf"))
    a = md_rows(folder / "3. 직종별 조사노임.md")
    b = pdf_rows(pdf)
    clean = lambda v: [x.replace("　", "").strip() for x in v if x.replace("　", "").strip()]
    fa = [(c, n.replace(" ", ""), clean(v)) for c, n, v in a]
    fb = [(c, n.replace(" ", ""), clean(v)) for c, n, v in b]
    # 조사노임표는 쪽 18~ 이므로 md 표 줄 수 = 직종수. PDF 쪽 전체에서 잡힌 줄 중 표 줄만 남기려면 md 코드+이름으로 맞춤
    keyb = {(c, n): v for c, n, v in fb}
    bad = []
    for c, n, v in fa:
        pv = keyb.get((c, n))
        if pv is None:
            bad.append(("PDF 에 없음", c, n))
        elif pv != v:
            bad.append(("값 다름", c, n, v, pv))
    return len(fa), bad


MONTH = re.compile(r"(\d{4})\s*[년.]\s*(\d{1,2})\s*월")


def load_rounds():
    """{적용일: {'months': [조사월 3개], 'rows': {직종명: (코드, [값 3개])}, 'avg': [전체평균 3개]}} — 2026-07-01 은 기본 폴더."""
    res = {}
    dirs = [
        (d.name, d) for d in sorted(BASE.iterdir()) if d.is_dir() and re.match(r"\d{4}-", d.name)
    ]
    dirs.append(("2026-07-01", BASE))
    for label, d in dirs:
        md = d / "3. 직종별 조사노임.md"
        lines = md.read_text(encoding="utf-8").split(chr(10))
        months = None
        blocks, cur = [], []
        for ln in lines:
            if ln.startswith("|"):
                cur.append(ln)
            elif cur:
                blocks.append(cur)
                cur = []
        if cur:
            blocks.append(cur)
        rows, avg = {}, None
        for blk in blocks:
            head = " ".join(blk[:3])
            if not ("업 종" in head and "조사노임" in head):
                continue
            hc = [c.strip() for c in blk[0].strip().strip("|").split("|")]
            ms = [MONTH.search(c) for c in hc]
            ms = [f"{m.group(1)}.{int(m.group(2))}" for m in ms if m]
            months = months or ms
            for ln in blk[2:]:
                cells = [c.strip() for c in ln.strip().strip("|").split("|")]
                if cells and cells[0].startswith("전체평균"):
                    avg = [c for c in cells[1:] if c]
                for i, c in enumerate(cells):
                    m = CODE.match(c)
                    if m and i <= 2:
                        v = cells[i + 1 :]
                        # 값 칸: 조사노임 · 변동계수 · 앞 조사 · 앞앞 조사 → 변동계수 뺌
                        if len(v) == 4:
                            v = [v[0], v[2], v[3]]
                        rows[c.split(".", 1)[1].replace(" ", "")] = (int(m.group(1)), v)
                        break
        res[label] = {"months": months, "rows": rows, "avg": avg}
    return res


def cross_check():
    rs = load_rounds()
    seen = {}
    diffs = []
    for label, r in rs.items():
        for name, (code, vals) in r["rows"].items():
            for mth, v in zip(r["months"], vals):
                k = (name, mth)
                if k in seen and seen[k][1] != v:
                    diffs.append((name, mth, seen[k], (label, v)))
                seen.setdefault(k, (label, v))
    return rs, diffs


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "cross":
        rs, diffs = cross_check()
        for label, r in rs.items():
            print(label, r["months"], "직종", len(r["rows"]), "전체평균", r["avg"])
        print("회차 사이 다른 칸", len(diffs))
        for d in diffs[:60]:
            print("  ", d)
        sys.exit()
    for d in sorted(p for p in BASE.iterdir() if p.is_dir() and re.match(r"\d{4}-", p.name)):
        if not list(d.glob("*.pdf")):
            continue
        n, bad = check_round(d)
        print(d.name, "직종 줄", n, "이상", len(bad), bad[:4])