Merge develop into feature/fix-build-error

# Conflicts:
#	docs/RELEASE-NOTES.md
This commit is contained in:
dnlee 2026-04-17 13:27:53 +09:00
커밋 846c63eae9
18개의 변경된 파일16739개의 추가작업 그리고 57980개의 파일을 삭제

3
.gitignore vendored
파일 보기

@ -79,6 +79,9 @@ prediction/image/**/*.pth
frontend/public/hns-manual/pages/ frontend/public/hns-manual/pages/
frontend/public/hns-manual/images/ frontend/public/hns-manual/images/
# HNS import pipeline outputs (local, 1회성 생성물)
backend/scripts/hns-import/out/
# Claude Code (team workflow tracked, override global gitignore) # Claude Code (team workflow tracked, override global gitignore)
!.claude/ !.claude/
.claude/settings.local.json .claude/settings.local.json

파일 보기

@ -22,6 +22,7 @@
"pg": "^8.19.0" "pg": "^8.19.0"
}, },
"devDependencies": { "devDependencies": {
"@anthropic-ai/sdk": "^0.89.0",
"@types/bcrypt": "^6.0.0", "@types/bcrypt": "^6.0.0",
"@types/cookie-parser": "^1.4.10", "@types/cookie-parser": "^1.4.10",
"@types/cors": "^2.8.17", "@types/cors": "^2.8.17",
@ -34,6 +35,37 @@
"typescript": "^5.7.3" "typescript": "^5.7.3"
} }
}, },
"node_modules/@anthropic-ai/sdk": {
"version": "0.89.0",
"resolved": "https://registry.npmjs.org/@anthropic-ai/sdk/-/sdk-0.89.0.tgz",
"integrity": "sha512-nyGau0zex62EpU91hsHa0zod973YEoiMgzWZ9hC55WdiOLrE4AGpcg4wXI7lFqtvMLqMcLfewQU9sHgQB6psow==",
"dev": true,
"license": "MIT",
"dependencies": {
"json-schema-to-ts": "^3.1.1"
},
"bin": {
"anthropic-ai-sdk": "bin/cli"
},
"peerDependencies": {
"zod": "^3.25.0 || ^4.0.0"
},
"peerDependenciesMeta": {
"zod": {
"optional": true
}
}
},
"node_modules/@babel/runtime": {
"version": "7.29.2",
"resolved": "https://registry.npmjs.org/@babel/runtime/-/runtime-7.29.2.tgz",
"integrity": "sha512-JiDShH45zKHWyGe4ZNVRrCjBz8Nh9TMmZG1kh4QTK8hCBTWBi8Da+i7s1fJw7/lYpM4ccepSNfqzZ/QvABBi5g==",
"dev": true,
"license": "MIT",
"engines": {
"node": ">=6.9.0"
}
},
"node_modules/@esbuild/aix-ppc64": { "node_modules/@esbuild/aix-ppc64": {
"version": "0.27.3", "version": "0.27.3",
"resolved": "https://registry.npmjs.org/@esbuild/aix-ppc64/-/aix-ppc64-0.27.3.tgz", "resolved": "https://registry.npmjs.org/@esbuild/aix-ppc64/-/aix-ppc64-0.27.3.tgz",
@ -1689,6 +1721,20 @@
"bignumber.js": "^9.0.0" "bignumber.js": "^9.0.0"
} }
}, },
"node_modules/json-schema-to-ts": {
"version": "3.1.1",
"resolved": "https://registry.npmjs.org/json-schema-to-ts/-/json-schema-to-ts-3.1.1.tgz",
"integrity": "sha512-+DWg8jCJG2TEnpy7kOm/7/AxaYoaRbjVB4LFZLySZlWn8exGs3A4OLJR966cVvU26N7X9TWxl+Jsw7dzAqKT6g==",
"dev": true,
"license": "MIT",
"dependencies": {
"@babel/runtime": "^7.18.3",
"ts-algebra": "^2.0.0"
},
"engines": {
"node": ">=16"
}
},
"node_modules/jsonwebtoken": { "node_modules/jsonwebtoken": {
"version": "9.0.3", "version": "9.0.3",
"resolved": "https://registry.npmjs.org/jsonwebtoken/-/jsonwebtoken-9.0.3.tgz", "resolved": "https://registry.npmjs.org/jsonwebtoken/-/jsonwebtoken-9.0.3.tgz",
@ -2618,6 +2664,13 @@
"node": ">=0.6" "node": ">=0.6"
} }
}, },
"node_modules/ts-algebra": {
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/ts-algebra/-/ts-algebra-2.0.0.tgz",
"integrity": "sha512-FPAhNPFMrkwz76P7cdjdmiShwMynZYN6SgOujD1urY4oNm80Ou9oMdmbR45LotcKOXoy7wSmHkRFE6Mxbrhefw==",
"dev": true,
"license": "MIT"
},
"node_modules/tsx": { "node_modules/tsx": {
"version": "4.21.0", "version": "4.21.0",
"resolved": "https://registry.npmjs.org/tsx/-/tsx-4.21.0.tgz", "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.21.0.tgz",

파일 보기

@ -23,6 +23,7 @@
"pg": "^8.19.0" "pg": "^8.19.0"
}, },
"devDependencies": { "devDependencies": {
"@anthropic-ai/sdk": "^0.89.0",
"@types/bcrypt": "^6.0.0", "@types/bcrypt": "^6.0.0",
"@types/cookie-parser": "^1.4.10", "@types/cookie-parser": "^1.4.10",
"@types/cors": "^2.8.17", "@types/cors": "^2.8.17",

파일 보기

@ -0,0 +1,140 @@
# HNS 물질 Import 파이프라인
`C:\Projects\MeterialDB\유해물질 화물적부도 검색툴.xlsm` 외부 자료를 `HNS_SUBSTANCE` DB로 변환하는 1회성 파이프라인.
## 파이프라인 구조
```
[Excel xlsm] [PDF 물질정보집 (193종)]
├─ (1) extract-excel.py (2b) extract-pdf.py
│ → out/base.json → out/pdf-data.json
└─ (2a) extract-images.py ──────────────────────────┐
→ out/images/*.png │
↓ │
(3) ocr-images.ts │
→ out/ocr.json │
↓ ↓
(4) merge-data.ts ←──────────────┘
→ frontend/src/data/hnsSubstanceData.json
(5) tsx src/db/seedHns.ts
→ HNS_SUBSTANCE 테이블
```
**병합 우선순위**: `pdf-data.json` > `base.json` > `ocr.json`
## 전제 조건
- Python 3.9+ with `openpyxl`, `PyMuPDF(fitz)`
- Node.js 20
- `ANTHROPIC_API_KEY` 환경변수 (Claude Vision API, OCR 실행 시에만 필요)
- Excel 원본: `C:\Projects\MeterialDB\유해물질 화물적부도 검색툴.xlsm`
- PDF 원본: `C:\Projects\MeterialDB\해상화학사고_대응_물질정보집.pdf`
## 실행 순서
### 1) Excel 메타 시트 파싱
```bash
cd backend
python scripts/hns-import/extract-excel.py
```
- 입력: `C:\Projects\MeterialDB\유해물질 화물적부도 검색툴.xlsm`
- 처리 시트: `화물적부도 화물코드`(1,345개), `동의어`(215개), `IBC CODE`(분류)
- 출력: `scripts/hns-import/out/base.json`
### 2a) 이미지 225개 추출 (선택 — OCR 실행 시만 필요)
```bash
python scripts/hns-import/extract-images.py
```
- 출력: `out/images/{nameKr}.png`, `out/image-map.json`
### 2b) PDF 물질정보집 파싱 ★ 권장
```bash
python scripts/hns-import/extract-pdf.py
```
- 입력: `C:\Projects\MeterialDB\해상화학사고_대응_물질정보집.pdf`
- 해양경찰청 발행 193종, 텍스트 직접 추출 (OCR 불필요)
- 출력: `scripts/hns-import/out/pdf-data.json`
### 3) Claude Vision OCR (선택 — pdf-data.json 없을 때 보조)
```bash
export ANTHROPIC_API_KEY="sk-ant-..."
cd backend
npx tsx scripts/hns-import/ocr-images.ts
```
- 이미지 한 장당 Claude API 1회 호출, 동시 5개 병렬
- 출력: `out/ocr.json` `{ [nameKr]: OcrResult }`
### 4) 최종 JSON 병합
```bash
cd backend
npx tsx scripts/hns-import/merge-data.ts
```
- 입력: `out/base.json` + `out/pdf-data.json` + `out/ocr.json` (없으면 건너뜀)
- 출력: `frontend/src/data/hnsSubstanceData.json` (전량 덮어쓰기)
### 5) DB 재시드
```bash
cd backend
npx tsx src/db/seedHns.ts
```
- 기존 `DELETE FROM HNS_SUBSTANCE` → 새 514종 INSERT
## 빠른 재실행 (PDF 추출 → 병합 → 시드)
```bash
cd backend
python scripts/hns-import/extract-pdf.py && \
npx tsx scripts/hns-import/merge-data.ts && \
npx tsx src/db/seedHns.ts
```
## 현재 데이터 현황 (2024-04 기준)
| 항목 | 이전 (OCR) | 현재 (PDF) |
|------|-----------|-----------|
| 총 물질 수 | 514종 | 514종 |
| 상세정보 보유 (인화점 있음) | 152종 | 195종 |
| NFPA 코드 있음 | ~150종 | 201종 |
| CAS 번호 있음 | ~380종 | 504종 |
| 해양거동 있음 | ~0종 | 206종 |
| 한국어 유사명 있음 | ~200종 | 449종 |
## 재실행 안내
- `out/` 디렉토리는 `.gitignore` 처리되어 커밋되지 않음
- OCR 결과는 비결정적이므로 재실행 시 약간 달라질 수 있음
- PDF 추출은 결정적(동일 입력 → 동일 출력)이므로 재실행 안전
## 알려진 이슈
### 1) PDF 매칭 실패 35종
PDF 국문명과 base.json 국문명이 달라 매칭되지 않는 항목이 35개 존재.
`out/pdf-unmatched.json`에서 목록 확인 가능. 해당 항목은 OCR 데이터로 보조.
**원인:**
- 영문제품명이 국문명으로 등록된 경우 (예: `DER 383 Epoxy resin``디이알 383`)
- 동일 CAS 충돌 (예: `컨덴세이트``나프타`가 같은 CAS)
- 표기 차이 (예: `아이소파-G``아이소파 G`)
### 2) 2열 레이아웃 파싱 노이즈 (약 9건)
PDF 물질특성 블록이 2열로 구성되어 있어, 일부 항목에서 비중 값이 온도값으로 오추출될 수 있음.
영향 범위 최소 (벤젠 등 9종, 값이 100 이상이면 의심).
### 3) SEBC/CAS/UN 번호 varchar 길이 초과
`base.json` 생성 시 Excel에서 복수 CAS/UN 번호를 줄바꿈으로 결합해 저장하여, `HNS_SUBSTANCE` 테이블의 `VARCHAR(20)` 등 제약을 초과했음. 현재는 [`seedHns.ts`](../../../backend/src/db/seedHns.ts) 의 `firstToken()` 헬퍼로 첫 토큰만 검색 컬럼에 저장하고 원본 전체는 `DATA` JSONB에 보존.

파일 보기

@ -0,0 +1,236 @@
"""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()

파일 보기

@ -0,0 +1,170 @@
"""물질별 시트에서 메인 카드 이미지(100KB+) 추출.
엑셀 워크시트 drawing image 관계 체인을 추적해
물질 시트의 핵심 이미지만 out/images/{nameKr}.png 저장.
동시에 out/image-map.json 생성 (파일명 시트명/국문명 매핑).
"""
from __future__ import annotations
import io
import json
import os
import re
import sys
import zipfile
from pathlib import Path
from xml.etree import ElementTree as ET
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
SCRIPT_DIR = Path(__file__).parent.resolve()
OUT_DIR = SCRIPT_DIR / 'out'
IMG_DIR = OUT_DIR / 'images'
IMG_DIR.mkdir(parents=True, exist_ok=True)
SOURCE_XLSX = Path(os.environ.get(
'HNS_SOURCE_XLSX',
r'C:\Projects\MeterialDB\유해물질 화물적부도 검색툴.xlsm',
))
NS = {
'm': 'http://schemas.openxmlformats.org/spreadsheetml/2006/main',
'r': 'http://schemas.openxmlformats.org/officeDocument/2006/relationships',
'pr': 'http://schemas.openxmlformats.org/package/2006/relationships',
}
# 메타 시트(데이터 시트)는 스킵
SKIP_SHEETS = {
'화물적부도 화물코드',
'항구별 코드',
'동의어',
'IBC CODE',
'경계선',
}
# 지침서 번호 시트(115~171) 패턴: 순수 숫자
SKIP_PATTERN = re.compile(r'^\d{3}$')
# 최소 이미지 크기 (주요 카드만 대상, 작은 아이콘 제외)
MIN_IMAGE_SIZE = 50_000 # 50 KB
def safe_filename(name: str) -> str:
name = name.strip().rstrip(',').strip()
name = re.sub(r'[<>:"/\\|?*]', '_', name)
return name
def norm_path(p: str) -> str:
return os.path.normpath(p).replace(os.sep, '/')
def main() -> None:
print(f'[읽기] {SOURCE_XLSX}')
if not SOURCE_XLSX.exists():
raise SystemExit(f'소스 파일 없음: {SOURCE_XLSX}')
image_map: dict[str, dict] = {}
saved = 0
skipped = 0
missing = 0
with zipfile.ZipFile(SOURCE_XLSX) as z:
# 1) workbook.xml → sheet 목록
with z.open('xl/workbook.xml') as f:
wb_root = ET.parse(f).getroot()
sheets = []
for s in wb_root.findall('m:sheets/m:sheet', NS):
sheets.append({
'name': s.get('name'),
'rid': s.get('{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id'),
})
with z.open('xl/_rels/workbook.xml.rels') as f:
rels_root = ET.parse(f).getroot()
rid_target = {r.get('Id'): r.get('Target') for r in rels_root.findall('pr:Relationship', NS)}
for s in sheets:
s['target'] = rid_target.get(s['rid'])
print(f'[시트] 총 {len(sheets)}')
for s in sheets:
name = s['name']
if name in SKIP_SHEETS or SKIP_PATTERN.match(name or ''):
skipped += 1
continue
sheet_file = 'xl/' + s['target']
rels_file = os.path.dirname(sheet_file) + '/_rels/' + os.path.basename(sheet_file) + '.rels'
try:
with z.open(rels_file) as f:
srels = ET.parse(f).getroot()
except KeyError:
missing += 1
continue
# 시트 → drawing
drawing_rel = None
for r in srels.findall('pr:Relationship', NS):
t = r.get('Target') or ''
if 'drawing' in (r.get('Type') or '').lower() and 'drawings/' in t:
drawing_rel = t
break
if not drawing_rel:
missing += 1
continue
drawing_path = norm_path(os.path.join(os.path.dirname(sheet_file), drawing_rel))
drawing_rels_path = os.path.dirname(drawing_path) + '/_rels/' + os.path.basename(drawing_path) + '.rels'
try:
with z.open(drawing_rels_path) as f:
drels = ET.parse(f).getroot()
except KeyError:
missing += 1
continue
# drawing → images
image_paths: list[str] = []
for r in drels.findall('pr:Relationship', NS):
t = r.get('Target') or ''
if 'image' in t.lower():
img_path = norm_path(os.path.join(os.path.dirname(drawing_path), t))
image_paths.append(img_path)
if not image_paths:
missing += 1
continue
# 가장 큰 이미지 선택 (실제 카드 이미지는 100KB+, 아이콘은 수 KB)
sized = [(z.getinfo(p).file_size, p) for p in image_paths]
sized.sort(reverse=True)
largest_size, largest_path = sized[0]
if largest_size < MIN_IMAGE_SIZE:
missing += 1
continue
# 저장
safe = safe_filename(name)
ext = os.path.splitext(largest_path)[1].lower() or '.png'
out_name = f'{safe}{ext}'
out_path = IMG_DIR / out_name
with z.open(largest_path) as fin, open(out_path, 'wb') as fout:
fout.write(fin.read())
image_map[out_name] = {
'sheetName': name,
'nameKr': safe,
'source': largest_path,
'sizeBytes': largest_size,
}
saved += 1
if saved % 25 == 0:
print(f' {saved}개 저장 완료')
print(f'\n[결과] 저장 {saved} / 스킵(메타) {skipped} / 이미지없음 {missing}')
map_path = OUT_DIR / 'image-map.json'
with open(map_path, 'w', encoding='utf-8') as f:
json.dump(image_map, f, ensure_ascii=False, indent=2)
print(f'[완료] 매핑 파일: {map_path}')
if __name__ == '__main__':
main()

파일 보기

@ -0,0 +1,707 @@
"""PDF 물질정보집 → pdf-data.json 변환.
원본: C:\\Projects\\MeterialDB\\해상화학사고_대응_물질정보집.pdf
해양경찰청 발행 193 물질 정보
PDF 구조:
- 페이지 1-21: 표지/머리말/목차
- 페이지 22-407: 193 × 2페이지 물질 카드
- 요약 카드 (홀수 순서): 인화점·발화점·증기압·증기밀도·폭발범위·NFPA·해양거동
- 상세 카드 (짝수 순서): 유사명·CAS·UN·GHS분류·물질특성·인체유해성·응급조치
- 물질 NO(1-193) 0-인덱스 시작 페이지: 21 + (NO-1) * 2
출력: out/pdf-data.json
{ [nameKr]: OcrResult } merge-data.ts 동일한 구조
"""
from __future__ import annotations
import io
import json
import os
import re
import sys
from pathlib import Path
import fitz # PyMuPDF
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)
PDF_PATH = Path(os.environ.get(
'HNS_PDF_PATH',
r'C:\Projects\MeterialDB\해상화학사고_대응_물질정보집.pdf',
))
# 전각 문자 → 반각 변환 테이블
_FULLWIDTH = str.maketrans(
'()tC°℃ ,',
'()tC℃℃ ,',
)
def clean(s: str) -> str:
"""텍스트 정리."""
if not s:
return ''
s = s.translate(_FULLWIDTH)
# 온도 기호 통일: 仁/七/부/사 → ℃ (OCR 오인식)
s = re.sub(r'(?<=[0-9])\s*[仁七부사   ](?=\s|$|이|이하)', '', s)
s = re.sub(r'(?<=[0-9])\s*[tT](?=\s|$|이|이하)', '', s)
s = re.sub(r'\s+', ' ', s)
return s.strip()
def norm_key(s: str) -> str:
"""정규화 키: 공백/특수문자 제거 + 소문자."""
if not s:
return ''
return re.sub(r'[\s,./\-_()\[\]··]+', '', s).lower()
def normalize_cas(raw: str) -> str:
"""CAS 번호 정규화: OCR 노이즈 제거 후 X-XX-X 형식 반환."""
if not raw:
return ''
# 혼합물
if '혼합물' in raw:
return ''
# 특수 대시 → -
s = raw.replace('', '-').replace('', '-').replace('', '-')
# OCR 오인식: 이,이, 공백 등 → 0
s = re.sub(r'[이oO]', '0', s)
s = re.sub(r'["\'\s ]', '', s) # 잡자 제거
# CAS 포맷 검증 후 반환
m = re.match(r'^(\d{2,7}-\d{2}-\d)$', s)
if m:
return m.group(1).lstrip('0') or '0' # 앞자리 0 제거
# 완전히 일치 안 하면 숫자+대시만 남기고 검증
s2 = re.sub(r'[^0-9\-]', '', s)
m2 = re.match(r'^(\d{2,7}-\d{2}-\d)$', s2)
if m2:
return m2.group(1).lstrip('0') or '0'
return ''
def find_cas_in_text(text: str) -> str:
"""텍스트에서 CAS 번호 패턴 검색."""
# 표준 CAS 패턴: 숫자-숫자2자리-숫자1자리
candidates = re.findall(r'\b(\d{1,7}[\-—-\s]{1,2}\d{2}[\-—-\s]{1,2}\d)\b', text)
for c in candidates:
cas = normalize_cas(c)
if cas and len(cas) >= 5:
return cas
return ''
def parse_nfpa(text: str) -> dict | None:
"""NFPA 코드 파싱: '건강 : 3 화재 : 0 반응 : 1' 형태."""
m = re.search(r'건강\s*[:]\s*(\d)\s*화재\s*[:]\s*(\d)\s*반응\s*[:]\s*(\d)', text)
if m:
return {
'health': int(m.group(1)),
'fire': int(m.group(2)),
'reactivity': int(m.group(3)),
'special': '',
}
# 대안 패턴: 줄바꿈 포함
m2 = re.search(r'건강\s*[:]\s*(\d).*?화재\s*[:]\s*(\d).*?반응\s*[:]\s*(\d)', text, re.DOTALL)
if m2:
return {
'health': int(m2.group(1)),
'fire': int(m2.group(2)),
'reactivity': int(m2.group(3)),
'special': '',
}
return None
def extract_field_after(text: str, label: str, max_chars: int = 80) -> str:
"""레이블 직후 값 추출 (단순 패턴)."""
idx = text.find(label)
if idx < 0:
return ''
snippet = text[idx + len(label): idx + len(label) + max_chars + 50]
# 첫 비공백 줄 추출
lines = snippet.split('\n')
for line in lines:
v = clean(line)
if v and v not in (':', '', ''):
return v[:max_chars]
return ''
def parse_summary_card(text: str, index_entry: dict) -> dict:
"""요약 카드(첫 번째 페이지) 파싱."""
result: dict = {}
# 인화점
m = re.search(r'인화점\s*\n([^\n화발증폭위※]+)', text)
if m:
val = clean(m.group(1))
if val and '위험' not in val and len(val) < 40:
result['flashPoint'] = val
# 발화점
m = re.search(r'발화점\s*\n([^\n화발증폭위※인]+)', text)
if m:
val = clean(m.group(1))
if val and len(val) < 40:
result['autoIgnition'] = val
# 증기압 (요약 카드에서는 값이 더 명확하게 나옴)
m = re.search(r'(?:증기압|흥기압)\s*\n?([^\n증기밀도폭발인화발화]+)', text)
if m:
val = clean(m.group(1))
# 파편화된 텍스트 제거
if val and re.search(r'\d', val) and len(val) < 60:
result['vaporPressure'] = val
# 증기밀도 숫자값
m = re.search(r'증기밀도\s*\n?([0-9][^\n]{0,20})', text)
if m:
val = clean(m.group(1))
if val and len(val) < 20:
result['vaporDensity'] = val
# 폭발범위 (2열 레이아웃으로 값이 레이블에서 멀리 떨어질 수 있어 전문 탐색도 병행)
m = re.search(r'폭발범위\s*\n([^\n위험인화발화※]+)', text)
if m:
val = clean(m.group(1))
if val and '%' in val and len(val) < 30:
result['explosionRange'] = val
# 2열 레이아웃 폴백: 텍스트 전체에서 "숫자~숫자%" 패턴 검색
if not result.get('explosionRange'):
m = re.search(r'(\d+[\.,]?\d*\s*~\s*\d+[\.,]?\d*\s*%)', text)
if m:
result['explosionRange'] = clean(m.group(1))
# 화재시 대피거리
m = re.search(r'화재시\s*대피거리\s*\n?([^\n]+)', text)
if m:
val = clean(m.group(1))
if val:
result['responseDistanceFire'] = val
# 해양거동
m = re.search(r'해양거동\s*\n([^\n상온이격방호방제]+)', text)
if not m:
m = re.search(r'해양거동\s+([^\n]+)', text)
if m:
val = clean(m.group(1))
if val and len(val) < 80:
result['marineResponse'] = val
# 상온상태
m = re.search(r'상온상태\s*\n([^\n이격방호비중색상휘발냄새]+)', text)
if m:
val = clean(m.group(1))
if val and len(val) < 60:
result['state'] = val
# 냄새
m = re.search(r'냄새\s*\n([^\n이격방호색상비중상온휘발]+)', text)
if m:
val = clean(m.group(1))
if val and len(val) < 60:
result['odor'] = val
# 비중
m = re.search(r'비중\s*\n[^\n]*\n([0-9][^\n]{0,25})', text)
if not m:
m = re.search(r'비중\s*\n([0-9][^\n]{0,25})', text)
if m:
val = clean(m.group(1))
if val and len(val) < 30:
result['density'] = val
# 색상
m = re.search(r'색상\s*\n([^\n이격방호냄새비중상온휘발]+)', text)
if m:
val = clean(m.group(1))
if val and len(val) < 40:
result['color'] = val
# 이격거리 / 방호거리 거리 숫자 추출
m_hot = re.search(r'(?:이격거리|Hot\s*Zone).*?\n([^\n방호거리]+(?:\d+m|반경[^\n]+))', text, re.IGNORECASE)
if m_hot:
result['responseDistanceSpillDay'] = clean(m_hot.group(1))
m_warm = re.search(r'(?:방호거리|Warm\s*Zone).*?\n([^\n이격거리]+(?:\d+m|방향[^\n]+))', text, re.IGNORECASE)
if m_warm:
result['responseDistanceSpillNight'] = clean(m_warm.group(1))
return result
def parse_detail_card(text: str) -> dict:
"""상세 카드(두 번째 페이지) 파싱."""
result: dict = {}
# ── nameKr 헤더에서 추출 ──────────────────────────────────────────
# 형식: "001 과산화수소" or "0이 과산화수소"
first_lines = text.strip().split('\n')[:4]
for line in first_lines:
line = line.strip()
# 숫자/OCR숫자로 시작하고 뒤에 한글이 오는 패턴
m = re.match(r'^[0-9이이아오-]{2,3}\s+([\w\s\-,./()]+)$', line)
if m:
candidate = clean(m.group(1).strip())
if candidate and re.search(r'[가-힣A-Za-z]', candidate):
result['nameKr'] = candidate
break
# ── 분류 ──────────────────────────────────────────────────────────
m = re.search(r'(?:유해액체물질|위험물질|석유\s*및|해양환경관리법)[^\n]{0,60}', text)
if m:
result['hazardClass'] = clean(m.group(0))
# ── 물질요약 ───────────────────────────────────────────────────────
# 물질요약 레이블 이후 ~ 유사명/CAS 번호 전까지
m = re.search(r'(?:물질요약|= *닐으서|진 O야)(.*?)(?=유사명|CAS|$)', text, re.DOTALL)
if not m:
# 분류값 이후 ~ 유사명 전
m = re.search(r'(?:유해액체물질|석유 및)[^\n]*\n(.*?)(?=유사명|CAS)', text, re.DOTALL)
if m:
summary = re.sub(r'\s+', ' ', m.group(1)).strip()
if summary and len(summary) > 15:
result['materialSummary'] = summary[:500]
# ── 유사명 ─────────────────────────────────────────────────────────
m = re.search(r'유사명\s*\n?(.*?)(?=CAS|UN\s*번호|\d{4,7}-\d{2}-\d|분자식|$)', text, re.DOTALL)
if m:
synonyms_raw = re.sub(r'\s+', ' ', m.group(1)).strip()
# CAS 번호 형태면 제외
if synonyms_raw and not re.match(r'^\d{4,7}-\d{2}-\d', synonyms_raw) and len(synonyms_raw) < 300:
result['synonymsKr'] = synonyms_raw
# ── CAS 번호 ────────────────────────────────────────────────────────
# 1순위: "CAS번호" / "CAS 번호" 직후 줄
m = re.search(r'CAS\s*번호\s*\n\s*([^\n분자NFPA용도인화발화물질]+)', text)
if not m:
m = re.search(r'CAS\s*번호\s*([0-9][^\n분자NFPA용도인화발화물질]{4,20})', text)
if m:
cas = normalize_cas(m.group(1).strip().split()[0])
if cas:
result['casNumber'] = cas
# 2순위: 텍스트 전체에서 CAS 패턴 검색
if not result.get('casNumber'):
cas = find_cas_in_text(text)
if cas:
result['casNumber'] = cas
# ── UN 번호 ─────────────────────────────────────────────────────────
# NFPA 코드 이후 줄에 있는 4자리 숫자
m = re.search(r'(?:UN\s*번호|UN번호)\s*\n?\s*([0-9]{3,4})', text)
if not m:
# NFPA 다음 4자리 숫자
m = re.search(r'반응\s*[:]\s*\d\s*\n\s*([0-9]{3,4})\s*\n', text)
if m:
result['unNumber'] = m.group(1).strip()
# ── NFPA 코드 ───────────────────────────────────────────────────────
nfpa = parse_nfpa(text)
if nfpa:
result['nfpa'] = nfpa
# ── 용도 ────────────────────────────────────────────────────────────
m = re.search(r'용도\s*\n(.*?)(?=물질특성|인체\s*유해|인체유해|흡입노출|보호복|초동|$)', text, re.DOTALL)
if m:
usage = re.sub(r'\s+', ' ', m.group(1)).strip()
# GHS 마크(특수문자 블록) 제거
usage = re.sub(r'<[^>]*>|[♦◆◇△▲▼▽★☆■□●○◐◑]+', '', usage).strip()
if usage and len(usage) < 200:
result['usage'] = usage
# ── 물질특성 블록 ───────────────────────────────────────────────────
props_start = text.find('물질특성')
props_end = text.find('인체 유해성')
if props_end < 0:
props_end = text.find('인체유해성')
if props_end < 0:
props_end = text.find('흡입노출')
props_text = text[props_start:props_end] if 0 <= props_start < props_end else text
# 인화점 (상세) — 단일 알파벳(X/O 등 위험도 마크) 제외, 숫자 포함 값만 허용
m = re.search(r'인화점\s+([^\n발화끓는수용상온]+)', props_text)
if not m:
m = re.search(r'인화점\s*\n\s*([^\n발화끓는수용상온]+)', props_text)
if m:
val = clean(m.group(1))
if val and len(val) < 40 and (re.search(r'\d', val) or re.search(r'없음|해당없음|N/A', val)):
result['flashPoint'] = val
# 발화점 (상세) — 숫자 포함 값만 허용
m = re.search(r'발화점\s+([^\n인화끓는수용상온]+)', props_text)
if not m:
m = re.search(r'발화점\s*\n\s*([^\n인화끓는수용상온]+)', props_text)
if m:
val = clean(m.group(1))
if val and len(val) < 40 and (re.search(r'\d', val) or re.search(r'없음|해당없음|N/A', val)):
result['autoIgnition'] = val
# 끓는점
m = re.search(r'끓는점\s+([^\n인화발화수용상온]+)', props_text)
if not m:
m = re.search(r'끓는점\s*\n\s*([^\n인화발화수용상온]+)', props_text)
if m:
val = clean(m.group(1))
if val and len(val) < 40:
result['boilingPoint'] = val
# 수용해도
m = re.search(r'수용해도\s+([^\n인화발화끓는상온]+)', props_text)
if not m:
m = re.search(r'수용해도\s*\n\s*([^\n인화발화끓는상온]+)', props_text)
if m:
val = clean(m.group(1))
if val and len(val) < 50:
result['solubility'] = val
# 상온상태 (상세)
m = re.search(r'상온상태\s+([^\n색상냄새비중증기인화발화]+)', props_text)
if not m:
m = re.search(r'상온상태\s*\n\s*([^\n색상냄새비중증기인화발화]+)', props_text)
if m:
val = clean(m.group(1)).strip('()')
if val and len(val) < 60:
result['state'] = val
# 색상 (상세)
m = re.search(r'색상\s+([^\n상온냄새비중증기인화발화]+)', props_text)
if not m:
m = re.search(r'색상\s*\n\s*([^\n상온냄새비중증기인화발화]+)', props_text)
if m:
val = clean(m.group(1))
if val and len(val) < 40:
result['color'] = val
# 냄새 (상세)
m = re.search(r'냄새\s+([^\n상온색상비중증기인화발화]+)', props_text)
if not m:
m = re.search(r'냄새\s*\n\s*([^\n상온색상비중증기인화발화]+)', props_text)
if m:
val = clean(m.group(1))
if val and len(val) < 60:
result['odor'] = val
# 비중 (상세)
m = re.search(r'비중\s+([0-9][^\n증기점도휘발]{0,25})', props_text)
if not m:
m = re.search(r'비중\s*\n\s*([0-9][^\n증기점도휘발]{0,25})', props_text)
if m:
val = clean(m.group(1))
if val and len(val) < 30:
result['density'] = val
# 증기압 (상세)
m = re.search(r'증기압\s+([^\n증기밀도점도휘발]{3,40})', props_text)
if not m:
m = re.search(r'증기압\s*\n\s*([^\n증기밀도점도휘발]{3,40})', props_text)
if m:
val = clean(m.group(1))
if val and re.search(r'\d', val):
result['vaporPressure'] = val
# 증기밀도 (상세)
m = re.search(r'증기밀도\s+([0-9,\.][^\n]{0,15})', props_text)
if not m:
m = re.search(r'증기밀도\s*\n\s*([0-9,\.][^\n]{0,15})', props_text)
if m:
val = clean(m.group(1))
if val and len(val) < 20:
result['vaporDensity'] = val
# 점도
m = re.search(r'점도\s+([0-9][^\n]{0,25})', props_text)
if not m:
m = re.search(r'점도\s*\n\s*([0-9][^\n]{0,25})', props_text)
if m:
val = clean(m.group(1))
if val and len(val) < 30:
result['viscosity'] = val
# ── 인체유해성 블록 ─────────────────────────────────────────────────
hazard_start = max(text.find('인체 유해성'), text.find('인체유해성'))
if hazard_start < 0:
hazard_start = text.find('급성독성')
response_start = text.find('초동대응')
hazard_text = text[hazard_start:response_start] if 0 <= hazard_start < response_start else ''
# IDLH
m = re.search(r'I?DLH[^\n]{0,20}\n?\s*([0-9][^\n]{0,20})', hazard_text or text)
if m:
val = clean(m.group(1))
if val and re.search(r'\d', val):
result['idlh'] = val
# TWA
m = re.search(r'TWA[^\n]{0,20}\n?\s*([0-9][^\n]{0,20})', hazard_text or text)
if m:
val = clean(m.group(1))
if val and re.search(r'\d', val):
result['twa'] = val
# ── 응급조치 ─────────────────────────────────────────────────────────
fa_start = text.find('흡입노출')
fa_end = text.find('초동대응')
if fa_start >= 0:
fa_text = text[fa_start: fa_end if fa_end > fa_start else fa_start + 600]
fa = re.sub(r'\s+', ' ', fa_text).strip()
result['msds'] = {
'firstAid': fa[:600],
'spillResponse': '',
'hazard': '',
'fireFighting': '',
'exposure': '',
'regulation': '',
}
# ── 초동대응 - 이격거리/방호거리 (상세카드에서) ─────────────────────
m = re.search(r'초기\s*이격거리[^\n]{0,10}m[^\n]{0,5}\n?\s*([0-9]+)', text)
if m:
result['responseDistanceSpillDay'] = m.group(1) + 'm'
m = re.search(r'방호거리[^\n]{0,10}m[^\n]{0,5}\n?\s*([0-9]+)', text)
if m:
result['responseDistanceSpillNight'] = m.group(1) + 'm'
m = re.search(r'화재\s*시\s*대피거리[^\n]{0,10}m[^\n]{0,5}\n?\s*([0-9]+)', text)
if m:
result['responseDistanceFire'] = m.group(1) + 'm'
# ── GHS 분류 ─────────────────────────────────────────────────────────
ghs_items = re.findall(
r'(?:인화성[^\n(]{2,40}|급성독성[^\n(]{2,40}|피부부식[^\n(]{2,40}|'
r'\s*손상[^\n(]{2,40}|발암성[^\n(]{2,40}|생식독성[^\n(]{2,40}|'
r'수생환경[^\n(]{2,40}|특정표적[^\n(]{2,40}|흡인유해[^\n(]{2,40})',
text,
)
if ghs_items:
result['ghsClass'] = ' / '.join(clean(g) for g in ghs_items[:6])
return result
def parse_index_pages(pdf: fitz.Document) -> dict[int, dict]:
"""목차 페이지(4-21)에서 NO → {nameKr, nameEn, casNumber} 매핑 구축."""
index: dict[int, dict] = {}
for page_idx in range(3, 21):
page = pdf[page_idx]
text = page.get_text()
lines = [ln.strip() for ln in text.split('\n') if ln.strip()]
for i, line in enumerate(lines):
if not re.match(r'^\d{1,3}$', line):
continue
no = int(line)
if not (1 <= no <= 193):
continue
if no in index:
continue
# 탐색 창: NO 앞 1~4줄
cas, name_en, name_kr = '', '', ''
if i >= 1:
# CAS 줄: 숫자-숫자-숫자 패턴 (OCR 노이즈 허용)
raw_cas = lines[i - 1]
cas = normalize_cas(raw_cas) if re.match(r'^[0-9이이아oO-\-—-"\'\. ]{5,30}$|혼합물', raw_cas) else ''
if not cas and '혼합물' in raw_cas:
cas = '혼합물'
if cas or '혼합물' in (lines[i - 1] if i >= 1 else ''):
if i >= 2:
name_en = lines[i - 2]
if i >= 3:
name_kr = lines[i - 3]
elif i >= 2:
# CAS가 없는 경우(매칭 실패) - 줄 이동해서 재탐색
raw_cas2 = lines[i - 2] if i >= 2 else ''
cas = normalize_cas(raw_cas2) if re.match(r'^[0-9이이아oO-\-—-"\'\. ]{5,30}$|혼합물', raw_cas2) else ''
if cas or '혼합물' in raw_cas2:
name_en = lines[i - 1] if i >= 1 else ''
# name_kr는 찾기 어려움
if not name_kr and i >= 3:
# 이름이 공백/짧으면 더 위 줄에서 찾기
for j in range(3, min(6, i + 1)):
cand = lines[i - j]
if re.search(r'[가-힣]', cand) and len(cand) > 1:
name_kr = cand
break
index[no] = {
'no': no,
'nameKr': name_kr,
'nameEn': name_en,
'casNumber': cas if cas != '혼합물' else '',
}
return index
def extract_name_from_summary(text: str) -> tuple[str, str]:
"""요약 카드에서 nameKr, nameEn 추출."""
name_kr, name_en = '', ''
lines = text.strip().split('\n')
# 1~6번 줄에서 한글 이름 탐색 (헤더 "해상화학사고 대응 물질정보집" 이후)
found_header = False
for line in lines:
line = line.strip()
if not line:
continue
# 제목 줄 건너뜀
if '해상화학사고' in line or '대응' in line or '물질정보집' in line:
found_header = True
continue
# 3자리 번호 줄 건너뜀
if re.match(r'^\d{1,3}$', line):
continue
# 한글이 있으면 nameKr 후보
if re.search(r'[가-힣]', line) and len(line) > 1 and '위험' not in line and '분류' not in line:
if not name_kr:
name_kr = clean(line)
# 영문명: (영문명) 형태
m_en = re.search(r'[(]([A-Za-z][^)]{3,60})[)]', line)
if m_en and not name_en:
name_en = clean(m_en.group(1))
if name_kr and name_en:
break
return name_kr, name_en
def parse_substance(pdf: fitz.Document, no: int, index_entry: dict) -> dict | None:
"""물질 번호 no에 해당하는 2페이지를 파싱하여 통합 레코드 반환."""
start_idx = 21 + (no - 1) * 2
if start_idx + 1 >= pdf.page_count:
return None
summary_text = pdf[start_idx].get_text()
detail_text = pdf[start_idx + 1].get_text()
summary = parse_summary_card(summary_text, index_entry)
detail = parse_detail_card(detail_text)
# nameKr 결정 우선순위: 인덱스 > 상세카드 헤더 > 요약카드
name_kr = index_entry.get('nameKr', '')
if not name_kr:
name_kr = detail.get('nameKr', '')
if not name_kr:
name_kr, _ = extract_name_from_summary(summary_text)
# nameEn
name_en = index_entry.get('nameEn', '')
# 통합: detail 우선, 없으면 summary
merged: dict = {
'nameKr': name_kr,
'nameEn': name_en,
}
for key in ['casNumber', 'unNumber', 'usage', 'synonymsKr',
'flashPoint', 'autoIgnition', 'boilingPoint', 'density', 'solubility',
'vaporPressure', 'vaporDensity', 'volatility', 'explosionRange',
'state', 'color', 'odor', 'viscosity', 'idlh', 'twa',
'responseDistanceFire', 'responseDistanceSpillDay', 'responseDistanceSpillNight',
'marineResponse', 'hazardClass', 'ghsClass', 'materialSummary', 'msds']:
detail_val = detail.get(key)
summary_val = summary.get(key)
if detail_val:
merged[key] = detail_val
elif summary_val:
merged[key] = summary_val
# CAS: 인덱스 우선
if index_entry.get('casNumber') and not merged.get('casNumber'):
merged['casNumber'] = index_entry['casNumber']
# NFPA: detail 우선
if 'nfpa' in detail:
merged['nfpa'] = detail['nfpa']
if 'msds' not in merged:
merged['msds'] = {
'firstAid': '', 'spillResponse': '', 'hazard': '',
'fireFighting': '', 'exposure': '', 'regulation': '',
}
merged['_no'] = no
merged['_pageIdx'] = start_idx
return merged
def main() -> None:
if not PDF_PATH.exists():
raise SystemExit(f'PDF 파일 없음: {PDF_PATH}')
print(f'[읽기] {PDF_PATH}')
pdf = fitz.open(str(PDF_PATH))
print(f'[PDF] 총 {pdf.page_count}페이지')
# 1. 인덱스 파싱
print('[인덱스] 목차 페이지 파싱 중...')
index = parse_index_pages(pdf)
print(f'[인덱스] {len(index)}개 항목 발견')
# 2. 물질 카드 파싱
results: dict[str, dict] = {}
failed: list[int] = []
for no in range(1, 194):
entry = index.get(no, {'no': no, 'nameKr': '', 'nameEn': '', 'casNumber': ''})
try:
rec = parse_substance(pdf, no, entry)
if rec:
name_kr = rec.get('nameKr', '')
if name_kr:
key = name_kr
if key in results:
key = f'{name_kr}_{no}'
results[key] = rec
else:
print(f' [경고] NO={no} nameKr 없음 - 건너뜀')
failed.append(no)
except Exception as e:
print(f' [오류] NO={no}: {e}')
failed.append(no)
pdf.close()
# 3. 저장
out_path = OUT_DIR / 'pdf-data.json'
with open(out_path, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
size_kb = out_path.stat().st_size / 1024
print(f'\n[완료] {out_path} ({size_kb:.0f} KB, {len(results)}종)')
if failed:
print(f'[경고] 파싱 실패 {len(failed)}종: {failed}')
# 4. 통계
with_flash = sum(1 for v in results.values() if v.get('flashPoint'))
with_nfpa = sum(1 for v in results.values() if v.get('nfpa'))
with_cas = sum(1 for v in results.values() if v.get('casNumber'))
with_syn = sum(1 for v in results.values() if v.get('synonymsKr'))
print(f'[통계] 인화점: {with_flash}종, NFPA: {with_nfpa}종, CAS: {with_cas}종, 유사명: {with_syn}')
# 5. 샘플 출력
print('\n[샘플] 주요 항목:')
sample_keys = ['과산화수소', '나프탈렌', '벤젠', '톨루엔']
for k in sample_keys:
if k in results:
v = results[k]
print(f' {k}: fp={v.get("flashPoint","")} nfpa={v.get("nfpa")} cas={v.get("casNumber","")}')
if __name__ == '__main__':
main()

파일 보기

@ -0,0 +1,23 @@
"""배치 JSON을 ocr.json에 병합."""
import json
import sys
from pathlib import Path
SCRIPT_DIR = Path(__file__).parent.resolve()
OUT_DIR = SCRIPT_DIR / 'out'
OCR_PATH = OUT_DIR / 'ocr.json'
BATCH_PATH = OUT_DIR / 'batch.json'
with open(OCR_PATH, encoding='utf-8') as f:
ocr = json.load(f)
with open(BATCH_PATH, encoding='utf-8') as f:
batch = json.load(f)
added = [k for k in batch if k not in ocr]
updated = [k for k in batch if k in ocr]
ocr.update(batch)
with open(OCR_PATH, 'w', encoding='utf-8') as f:
json.dump(ocr, f, ensure_ascii=False, indent=2)
print(f'merged: +{len(added)} added, ~{len(updated)} updated, total {len(ocr)}')

파일 보기

@ -0,0 +1,362 @@
/**
* base.json + pdf-data.json + ocr.json frontend/src/data/hnsSubstanceData.json
*
* 우선순위: pdf-data (PDF , ) > base.json > ocr.json ( OCR, )
* :
* 1. CAS ( )
* 2. (nameKr)
* 3. (synonymsKr)
*/
import { readFileSync, writeFileSync, existsSync } from 'node:fs';
import { resolve, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
const __dirname = dirname(fileURLToPath(import.meta.url));
const OUT_DIR = resolve(__dirname, 'out');
const BASE_PATH = resolve(OUT_DIR, 'base.json');
const PDF_PATH = resolve(OUT_DIR, 'pdf-data.json');
const OCR_PATH = resolve(OUT_DIR, 'ocr.json');
const TARGET_PATH = resolve(__dirname, '../../../frontend/src/data/hnsSubstanceData.json');
function normalizeName(s: string | undefined): string {
if (!s) return '';
return s
.replace(/\s+/g, '')
.replace(/[,.·/\-_()[\]]/g, '')
.toLowerCase();
}
function normalizeCas(s: string | undefined): string {
if (!s) return '';
// 앞자리 0 제거 후 정규화
return s
.replace(/[^0-9\-]/g, '')
.replace(/^0+/, '')
.trim();
}
interface NfpaBlock {
health: number;
fire: number;
reactivity: number;
special: string;
}
interface MsdsBlock {
hazard: string;
firstAid: string;
fireFighting: string;
spillResponse: string;
exposure: string;
regulation: string;
}
interface BaseRecord {
id: number;
abbreviation: string;
nameKr: string;
nameEn: string;
synonymsEn: string;
synonymsKr: string;
unNumber: string;
casNumber: string;
transportMethod: string;
sebc: string;
usage: string;
state: string;
color: string;
odor: string;
flashPoint: string;
autoIgnition: string;
boilingPoint: string;
density: string;
solubility: string;
vaporPressure: string;
vaporDensity: string;
explosionRange: string;
nfpa: NfpaBlock;
hazardClass: string;
ergNumber: string;
idlh: string;
aegl2: string;
erpg2: string;
responseDistanceFire: string;
responseDistanceSpillDay: string;
responseDistanceSpillNight: string;
marineResponse: string;
ppeClose: string;
ppeFar: string;
msds: MsdsBlock;
ibcHazard: string;
ibcShipType: string;
ibcTankType: string;
ibcDetection: string;
ibcFireFighting: string;
ibcMinRequirement: string;
emsCode: string;
emsFire: string;
emsSpill: string;
emsFirstAid: string;
cargoCodes: Array<{ code: string; name: string; company: string; source: string }>;
portFrequency: Array<{ port: string; portCode: string; lastImport: string; frequency: string }>;
}
interface PdfResult {
[key: string]: unknown;
casNumber?: string;
nameKr?: string;
nfpa?: Partial<NfpaBlock>;
msds?: Partial<MsdsBlock>;
}
interface OcrResult {
[key: string]: unknown;
}
function firstString(...values: Array<unknown>): string {
for (const v of values) {
if (typeof v === 'string' && v.trim().length > 0) return v.trim();
}
return '';
}
function pickNfpa(source: PdfResult | OcrResult): NfpaBlock | null {
const n = source.nfpa as Partial<NfpaBlock> | undefined;
if (!n || typeof n !== 'object') return null;
const h = Number(n.health);
const f = Number(n.fire);
const r = Number(n.reactivity);
if ([h, f, r].some((x) => !Number.isFinite(x))) return null;
return {
health: h,
fire: f,
reactivity: r,
special: typeof n.special === 'string' ? n.special : '',
};
}
function pickMsds(
pdf: PdfResult | undefined,
ocr: OcrResult | undefined,
base: MsdsBlock,
): MsdsBlock {
const p = (pdf?.msds ?? {}) as Partial<MsdsBlock>;
const o = (ocr?.msds ?? {}) as Partial<MsdsBlock>;
return {
hazard: firstString(base.hazard, p.hazard, o.hazard),
firstAid: firstString(base.firstAid, p.firstAid, o.firstAid),
fireFighting: firstString(base.fireFighting, p.fireFighting, o.fireFighting),
spillResponse: firstString(base.spillResponse, p.spillResponse, o.spillResponse),
exposure: firstString(base.exposure, p.exposure, o.exposure),
regulation: firstString(base.regulation, p.regulation, o.regulation),
};
}
function merge(
base: BaseRecord,
pdf: PdfResult | undefined,
ocr: OcrResult | undefined,
): BaseRecord {
const nfpaFromPdf = pdf ? pickNfpa(pdf) : null;
const nfpaFromOcr = ocr ? pickNfpa(ocr) : null;
// pdf NFPA 우선, 없으면 ocr, 없으면 base
const nfpa = nfpaFromPdf ?? nfpaFromOcr ?? base.nfpa;
return {
...base,
// pdf > base > ocr 우선순위
unNumber: firstString(pdf?.unNumber, base.unNumber, ocr?.unNumber),
casNumber: firstString(pdf?.casNumber, base.casNumber, ocr?.casNumber),
synonymsKr: firstString(pdf?.synonymsKr, base.synonymsKr, ocr?.synonymsKr),
transportMethod: firstString(base.transportMethod, pdf?.transportMethod, ocr?.transportMethod),
sebc: firstString(base.sebc, pdf?.sebc, ocr?.sebc),
usage: firstString(pdf?.usage, base.usage, ocr?.usage),
state: firstString(pdf?.state, base.state, ocr?.state),
color: firstString(pdf?.color, base.color, ocr?.color),
odor: firstString(pdf?.odor, base.odor, ocr?.odor),
flashPoint: firstString(pdf?.flashPoint, base.flashPoint, ocr?.flashPoint),
autoIgnition: firstString(pdf?.autoIgnition, base.autoIgnition, ocr?.autoIgnition),
boilingPoint: firstString(pdf?.boilingPoint, base.boilingPoint, ocr?.boilingPoint),
density: firstString(pdf?.density, base.density, ocr?.density),
solubility: firstString(pdf?.solubility, base.solubility, ocr?.solubility),
vaporPressure: firstString(pdf?.vaporPressure, base.vaporPressure, ocr?.vaporPressure),
vaporDensity: firstString(pdf?.vaporDensity, base.vaporDensity, ocr?.vaporDensity),
explosionRange: firstString(pdf?.explosionRange, base.explosionRange, ocr?.explosionRange),
nfpa,
hazardClass: firstString(pdf?.hazardClass, base.hazardClass, ocr?.hazardClass),
ergNumber: firstString(base.ergNumber, pdf?.ergNumber, ocr?.ergNumber),
idlh: firstString(pdf?.idlh, base.idlh, ocr?.idlh),
aegl2: firstString(base.aegl2, pdf?.aegl2, ocr?.aegl2),
erpg2: firstString(base.erpg2, pdf?.erpg2, ocr?.erpg2),
responseDistanceFire: firstString(pdf?.responseDistanceFire, base.responseDistanceFire, ocr?.responseDistanceFire),
responseDistanceSpillDay: firstString(pdf?.responseDistanceSpillDay, base.responseDistanceSpillDay, ocr?.responseDistanceSpillDay),
responseDistanceSpillNight: firstString(pdf?.responseDistanceSpillNight, base.responseDistanceSpillNight, ocr?.responseDistanceSpillNight),
marineResponse: firstString(pdf?.marineResponse, base.marineResponse, ocr?.marineResponse),
ppeClose: firstString(base.ppeClose, pdf?.ppeClose, ocr?.ppeClose),
ppeFar: firstString(base.ppeFar, pdf?.ppeFar, ocr?.ppeFar),
msds: pickMsds(pdf, ocr, base.msds),
emsCode: firstString(base.emsCode, pdf?.emsCode, ocr?.emsCode),
emsFire: firstString(base.emsFire, pdf?.emsFire, ocr?.emsFire),
emsSpill: firstString(base.emsSpill, pdf?.emsSpill, ocr?.emsSpill),
emsFirstAid: firstString(base.emsFirstAid, pdf?.emsFirstAid, ocr?.emsFirstAid),
};
}
function main() {
if (!existsSync(BASE_PATH)) {
console.error(`base.json 없음: ${BASE_PATH}`);
console.error('→ extract-excel.py 를 먼저 실행하세요.');
process.exit(1);
}
const base: BaseRecord[] = JSON.parse(readFileSync(BASE_PATH, 'utf-8'));
// PDF 데이터 로드
const pdfRaw: Record<string, PdfResult> = existsSync(PDF_PATH)
? JSON.parse(readFileSync(PDF_PATH, 'utf-8'))
: {};
// OCR 데이터 로드
const ocr: Record<string, OcrResult> = existsSync(OCR_PATH)
? JSON.parse(readFileSync(OCR_PATH, 'utf-8'))
: {};
console.log(
`[입력] base ${base.length}종, pdf ${Object.keys(pdfRaw).length}종, ocr ${Object.keys(ocr).length}`,
);
// ── PDF 인덱스 구축 ─────────────────────────────────────────────────
// 1) nameKr 정규화 인덱스
const pdfByName = new Map<string, PdfResult>();
// 2) CAS 번호 인덱스
const pdfByCas = new Map<string, PdfResult>();
for (const [key, value] of Object.entries(pdfRaw)) {
const normKey = normalizeName(key);
if (normKey) pdfByName.set(normKey, value);
const cas = normalizeCas(value.casNumber);
if (cas) {
if (!pdfByCas.has(cas)) pdfByCas.set(cas, value);
}
}
// ── OCR 인덱스 구축 ─────────────────────────────────────────────────
const ocrByName = new Map<string, OcrResult>();
const ocrNormToOrig = new Map<string, string>();
for (const [key, value] of Object.entries(ocr)) {
const normKey = normalizeName(key);
if (normKey) {
ocrByName.set(normKey, value);
ocrNormToOrig.set(normKey, key);
}
}
// ── 병합 ──────────────────────────────────────────────────────────
let pdfMatchedByName = 0;
let pdfMatchedByCas = 0;
let pdfMatchedBySynonym = 0;
let ocrMatched = 0;
const pdfUnmatched = new Set(Object.keys(pdfRaw));
const ocrUnmatched = new Set(ocrByName.keys());
const merged = base.map((record) => {
let pdfResult: PdfResult | undefined;
let ocrResult: OcrResult | undefined;
// ── PDF 매칭 ────────────────────────────────────────────────────
// 1. CAS 번호 매칭 (가장 정확)
const baseCas = normalizeCas(record.casNumber);
if (baseCas) {
pdfResult = pdfByCas.get(baseCas);
if (pdfResult) {
pdfMatchedByCas++;
const origKey = pdfResult.nameKr as string | undefined;
if (origKey) pdfUnmatched.delete(origKey);
}
}
// 2. nameKr 정규화 매칭
if (!pdfResult) {
const normKr = normalizeName(record.nameKr);
pdfResult = pdfByName.get(normKr);
if (pdfResult) {
pdfMatchedByName++;
const origKey = pdfResult.nameKr as string | undefined;
if (origKey) pdfUnmatched.delete(origKey);
}
}
// 3. synonymsKr 동의어 매칭
if (!pdfResult && record.synonymsKr) {
const synonyms = record.synonymsKr.split(' / ');
for (const syn of synonyms) {
const normSyn = normalizeName(syn);
if (!normSyn) continue;
pdfResult = pdfByName.get(normSyn);
if (pdfResult) {
pdfMatchedBySynonym++;
const origKey = pdfResult.nameKr as string | undefined;
if (origKey) pdfUnmatched.delete(origKey);
break;
}
}
}
// ── OCR 매칭 (PDF 없는 경우 보조) ────────────────────────────────
const normKr = normalizeName(record.nameKr);
const ocrByNameResult = ocrByName.get(normKr);
if (ocrByNameResult) {
ocrResult = ocrByNameResult;
ocrMatched++;
ocrUnmatched.delete(normKr);
}
if (!ocrResult && record.synonymsKr) {
const synonyms = record.synonymsKr.split(' / ');
for (const syn of synonyms) {
const normSyn = normalizeName(syn);
if (!normSyn) continue;
const synOcrResult = ocrByName.get(normSyn);
if (synOcrResult) {
ocrResult = synOcrResult;
ocrMatched++;
ocrUnmatched.delete(normSyn);
break;
}
}
}
return merge(record, pdfResult, ocrResult);
});
// ── 통계 출력 ──────────────────────────────────────────────────────
const pdfTotal = pdfMatchedByCas + pdfMatchedByName + pdfMatchedBySynonym;
console.log(
`[PDF 매칭] 총 ${pdfTotal}종 (CAS: ${pdfMatchedByCas}, 국문명: ${pdfMatchedByName}, 동의어: ${pdfMatchedBySynonym})`,
);
console.log(`[OCR 매칭] ${ocrMatched}`);
if (pdfUnmatched.size > 0) {
const unmatchedList = Array.from(pdfUnmatched).sort();
const unmatchedPath = resolve(OUT_DIR, 'pdf-unmatched.json');
writeFileSync(
unmatchedPath,
JSON.stringify({ count: unmatchedList.length, keys: unmatchedList }, null, 2),
'utf-8',
);
console.warn(
`[경고] PDF 매칭 실패 ${unmatchedList.length}개 → ${unmatchedPath}`,
);
unmatchedList.slice(0, 10).forEach((k) => console.warn(` - ${k}`));
if (unmatchedList.length > 10) console.warn(` ... +${unmatchedList.length - 10}`);
}
writeFileSync(TARGET_PATH, JSON.stringify(merged, null, 2), 'utf-8');
const sizeKb = (JSON.stringify(merged).length / 1024).toFixed(0);
console.log(`[완료] ${TARGET_PATH} (${sizeKb} KB, ${merged.length}종)`);
console.log(` 상세 정보 보유: ${merged.filter((r) => r.flashPoint).length}`);
console.log(` NFPA 있음: ${merged.filter((r) => r.nfpa.health || r.nfpa.fire || r.nfpa.reactivity).length}`);
}
main();

파일 보기

@ -0,0 +1,300 @@
/**
* Claude Vision API HNS JSON .
*
* 입력: out/images/*.png (222)
* 출력: out/ocr.json { [nameKr]: Partial<HNSSearchSubstance> }
*
* 환경변수: ANTHROPIC_API_KEY
* 모델: claude-sonnet-4-5 (Vision + )
* 동시성: 5, 3
*
* ocr.json .
*/
import 'dotenv/config';
import Anthropic from '@anthropic-ai/sdk';
import { readFileSync, readdirSync, writeFileSync, existsSync } from 'node:fs';
import { resolve, dirname, basename, extname } from 'node:path';
import { fileURLToPath } from 'node:url';
const __dirname = dirname(fileURLToPath(import.meta.url));
const OUT_DIR = resolve(__dirname, 'out');
const IMG_DIR = process.env.HNS_OCR_IMG_DIR
? resolve(process.env.HNS_OCR_IMG_DIR)
: resolve(OUT_DIR, 'images');
const OCR_PATH = process.env.HNS_OCR_OUT
? resolve(process.env.HNS_OCR_OUT)
: resolve(OUT_DIR, 'ocr.json');
const FAIL_PATH = process.env.HNS_OCR_FAIL
? resolve(process.env.HNS_OCR_FAIL)
: resolve(OUT_DIR, 'ocr-failures.json');
const OCR_LIMIT = process.env.HNS_OCR_LIMIT ? parseInt(process.env.HNS_OCR_LIMIT, 10) : undefined;
const OCR_ONLY = process.env.HNS_OCR_ONLY
? process.env.HNS_OCR_ONLY.split(',').map((s) => s.trim()).filter(Boolean)
: undefined;
const CONCURRENCY = 5;
const MAX_RETRIES = 3;
const MODEL = process.env.HNS_OCR_MODEL ?? 'claude-sonnet-4-5';
const SYSTEM_PROMPT = `당신은 한국 해양 방제용 HNS 비상대응 카드 이미지를 구조화 JSON으로 추출하는 전문 파서입니다.
릿 :
- 상단: 국문명,
- 물질특성: CAS번호, UN번호, , , (///), , //, //, //, /, NFPA (//), GHS , ERG
- 대응방법: 주요 (PPE /), (EmS F-x), (EmS S-x), ,
- 인체유해성: TWA / STEL / AEGL-2 / IDLH, /// ·
JSON **** . "" null.
(: "80℃", "2,410 mmHg (25℃)").
NFPA // 0~4 . special ( ).
** JSON ** ( ).
:
{
"transportMethod": "",
"state": "",
"color": "",
"odor": "",
"flashPoint": "",
"autoIgnition": "",
"boilingPoint": "",
"density": "",
"solubility": "",
"vaporPressure": "",
"vaporDensity": "",
"explosionRange": "",
"nfpa": { "health": 0, "fire": 0, "reactivity": 0, "special": "" },
"hazardClass": "",
"ergNumber": "",
"idlh": "",
"aegl2": "",
"erpg2": "",
"twa": "",
"stel": "",
"responseDistanceFire": "",
"responseDistanceSpillDay": "",
"responseDistanceSpillNight": "",
"marineResponse": "",
"ppeClose": "",
"ppeFar": "",
"msds": {
"hazard": "",
"firstAid": "",
"fireFighting": "",
"spillResponse": "",
"exposure": "",
"regulation": ""
},
"emsCode": "",
"emsFire": "",
"emsSpill": "",
"emsFirstAid": "",
"sebc": ""
}`;
interface OcrResult {
transportMethod?: string;
state?: string;
color?: string;
odor?: string;
flashPoint?: string;
autoIgnition?: string;
boilingPoint?: string;
density?: string;
solubility?: string;
vaporPressure?: string;
vaporDensity?: string;
explosionRange?: string;
nfpa?: { health: number; fire: number; reactivity: number; special: string };
hazardClass?: string;
ergNumber?: string;
idlh?: string;
aegl2?: string;
erpg2?: string;
twa?: string;
stel?: string;
responseDistanceFire?: string;
responseDistanceSpillDay?: string;
responseDistanceSpillNight?: string;
marineResponse?: string;
ppeClose?: string;
ppeFar?: string;
msds?: {
hazard?: string;
firstAid?: string;
fireFighting?: string;
spillResponse?: string;
exposure?: string;
regulation?: string;
};
emsCode?: string;
emsFire?: string;
emsSpill?: string;
emsFirstAid?: string;
sebc?: string;
}
function loadExisting<T>(path: string, fallback: T): T {
if (!existsSync(path)) return fallback;
try {
return JSON.parse(readFileSync(path, 'utf-8'));
} catch {
return fallback;
}
}
function extractJson(text: string): OcrResult | null {
const cleaned = text.replace(/```json\s*/gi, '').replace(/```\s*$/g, '').trim();
const firstBrace = cleaned.indexOf('{');
const lastBrace = cleaned.lastIndexOf('}');
if (firstBrace < 0 || lastBrace < 0) return null;
try {
return JSON.parse(cleaned.slice(firstBrace, lastBrace + 1));
} catch {
return null;
}
}
async function callVision(client: Anthropic, imagePath: string): Promise<OcrResult> {
const imageData = readFileSync(imagePath).toString('base64');
const ext = extname(imagePath).slice(1).toLowerCase();
const mediaType = (ext === 'jpg' || ext === 'jpeg' ? 'image/jpeg' : 'image/png') as
| 'image/png'
| 'image/jpeg';
const response = await client.messages.create({
model: MODEL,
max_tokens: 4096,
system: [
{
type: 'text',
text: SYSTEM_PROMPT,
cache_control: { type: 'ephemeral' },
},
],
messages: [
{
role: 'user',
content: [
{
type: 'image',
source: { type: 'base64', media_type: mediaType, data: imageData },
},
{
type: 'text',
text: '이 HNS 비상대응 카드 이미지에서 모든 필드를 추출해 JSON으로 반환하세요.',
},
],
},
],
});
const textBlock = response.content.find((b) => b.type === 'text');
if (!textBlock || textBlock.type !== 'text') {
throw new Error('응답에 텍스트 블록 없음');
}
const result = extractJson(textBlock.text);
if (!result) {
throw new Error(`JSON 파싱 실패: ${textBlock.text.slice(0, 200)}`);
}
return result;
}
async function processWithRetry(
client: Anthropic,
imagePath: string,
nameKr: string,
): Promise<OcrResult> {
let lastErr: unknown;
for (let attempt = 1; attempt <= MAX_RETRIES; attempt++) {
try {
return await callVision(client, imagePath);
} catch (err) {
lastErr = err;
const wait = 1000 * Math.pow(2, attempt - 1);
console.warn(`[${nameKr}] 시도 ${attempt} 실패, ${wait}ms 후 재시도: ${String(err).slice(0, 120)}`);
await new Promise((r) => setTimeout(r, wait));
}
}
throw lastErr;
}
async function runPool<T>(items: T[], worker: (item: T, idx: number) => Promise<void>) {
let cursor = 0;
const workers = Array.from({ length: CONCURRENCY }, async () => {
while (cursor < items.length) {
const idx = cursor++;
await worker(items[idx], idx);
}
});
await Promise.all(workers);
}
async function main() {
if (!process.env.ANTHROPIC_API_KEY) {
console.error('ANTHROPIC_API_KEY 환경변수가 없습니다.');
process.exit(1);
}
const client = new Anthropic();
if (!existsSync(IMG_DIR)) {
console.error(`이미지 디렉토리 없음: ${IMG_DIR}`);
process.exit(1);
}
const allImages = readdirSync(IMG_DIR).filter((f) => /\.(png|jpg|jpeg)$/i.test(f));
const images = OCR_ONLY
? allImages.filter((f) => OCR_ONLY.includes(basename(f, extname(f))))
: allImages;
const existing: Record<string, OcrResult> = loadExisting(OCR_PATH, {});
const failures: Record<string, string> = loadExisting(FAIL_PATH, {});
let pending = images.filter((f) => {
const nameKr = basename(f, extname(f));
return !(nameKr in existing);
});
if (OCR_LIMIT && Number.isFinite(OCR_LIMIT)) {
pending = pending.slice(0, OCR_LIMIT);
}
console.log(`[OCR] 전체 ${allImages.length}개 중 대상 ${images.length}개, 이미 처리 ${Object.keys(existing).length}개, 이번 실행 ${pending.length}`);
console.log(`[모델] ${MODEL}, 동시 ${CONCURRENCY}, 재시도 최대 ${MAX_RETRIES}`);
console.log(`[출력] ${OCR_PATH}`);
let done = 0;
let failed = 0;
await runPool(pending, async (file, idx) => {
const nameKr = basename(file, extname(file));
const path = resolve(IMG_DIR, file);
try {
const result = await processWithRetry(client, path, nameKr);
existing[nameKr] = result;
delete failures[nameKr];
done++;
if (done % 10 === 0 || done === pending.length) {
writeFileSync(OCR_PATH, JSON.stringify(existing, null, 2), 'utf-8');
console.log(` 진행 ${done}/${pending.length} (실패 ${failed}) - 중간 저장`);
}
} catch (err) {
failed++;
failures[nameKr] = String(err).slice(0, 500);
console.error(`[실패] ${nameKr}: ${String(err).slice(0, 200)}`);
}
});
writeFileSync(OCR_PATH, JSON.stringify(existing, null, 2), 'utf-8');
writeFileSync(FAIL_PATH, JSON.stringify(failures, null, 2), 'utf-8');
console.log(`\n[완료] 성공 ${Object.keys(existing).length} / 실패 ${Object.keys(failures).length}`);
console.log(` OCR 결과: ${OCR_PATH}`);
if (Object.keys(failures).length > 0) {
console.log(` 실패 목록: ${FAIL_PATH} (재실행하면 실패분만 재시도)`);
}
}
main().catch((err) => {
console.error(err);
process.exit(1);
});

파일 보기

@ -0,0 +1,324 @@
"""로컬 EasyOCR 기반 HNS 카드 이미지 파싱.
전용 venv(.venv) 설치된 easyocr을 사용한다.
1. 이미지 EasyOCR (bbox, text, conf) 리스트
2. y좌표로 그룹화 x좌표 정렬
3. 레이블 키워드 기반 필드 매핑 (정규식)
4. 결과를 out/ocr.json 누적 저장 (재실행 가능)
실행:
cd backend/scripts/hns-import
source .venv/Scripts/activate # Windows Git Bash
python ocr-local.py [--limit N] [--only 벤젠,톨루엔,...]
"""
from __future__ import annotations
import argparse
import io
import json
import os
import re
import sys
from pathlib import Path
from typing import Any
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
SCRIPT_DIR = Path(__file__).parent.resolve()
OUT_DIR = SCRIPT_DIR / 'out'
IMG_DIR = OUT_DIR / 'images'
OCR_PATH_DEFAULT = OUT_DIR / 'ocr.json'
FAIL_PATH_DEFAULT = OUT_DIR / 'ocr-failures.json'
# ────────── 필드 레이블 패턴 (EasyOCR 오인식 변형 포함) ──────────
# 각 필드의 후보 레이블 문자열(공백 제거 후 비교). 한글 OCR이 종종 비슷한 글자로 오인식되므로
# 대표적인 변형도 함께 등록 (예: "인화점" ↔ "인회점", "끓는점" ↔ "꿈는점" ↔ "끝는점").
LABEL_CANDIDATES: dict[str, list[str]] = {
'casNumber': ['CAS번호', 'CASNO', 'CAS'],
'unNumber': ['UN번호', 'UNNO', 'UN'],
'transportMethod': ['운송방법', '운승방벌', '운송방벌', '운송방립', '운송'],
'usage': ['용도'],
'state': ['성상', '상태', '형태'],
'color': ['색상', ''],
'odor': ['냄새'],
'flashPoint': ['인화점', '인회점', '인하점', '인호점'],
'autoIgnition': ['발화점', '발회점', '발하점'],
'boilingPoint': ['끓는점', '꿈는점', '끝는점', '끊는점'],
'density': ['비중'],
'solubility': ['용해도', '용해'],
'vaporPressure': ['증기압', '증기압력'],
'vaporDensity': ['증기밀도'],
'explosionRange': ['폭발범위', '곡발범위', '폭범위', '폭발한계'],
'idlh': ['IDLH'],
'aegl2': ['AEGL-2', 'AEGL2'],
'erpg2': ['ERPG-2', 'ERPG2'],
'twa': ['TWA'],
'stel': ['STEL'],
'ergNumber': ['ERG번호', 'ERG'],
'hazardClass': ['위험분류', '위험', '분류'],
'synonymsKr': ['유사명'],
'responseDistanceFire': ['대피거리', '머피거리'],
'ppeClose': ['근거리(레벨A)', '근거리레벨A', '근거리', '레벨A'],
'ppeFar': ['원거리(레벨C)', '원거리레벨C', '원거리', '레벨C'],
'emsFire': ['화재(F-E)', '화재(F-C)', '화재(F-D)', '화재대응'],
'emsSpill': ['유출(S-U)', '유출(S-O)', '유출(S-D)', '해상유출'],
'marineResponse': ['해상대응', '해상'],
}
def _norm_label(s: str) -> str:
"""공백/특수문자 제거 후 비교용 정규화."""
return re.sub(r'[\s,.·()\[\]:;\'"-]+', '', s).strip()
LABEL_INDEX: dict[str, str] = {}
for _field, _candidates in LABEL_CANDIDATES.items():
for _cand in _candidates:
LABEL_INDEX[_norm_label(_cand)] = _field
# NFPA 셀 값(한 자릿수 0~4) 추출용
NFPA_VALUE_RE = re.compile(r'^[0-4]$')
def group_rows(items: list[dict], y_tolerance_ratio: float = 0.6) -> list[list[dict]]:
"""텍스트 조각들을 y 좌표 기준으로 행 단위로 그룹화 (글자 높이 비례 허용치)."""
if not items:
return []
heights = [it['y1'] - it['y0'] for it in items]
median_h = sorted(heights)[len(heights) // 2]
y_tol = max(8, median_h * y_tolerance_ratio)
sorted_items = sorted(items, key=lambda it: it['cy'])
rows: list[list[dict]] = []
for it in sorted_items:
if rows and abs(it['cy'] - rows[-1][-1]['cy']) <= y_tol:
rows[-1].append(it)
else:
rows.append([it])
for row in rows:
row.sort(key=lambda it: it['cx'])
return rows
def _match_label(text: str) -> str | None:
key = _norm_label(text)
if not key:
return None
# 정확 일치 우선
if key in LABEL_INDEX:
return LABEL_INDEX[key]
# 접두 일치 (OCR이 뒤에 잡티를 붙이는 경우)
for cand_key, field in LABEL_INDEX.items():
if len(cand_key) >= 2 and key.startswith(cand_key):
return field
return None
def parse_card(items: list[dict]) -> dict[str, Any]:
"""OCR 결과 목록을 필드 dict로 변환."""
rows = group_rows(items)
result: dict[str, Any] = {}
# 1) 행 내 "레이블 → 값" 쌍 추출
# 같은 행에서 레이블 바로 뒤의 첫 non-label 텍스트를 값으로 사용.
for row in rows:
# 여러 레이블이 같은 행에 있을 수 있음 (2컬럼 표 구조)
idx = 0
while idx < len(row):
field = _match_label(row[idx]['text'])
if field:
# 다음 non-label 조각을 값으로 취함
value_parts: list[str] = []
j = idx + 1
while j < len(row):
nxt = row[j]
if _match_label(nxt['text']):
break
value_parts.append(nxt['text'])
j += 1
if value_parts and field not in result:
value = ' '.join(value_parts).strip()
if value and value not in ('-', '', 'N/A'):
result[field] = value
idx = j
else:
idx += 1
# 2) NFPA 추출: "NFPA" 단어 주변의 0~4 숫자 3개
nfpa_idx_row: int | None = None
for ri, row in enumerate(rows):
for cell in row:
if re.search(r'NFPA', cell['text']):
nfpa_idx_row = ri
break
if nfpa_idx_row is not None:
break
if nfpa_idx_row is not None:
# 해당 행 + 다음 2개 행에서 0~4 숫자 수집
candidates: list[int] = []
for ri in range(nfpa_idx_row, min(nfpa_idx_row + 3, len(rows))):
for cell in rows[ri]:
m = NFPA_VALUE_RE.match(cell['text'].strip())
if m:
candidates.append(int(cell['text'].strip()))
if len(candidates) >= 3:
break
if len(candidates) >= 3:
break
if len(candidates) >= 3:
result['nfpa'] = {
'health': candidates[0],
'fire': candidates[1],
'reactivity': candidates[2],
'special': '',
}
# 3) EmS 코드 (F-x / S-x 패턴)
all_text = ' '.join(cell['text'] for row in rows for cell in row)
f_match = re.search(r'F\s*-\s*([A-Z])', all_text)
s_match = re.search(r'S\s*-\s*([A-Z])', all_text)
if f_match or s_match:
parts = []
if f_match:
parts.append(f'F-{f_match.group(1)}')
if s_match:
parts.append(f'S-{s_match.group(1)}')
if parts:
result['emsCode'] = ', '.join(parts)
# 4) ERG 번호 (3자리 숫자, P 접미사 가능, "ERG" 키워드 근처)
erg_match = re.search(r'ERG[^\d]{0,10}(\d{3}P?)', all_text)
if erg_match:
result['ergNumber'] = erg_match.group(1)
# 5) EmS F-x / S-x 코드 뒤의 본문 (생략 - 이미지 내 텍스트 밀도가 낮아 행 단위로 이미 잡힘)
return result
def _preprocess_image(pil_img, upscale: float = 2.5):
"""한글 OCR 정확도 향상을 위한 업스케일 + 샤프닝 + 대비 향상."""
from PIL import Image, ImageEnhance, ImageFilter
import numpy as np
if pil_img.mode != 'RGB':
pil_img = pil_img.convert('RGB')
# 1) 업스케일 (LANCZOS)
w, h = pil_img.size
pil_img = pil_img.resize((int(w * upscale), int(h * upscale)), Image.LANCZOS)
# 2) 대비 향상
pil_img = ImageEnhance.Contrast(pil_img).enhance(1.3)
# 3) 샤프닝
pil_img = pil_img.filter(ImageFilter.UnsharpMask(radius=1.5, percent=150, threshold=2))
return np.array(pil_img)
def run_ocr(image_path: Path, reader, upscale: float = 2.5) -> list[dict]:
# OpenCV가 Windows에서 한글 경로를 못 읽으므로 PIL로 로드 후 전처리
from PIL import Image
with Image.open(image_path) as pil:
img = _preprocess_image(pil, upscale=upscale)
raw = reader.readtext(img, detail=1, paragraph=False)
items: list[dict] = []
for bbox, text, conf in raw:
if not text or not str(text).strip():
continue
xs = [p[0] for p in bbox]
ys = [p[1] for p in bbox]
items.append({
'text': str(text).strip(),
'cx': sum(xs) / 4.0,
'cy': sum(ys) / 4.0,
'x0': min(xs),
'x1': max(xs),
'y0': min(ys),
'y1': max(ys),
'conf': float(conf),
})
return items
def load_json(path: Path, fallback):
if not path.exists():
return fallback
try:
return json.loads(path.read_text(encoding='utf-8'))
except Exception:
return fallback
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument('--limit', type=int, default=None)
parser.add_argument('--only', type=str, default=None,
help='파이프(|)로 구분된 물질명 리스트')
parser.add_argument('--img-dir', type=Path, default=IMG_DIR)
parser.add_argument('--out', type=Path, default=OCR_PATH_DEFAULT)
parser.add_argument('--fail', type=Path, default=FAIL_PATH_DEFAULT)
parser.add_argument('--debug', action='store_true',
help='파싱 중간 결과(row 단위) 함께 출력')
args = parser.parse_args()
import easyocr # noqa: WPS433
print('[로딩] EasyOCR 모델 (ko + en)... (최초 실행 시 수 분 소요)')
reader = easyocr.Reader(['ko', 'en'], gpu=False, verbose=False)
print('[로딩] 완료')
images = sorted([p for p in args.img_dir.iterdir() if p.suffix.lower() in {'.png', '.jpg', '.jpeg'}])
if args.only:
only_set = {s.strip() for s in args.only.split('|') if s.strip()}
images = [p for p in images if p.stem in only_set]
existing: dict[str, Any] = load_json(args.out, {})
failures: dict[str, str] = load_json(args.fail, {})
pending = [p for p in images if p.stem not in existing]
if args.limit:
pending = pending[: args.limit]
print(f'[대상] {len(images)}개 중 대기 {len(pending)}개, 이미 처리 {len(existing)}')
ok = 0
fail = 0
for i, path in enumerate(pending, start=1):
name = path.stem
try:
items = run_ocr(path, reader)
parsed = parse_card(items)
if args.debug:
print(f'\n--- {name} (텍스트 {len(items)}개) ---')
for row in group_rows(items):
print(' |', ''.join(f'{c["text"]}' for c in row))
print(f' → parsed: {parsed}')
existing[name] = parsed
if name in failures:
del failures[name]
ok += 1
except Exception as e: # noqa: BLE001
failures[name] = f'{type(e).__name__}: {e}'[:500]
fail += 1
print(f'[실패] {name}: {e}')
if i % 10 == 0 or i == len(pending):
args.out.write_text(json.dumps(existing, ensure_ascii=False, indent=2), encoding='utf-8')
args.fail.write_text(json.dumps(failures, ensure_ascii=False, indent=2), encoding='utf-8')
print(f' 진행 {i}/{len(pending)} (성공 {ok}, 실패 {fail}) - 중간 저장')
args.out.write_text(json.dumps(existing, ensure_ascii=False, indent=2), encoding='utf-8')
args.fail.write_text(json.dumps(failures, ensure_ascii=False, indent=2), encoding='utf-8')
print(f'\n[완료] 성공 {ok} / 실패 {fail}')
print(f' 결과: {args.out}')
if failures:
print(f' 실패 목록: {args.fail}')
if __name__ == '__main__':
main()

파일 보기

@ -0,0 +1,29 @@
# extract-excel.py 용
openpyxl>=3.1.0
# ocr-local.py 용 (EasyOCR 기반 로컬 OCR, 대안 파이프라인)
easyocr==1.7.2
filelock==3.19.1
fsspec==2025.10.0
ImageIO==2.37.2
Jinja2==3.1.6
lazy-loader==0.5
MarkupSafe==3.0.3
mpmath==1.3.0
networkx==3.2.1
ninja==1.13.0
numpy==2.0.2
opencv-python-headless==4.13.0.92
packaging==26.1
pillow==11.3.0
pyclipper==1.3.0.post6
python-bidi==0.6.7
PyYAML==6.0.3
scikit-image==0.24.0
scipy==1.13.1
shapely==2.0.7
sympy==1.14.0
tifffile==2024.8.30
torch==2.8.0
torchvision==0.23.0
typing_extensions==4.15.0

파일 보기

@ -24,8 +24,14 @@ async function seedHnsSubstances() {
let inserted = 0 let inserted = 0
// varchar 길이 제한에 맞춰 첫 번째 토큰만 검색 컬럼에 저장 (원본은 DATA JSONB에 보존)
const firstToken = (v: unknown, max: number): string | null => {
if (v == null) return null
const s = String(v).split(/[\n,;/]/)[0].trim()
return s ? s.slice(0, max) : null
}
for (const s of HNS_SEARCH_DB) { for (const s of HNS_SEARCH_DB) {
// 검색용 컬럼 추출, 나머지는 DATA JSONB로 저장
const { abbreviation, nameKr, nameEn, unNumber, casNumber, sebc, ...detailData } = s const { abbreviation, nameKr, nameEn, unNumber, casNumber, sebc, ...detailData } = s
await client.query( await client.query(
@ -39,7 +45,16 @@ async function seedHnsSubstances() {
CAS_NO = EXCLUDED.CAS_NO, CAS_NO = EXCLUDED.CAS_NO,
SEBC = EXCLUDED.SEBC, SEBC = EXCLUDED.SEBC,
DATA = EXCLUDED.DATA`, DATA = EXCLUDED.DATA`,
[s.id, abbreviation, nameKr, nameEn, unNumber, casNumber, sebc, JSON.stringify(detailData)] [
s.id,
firstToken(abbreviation, 50),
firstToken(nameKr, 200) ?? '',
firstToken(nameEn, 200),
firstToken(unNumber, 10),
firstToken(casNumber, 20),
firstToken(sebc, 50),
JSON.stringify(detailData),
]
) )
inserted++ inserted++

파일 보기

@ -4,6 +4,9 @@
## [Unreleased] ## [Unreleased]
### 추가
- HNS: 물질 DB 데이터 확장 및 데이터 구조 개선 (PDF 추출 스크립트, 병합 스크립트 개선, 물질 상세 패널 업데이트)
### 변경 ### 변경
- 디자인 시스템: color 토큰 Definition 팔레트로 마이그레이션 (bg/stroke/fg 쿨톤 전환, Primary #0099DD 적용) - 디자인 시스템: color 토큰 Definition 팔레트로 마이그레이션 (bg/stroke/fg 쿨톤 전환, Primary #0099DD 적용)

파일 보기

@ -1807,4 +1807,3 @@ ${styles}
</div> </div>
); );
} }

파일 보기

@ -18,6 +18,20 @@ export function HmsDetailPanel({
'🔗 화물적부도·항구별 코드', '🔗 화물적부도·항구별 코드',
]; ];
const nfpa = s.nfpa; const nfpa = s.nfpa;
// eslint-disable-next-line @typescript-eslint/no-unused-vars
const sebcColor = !s.sebc
? 'var(--fg-sub)'
: s.sebc.startsWith('G')
? 'var(--color-accent)'
: s.sebc.startsWith('E')
? 'var(--color-accent)'
: s.sebc.startsWith('F')
? 'var(--color-caution)'
: s.sebc.startsWith('D')
? 'var(--color-accent)'
: s.sebc.startsWith('S')
? 'var(--color-accent)'
: 'var(--fg-sub)';
return ( return (
<div <div

파일 크기가 너무 크기때문에 변경 상태를 표시하지 않습니다. Load Diff

파일 보기

@ -282,6 +282,8 @@ export interface HNSSearchSubstance {
hazardClass: string; hazardClass: string;
ergNumber: string; ergNumber: string;
idlh: string; idlh: string;
twa: string;
stel: string;
aegl2: string; aegl2: string;
erpg2: string; erpg2: string;
responseDistanceFire: string; responseDistanceFire: string;