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

120 lines
5.6 KiB
Python

import os, sys, re, json, glob
sys.stdout.reconfigure(encoding='utf-8')
base_dir = 'resources/knowledge/original/원가계산/건설공사_표준품셈'
divisions = ['02_토목부문', '03_건축부문', '04_기계설비부문', '05_유지관리부문']
risk_tables = []
for div in divisions:
div_path = os.path.join(base_dir, div)
if not os.path.exists(div_path):
continue
md_files = sorted(glob.glob(os.path.join(div_path, '*.md')))
for fpath in md_files:
fname = os.path.basename(fpath)
if fname.startswith('_') or '개정사항' in fname:
continue
with open(fpath, 'r', encoding='utf-8') as f:
lines = f.readlines()
current_sec = 'UNKNOWN'
in_table = False
headers = []
table_rows = []
start_line = 0
for idx, line in enumerate(lines):
line_s = line.strip()
sec_m = re.search(r'(\d+-\d+(?:-\d+)?(?:\s+[^\n(]+)?)', line_s)
if sec_m and not line_s.startswith('|'):
current_sec = sec_m.group(1).strip()
if line_s.startswith('|') and not line_s.startswith('|---'):
cols = [c.strip() for c in line_s.split('|')[1:-1]]
if not in_table:
# check if table starts
headers = cols
in_table = True
start_line = idx + 1
table_rows = []
continue
if in_table:
table_rows.append((idx + 1, cols))
else:
if in_table and (line_s == '' or not line_s.startswith('|')):
# Evaluate table for risk criteria
# 1. Condition or remarks across multiple rows
# 2. No classification code, split only by condition
# 3. Efficiency, coefficient, factor tables
header_text = " ".join(headers)
table_all_text = header_text + " " + " ".join([" ".join(r[1]) for r in table_rows])
is_risk = False
risk_reasons = []
# Rule 1: Efficiency / coefficient / factor table
if any(k in header_text for k in ['효율', '작업효율', '계수', '할증', '보정', '손실률', '환산', '증감']):
is_risk = True
risk_reasons.append('효율/계수/할증 표')
elif any(k in current_sec for k in ['효율', '계수', '할증']):
is_risk = True
risk_reasons.append('효율/계수 섹션')
# Rule 2: Remarks column present with multiple rows
if any('비고' in h or '비 고' in h for h in headers) and len(table_rows) >= 2:
rem_idx = [i for i, h in enumerate(headers) if '비고' in h or '비 고' in h][0]
non_empty_rem = [r[1][rem_idx] for r in table_rows if rem_idx < len(r[1]) and r[1][rem_idx]]
if len(non_empty_rem) >= 1:
is_risk = True
risk_reasons.append('비고 열 보유 다중행')
# Rule 3: Conditions branching (no job category/worker in headers, just conditions)
condition_kws = ['조건', '지형', '토질', '수심', '암질', '두께', '경사', '규격', '차단', '구분']
if any(k in header_text for k in condition_kws) and not any(k in header_text for k in ['직종', '인부', '기술자', '기능사']):
# Table has conditions, possibly matrix/coefficient
if len(table_rows) >= 2:
is_risk = True
risk_reasons.append('조건 분기 표 (직종 없음)')
# Rule 4: Col 0 has merged or multi-level values
col0_vals = [r[1][0] for r in table_rows if len(r[1]) > 0]
if any(v == '' for v in col0_vals) or any('→' in v for v in col0_vals):
is_risk = True
risk_reasons.append('Col 0 빈칸/연결(병합/다단)')
if is_risk:
risk_tables.append({
'division': div,
'file': fname,
'sec': current_sec,
'start_line': start_line,
'rows': len(table_rows),
'headers': headers,
'reasons': risk_reasons,
'col0_sample': col0_vals[:4]
})
in_table = False
headers = []
table_rows = []
print(f"Total Risk Tables found: {len(risk_tables)}")
by_file = {}
for rt in risk_tables:
k = f"{rt['division']}/{rt['file']}"
by_file.setdefault(k, []).append(rt)
for k, v in by_file.items():
print(f"\n[{k}] : {len(v)} risk tables")
for item in v[:5]:
print(f" - L{item['start_line']} {item['sec']} ({', '.join(item['reasons'])}) : {item['headers']}")
if len(v) > 5:
print(f" ... and {len(v)-5} more")
with open('scratch/risk_tables_list.json', 'w', encoding='utf-8') as f:
json.dump(risk_tables, f, ensure_ascii=False, indent=2)