检测数据清洗格式转换：手工改完说不清动了哪一处怎么办

仪器导出的检测明细交到手上常常能看不能用编号里藏着复制时带上的零宽字符客户名混着全角空格一列里既有12.3 mg/L又有0.523日期有2026/10/06也有2026.10.6全角 . 算不出均值。手工改到第三遍自己都说不清哪一格动过。为了解决这个问题Python 提供了 csv 模块加内置的re与字符串处理不用装库把每一列该怎么洗写成一张规则表逐格过规则每改一处记一条「原值 → 新值 → 规则」。本文按四步走配置规则表、读文件、逐格清洗、输出。重点不是把数据洗好看而是洗过的地方怎么留痕、认不出的行怎么办。脚本跑完自动回读校验30 项断言钉住每条规则。先说边界这三件事别做清洗工具只做改写法这一层三件事别做单位换算改的是数值本身要改口径得甲方书面确认、缺失值填补补出来的值没有来源、业务去重合并同一样品两条结果合并哪条是业务判断。第一步把哪些列要洗、怎么洗写成表两张表。第一张管怎么把文件读进来编码、分隔符、表头行各不一样一种格式一条。出处run.py顶部的格式表FORMATS配置区FORMATS [ { 名称: 仪器导出_逗号分隔, 后缀: (.csv,), 编码: (utf-8-sig, gbk), 分隔符: ,, 必需列: (样品编号, 客户名称, 检测项目, 结果, 单位, 检测日期), 判空列: 检测项目, 列映射: { 样品编号: 样品编号, 客户名称: 客户名称, 检测项目: 检测项目, 结果: 结果, 单位: 单位, 检测日期: 检测日期, }, }, { 名称: 仪器导出_制表符分隔, 后缀: (.txt,), 编码: (gbk, utf-8-sig), 分隔符: \t, 必需列: (样品编号, 客户名称, 检测项目, 结果, 单位, 检测日期), 判空列: 检测项目, 列映射: { 样品编号: 样品编号, 客户名称: 客户名称, 检测项目: 检测项目, 结果: 结果, 单位: 单位, 检测日期: 检测日期, }, }, ]第二张表是主角一列一条规则动作按顺序执行加字段只加一行。出处run.py顶部的清洗规则表RULES紧跟FORMATSRULES [ {列: 样品编号, 动作: [去零宽, 去空白, 全角转半角], 必填: True}, {列: 客户名称, 动作: [去零宽, 去空白, 全角转半角], 必填: True}, {列: 检测项目, 动作: [去零宽, 去空白, 全角转半角, 别名归一], 必填: True, 映射: {含量: 含量(%), 含量%: 含量(%), 有关物质: 有关物质(%)}}, {列: 结果, 动作: [去零宽, 去空白, 全角转半角]}, {列: 单位, 动作: [去零宽, 去空白, 全角转半角, 别名归一], 映射: {mg/l: mg/L, MG/L: mg/L, ug/ml: μg/mL}}, {列: 检测日期, 动作: [去零宽, 去空白, 全角转半角, 日期归一], 必填: True}, ]这张表里没有任何推断性动作别名归一只认写好的映射表表外的值不动。三个常量哪些文字算结论、结果分哪几态、带单位怎么认。出处run.py的受控词表与状态常量区紧跟RULESDECISIONS {未检出, ND} STATUS_NUM, STATUS_LIMIT, STATUS_TEXT, STATUS_EMPTY 数值, 低于下限, 文字结论, 空 NUM_UNIT re.compile(r^(?Psign[≤≥]??)(?Pnum-?\d(?:\.\d)?)(?Punit[^\d]*)$)DECISIONS只有两个词不是偷懒未检出、ND是检出结论待检、未测是还没做——混在一起以后统计合格率会多出一批假结论。第二步读文件认格式、切表头、切表尾编码与表头行不靠文件名、不靠行号靠表头特征列后缀先粗筛必需列全中才算认对格式。出处run.py的read_text()与pick_format()读文件一节def read_text(path: Path, encodings) - str: 按候选编码依次试。GBK 文件用 utf-8 读会直接抛 UnicodeDecodeError不是乱码。 last None for enc in encodings: try: return path.read_text(encodingenc) except UnicodeDecodeError as exc: last exc raise last def pick_format(path: Path): 先看后缀再用表头特征列确认——同样后缀、表头不同的两种格式就靠这一步分开。 lines None for fmt in FORMATS: if path.suffix.lower() not in fmt[后缀]: continue if lines is None: lines read_text(path, fmt[编码]).splitlines() for raw in lines[:10]: cells [c.strip() for c in raw.split(fmt[分隔符])] if all(col in cells for col in fmt[必需列]): return fmt, lines return None, None表尾最容易出事说明行切不掉会被当成一条数据。靠判空列切——挑数据行一定有值、说明行一定没值的那列。别拿样品编号或客户名称判空缺值的那两列本身就是要点出来的挂起项。出处run.py的read_export()紧跟pick_format()def read_export(path: Path, fmt, lines) - list[dict]: 切出表头行之后的数据行按判空列丢掉表尾说明行。 header_at None for i, raw in enumerate(lines): cells [c.strip() for c in raw.split(fmt[分隔符])] if all(col in cells for col in fmt[必需列]): header_at i break if header_at is None: return [] rows list(csv.reader(lines[header_at:], delimiterfmt[分隔符])) header [c.strip() for c in rows[0]] out [] for line_no, cells in enumerate(rows[1:], startheader_at 2): rec {h: (cells[i] if i len(cells) else ) for i, h in enumerate(header)} if not rec.get(fmt[判空列], ).strip(): continue # 表尾说明行 / 空行判空列一空就到头 item {dst: rec.get(src, ) for dst, src in fmt[列映射].items()} item[_来源文件] path.name item[_行号] line_no out.append(item) return out_来源文件与_行号给留痕用谁问这条数据动过没有能定位回原始文件第几行。第三步逐格清洗每改一处记一条出处run.py的drop_zero_width()、strip_ws()、to_halfwidth()清洗一节开头三个相邻def drop_zero_width(value: str) - str: 去掉零宽字符——从系统或网页里复制出来的值很常见肉眼完全看不见。 return re.sub(r[\u200b\u200c\u200d\ufeff], , str(value)) def strip_ws(value: str) - str: 去掉所有空白含全角空格 U3000 与不间断空格 U00A0strip() 都不管。 return re.sub(r[\s\u00a0\u3000], , str(value)) def to_halfwidth(value: str) - str: 全角转半角只动 0xFF01-0xFF5E到与全角空格其他一个字符都不碰。 不图省事用 unicodedata.normalize(NFKC)它顺手会把 µ(U00B5) 换成 μ(U03BC)、 把 Ⅲ 换成 III单位名一旦被悄悄改掉两批数据就合不到一起了。 out [] for ch in str(value): code ord(ch) if code 0x3000: out.append( ) elif 0xFF01 code 0xFF5E: out.append(chr(code - 0xFEE0)) else: out.append(ch) return .join(out)日期归一只做能认出来的格式归一认不出返回 None绝不兜成今天。出处run.py的parse_date()紧跟to_halfwidth()def parse_date(raw: str): 2026-10-06 / 2026/10/06 / 2026.10.6 / 20261006 都认认不出返回 None不兜今天。 text str(raw).strip() for pattern in (%Y-%m-%d, %Y/%m/%d, %Y.%m.%d, %Y%m%d): try: return datetime.strptime(text, pattern).strftime(%Y-%m-%d) except ValueError: continue return None清洗引擎照着规则表一格一格过值一变就追加一条留痕。出处run.py的CELL_ACTIONS与clean_cell()紧跟parse_date()CELL_ACTIONS {去零宽: drop_zero_width, 去空白: strip_ws, 全角转半角: to_halfwidth} def clean_cell(value, rule: dict, changes: list, line_no) - tuple[str, str]: 按规则里的动作顺序清洗一格每改一处记一条留痕。返回 (清洗后的值, 挂起原因)。 cur str(value if value is not None else ) col rule[列] for action in rule[动作]: if action 别名归一: new rule.get(映射, {}).get(cur, cur) # 映射表里没有的一个字都不动 if new ! cur: changes.append((line_no, col, action, cur, new)) cur new elif action 日期归一: new parse_date(cur) if new is None: return cur, f检测日期认不出{cur} if new ! cur: changes.append((line_no, col, action, cur, new)) cur new else: new CELL_ACTIONS[action](cur) if new ! cur: changes.append((line_no, col, action, cur, new)) cur new return cur, 五个动作看着都简单但每个背后都有坑动作为什么这么写去零宽肉眼看不见会让看起来一样的编号对不上去空白strip()管不到全角空格但它动内部空格样品名要另配只去首尾全角转半角只动 UFF01–UFF5ENFKC 会把 µ 换成 μ、罗马数字拆开别名归一映射表由人给表里没有的不动日期归一认不出的挂起不兜今天结果列不能只存一个数值0.01按 0.01 算、未检出按 0 算整列均值被悄悄拉低还不报错。所以拆成原值、数值、状态三份。出处run.py的split_result()紧跟clean_cell()def split_result(text: str): 结果原值 → (数值, 单位, 状态, 挂起原因)。四态数值 / 低于下限 / 文字结论 / 空。 if not text: return None, , STATUS_EMPTY, if , in text: return None, , , 结果含逗号分不清千分位还是分隔符 m NUM_UNIT.match(text) if m: unit m.group(unit).strip() if m.group(sign): return None, unit, STATUS_LIMIT, return float(m.group(num)), unit, STATUS_NUM, if text in DECISIONS: return None, , STATUS_TEXT, return None, , , f结果认不出且不在受控词表内{text}3,250要单独说逗号可能是千分位也可能是分隔符——分不清就不许猜直接挂起。结果里带的单位拆到单位列但不做任何换算。最后把关的是clean_row()任何一列认不出就整行挂起前面已改好的格子也一并作废——留痕表里不能留下对不上明细的孤儿记录。出处run.py的clean_row()紧跟split_result()def clean_row(row: dict) - tuple[dict | None, list, str]: 清洗一行先过规则表再拆结果列。任何一列认不出就整行挂起——不改也不记留痕。 line_no row.get(_行号) changes: list [] out {来源文件: row.get(_来源文件, ), 行号: line_no} for rule in RULES: col rule[列] value, reason clean_cell(row.get(col, ), rule, changes, line_no) if reason: return None, changes, reason if rule.get(必填) and not value: return None, changes, f{col}为空 out[col] value value, unit, status, reason split_result(out[结果]) if reason: return None, changes, reason out[结果数值], out[结果状态] value, status out[单位] unit or out[单位] # 结果里带了单位就用它没带才用单位列 return out, changes, 挂起行的留痕一并作废整行不产出留痕表里就不能留下对不上明细的孤儿记录。第四步输出四份文件输出四份数据、留痕、挂起、台账。留痕表是真正的价值谁问你动了哪些格子原值新值并排列着一句不用解释。出处run.py的run_all()输出一节def run_all(): 扫 01_raw_data/ 全部文件读 → 清洗 → 归堆。返回 (明细, 留痕, 挂起, 台账)。 clean_rows, changes, pending, ledger [], [], [], [] for path in sorted(RAW_DIR.iterdir()): if not path.is_file(): continue fmt, lines pick_format(path) if fmt is None: # 认不出的文件照样登记不安静跳过 pending.append({来源文件: path.name, 行号: , 挂起原因: 文件格式认不出}) ledger.append({来源文件: path.name, 读出行数: 0, 输出行数: 0, 留痕条数: 0, 挂起条数: 1, 判定: 挂起}) continue read_n out_n chg_n pend_n 0 for row in read_export(path, fmt, lines): read_n 1 cleaned, cell_changes, reason clean_row(row) if cleaned is None: pend_n 1 pending.append({来源文件: path.name, 行号: row.get(_行号), 挂起原因: reason}) continue # 整行作废前面已改好的格子也一起作废 out_n 1 clean_rows.append(cleaned) for line_no, col, action, old, new in cell_changes: changes.append({来源文件: path.name, 行号: line_no, 列: col, 动作: action, 原值: old, 新值: new}) chg_n len(cell_changes) ledger.append({来源文件: path.name, 读出行数: read_n, 输出行数: out_n, 留痕条数: chg_n, 挂起条数: pend_n, 判定: PASS if out_n else 挂起}) return clean_rows, changes, pending, ledger写文件一律utf-8-sig另算明细指纹用跑两遍证明清洗是纯函数。出处run.py的write_csv()与digest()紧跟run_all()def write_csv(path: Path, header: list, rows: list) - None: 一律 utf-8-sigExcel 双击打开不乱码。 with open(path, w, encodingutf-8-sig, newline) as f: w csv.writer(f) w.writerow(header) for row in rows: w.writerow([ if row.get(c) is None else row.get(c) for c in header]) def digest(rows: list) - str: 把明细摊成一行行文本算指纹用来证明同一份输入跑两遍结果一字不差。 buf io.StringIO() w csv.DictWriter(buf, fieldnamesCLEAN_COLS, extrasactionignore, lineterminator\n) for row in rows: w.writerow({c: ( if row.get(c) is None else row.get(c)) for c in CLEAN_COLS}) return hashlib.sha1(buf.getvalue().encode(utf-8)).hexdigest()run.py另有一段工程外壳正文不贴回读断言、运行日志、结束暂停。实测跑了什么五个输入文件读出 18 行输出 14 行、留痕 17 条、挂起 5 条留痕按动作分布是 1/5/4/5/2。抓到的典型全角 . 与半全角混写的0.0都转成了能算的数值三种日期写法含全角横线归一成同一个日期12.3 mg/L拆成数值与单位。五条挂起日期写待定、结果写待检词表外、结果写3,250、客户名称空、一个不是数据文件的说明文本。一条没进明细也一条没被丢掉。总结清洗这类活难点不是会不会写正则而是边界划在哪只改写法不改含义换算、补值、合并都不进工具每处改动都留痕原值列永远保留认不出的一律挂起不猜、不丢、不兜默认值。守住这条线工具再简单也拿得给 QC 看。完整源码本文配套示例已开源五个不同格式的输入文件、五类挂起、30 项回读断言都在里面克隆下来直接复跑huang_jianhua0101/examples - Gitee.com关于我在实验室一线待了 13 年9 年制药 4 年第三方检测做的一直是实验室信息化。做过 STARLIMS 的甲方 PM——一期、二期两轮上线都由我主导招标到 3Q 验证到验收全流程也在系统上自己做过二次开发——把纸质的账号申请流程搬到线上跑在 STARLIMS 之前还有 6 年多 CS 架构 LIMS 的使用与运维经验其中一段经 Citrix 远程接入。现在专做实验室里那些重复劳动报表自动生成、仪器数据对接、合规文档批量处理。本科物理化学、硕士计算机化学既听得懂 QA 说的变更控制也看得懂仪器导出的原始数据长什么样。SOP、偏差、OOS、样本流转这些词不用你解释。现在主要做这几类- 检验报告与台账批量生成模板不动数据自动填格式一步不错- 仪器数据对接色谱、光谱、酶标仪导出的原始文件解析、清洗、入库、转成报表- 合规文档自动化SOP、验证方案、批记录这类重复文档的批量生成与核对- 数据完整性核查按 ALCOA 逐条核对原始数据与记录是否对得上手里有这类活儿卡着或者只是想问问能不能自动化都欢迎评论区聊先把问题说清楚再谈怎么做。