"""Excel 메타 시트 → base.json 변환. 처리 시트: - 화물적부도 화물코드: 1,345개 기본 레코드 - 동의어: 215개 한/영 유사명 - IBC CODE: IMO IBC 분류 출력: HNSSearchSubstance 스키마(frontend/src/common/types/hns.ts)에 맞춘 JSON 배열. """ from __future__ import annotations import io import json import os import re import sys from collections import defaultdict from pathlib import Path import openpyxl sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') SCRIPT_DIR = Path(__file__).parent.resolve() OUT_DIR = SCRIPT_DIR / 'out' OUT_DIR.mkdir(exist_ok=True) SOURCE_XLSX = Path(os.environ.get( 'HNS_SOURCE_XLSX', r'C:\Projects\MeterialDB\유해물질 화물적부도 검색툴.xlsm', )) def norm_key(s: str | None) -> str: if not s: return '' return re.sub(r'\s+', '', str(s)).strip().lower() def split_synonyms(raw: str | None) -> str: if not raw: return '' # 원본은 "·" 또는 "/" 구분, 개행 포함 parts = re.split(r'[·/\n]+', str(raw)) cleaned = [p.strip() for p in parts if p and p.strip()] return ' / '.join(cleaned) def clean_text(v) -> str: if v is None: return '' return str(v).strip() def main() -> None: print(f'[읽기] {SOURCE_XLSX}') if not SOURCE_XLSX.exists(): raise SystemExit(f'소스 파일 없음: {SOURCE_XLSX}') wb = openpyxl.load_workbook(SOURCE_XLSX, read_only=True, data_only=True, keep_vba=False) # ────────── 화물적부도 화물코드 ────────── ws = wb['화물적부도 화물코드'] rows = list(ws.iter_rows(values_only=True)) # 헤더 row6: 연번, 약자/제품명, 영어명, 영문명 동의어, 국문명, 국문명 동의어, 주요 사용처, UN번호, CAS번호 cargo_rows = [r for r in rows[6:] if r[0] is not None and isinstance(r[0], (int, float))] print(f'[화물적부도] 데이터 행 {len(cargo_rows)}개') # ────────── 동의어 시트 ────────── ws_syn = wb['동의어'] syn_rows = list(ws_syn.iter_rows(values_only=True)) # 헤더 row2: 연번, 국문명, 영문명, cas, un, 한글 유사명, 영문 유사명 syn_map: dict[str, dict] = {} for r in syn_rows[2:]: if not r or r[0] is None: continue name_kr = clean_text(r[1]) cas = clean_text(r[3]) if not name_kr and not cas: continue key = norm_key(name_kr) or norm_key(cas) syn_map[key] = { 'synonymsKr': split_synonyms(r[5]) if len(r) > 5 else '', 'synonymsEn': split_synonyms(r[6]) if len(r) > 6 else '', } print(f'[동의어] {len(syn_map)}개') # ────────── IBC CODE 시트 ────────── ws_ibc = wb['IBC CODE'] ibc_map: dict[str, dict] = {} for i, r in enumerate(ws_ibc.iter_rows(values_only=True)): if i < 2: continue # header 2 rows if not r or not r[0]: continue name_en = clean_text(r[0]) key = norm_key(name_en) if not key: continue ibc_map[key] = { 'ibcHazard': clean_text(r[2]), # 위험성 S/P 'ibcShipType': clean_text(r[3]), # 선박형식 'ibcTankType': clean_text(r[4]), # 탱크형식 'ibcDetection': clean_text(r[10]) if len(r) > 10 else '', # 탐지장치 'ibcFireFighting': clean_text(r[12]) if len(r) > 12 else '', # 화재대응 'ibcMinRequirement': clean_text(r[14]) if len(r) > 14 else '', # 구체적운영상 요건 } print(f'[IBC CODE] {len(ibc_map)}개') wb.close() # ────────── 통합 레코드 생성 ────────── # 동일 CAS/국문명 기준으로 cargoCodes 그룹화 groups: dict[str, list] = defaultdict(list) for r in cargo_rows: _, abbr, name_en, syn_en, name_kr, syn_kr, usage, un, cas = r[:9] # 그룹 키: CAS 우선, 없으면 국문명 cas_s = clean_text(cas) group_key = cas_s if cas_s else norm_key(name_kr) groups[group_key].append({ 'abbreviation': clean_text(abbr), 'nameKr': clean_text(name_kr), 'nameEn': clean_text(name_en), 'synonymsKr': split_synonyms(syn_kr), 'synonymsEn': split_synonyms(syn_en), 'usage': clean_text(usage), 'unNumber': clean_text(un), 'casNumber': cas_s, }) records: list[dict] = [] next_id = 1 for group_key, entries in groups.items(): # 대표 레코드: 가장 먼저 등장 (동의어 필드가 있는 걸 우선) primary = max(entries, key=lambda e: (bool(e['synonymsKr']), bool(e['synonymsEn']), len(e['nameKr']))) name_kr_key = norm_key(primary['nameKr']) name_en_key = norm_key(primary['nameEn']) # 동의어 병합 syn_extra = syn_map.get(name_kr_key, {}) synonyms_kr = ' / '.join(filter(None, [primary['synonymsKr'], syn_extra.get('synonymsKr', '')])) synonyms_en = ' / '.join(filter(None, [primary['synonymsEn'], syn_extra.get('synonymsEn', '')])) # IBC 병합 (영문명 기준) ibc = ibc_map.get(name_en_key, {}) # cargoCodes 집계 cargo_codes = [ { 'code': e['abbreviation'], 'name': e['nameEn'] or e['nameKr'], 'company': '국제공통', 'source': '적부도', } for e in entries if e['abbreviation'] ] record = { 'id': next_id, 'abbreviation': primary['abbreviation'], 'nameKr': primary['nameKr'], 'nameEn': primary['nameEn'], 'synonymsKr': synonyms_kr, 'synonymsEn': synonyms_en, 'unNumber': primary['unNumber'], 'casNumber': primary['casNumber'], 'transportMethod': '', 'sebc': '', # 물리·화학 (OCR 단계에서 채움) 'usage': primary['usage'], 'state': '', 'color': '', 'odor': '', 'flashPoint': '', 'autoIgnition': '', 'boilingPoint': '', 'density': '', 'solubility': '', 'vaporPressure': '', 'vaporDensity': '', 'explosionRange': '', # 위험도 'nfpa': {'health': 0, 'fire': 0, 'reactivity': 0, 'special': ''}, 'hazardClass': '', 'ergNumber': '', 'idlh': '', 'aegl2': '', 'erpg2': '', # 방제 'responseDistanceFire': '', 'responseDistanceSpillDay': '', 'responseDistanceSpillNight': '', 'marineResponse': '', 'ppeClose': '', 'ppeFar': '', # MSDS 'msds': { 'hazard': '', 'firstAid': '', 'fireFighting': '', 'spillResponse': '', 'exposure': '', 'regulation': '', }, # IBC 'ibcHazard': ibc.get('ibcHazard', ''), 'ibcShipType': ibc.get('ibcShipType', ''), 'ibcTankType': ibc.get('ibcTankType', ''), 'ibcDetection': ibc.get('ibcDetection', ''), 'ibcFireFighting': ibc.get('ibcFireFighting', ''), 'ibcMinRequirement': ibc.get('ibcMinRequirement', ''), # EmS (OCR에서 채움) 'emsCode': '', 'emsFire': '', 'emsSpill': '', 'emsFirstAid': '', # cargoCodes / portFrequency 'cargoCodes': cargo_codes, 'portFrequency': [], } records.append(record) next_id += 1 print(f'[통합] 그룹화 결과 {len(records)}종 (화물적부도 {len(cargo_rows)}행 기준)') # 저장 out_path = OUT_DIR / 'base.json' with open(out_path, 'w', encoding='utf-8') as f: json.dump(records, f, ensure_ascii=False, indent=2) print(f'[완료] {out_path} ({out_path.stat().st_size / 1024:.0f} KB)') if __name__ == '__main__': main()