简介这是一个面向医学信息化与科研分析场景的CMID医学数据提取JSON数据文件适合处理临床信息、电子病历或构建医疗数据接口的数据工程师与研究人员使用。该JSON文件以键值对形式组织完整呈现患者基本信息、病史记录、检验结果、影像资料等多模块的层级嵌套结构清晰展示字段标识、数据类型与关联方式可帮助读者快速理解医学数据的标准化表达与交换逻辑。资源包为zip压缩格式整体仅1.05MB内含1个JSON数据文件规模精炼、便于下载解析可直接作为真实医疗数据样例用于解析测试或流程验证。目前已有169人学习或下载适合作为医学数据抽取、清洗及字段映射的参考样例也可用于相关算法开发、接口联调与数据交换演练对掌握医学信息数字化处理有一定参考价值。1. cmid医学数据提取JSON先明确交付的是“文件”还是“管道”cmid医学数据提取json数据文件这几年做医疗信息化项目时经常被当成一个简单脚本看待拿到一堆cmid系统导出的文本或表格转成json文件就算完事。实际跑过几轮就会发现cmid数据很少是干净的二维表里面可能混着分号分隔的检查项、嵌套的JSON字符串、甚至跨多行的文本块。这篇文章适合数据工程师、医学研究助理和系统集成岗位目标是用最稳妥的方式把cmid源文件落成结构清晰、后续能直接进分析管线或模型输入的json数据文件。开篇先把预期管理好你要交付的是一次性提取脚本还是一套能反复处理增量数据的解析管道这决定了后续章节里每一步的取舍。2. 界定cmid源文件形态先建schema再写解析器避免边写边改我见过太多“拿到文件直接json.loads”的现场十次里有八次在第二列就报错。医学数据提取JSON的第一步不是写代码而是确认源头文件到底长什么样。cmid这类系统导出的数据通常不是单一格式常见形态能归成四类第一类是标准CSV/TSV导出每行一条记录列名是中文表头第二类是数据库dump出来的文本某些字段本身就是序列化后的JSON字符串第三类是诊断描述、主诉、既往史这类长文本可能包含换行和Tab第四类是多个分表导出后手工拼接的文件行之间逻辑关联靠外键字段维系。拿到文件后先在编辑器里看前100行的原始字节注意编码、分隔符、换行符再决定解析方案。2.1 用五分钟审计法确定解析策略我一般会在写解析代码前抽出五分钟做四个检查第一用file命令看编码cmid老系统经常导出GBK直接按UTF-8读必然乱码第二看首行是否是列名以及列名里有没有换行和空格第三统计每条记录的分隔符数量是否一致如果不一致说明存在嵌套分隔符第四抽查三到五条记录里最长的一条确认是否存在跨行字段。把这些结论记下来再对应选解析策略CSV还是split()要不要csv.reader长文本字段要不要先做块合并。这里有个常见的思维误区源文件里的“一列”不一定等于JSON里的“一个字段”。cmid导出的检查结果列可能是“白细胞计数:6.8×10^9/L;中性粒细胞百分比:65.2%”这样的合并串需要先定义拆分规则再映射到目标schema。如果上来就按原始列名生成JSON键后面清洗时反而得多写一层转换。所以审计之后先画一张字段映射表比直接开编辑器更重要。2.2 输出JSON的层级设计患者、就诊、观测三层分开医学数据的JSON结构不能做成一张大宽表。cmid数据天然存在层级一个患者有多次就诊一次就诊有多个检查项一个检查项又有数值、单位、参考范围、异常标记等属性。把这三层塞进同一个扁平对象里后续做筛选和聚合时处处受挫。我常用的输出结构是患者-就诊-观测三层患者层放基本信息和唯一标识就诊层放入院时间、科室、主诊断观测层放具体检查项的明细列表。{ patient_id: P20240308001, gender: F, birth_year: 1968, visits: [ { visit_id: V20240308001, admission_time: 2024-03-08T08:12:0008:00, department: 心血管内科, diagnosis: 冠心病, observations: [ { item_code: WBC, item_name: 白细胞计数, value: 6.8, unit: 10^9/L, abnormal_flag: false } ] } ] }这段结构里value字段只放数值单位单独放在unit这样建模工具在统计均值、方差时不至于把“6.8×10^9/L”当成字符串。item_code用标准化代码而不是中文名是因为同一项在不同科室导出时可能叫“白细胞计数”也可能叫“白细胞”代码才是稳定的关联键。abnormal_flag直接计算出来落库省得下游每次都对比参考范围。设计阶段就把这些想好解析代码写起来只是机械翻译。2.3 字段语义映射表把业务含义固化到代码里设计完层级结构后我会先整理一份字段映射表每一条包括源字段名、样例值、目标字段名、清洗规则四列。这一步是给整个提取过程立规矩。比如源字段叫“就诊时间”样例值是“2024/3/8 8:12”目标字段是admission_time清洗规则是转成ISO 8601格式源字段叫“体温”样例值是“36.8 ℃”目标字段是body_temperature_celsius清洗规则是去掉单位后转float再与单位字段分开。源字段样例值目标字段清洗规则门诊号20240308-001patient_id去空格、去横线就诊时间2024/3/8 8:12admission_time解析后转ISO 8601白细胞计数6.8×10^9/La白细胞_数值提取浮点数单位10^9/Lunit作为独立字段保留吸烟史无/有/未知smoking_history映射为0/1/null注意表格最后一行医学数据里大量存在“有”“无”“未查”“不详”这类枚举值如果不定义映射规则到后端就是一团乱麻。我习惯把这类映射单独放在一个enum_mapping.py里统一维护而不是散落在解析函数内改口径时只需要换一个字典。设计阶段还有一个容易忽略的细节要不要保留原始文本。我建议在JSON里留一个source_text字段存原始行的完整内容。这样下游如果发现转换后的字段对不上还能回头查原文否则数据链路一旦断裂再找原始记录的成本非常高。3. 用Python把cmid原始行转成JSON主解析循环与三个可调参数结构定义清楚后解析代码就有了明确目标。下面这套代码是我处理cmid TSV文件时常用的骨架按行读取、按列拆分、逐条构造JSON记录并写入文件。它不依赖pandas原因是cmid文件经常存在跨行字段和复杂的嵌套分隔符用pandas逐行处理反而容易绕圈子。3.1 最小解析循环一行文本到一条JSON记录import json import csv from datetime import datetime ENCODING utf-8 DELIMITER \t INPUT_FILE cmid_export_20240308.txt OUTPUT_FILE cmid_export_20240308.jsonl def parse_time(value: str) - str | None: 把多种时间写法统一成ISO 8601解析失败返回None if not value or value.strip() : return None normalized value.strip().replace(/, -) for fmt in (%Y-%m-%d %H:%M:%S, %Y-%m-%d %H:%M, %Y-%m-%d): try: return datetime.strptime(normalized, fmt).isoformat() except ValueError: continue return None def clean_text(value: str) - str: 清除字段里的不可见字符但保留中文和基本标点 return .join(value.split()) def process_line(row: dict) - dict: 把源文件一行映射为输出JSON对象 return { patient_id: clean_text(row.get(患者编号, )), visit_id: clean_text(row.get(就诊编号, )), admission_time: parse_time(row.get(就诊时间, )), department: clean_text(row.get(科室, )), diagnosis: clean_text(row.get(主诊断, )), source_text: json.dumps(row, ensure_asciiFalse), } def main() - None: output_handle open(OUTPUT_FILE, w, encodingENCODING) with open(INPUT_FILE, r, encodingENCODING) as f: reader csv.DictReader(f, delimiterDELIMITER) for line_number, row in enumerate(reader, start1): try: record process_line(row) output_handle.write(json.dumps(record, ensure_asciiFalse) \n) except Exception as exc: print(f行号 {line_number} 解析失败: {exc}) output_handle.close() if __name__ __main__: main()这段代码有三个关键点。第一逐行读取不一次性把整个文件加载进内存遇到几千MB的cmid导出文件逐行写JSON Lines的写法不会因内存问题重启。第二csv.DictReader按第一行表头映射列名省去手动指定索引的麻烦源文件如果没有表头需要先补一行或者改用fieldnames参数。第三输出文件名是.jsonl而不是.json因为每行一条记录后续既可以被json.loads逐行读也可以被大数据工具直接按行消费。3.2 嵌套JSON字符串json.loads和ast.literal_eval怎么选cmid文件里偶尔会出现整列都是JSON字符串的情况比如某个“检查详情”字段存的是{heart_rate: 88, blood_pressure: {systolic: 135, diastolic: 85}}。这种字段不能直接当成普通文本输出需要先解析再展开。json.loads是首选它只接受合法JSON字符串并严格按照JSON规范解析。若调用ast.literal_eval它支持Python字面量能容忍单引号和True/False这类写法但解析速度较慢且接受不规范的输入可能会掩盖源数据的问题。我的习惯是先用json.loads尝试如果失败就打印出错行号和原始片段人工确认是源文件损坏还是格式本身就不是JSON再决定是否启用容错分支。不要在解析时默默吞掉异常。下面是嵌套字段的处理示例def parse_nested_field(value: str, line_number: int): if not value or value.strip() : return None try: return json.loads(value) except json.JSONDecodeError as exc: print(f行号 {line_number} 的嵌套JSON字段无法解析: {exc}) print(f原始片段: {value[:200]}) return None3.3 写盘方式单JSON数组还是JSON Lines解析完的记录文件必须以JSON Lines写盘还是有需求时合并成数组取决于下游如何使用。如果下游是数据建模工具或脚本逐行读JSON Lines最省事读一行处理一行如果要对接传统API需要一次性返回一个数组再包一层写数组也很快。我平常输出JSON Lines原因只有一个即使转换到一半进程被中断已完成的行也不会丢调试时还能用tail直接看最后几条记录校验进度。def write_records(records: list, output_path: str) - None: with open(output_path, w, encodingutf-8) as f: json.dump(records, f, ensure_asciiFalse, indent2)这段适合已经确认记录数不大、内存完全放得下的场景。参数ensure_asciiFalse一定不能省否则中文会被转成\uXXXX文件体积变大肉眼排查问题也不方便。indent2只用于小规模交付和人工核对几百MB级别的文件不要加缩进磁盘占用和解析性能都不划算。3.4 列名清洗中文字段名映射到英文键cmid系统导出时列名往往是中文直接当JSON键保存在技术债上站不住。中文键在Windows文件系统、老旧JSON解析库和部分建模工具里都可能出问题。我按映射规则把中文列名统一成英文键名规则是全部小写、蛇形命名、语义尽量短。这个映射表建议放在独立JSON里维护解析代码只管调用不把映射规则硬编码在业务逻辑中。FIELD_MAPPING { 患者编号: patient_id, 就诊编号: visit_id, 就诊时间: admission_time, 科室: department, 主诊断: diagnosis, } def map_row_to_record(row: dict) - dict: return { FIELD_MAPPING[key]: value for key, value in row.items() if key in FIELD_MAPPING }映射逻辑里最容易忽略的是“源文件列顺序不固定”的坑。以前我直接用row[0]、row[1]按下标取值换一版导出文件就全乱了。用DictReader配合FIELD_MAPPING后列顺序变了代码照样跑通这是cmid这类经常重导出的场景里必须养成的习惯。4. 时间、单位、空值的归一化让JSON在建模阶段少返工结构化的JSON不等于能直接建模的JSON。cmid数据里最消耗时间的就是时间格式、单位体系和空值语义三轮清洗。提取过程中把这三个口径统一好下游建模阶段的返工量能降低一大半。4.1 时间字段六种日期写法统一成ISO 8601不同科室导出的日期格式五花八门有人写“2024/3/8 8:12”有人写“20240308081200”还有人写“2024-03-08 08:12:00.000”。这些看着都像时间建模阶段如果混着用排序、差值计算全是错的。我在parse_time函数里按优先级依次尝试多种格式返回统一的ISO 8601字符串。注意一个边界cmid老系统导出时年份常常只有两位比如“24-03-08”需要显式定义“00-69映射为20xx70-99映射为19xx”还是直接报错。我选择报错并记录因为这种二义性不该由程序猜。4.2 数值与单位别把“mg/dl”和“mmol/l”存进同一个键医学数据的数值必须和单位绑在一起看。血糖“6.8”后面可能是mmol/L也可能是mg/dL两者含义完全不同。如果JSON里只存数值不存单位后续做统计时会把两个数量级的数据混在一起。我的方案是每个观测项输出value和unit两个字段同时在清洗阶段定义目标单位约定。UNIT_STANDARD { 血糖: mmol/L, 白细胞计数: 10^9/L, 血红蛋白: g/L, } def normalize_value(item_name: str, value_str: str, unit: str): if item_name not in UNIT_STANDARD: return float(value_str), unit if unit UNIT_STANDARD[item_name]: return float(value_str), unit return float(value_str), unit这个示例只做了最简单的“保留原始单位”没有做真正的单位换算因为不同项目的换算系数差异很大需求场景又不一致。换算逻辑必须由临床人员确认后单独实现不要在提取脚本里默认换算。4.3 空值与缺失null、0、空字符串、字符串“null”要分开源文件里表示“没记录”的方式非常混乱。有的填“-”有的填“未查”有的直接留空还有些系统会把空值序列化成字符串null。这四个状态如果统一转成None下游无法区分“数据缺失”和“检查结果为空”的临床语义。我定义的口径是空白和“-”转None“未查”“未做”转None并附一条missing_note字段数值0保留为0因为0在很多检验项目里有明确临床意义字符串null按缺失处理并在日志里警告。EMPTY_VALUES {, -, 未查, 不详, null, N/A} def handle_missing(value: str) - tuple: if value is None: return None, missing_blank stripped value.strip() if stripped in EMPTY_VALUES: return None, missing_known return value.strip(), None这个函数返回两个值第一个是清洗后的数据第二个是缺失原因标记。这个标记不一定要写入最终JSON但调试阶段有它能很清楚地看出哪些字段天然缺失率高哪些字段是清洗规则误伤。4.4 枚举值与修饰符不忽略“200”这类前缀检验结果中经常出现“200”“0.5”“”这类修饰符。简单的float()转换会直接抛异常如果直接把整行丢掉那这行数据就废了。我按“数值前缀修饰符拆分单独标记”来做解析出修饰符存入value_modifier字段剩余部分尝试转浮点转换失败的整条记录进异常队列而不是静默丢弃。def parse_qualified_value(raw: str): raw raw.strip() if raw.startswith(): return float(raw[1:]), if raw.startswith(): return float(raw[1:]), return float(raw), None这类修饰符在建模时可以按或处理也可以在筛选场景排除至少不会因为解析失败丢数据。把这个逻辑放在解析函数而不是清洗函数里是为了让报错尽早暴露必要时人工定义处理规则。5. cmid JSON提取避坑排查五个真实翻车现场与对策这里写五个我在处理cmid数据时踩过的具体问题每一个都有现象、原因和对策。读者做类似项目时可以直接拿这些现象对照排查。5.1 现象一打开文件全是乱码json.loads在“update_time”字段上抛错现象是json.loads抛UnicodeDecodeError文件内容在编辑器里显示乱码。原因是cmid老版本导出文件使用GBK编码而解析代码直接用utf-8打开。解决方法是先确认编码再统一转换为UTF-8。替换读取参数为encodinggbk或者在读取后显式转码。另外注意源文件可能存在混合编码一个文件里前半段是GBK后半段混入UTF-8这种情况逐行按errorsreplace处理同时输出警告行号。5.2 现象二同一患者同一天出现两条重复visit_id现象是计数时发现visit_id重复导致患者就诊次数虚高。原因是源系统存在跨科室转科记录同一次住院过程在导出时被拆成两条记录。解决方法是先确认业务口径是“按科室拆分”还是“合并为一次就诊”。我一般会在清洗层加一步按patient_id和visit_id去重如果多条记录时间范围重叠则合并diagnosis和observations数组直到业务方给明确口径为止。5.3 现象三JSON字符串里有未转义的换行行数对不上现象是程序按行处理时每处理几条就报一次“行数错位”且报错位置没有固定规律。原因是某个长文本字段内部包含原始换行符导出时没有转成\n文件里实际是多行。解决方法是不能按\n机械切分需要按记录边界识别。我改用状态机方式先判断当前行是否能解析成一条完整记录如果不能则等下一行拼接直到字段数完整为止。def iter_complete_records(file_handle): buffer for line in file_handle: buffer line # 尝试用当前分隔符解析成功则yield失败继续累积 try: record parse_record(buffer) yield record buffer except IncompleteRecordError: continue这种方法牺牲了一点速度换来的是能不丢数据地处理跨行文本。代价是解析函数必须能够判断完整性否则最后一个缓冲块永远无法退出循环。5.4 现象四内存溢出进程跑了一半被系统杀掉现象是处理500MB以上的导出文件时程序在写入阶段内存持续上涨直到OOM。原因是解析完成后先把所有记录list收集起来最后统一json.dump到文件。解决方法是按第一条目的JSON Lines方案逐行写入只在内存中保留当前记录和必要的映射表。如果下游一定要单个JSON数组可以等全部行写完后再用文件流的方式拼接[、,、]不要先把所有内容加载到内存。5.5 现象五字段值是“6.8×10^9/L”直接float转换失败现象是解析白细胞计数时float()抛ValueError。原因是源文件把单位拼在数值后面而且用了乘号“×”而不是字母“x”。解决方法是先提取数字部分再单独解析单位同时注意中文全角字符和半角字符混用的情况比如“”是全角数字需要先translate或替换成全角转半角。def extract_number_and_unit(raw: str): normalized raw.replace(×, x).replace(, /) match re.search(r([-]?\d\\.?\d*)\\s*([A-Za-z%/μ^]), normalized) if not match: return None, None return float(match.group(1)), match.group(2)正则表达式要按实际数据调整尤其单位里可能包含下标、上标字符。这个函数的返回值也要配合前面的UNIT_STANDARD使用否则仅解析出来没有校准单位还是乱的。6. 抽读回验JSON文件质量用一个校验脚本收尾全量解析跑完后别急着通知下游先做一轮抽查回读。我的习惯是写一个小脚本读取生成的JSON Lines文件统计三个数字总记录数、关键字段缺失率、重复记录数。总和源文件行数对不上逐批排查缺失率异常高优先看清洗函数是否误伤重复率异常回到5.2的合并逻辑。jq -r .patient_id | .admission_time cmid_export_20240308.jsonl | sort | uniq -d | head -20这段命令利用jq抽取患者和就诊时间排序后找重复组合输出前20条。运行前先确认jq已经安装没有安装就改用Python脚本逐行统计。这是交付前最小成本的防线。最后一章还要教一个习惯把解析脚本、映射表、清洗规则、异常行清单放在同一个目录里用日期命名版本。这样三个月后源系统改了导出格式你能快速定位是哪个环节出了问题。我自己曾经因为不保留异常行清单事后返工时不得不把旧文件重新解析一遍白白浪费一整天。这个习惯后来帮我省下的时间远超写脚本的时间希望帮到你。本文还有配套的精品资源点击获取