脚本如何批量转换医疗数据格式

本文目录导读：

1. 核心工具选择（推荐）
2. 实战脚本示例（Windows/Linux/Mac均适用）
3. 专业级批量转换方案（处理大量文件）
4. 关键注意事项
5. 快速选择指南

针对医疗数据格式的批量转换,通常需要处理 HL7 v2.x/v3、FHIR、DICOM、CSV/Excel 以及国内特有的 电子病历（XML/JSON） 格式，以下提供基于不同场景的脚本解决方案，优先使用开源工具和 Python。

核心工具选择（推荐）

实战脚本示例（Windows/Linux/Mac均适用）

场景1：批量HL7 v2→ FHIR R4（最常用）

# install: pip install hl7apy fhir.resources
import os, json, glob
from hl7apy.parser import parse_message
from hl7apy.core import Message
from fhir.resources.patient import Patient
from fhir.resources.encounter import Encounter
from datetime import datetime
def hl7_to_fhir(hl7_text):
    # 1. 解析HL7消息
    msg = parse_message(hl7_text)
    # 2. 创建FHIR Patient资源
    patient = Patient.construct()
    # 从PID段提取姓名（HL7中的PID-5）
    pid_seg = msg.MSH.PID if hasattr(msg.MSH, 'PID') else msg.PID
    patient.id = f"patient-{pid_seg.PID_3[0].value}"  # PID-3 MRN
    patient.name = [{"family": pid_seg.PID_5[0].P_1.value, 
                     "given": [pid_seg.PID_5[0].P_2.value]}]
    # 3. 创建Encounter（就诊记录）
    encounter = Encounter.construct()
    encounter.patient = {"reference": f"Patient/{patient.id}"}
    return [patient.dict(), encounter.dict()]
# 批量处理指定文件夹所有.hl7文件
input_dir = "./hl7_files"
output_dir = "./fhir_output"
os.makedirs(output_dir, exist_ok=True)
for file_path in glob.glob(f"{input_dir}/*.hl7"):
    with open(file_path, 'r') as f:
        hl7_content = f.read()
    fhir_bundle = hl7_to_fhir(hl7_content)
    # 按原始文件名输出JSON
    base_name = os.path.basename(file_path).replace('.hl7', '.json')
    with open(f"{output_dir}/{base_name}", 'w') as f:
        json.dump(fhir_bundle, f, indent=2)
    print(f"Converted: {file_path} -> {base_name}")

场景2：批量DICOM→ NIfTI（神经影像）

# 使用dcm2niix命令行工具（跨平台，速度极快）
# 安装：https://github.com/rordenlab/dcm2niix/releases
# Windows命令（PowerShell）
Get-ChildItem -Path "D:\raw_dicom" -Directory | ForEach-Object {
    $output_name = $_.Name + "_converted"
    & dcm2niix -z y -o "D:\nifti_output" -f $output_name $_.FullName
}
# Linux/Mac bash
for dir in /path/to/raw_dicom/*/; do
    base=$(basename "$dir")
    dcm2niix -z y -o /path/to/nifti_output -f "${base}_converted" "$dir"
done

场景3：CSV表格→ FHIR Bundle（批量结构数据转换）

# 假设CSV列：patient_id, name_family, name_given, gender, birth_date
import pandas as pd
from fhir.resources.patient import Patient
from datetime import date
def csv_row_to_patient(row):
    p = Patient.construct()
    p.id = f"pat-{row['patient_id']}"
    p.name = [{"family": row['name_family'], "given": [row['name_given']]}]
    p.gender = row['gender'].lower()[:4]  # male/female
    p.birthDate = row['birth_date']  # YYYY-MM-DD格式
    return p
df = pd.read_csv("patients.csv")
fhir_bundle = {"resourceType": "Bundle", "entry": []}
for _, row in df.iterrows():
    patient_resource = csv_row_to_patient(row)
    fhir_bundle["entry"].append({
        "fullUrl": f"urn:uuid:{patient_resource.id}",
        "resource": patient_resource.dict()
    })
# 输出批量FHIR Bundle
import json
with open("fhir_export.json", "w") as f:
    json.dump(fhir_bundle, f, indent=2, default=str)

专业级批量转换方案（处理大量文件）

基于Apache NiFi（图形化数据流）

• 优势：支持HL7/FHIR/DICOM 300+种处理器
• 配置：NiFi → GetFile → Processors(HAPI HL7v2 to FHIR) → PutFile
• 适合企业级ETL管道

命令行工具链（Linux/Unix环境）

# 安装：sudo apt install python3-pip mediainfo
# 发现所有医疗格式文件
find /data -type f \( -name "*.hl7" -o -name "*.dcm" -o -name "*.xml" \) | while read file; do
    case "$file" in
        *.hl7)
            python3 convert_hl7_to_fhir.py "$file"
            ;;
        *.dcm)
            dcm2niix -z y -o /output "$file"
            ;;
        *.xml)
            python3 convert_xml_to_json.py "$file"
            ;;
    esac
done

Docker批量处理容器

FROM python:3.11-slim
WORKDIR /app
RUN pip install pydicom hl7apy fhir.resources pandas
COPY . .
CMD ["python", "bulk_convert.py", "--input", "/data", "--format", "fhir"]

关键注意事项

1. 数据合规性：

   • 转换后需脱敏：对HL7中的PID-19（SSN）、FHIR中的identifier等敏感字段模糊化
   • 批量转换前备份原始数据

2. 编码处理：

   • HL7 v2默认ASCII，包含特殊字符|^~\&，转换时不可丢失分隔符
   • 中国电子病历（XML）需指定UTF-8 with BOM

3. 大型文件处理：

   • 1000+文件建议：find + xargs 或 python multiprocessing
   • 影像数据（DICOM）使用 dcm2niix 多线程模式：-m y

4. 日志与错误重试：

   import logging
   logging.basicConfig(filename='convert_errors.log', level=logging.ERROR)
   try:
       # 转换单条
   except Exception as e:
       logging.error(f"File {file} failed: {str(e)}")

快速选择指南

如果需要针对特定格式（如美国CCDA文档、中国卫宁电子病历、PACS DICOM标签转换）的详细脚本，可以提供更多上下文信息（字段映射关系、输出具体要求），我可以生成更精准的转换代码。