269 lines
10 KiB
Python
269 lines
10 KiB
Python
import json
|
||
import sys
|
||
import re
|
||
from pathlib import Path
|
||
from collections import defaultdict, Counter
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||
from scratch.enrich_tables import tables, get_chapter
|
||
|
||
sys.stdout.reconfigure(encoding='utf-8')
|
||
|
||
# 1. Load data files
|
||
with open("resources/knowledge/original/행정규칙/임도 품셈 적용기준 (현 산림사업 표준품셈)/첨부/(산림청고시 제2025-82호) 산림사업 표준품셈.md", "r", encoding="utf-8") as f:
|
||
md_lines = f.readlines()
|
||
|
||
with open('resources/data_work_item_master/work_item_master_2026-01-01.json', 'r', encoding='utf-8') as f:
|
||
master = json.load(f)
|
||
|
||
with open('resources/data_cost_input_value/pum_forest_2026.json', 'r', encoding='utf-8') as f:
|
||
pum_data = json.load(f)
|
||
|
||
with open('resources/data_cost_input_value/coef_2026.json', 'r', encoding='utf-8') as f:
|
||
coef_data = json.load(f)
|
||
|
||
pum_tables = {t['table_id']: t for t in pum_data.get('variables', {}).get('pum', {}).get('tables', [])}
|
||
|
||
master_tables = {}
|
||
table_wi_map = {}
|
||
for wi in master['work_items']:
|
||
for t in wi.get('tables', []):
|
||
tid = t.get('pum_table_id')
|
||
if tid:
|
||
master_tables[tid] = t
|
||
table_wi_map[tid] = wi
|
||
|
||
orphan_tables = {t.get('pum_table_id'): t for t in master.get('orphan_tables', []) if t.get('pum_table_id')}
|
||
|
||
print("=== STARTING COMPREHENSIVE 5-ITEM AUDIT ===")
|
||
|
||
# --- 1. MD vs Master Presence (표·절·갈래 누락) ---
|
||
all_tids = [f"F{i+1:04d}" for i in range(476)]
|
||
audit_item_1 = {
|
||
'missing_tables': [], # completely missing from master
|
||
'orphan_tables': [], # in orphan_tables (not in work_items)
|
||
'wi_tables_count': len(master_tables),
|
||
'orphan_tables_count': len(orphan_tables)
|
||
}
|
||
|
||
for tid in all_tids:
|
||
if tid not in master_tables and tid not in orphan_tables:
|
||
t = pum_tables.get(tid, {})
|
||
audit_item_1['missing_tables'].append({
|
||
'table_id': tid,
|
||
'line': t.get('line'),
|
||
'section': t.get('section'),
|
||
'reason': '고시 본칙 경과조치 표(Line 31)로 공종 축 제외'
|
||
})
|
||
elif tid in orphan_tables:
|
||
t = pum_tables.get(tid, {})
|
||
ot = orphan_tables[tid]
|
||
audit_item_1['orphan_tables'].append({
|
||
'table_id': tid,
|
||
'line': t.get('line'),
|
||
'section': t.get('section'),
|
||
'pum_form': ot.get('pum_form'),
|
||
'reason': '부록 설계사례(Line 8031~9106) 예시표로 단가산출 공종축 미배정'
|
||
})
|
||
|
||
print(f"Item 1: Missing tables: {len(audit_item_1['missing_tables'])}, Orphan tables: {len(audit_item_1['orphan_tables'])}")
|
||
|
||
# --- 2. 값이 다른 칸 (숫자 불일치) ---
|
||
# Raw cells match 100%. Check [주] numerical values and constants in coef vs MD
|
||
audit_item_2 = []
|
||
|
||
# Check L, C values in coef vs F0004
|
||
f0004 = pum_tables.get('F0004')
|
||
soil_L_records = coef_data.get('variables', {}).get('coef_soil_L', {}).get('records', [])
|
||
soil_C_records = coef_data.get('variables', {}).get('coef_soil_C', {}).get('records', [])
|
||
|
||
soil_diffs = []
|
||
if f0004:
|
||
for idx, r in enumerate(f0004.get('rows', [])):
|
||
sname = r[0].strip().replace(' ', '')
|
||
l_str = r[1].strip()
|
||
c_str = r[2].strip()
|
||
# Find in coef records
|
||
matched_l = None
|
||
if idx < len(soil_L_records):
|
||
matched_l = soil_L_records[idx]
|
||
matched_c = None
|
||
if idx < len(soil_C_records):
|
||
matched_c = soil_C_records[idx]
|
||
|
||
# Parse min, max from l_str
|
||
lm = re.match(r'([0-9\.]+)(?:[～~∼\-]([0-9\.]+))?', l_str)
|
||
if lm and matched_l:
|
||
orig_min = float(lm.group(1))
|
||
orig_max = float(lm.group(2)) if lm.group(2) else orig_min
|
||
if abs(matched_l.get('min', 0) - orig_min) > 1e-4 or abs(matched_l.get('max', 0) - orig_max) > 1e-4:
|
||
soil_diffs.append({'row': idx, 'soil': sname, 'md_L': l_str, 'coef_L': matched_l})
|
||
|
||
audit_item_2.append({'category': 'soil_conversion_L_C', 'diffs': soil_diffs})
|
||
print(f"Item 2: Soil conversion discrepancies: {len(soil_diffs)}")
|
||
|
||
# --- 3. 밑수·단위 (md 에는 있는데 마스터에서 비었거나 다른 것) ---
|
||
basis_pattern = re.compile(r'\[\s*([0-9\.\,]*)\s*([a-zA-Z㎡㎥㏊ha본개공mkm㎏gtonLℓ대인組]+)\s*당\s*\]')
|
||
basis_pattern2 = re.compile(r'\(단위\s*:\s*([^\)]+)\)')
|
||
|
||
audit_item_3 = []
|
||
for t in tables:
|
||
tid = t['table_id']
|
||
mt = master_tables.get(tid) or orphan_tables.get(tid)
|
||
if not mt:
|
||
continue
|
||
|
||
m_qty = mt.get('basis_quantity')
|
||
m_unit = mt.get('basis_unit')
|
||
pform = mt.get('pum_form')
|
||
|
||
# MD search
|
||
found_q = None
|
||
found_u = None
|
||
found_src = None
|
||
for pl in reversed(t['pre_context']):
|
||
m = basis_pattern.search(pl)
|
||
if m:
|
||
q_str = m.group(1).replace(',', '').strip()
|
||
found_q = float(q_str) if q_str else 1.0
|
||
found_u = m.group(2).strip()
|
||
found_src = f"pre_context: '{pl}'"
|
||
break
|
||
m2 = basis_pattern2.search(pl)
|
||
if m2:
|
||
u_str = m2.group(1).strip()
|
||
# check if u_str has quantity e.g. "10㎡당"
|
||
qm = re.match(r'^([0-9\.\,]+)\s*([a-zA-Z㎡㎥㏊ha본개공mkm㎏gtonLℓ대인組]+)', u_str)
|
||
if qm:
|
||
found_q = float(qm.group(1).replace(',', ''))
|
||
found_u = qm.group(2).replace('당', '')
|
||
else:
|
||
found_q = 1.0
|
||
found_u = u_str.replace('당', '')
|
||
found_src = f"unit_parenthesis: '{pl}'"
|
||
break
|
||
|
||
if found_u:
|
||
# compare
|
||
if m_qty is None or m_unit is None:
|
||
audit_item_3.append({
|
||
'table_id': tid,
|
||
'chapter': t['chapter'],
|
||
'section': t['section'],
|
||
'line': t['start_line'],
|
||
'issue': 'BASIS_NULL_IN_MASTER',
|
||
'md_qty': found_q,
|
||
'md_unit': found_u,
|
||
'md_src': found_src,
|
||
'master_qty': m_qty,
|
||
'master_unit': m_unit,
|
||
'pum_form': pform
|
||
})
|
||
else:
|
||
# check numeric or unit mismatch
|
||
u_norm_md = found_u.replace('㎡','m2').replace('㎥','m3').replace('인/','').replace('당','')
|
||
u_norm_m = str(m_unit).replace('㎡','m2').replace('㎥','m3').replace('인/','').replace('당','')
|
||
if abs(float(m_qty) - float(found_q)) > 1e-4 or u_norm_md != u_norm_m:
|
||
audit_item_3.append({
|
||
'table_id': tid,
|
||
'chapter': t['chapter'],
|
||
'section': t['section'],
|
||
'line': t['start_line'],
|
||
'issue': 'BASIS_VALUE_OR_UNIT_DIFF',
|
||
'md_qty': found_q,
|
||
'md_unit': found_u,
|
||
'md_src': found_src,
|
||
'master_qty': m_qty,
|
||
'master_unit': m_unit,
|
||
'pum_form': pform
|
||
})
|
||
|
||
print(f"Item 3: Basis issues: {len(audit_item_3)}")
|
||
|
||
# --- 4. [주]의 계수·조건 누락 (표 아래 붙은 주석 계수가 마스터에 안 들어온 것) ---
|
||
audit_item_4 = []
|
||
for t in tables:
|
||
tid = t['table_id']
|
||
notes = t['notes']
|
||
if not notes:
|
||
continue
|
||
mt = master_tables.get(tid) or orphan_tables.get(tid)
|
||
if not mt:
|
||
continue
|
||
|
||
master_note = mt.get('condition_note')
|
||
# check if notes contain important pricing multipliers (%, 가산, 할증, 손료, 공제, 소요량, 레미콘, 자재 등)
|
||
cost_keywords = ['%', '가산', '할증', '손료', '손율', '공제', '소요', '포함', '별도', '레미콘', '감한다', '곱한다', '적용한다']
|
||
notes_text = " ".join(notes)
|
||
has_cost_keyword = any(k in notes_text for k in cost_keywords)
|
||
|
||
if has_cost_keyword:
|
||
# Check if master has condition_note
|
||
if not master_note:
|
||
audit_item_4.append({
|
||
'table_id': tid,
|
||
'chapter': t['chapter'],
|
||
'section': t['section'],
|
||
'line': t['start_line'],
|
||
'issue': 'CRITICAL_NOTE_MISSING_IN_MASTER',
|
||
'notes_count': len(notes),
|
||
'notes_preview': notes[:3],
|
||
'pum_form': mt.get('pum_form')
|
||
})
|
||
else:
|
||
# master has note, check if all notes captured
|
||
pass
|
||
|
||
print(f"Item 4: Critical notes missing in master: {len(audit_item_4)}")
|
||
|
||
# --- 5. 갈래 (variants) 및 pum_form 오분류 건 ---
|
||
# Tables where pum_form is reference but contain actual requirements/costs
|
||
audit_item_5 = []
|
||
reference_suspects = []
|
||
for t in tables:
|
||
tid = t['table_id']
|
||
mt = master_tables.get(tid)
|
||
if not mt:
|
||
continue
|
||
pform = mt.get('pum_form')
|
||
if pform == 'reference':
|
||
# Check if headers/rows contain units like 인, %, kg, ton, m, ㎡, ㎥, 대, etc.
|
||
hdr_str = " ".join(t['lines'][0:2])
|
||
# Also check if it belongs to core chapters
|
||
ch = t['chapter']
|
||
if any(k in ch for k in ['제5장', '제9장', '제12장', '제13장']):
|
||
reference_suspects.append({
|
||
'table_id': tid,
|
||
'chapter': ch,
|
||
'section': t['section'],
|
||
'line': t['start_line'],
|
||
'issue': 'REFERENCE_MISCLASSIFIED_SHOULD_BE_REQUIREMENT',
|
||
'pum_form': pform,
|
||
'basis': f"{mt.get('basis_quantity')} {mt.get('basis_unit')}"
|
||
})
|
||
|
||
print(f"Item 5: Suspect reference tables in core chapters: {len(reference_suspects)}")
|
||
|
||
# Save full results into json
|
||
audit_results = {
|
||
'summary': {
|
||
'total_tables_md': len(tables),
|
||
'item1_missing_tables': len(audit_item_1['missing_tables']),
|
||
'item1_orphan_tables': len(audit_item_1['orphan_tables']),
|
||
'item2_soil_diffs': len(soil_diffs),
|
||
'item3_basis_issues': len(audit_item_3),
|
||
'item4_missing_critical_notes': len(audit_item_4),
|
||
'item5_misclassified_references_core': len(reference_suspects)
|
||
},
|
||
'item1_presence': audit_item_1,
|
||
'item2_values': audit_item_2,
|
||
'item3_basis': audit_item_3,
|
||
'item4_notes': audit_item_4,
|
||
'item5_references': reference_suspects
|
||
}
|
||
|
||
with open('scratch/audit_results.json', 'w', encoding='utf-8') as f:
|
||
json.dump(audit_results, f, ensure_ascii=False, indent=2)
|
||
|
||
print("\nAudit results saved to scratch/audit_results.json.")
|