☰
检测数据清洗格式转换:手工改完说不清动了哪一处怎么办
2026/10/9 5:38:15 网站建设 项目流程

仪器导出的检测明细交到手上,常常"能看不能用":编号里藏着复制时带上的零宽字符,客户名混着全角空格,一列里既有12.3 mg/L又有0.523,日期有2026/10/06也有2026.10.6,全角 98.50 算不出均值。手工改到第三遍,自己都说不清哪一格动过。

为了解决这个问题,Python 提供了 csv 模块加内置的re与字符串处理:不用装库,把"每一列该怎么洗"写成一张规则表,逐格过规则,每改一处记一条「原值 → 新值 → 规则」。

本文按四步走:配置规则表、读文件、逐格清洗、输出。重点不是"把数据洗好看",而是"洗过的地方怎么留痕、认不出的行怎么办"。脚本跑完自动回读校验,30 项断言钉住每条规则。

先说边界:这三件事别做

清洗工具只做"改写法"这一层,三件事别做:单位换算(改的是数值本身,要改口径得甲方书面确认)、缺失值填补(补出来的值没有来源)、业务去重合并(同一样品两条结果,合并哪条是业务判断)。

第一步:把"哪些列要洗、怎么洗"写成表

两张表。第一张管"怎么把文件读进来":编码、分隔符、表头行各不一样,一种格式一条。

出处:run.py顶部的格式表FORMATS(配置区)

FORMATS = [ { "名称": "仪器导出_逗号分隔", "后缀": (".csv",), "编码": ("utf-8-sig", "gbk"), "分隔符": ",", "必需列": ("样品编号", "客户名称", "检测项目", "结果", "单位", "检测日期"), "判空列": "检测项目", "列映射": { "样品编号": "样品编号", "客户名称": "客户名称", "检测项目": "检测项目", "结果": "结果", "单位": "单位", "检测日期": "检测日期", }, }, { "名称": "仪器导出_制表符分隔", "后缀": (".txt",), "编码": ("gbk", "utf-8-sig"), "分隔符": "\t", "必需列": ("样品编号", "客户名称", "检测项目", "结果", "单位", "检测日期"), "判空列": "检测项目", "列映射": { "样品编号": "样品编号", "客户名称": "客户名称", "检测项目": "检测项目", "结果": "结果", "单位": "单位", "检测日期": "检测日期", }, }, ]

第二张表是主角:一列一条规则,动作按顺序执行,加字段只加一行。

出处:run.py顶部的清洗规则表RULES(紧跟FORMATS)

RULES = [ {"列": "样品编号", "动作": ["去零宽", "去空白", "全角转半角"], "必填": True}, {"列": "客户名称", "动作": ["去零宽", "去空白", "全角转半角"], "必填": True}, {"列": "检测项目", "动作": ["去零宽", "去空白", "全角转半角", "别名归一"], "必填": True, "映射": {"含量": "含量(%)", "含量%": "含量(%)", "有关物质": "有关物质(%)"}}, {"列": "结果", "动作": ["去零宽", "去空白", "全角转半角"]}, {"列": "单位", "动作": ["去零宽", "去空白", "全角转半角", "别名归一"], "映射": {"mg/l": "mg/L", "MG/L": "mg/L", "ug/ml": "μg/mL"}}, {"列": "检测日期", "动作": ["去零宽", "去空白", "全角转半角", "日期归一"], "必填": True}, ]

这张表里没有任何推断性动作:别名归一只认写好的映射表,表外的值不动。

三个常量:哪些文字算"结论"、结果分哪几态、带单位怎么认。

出处:run.py的受控词表与状态常量区(紧跟RULES)

DECISIONS = {"未检出", "ND"} STATUS_NUM, STATUS_LIMIT, STATUS_TEXT, STATUS_EMPTY = "数值", "低于下限", "文字结论", "空" NUM_UNIT = re.compile(r"^(?P<sign>[<>≤≥]?=?)(?P<num>-?\d+(?:\.\d+)?)(?P<unit>[^\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(encoding=enc) 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:], delimiter=fmt["分隔符"])) header = [c.strip() for c in rows[0]] out = [] for line_no, cells in enumerate(rows[1:], start=header_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: """去掉所有空白,含全角空格 U+3000 与不间断空格 U+00A0(strip() 都不管)。""" return re.sub(r"[\s\u00a0\u3000]+", "", str(value)) def to_halfwidth(value: str) -> str: """全角转半角:只动 0xFF01-0xFF5E(!到~)与全角空格,其他一个字符都不碰。 不图省事用 unicodedata.normalize("NFKC"):它顺手会把 µ(U+00B5) 换成 μ(U+03BC)、 把 Ⅲ 换成 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()管不到全角空格;但它动内部空格,样品名要另配"只去首尾"
全角转半角只动 U+FF01–U+FF5E;NFKC 会把 µ 换成 μ、罗马数字拆开
别名归一映射表由人给,表里没有的不动
日期归一认不出的挂起,不兜今天

结果列不能只存一个数值:<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-sig:Excel 双击打开不乱码。""" with open(path, "w", encoding="utf-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, fieldnames=CLEAN_COLS, extrasaction="ignore", 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。

抓到的典型:全角 98.50 与半全角混写的0.025都转成了能算的数值;三种日期写法(含全角横线)归一成同一个日期;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+ 逐条核对原始数据与记录是否对得上

手里有这类活儿卡着,或者只是想问问能不能自动化,都欢迎评论区聊,先把问题说清楚再谈怎么做。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询