Files
Aislo/resources/tester/scratch/audit_cell_values.py
T

125 lines
4.3 KiB
Python

import json
import sys
import re
from pathlib import Path
sys.stdout.reconfigure(encoding='utf-8')
# Load files
md_path = Path("resources/knowledge/original/행정규칙/임도 품셈 적용기준 (현 산림사업 표준품셈)/첨부/(산림청고시 제2025-82호) 산림사업 표준품셈.md")
with open(md_path, "r", encoding="utf-8") as f:
md_lines = f.readlines()
with open('resources/data_cost_input_value/pum_forest_2026.json', 'r', encoding='utf-8') as f:
pum_data = json.load(f)
pum_tables = pum_data.get('variables', {}).get('pum', {}).get('tables', [])
pum_map = {t['table_id']: t for t in pum_tables}
with open('resources/data_work_item_master/work_item_master_2026-01-01.json', 'r', encoding='utf-8') as f:
master = json.load(f)
master_tables = {}
for wi in master['work_items']:
for t in wi.get('tables', []):
tid = t.get('pum_table_id')
if tid:
master_tables[tid] = t
for t in master.get('orphan_tables', []):
tid = t.get('pum_table_id')
if tid:
master_tables[tid] = t
# Extract all markdown tables from MD
md_tables = []
current_table = []
start_line = -1
for idx, line in enumerate(md_lines):
line_str = line.strip()
if line_str.startswith('|') and line_str.endswith('|'):
if not current_table:
start_line = idx + 1
current_table.append((idx + 1, line))
else:
if current_table:
has_sep = any(re.match(r'^\|(?:\s*:?-+:?\s*\|)+$', l.strip()) for _, l in current_table)
if has_sep:
md_tables.append({
'start_line': start_line,
'end_line': current_table[-1][0],
'lines': [l for _, l in current_table]
})
current_table = []
if current_table:
has_sep = any(re.match(r'^\|(?:\s*:?-+:?\s*\|)+$', l.strip()) for _, l in current_table)
if has_sep:
md_tables.append({
'start_line': start_line,
'end_line': current_table[-1][0],
'lines': [l for _, l in current_table]
})
print(f"MD tables: {len(md_tables)}, pum_forest tables: {len(pum_tables)}")
# Function to parse table rows from md lines
def parse_md_cells(lines):
rows = []
for l in lines:
l_str = l.strip()
if re.match(r'^\|(?:\s*:?-+:?\s*\|)+$', l_str):
continue
cells = [c.strip() for c in l_str.split('|')[1:-1]]
rows.append(cells)
return rows
# Compare each table's cells between MD and pum_forest
cell_mismatches = []
row_count_mismatches = []
for i, mdt in enumerate(md_tables):
tid = f"F{i+1:04d}"
pt = pum_map.get(tid)
if not pt:
continue
md_rows = parse_md_cells(mdt['lines'])
# In pum_tables, headers is first row(s), rows is data rows
pt_headers = pt.get('headers', [])
pt_data_rows = pt.get('rows', [])
pt_all_rows = []
if pt_headers:
pt_all_rows.append(pt_headers)
pt_all_rows.extend(pt_data_rows)
# Check row counts
if len(md_rows) != len(pt_all_rows):
row_count_mismatches.append((tid, mdt['start_line'], len(md_rows), len(pt_all_rows)))
continue
for r_idx, (mr, pr) in enumerate(zip(md_rows, pt_all_rows)):
if len(mr) != len(pr):
cell_mismatches.append((tid, mdt['start_line'] + r_idx, f"Col count diff: md={len(mr)} vs pt={len(pr)}"))
continue
for c_idx, (mc, pc) in enumerate(zip(mr, pr)):
# Normalize strings for comparison
# remove spaces, trailing .0, etc.
mc_norm = mc.replace(" ", "").replace(",", "")
pc_norm = str(pc).replace(" ", "").replace(",", "")
if mc_norm != pc_norm:
# Check numeric equivalence
try:
if float(mc_norm) == float(pc_norm):
continue
except:
pass
cell_mismatches.append((tid, mdt['start_line'] + r_idx, f"Cell[{r_idx},{c_idx}] diff: md='{mc}' vs pt='{pc}'"))
print(f"Row count mismatches: {len(row_count_mismatches)}")
for r in row_count_mismatches[:10]:
print(f" {r[0]} line {r[1]}: md rows={r[2]} vs pt rows={r[3]}")
print(f"\nCell mismatches: {len(cell_mismatches)}")
for cm in cell_mismatches[:20]:
print(f" {cm[0]} line {cm[1]}: {cm[2]}")