简介数据提取是医疗信息化建设中最基础也最繁琐的环节。面对散落在多个系统、多张表中的临床记录如何将零散数据高效地聚合成规整的JSON文件是HIS/LIS/EMR系统集成和医学数据工程中的常见痛点。JSON因其灵活的嵌套结构天然适配医学数据中变长字段与复杂关联关系能够完整保留一条临床记录的语义边界。通过Python脚本连接数据库完成查询、清洗与序列化可以替代手工导表拼数据实现可重复的批量提取流程。从字段映射到去重策略从时间窗口增量抽取到编码处理再到用JSON Schema做输出自检每一步都直接影响下游系统解析效率和科研数据集的可用性。本文围绕cmid这一临床记录唯一标识系统梳理医学数据提取JSON数据文件时的技术选型与工程化细节为相关岗位提供一套可落地的实践参考。1. cmid医学数据提取json数据文件先把每条记录定位准在医学数据落库这件事上最烦的不是写代码而是源头数据根本没法直接用。cmid医学数据提取json数据文件这套资源解决的就是这类场景你能拿到一个带cmid临床记录唯一标识的医学数据源但散落在多个表、多个文件甚至多个系统里得按这个标识把记录抽出来、清洗掉冗余字段最后落成一个规整的JSON数据文件供下游做分析、上报或对接。适合HIS/LIS/EMR系统集成岗、医学数据工程师和做临床科研数据集的人用。它能帮你绕开“导出一堆Excel再手工拼”的原始状态把提取、转换、输出这一步固化成本地可重复跑的脚本参数调一次批量复用。2. 先摸底再动手数据源里cmid到底藏在哪2.1 三种常见数据源形态决定了提取脚本的写法CMID类医学数据提取源头一般逃不出三种形态关系型数据库MySQL、Oracle、SQL Server居多、接口返回的原始报文、以及一堆散落的Excel/CSV导出文件。形态不同提取脚本的入口就不同。关系型数据库是最常见的。病人基本信息、检查报告、医嘱、费用明细往往分开存放在不同表里靠patient_id或者visit_id这类字段关联。cmid在库表里一般不叫cmid可能叫record_no、case_no、exam_id甚至是一个复合主键。我拿到这类资源后的第一个动作不是急着写Python脚本而是先在库里跑一遍字段探查确认cmid这个标识到底落在哪张表的哪个字段上值域是否稳定。接口报文是另一种形态。很多医院集成平台的接口返回的是嵌套JSONcmid藏在body的某层结构里。这种数据没法直接“读表”要先落成原始报文文件再解析。Excel/CSV形态则常见于科研数据导出一个问题就能导出一堆文件文件名和cmid不一定一致需要靠文件内的某列做映射。数据源形态查找cmid的入口提取脚本的侧重点MySQL/Oracle/SQL Serverinformation_schema.columns 或直接看业务表主键SQL取数 Python封装JSON接口原始报文报文样例里的body嵌套字段JSON解析 字段重命名Excel/CSV导出文件表头列名或文件名规则行列转换 类型清洗如果是SQL Server我一般先跑这条语句确认cmid所在的表和列SELECT c.name AS column_name, t.name AS table_name FROM sys.columns c JOIN sys.tables t ON c.object_id t.object_id WHERE c.name LIKE %cmid% OR c.name LIKE %record% OR c.name LIKE %case% ORDER BY t.name;这段代码是在SQL Server的系统视图里按列名模糊搜索带cmid、record、case字样的字段返回字段名和所属表名。实际做医学数据提取时cmid这个命名字段并不一定精确存在用LIKE去匹配常见的临床记录编号命名习惯能快速定位到候选表。如果是MySQL就换成查询information_schema.columns表条件改成WHERE COLUMN_NAME LIKE %cmid%思路一样。定位到表之后还要确认一件事cmid是单值字段还是由多列拼出来的复合标识。复合标识在后续JSON输出时会影响unique_id的设计提前查清楚能省不少返工。2.2 字段级探查把要输出的JSON结构和源字段对应起来提取之前一定要做字段映射这是整个过程中最枯燥但最不能跳的一步。我见过太多人上来就写脚本结果输出的JSON里字段名和下游对接方的要求对不上白跑一遍。所谓字段映射就是确定最终JSON文件里每个键对应源数据的哪个字段以及类型怎么转。比如“患者姓名”对应源表的patient_name要转成string“检查日期”对应exam_date要转成日期字符串cmid本身对应源表的唯一标识字段输出时作为JSON的最外层键。这一步可以用一张映射表固化下来方便后续排查JSON输出键名源字段类型说明cmidexam_record.record_nostring唯一标识主键patient_namepatient_info.namestring姓名注意脱敏exam_dateexam_report.exam_datetimestring格式化为YYYY-MM-DD HH:MM:SSdiagnosisexam_report.diagnosis_textstring诊断文本itemsexam_report.item_detailsarray检查项详情嵌套结构字段探查时除了看类型还要看一下空值比例。如果某个字段的空值率超过30%就得和业务方确认这个字段是否为必填不是必填的话在JSON里是按缺省处理还是置为null。很多下游系统对这两种处理方式的要求完全不同置null有时候会直接导致反序列化报错这在后文避坑章里详细展开。我一般会顺带跑一次去重检查确认cmid在源表里是否有重复记录。用SQL看一眼就知道SELECT record_no, COUNT(*) FROM exam_record GROUP BY record_no HAVING COUNT(*) 1;这段SQL按record_no分组后过滤出重复的记录编号返回的是有重复值的编号和重复次数。这一步必须做因为cmid一旦重复生成JSON时就会发生同键覆盖直接丢数据。有重复的情况通常会出现在源表存在历史数据合并的场景两套系统的编号规则不一致导致同一个编号对应了两条不同记录。处理方式是在提取层加一个去重策略要么按时间保留最新一条要么把重复记录合并进items数组。2.3 变长字段和嵌套关系别平铺成一维表医学数据天然适合JSON而不是表格根本原因在于变长字段。一个患者的检查项数量不同诊断结论条数不同医嘱明细更是从三五条到几十条都有。用二维表存这种数据要么列数设得巨大留一堆空位要么拆成多张表再靠外键关联两头都麻烦。JSON的嵌套数组结构天生适配这种形态。cmid对应的一条记录检查项明细就放进items数组每项包含检查代码、结果值、参考范围、结果状态等子字段。这样输出的是一个自包含的JSON对象下游拿到就能直接解析不需要再走一次关联查询。在设计嵌套结构时我踩过的坑是没有限制数组长度。一条记录几十个检查项没问题但如果医嘱明细有上千条整个JSON对象会膨胀得很难看解析性能也受影响。所以我在提取脚本里一般会加一个可配置的明细条数上限超出部分单独出日志而不是一刀切截断。这个参数在真实场景里的默认值一般设在500左右具体看下游消费方的接口限制。3. Python提取脚本从数据库到规整JSON文件3.1 主流程脚本查询、清洗、序列化一气呵成摸排清楚源表结构后就可以把提取逻辑写成脚本。我一般用Python做这件事原因很简单pandas做清洗方便json模块做序列化顺手而且连接各种数据库都有现成驱动不用换语言。下面这段脚本是最基础的主流程适用于源数据落在MySQL或Oracle的情况。换成其他数据库只需要改连接串和方言即可import json import pandas as pd from datetime import datetime, date from decimal import Decimal DB_CONFIG { host: 127.0.0.1, port: 3306, user: reader, password: your_password, database: medical_db, charset: utf8mb4 } def fetch_rows(): import pymysql conn pymysql.connect(**DB_CONFIG) query SELECT t1.record_no AS cmid, t2.name AS patient_name, t1.exam_datetime, t1.diagnosis_text, t1.item_details FROM exam_record t1 LEFT JOIN patient_info t2 ON t1.patient_id t2.patient_id WHERE t1.exam_datetime %(start)s AND t1.exam_datetime %(end)s params {start: 2025-01-01 00:00:00, end: 2025-04-01 00:00:00} df pd.read_sql(query, conn, paramsparams) conn.close() return df def parse_items(raw): if raw is None: return [] if isinstance(raw, str): try: return json.loads(raw) except ValueError: return [] if isinstance(raw, list): return raw return [] def default_serializer(obj): if isinstance(obj, (datetime, date)): return obj.strftime(%Y-%m-%d %H:%M:%S) if isinstance(obj, Decimal): return float(obj) raise TypeError(fType {type(obj)} not serializable) def build_records(df): records [] seen set() for _, row in df.iterrows(): cmid str(row[cmid]) if cmid in seen: continue seen.add(cmind) records.append({ cmid: cmid, patient_name: row[patient_name] or , exam_datetime: row[exam_datetime], diagnosis: (row[diagnosis_text] or ).strip(), items: parse_items(row[item_details]) }) return records if __name__ __main__: source_df fetch_rows() result build_records(source_df) with open(cmid_records.json, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2, defaultdefault_serializer) print(ftotal records: {len(result)})这段脚本做的事情可以拆成四段来看。DB_CONFIG连接参数里charsetutf8mb4是必须的医学文本里经常出现特殊符号和生僻字只用utf8在某些情况下会丢字符。fetch_rows函数里用pandas的read_sql直接执行查询好处是结果自动变成DataFrame后续iterrows遍历起来方便注意SQL里的%(start)s和%(end)s是参数占位符由params字典传入避免SQL注入的同时也方便调整抽取时间窗口。build_records函数是清洗的核心。seen集合做cmid去重防止源表里本就存在重复记录导致输出重复row[patient_name] or 这段是把None转成空字符串保证JSON里不会出现null下游解析时少一个分支判断items字段用parse_items函数兜底因为item_details在源库里可能是JSON字符串、也可能是文本形式的列表统一解析成真正的list再输出。default_serializer函数是序列化的兜底方案datetime、date、Decimal这些类型json.dumps默认不认这里统一转成字符串或浮点数并给出明确报错提示避免脚本在数据量跑大时突然挂在某条记录的类型上。序列化参数里ensure_asciiFalse必须带上否则json.dumps会把中文全部转成\uXXXX的转义序列文件一打开全是乱码没法直接给下游用。indent2是缩进格式化便于人工抽查正式对接时如果对文件大小敏感可以改成indentNone配合separators参数压缩体积。3.2 时间窗口提取增量抽取参数怎么设医学数据提取和普通日志提取不一样的地方在于数据必须按时间窗口分批跑不能一次性全量。全量抽取在数据量大时会让数据库连接长时间占住也会让脚本内存吃紧。稳妥的做法是每次抽取一个时间区间用参数控制。我把时间窗口设计成这样import sys from datetime import datetime, timedelta if __name__ __main__: # 用法: python extract.py 2025-01-01 2025-01-31 start_date sys.argv[1] end_date sys.argv[2] start_dt datetime.strptime(start_date, %Y-%m-%d) end_dt datetime.strptime(end_date, %Y-%m-%d) timedelta(days1) source_df fetch_rows( startstart_dt.strftime(%Y-%m-%d %H:%M:%S), endend_dt.strftime(%Y-%m-%d %H:%M:%S) ) result build_records(source_df) output_name fcmid_records_{start_date}_to_{end_date}.json with open(output_name, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2, defaultdefault_serializer) print(ftotal records: {len(result)}, output: {output_name})这段脚本在命令行直接传起止日期。注意end_dt这里加了一天这是刻意为之因为业务数据里“当天”的语义是到当天23:59:59为止SQL里用小于号比较时如果直接传end_date会丢最后一小时的数据。用半开区间[ start, end )来切时间窗口是工程上最稳的做法。这个参数设计的好处是支持增量跑批。第一次全量跑历史区间之后每天只需要跑前一天的窗口比如python extract.py 2025-06-01 2025-06-01就能补齐一天的增量。把输出文件名带上日期区间即使重复跑也不会覆盖历史文件这在对接上游数据修正时特别有用。3.3 DataFrame逐行遍历太慢换成分批查询如果源表数据量超过百万行pandas的read_sql一次读全表内存肯定扛不住。而且build_records里的iterrows逐行遍历效率也一般这个写法在小数据集上没问题大数据量下就要换策略。常见的做法是分批查询。每次只取一个时间窗口内的数据窗口粒度从一天缩小到一小时或者用LIMIT OFFSET做分页。我更倾向于缩小时间窗口而不是用OFFSET因为OFFSET在深分页时数据库端会做大量无效扫描时间窗口走索引效率高得多。from datetime import datetime, timedelta def fetch_batch(start_dt, end_dt, batch_hours6): current start_dt while current end_dt: next_dt min(current timedelta(hoursbatch_hours), end_dt) yield current.strftime(%Y-%m-%d %H:%M:%S), next_dt.strftime(%Y-%m-%d %H:%M:%S) current next_dt for start, end in fetch_batch(start_dt, end_dt): batch_df fetch_rows(startstart, endend) # 每批独立处理结果是可累加的 batch_result build_records(batch_df) all_records.extend(batch_result)fetch_batch是一个生成器按batch_hours指定的时长切分时间段每次产出一个小窗口的起止时间。主循环里对每个小窗口调用fetch_rows拿到当前批次的DataFrame后处理完再合并到总结果集。这样单次查询的数据量被控制在几万行以内内存和数据库压力都可控。batch_hours这个参数要看库的性能来调。我在实际项目里一般从6小时开始试如果数据库负载低可以加到12小时减少循环次数如果查询变慢就缩到2小时。调参的观察指标是单次查询耗时控制在3秒以内比较理想。4. JSON再加工嵌套结构、去重与编码一把捋清4.1 把平铺字段改造成嵌套结构上一章的主流程脚本输出的JSON是最简结构但很多对接方要求的是嵌套格式比如把检查项目列表、诊断列表、化验结果分别放到独立的子对象里。这个时候就要在build_records里做结构重构。def build_nested_records(df): records [] seen set() for _, row in df.iterrows(): cmid str(row[cmid]) if cmid in seen: continue seen.add(cmid) record { cmid: cmid, patient: { name: row[patient_name] or , age: row[age] if row[age] is not None else -1 }, exam: { datetime: row[exam_datetime], diagnosis: (row[diagnosis_text] or ).strip() }, items: parse_items(row[item_details]) } records.append(record) return records嵌套结构的价值在于下游解析代码可以按固定路径取值。record[patient][name]这种访问方式比平铺字段更接近业务语义尤其在对接方用Java写反序列化类时嵌套结构能直接映射到对应的POJO类上。age字段这里故意在空值时输出-1而不是null或0因为0会被误当成真年龄-1在业务侧更容易被识别为无效值。这种改造方式有个前提源表字段本身已经包含了所需的全部信息。如果patient维度和exam维度不在同一张表里就得在SQL层先做JOIN或者分多次查询后在Python侧合并。合并的键仍然是cmid这也就是为什么第一步摸底cmid位置那么重要。4.2 数组字段的清洗重复项和脏数据怎么处理items数组里最容易出现两个问题重复项和脏字符串。重复项的来源常见于源表多个科室重复上报同一条检查数据脏字符串则常见于接口报文里混入了转义符或换行符。def clean_items(items): cleaned [] seen_keys set() for item in items: if not isinstance(item, dict): continue key f{item.get(item_code, )}-{item.get(item_name, )}-{item.get(result_value, )} if key in seen_keys: continue seen_keys.add(key) cleaned.append({ item_code: str(item.get(item_code, )).strip(), item_name: str(item.get(item_name, )).replace(\n, ).strip(), result_value: str(item.get(result_value, )).strip(), result_status: str(item.get(result_status, )).strip() }) return cleanedclean_items函数按“检查代码-检查名称-结果值”拼接成唯一键做去重相比于只按item_code去重能保留那些代码相同但结果不同的记录避免误杀。item_name里的换行符被替换成空格这个细节很重要因为JSON文件在Windows记事本里打开时换行符会破坏缩进结构人工检查时非常闹心。这条链路做完后用json模块验证一下结果是值得的。命令行里直接跑一句python -m json.tool文件就能做语法校验如果格式化通过说明JSON结构合法可以直接交给下游。4.3 编码与格式细节这些坑一次避开医学文本数据里最常见的编码问题就是GBK和UTF-8混用。源数据库如果是老系统数据可能是GBK存储的连接串里如果不指定charsetpandas读出来就是乱码。我习惯在连接参数里先试utf8mb4如果发现读取结果出现“锟斤拷”之类的字符再改成gbk重新读。另外有两件事我每次都会做。第一是统一输出时的换行符用Python写文件时默认会用在当前平台换行符Unix下是\nWindows下可能变成\r\n。对接方如果做了哈希校验换行不同会导致结果不一致。固定写法是用open时带上newline\nwith open(cmid_records.json, w, encodingutf-8, newline\n) as f: json.dump(result, f, ensure_asciiFalse, indent2, defaultdefault_serializer)第二是格式化工具的使用习惯。json.dump的indent2是给人类看的但对接方如果要求的是单行紧凑JSON体积能小一半以上。两种格式我都各存一份方便不同场景使用。紧凑格式只要把indent去掉加上separators(,, :)就行。5. 避坑与排查实录cmid提取中常见的五类问题5.1 输出的JSON里中文变成一串转义符现象json.dump之后打开输出文件看到中文全部变成了\u5f20\u4e09这样的字符完全没法直接读。原因json.dump默认的ensure_ascii参数是TruePython在做序列化时不放心非ASCII字符会把所有中文转为Unicode转义序列以保证纯ASCII输出。解决写文件时显式传ensure_asciiFalse。顺带检查打开文件的编码必须用utf-8读取才能正确显示中文。从那以后我每次写JSON输出文件都会把ensure_asciiFalse这一条写死在代码模板里不再依赖默认值。5.2 Java侧报JSON格式错误无法反序列化Date类型现象前端Java程序解析JSON时报错提示json parse error: cannot deserialize value of type java.util.Date from string但用Python读这个JSON完全没问题。原因Python的json序列化把datetime转成的字符串格式是“YYYY-MM-DD HH:MM:SS”而Java的Date反序列化默认用的是ISO8601格式两者在时间格式上不一致Java侧解析时就崩了。解决两种方案任选。一种是在Python侧把日期格式改成带T分隔符的ISO8601格式strftime(%Y-%m-%dT%H:%M:%S)另一种是在Java侧的ObjectMapper上配置日期格式。我一般选第一种因为改源头输出比改下游配置要简单而且对其他语言也更友好。5.3 cmid重复导致JSON里同键覆盖丢数据现象生成的文件校验时发现总记录数比源表少查日志发现部分cmid只输出了一条记录。原因build_records里用seen集合做了去重但去重策略是“保留先出现的记录”如果源表中同一个cmid对应多条不同内容的数据后出现的都被跳过了。解决先确认业务上cmid是否允许重复。如果不允许说明源表数据有垃圾需要找人修源如果允许说明cmid本身不是唯一键需要改成复合键去重或者在提取策略上做合并而不是去重。我在做这类提取时会先在SQL层用GROUP BY检查一遍重复率而不是直接交给Python去重。5.4 字段值为None导致下游数据库写入失败现象JSON文件里某些记录的某个字段是null下游入库时该字段非空约束报错。原因源表里该字段本身就为空Python读出来是Nonejson.dump序列化成了null。下游如果要求这个字段必填就会直接撞上约束。解决在build阶段就把None统一替换成业务约定的默认值。字符串字段用空字符串数字字段用-1或0日期字段用空字符串。替换逻辑写在开头提到的default_serializer里不生效因为那个函数只处理类型转换不处理空值空值替换必须在构造记录时就做。5.5 代码没问题但输出文件体积大一倍现象同样的数据量两次生成的JSON文件大小差很多对比后发现是中文被转成了转义序列。原因ensure_ascii没有设成False时每个中文字符被展开成6个字符的转义序列文件体积自然膨胀。中文占比越高膨胀越明显。解决用ensure_asciiFalse输出原生UTF-8文件。这种情况常见于把脚本从一台机器复制到另一台时配置丢失或者换了不同版本的代码模板。我后来养成了习惯输出后立即检查文件开头是否出现\u一出现就知道配置丢了。6. 输出自检用JSON Schema校验给文件上最后一道保险生成JSON文件之后直接交出去总让人不放心。字段缺失、类型偏差、日期格式错误这些问题人工抽查很难发现尤其当记录数上万时。我给输出文件加一层自检用JSON Schema做结构校验在交付前拦截问题。JSON Schema定义好每条记录的预期结构脚本跑完后自动校验一遍。下面这个schema覆盖了cmid、patient、exam、items四个核心字段SCHEMA { type: object, required: [cmid, patient, exam, items], properties: { cmid: {type: string, minLength: 1}, patient: { type: object, required: [name], properties: { name: {type: string}, age: {type: integer} } }, exam: { type: object, required: [datetime], properties: { datetime: {type: string, pattern: ^\\d{4}-\\d{2}-\\d{2}}, diagnosis: {type: string} } }, items: { type: array, items: { type: object, required: [item_code, item_name], properties: { item_code: {type: string}, item_name: {type: string}, result_value: {type: string} } } } } }校验逻辑放在主脚本末尾读回刚写出的JSON文件逐条比对结构。用jsonschema库实现很简单import jsonschema def validate_output(file_path): with open(file_path, r, encodingutf-8) as f: records json.load(f) errors [] for idx, record in enumerate(records): try: jsonschema.validate(instancerecord, schemaSCHEMA) except jsonschema.ValidationError as e: errors.append({index: idx, cmid: record.get(cmid), error: e.message}) if errors: print(fvalidation failed: {len(errors)} errors) for err in errors[:10]: print(err) raise SystemExit(1) print(fvalidation passed: {len(records)} records)validate_output函数读回刚才写出的JSON文件逐条用SCHEMA做校验。jsonschema.validate在实例与schema不符时抛异常捕获后记录出错位置、cmid和错误信息最多打印前10条方便定位。schema里对exam.datetime用了正则pattern校验日期字符串至少以“YYYY-MM-DD”开头这样Java侧再遇到日期解析问题之前就能发现。我在接到“cmid医学数据提取json数据文件”这类需求时会把这套自检固化成流程的固定一环。从那以后我每次生成完JSON都强制跑一遍schema校验和记录数比对确认无误后才算交付完成宁可多花这一分钟也不想让下游拿着有问题的文件来回扯皮。希望帮到你。本文还有配套的精品资源点击获取