237 lines
8.1 KiB
Python
237 lines
8.1 KiB
Python
"""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()
|