检测机构的检测数据是散在客户名下的。每个月客服都得给十几家客户各出一份汇总报表:这家按检测类别看合格率,那家要按委托单看检出情况;同一列,一家叫「检测类别」,另一家得叫「检测项目类别」。同一套数据,报表要出十几份,手工只能一家家筛。
为了解决这个问题,Python 提供了csv + openpyxl:csv 把三张表读进内存,openpyxl 按模板复制页签、按标签找格子填数。真正省事的不是「批量填表」,而是把每家客户的要求写成一张客户报表口径表——分组维度、表头叫法、要不要列明细,脚本读表决定怎么出。
本文按四步走:钉住路径字段、读三张表归堆、逐项判定汇总、按客户口径出页。
一、卡点不在「填表」,在这三件事
- 每家客户的口径不一样。分组维度、表头叫法、要不要明细,都是一家一个说法。
- 合格判定不是统一阈值。pH 判区间、限量项判上限、微生物判文字结论;结果值还有「<0.01」这种报告值。
- 有些项这个月还算不出来。报告还在审核、标准表里没登记、结果值整格空——既不能当合格,也不能悄悄丢掉。
二、四步走
步骤一:路径、字段、期间钉在一处
要钉的就是三件事:数据在哪、表头叫什么、统计哪一段时间。
出处:run.py顶部的路径与字段常量区
BASE = Path(__file__).resolve().parent RAW_DIR = BASE / "01_raw_data" OUT_DIR = BASE / "02_output" SRC_DIR = BASE / "source" TPL_FILE = SRC_DIR / "templates" / "客户检测数据汇总报表模板.xlsx" LOG_FILE = SRC_DIR / "run_log.txt" RESULT_FILE = RAW_DIR / "检测结果明细.csv" SPEC_FILE = RAW_DIR / "客户报表口径表.csv" RULE_FILE = RAW_DIR / "判定标准表.csv" REPORT_FILE = OUT_DIR / "客户检测数据汇总报表.xlsx" BAD_FILE = OUT_DIR / "不合格项明细.csv" PERIOD = "2026-09" # 统计期间 CLOSE_DATE = "2026-09-30" # 数据截止日:这天之后出的报告不算本期 # 结果明细的列名(跟着仪器导出的表头写,换表头就换这里) COL_ORDER = "委托单号" COL_CLIENT = "客户名称" COL_SAMPLE = "样品编号" COL_CATEGORY = "检测类别" COL_ITEM = "检测项目" COL_VALUE = "结果值" COL_UNIT = "单位" COL_STATUS = "报告状态" COL_REPORT = "报告日期" STATUS_DONE = "已出报告" STATUS_VOID = "已作废" INDICATORS = ["送检样品数", "检测项数", "已判定项数", "合格项数", "不合格项数", "未完成项数", "待确认项数", "合格率"]列名跟着仪器导出的表头写,换一张导出模板只改这一个常量区。PERIOD和CLOSE_DATE写成常量,昨天和今天跑出来的报表才会一模一样。
步骤二:读三张表,按客户归堆
三张表各有各的「一定有值的那一列」,拿它判空跳过表尾合计行;客户名先归一化再当键。
出处:norm()、clean_name()、read_table()(文件中部)
def norm(text): """比较用的归一化:去掉所有空白(含全角空格),统一成字符串。 客户名尾巴粘一个空格或全角空格(U+3000),不归一化就会把同一家客户拆成两三家。 """ return "".join(str(text or "").split()).replace("\u3000", "") def clean_name(text): """显示用客户名:只去掉首尾空白,中间的字保留原样。""" return str(text or "").strip()def read_table(path, key_col): """按 key_col 判空读一张 CSV:该列为空的行(表尾合计/说明行)一律跳过。 判空列必须一张表传一个——拿结果明细的「委托单号」去读口径表, 那张表根本没有这一列,每行都会被当成空行丢掉,整张表读空还不报错。 """ if not path.exists(): raise FileNotFoundError(f"找不到数据文件:{path.name}") with open(path, "r", encoding="utf-8-sig", newline="") as fh: raw_rows = list(csv.DictReader(fh)) rows = [] for row in raw_rows: if norm(row.get(key_col)) == "": continue rows.append({k: str(v or "").strip() for k, v in row.items()}) if not rows: raise ValueError(f"{path.name} 按「{key_col}」判空后一行都没读到,先看列名对不对") return rowskey_col当参数传,是这一节唯一要记的事。第一版我把判空列写死了,拿「委托单号」去读客户口径表——那张表没有这一列,每一行都被判成空行,整张口径表读成空的,代码一行错都不报,最后表现为所有客户都挂起。
步骤三:判定与汇总(关键一步)
这段分三层:解析结果值、按标准判定、同一样品同一项目去重取最新。
出处:parse_value()、judge()、pick_latest()(文件中部)
def parse_value(text): """解析结果值:返回 (符号, 数值);符号是「=」「<」「>」,读不出数值时返回原文。 仪器导出的结果值不只有纯数字:有「<0.01」这种报告值、有「未检出」这种文字结论, 也有整格空的(结果没传回来)。直接 float() 会在这些行上崩。 """ raw = norm(text) if raw == "": return None, None head, tail = raw[0], raw[1:] if head in "<<": try: return "<", float(tail) except ValueError: return raw, None if head in ">>": try: return ">", float(tail) except ValueError: return raw, None try: return "=", float(raw) except ValueError: return raw, Nonedef judge(value_text, rule): """按判定标准判一项结果:返回「合格」「不合格」,判不了返回 None。 判定方式不是只有上下限——只卡上限的(限量值)、只有下限的、纯文字的(微生物结论), 所以这里按 rule 里的方式分派,不写死一个比较符号。 """ way = norm(rule.get("判定方式")) low = norm(rule.get("下限")) high = norm(rule.get("上限")) expect = norm(rule.get("合格结论文字")) sign, num = parse_value(value_text) if sign is None: return None if way == "文字": return "合格" if sign == expect else "不合格" if sign in ("<", ">"): # 报告值只给了边界:<L 落在上限以内算合格,>L 落在下限以上算合格,跨界的说不清 if sign == "<": return "合格" if high and num <= float(high) else None return "合格" if low and num >= float(low) else None if num is None: return None if way == "上限": return "合格" if high and num <= float(high) else "不合格" if way == "下限": return "合格" if low and num >= float(low) else "不合格" if way == "区间": ok = True if low: ok = ok and num >= float(low) if high: ok = ok and num <= float(high) return "合格" if ok else "不合格" return Nonedef pick_latest(rows): """同一样品同一项目可能有好几条结果(复检重出):剔掉作废的,再取报告日期最新的一条。 作废件不能参与「取最新」——它是被撤回的结论,日期再新也不算数。 两条并列为最新(同一天出两份报告)说明数据有问题,返回 None 交给人工。 """ live = [r for r in rows if norm(r.get(COL_STATUS)) != STATUS_VOID] if not live: return None, "只剩被作废的结果" latest = max(live, key=lambda r: norm(r.get(COL_REPORT)) or "0000-00-00") same = [r for r in live if norm(r.get(COL_REPORT)) == norm(latest.get(COL_REPORT))] if len(same) > 1: return None, "同一报告日期有两条结果,需要人工确认" return latest, ""| 动作 | 为什么这么写 |
|---|---|
| 按表里的「判定方式」分派 | 判定不止上下限:有限量、有下限、有文字结论 |
| 「<L」只跟上限比 | 报告值只给了边界,<0.01对上限 0.01 就是合格 |
空值返回None,不给 0 | 空值当 0,微生物那项会变成「合格」 |
| 先剔作废再取最新 | 作废件是被撤回的结论,日期再新也不算数 |
| 同日期两条并列 → 挂起 | 一天出两份报告是数据有问题,宁可交人工 |
出处:summarize_client()(文件中部,判定与汇总的主函数)
def summarize_client(rows, spec, rules): """一家客户的汇总:按「样品 + 检测项目」去重,逐项判定,再按客户口径分组统计。 去重这步不能省——复检重出的样品在明细里留下两三条同项目的结果, 直接汇总等于把这些项算了两遍;取哪一条、作废的怎么处理都在 pick_latest() 里。 """ items = {} for row in rows: items.setdefault((norm(row.get(COL_SAMPLE)), norm(row.get(COL_ITEM))), []).append(row) live_items = {k: v for k, v in items.items() if any(norm(r.get(COL_STATUS)) != STATUS_VOID for r in v)} stat = {name: 0 for name in INDICATORS} stat["送检样品数"] = len(sample_set(rows)) stat["检测项数"] = len(live_items) detail, bad, pending = {}, [], [] dim_col = COL_CATEGORY if norm(spec["分组维度"]) == "检测类别" else COL_ORDER for (sample, item), group in live_items.items(): bucket = detail.setdefault(norm(group[0].get(dim_col)), {"检测项数": 0, "合格": 0, "不合格": 0}) bucket["检测项数"] += 1 latest, note = pick_latest(group) if latest is None: stat["待确认项数"] += 1 pending.append((sample, item, note)) continue if norm(latest.get(COL_STATUS)) != STATUS_DONE: # 报告还没出:不计入合格率分母,也不算丢——单独计一个数 stat["未完成项数"] += 1 continue rule = rules.get((norm(latest.get(COL_CATEGORY)), norm(latest.get(COL_ITEM)))) if rule is None: stat["待确认项数"] += 1 pending.append((sample, item, "判定标准表里没有这一项")) continue verdict = judge(latest.get(COL_VALUE), rule) if verdict is None: stat["待确认项数"] += 1 pending.append((sample, item, "结果值读不出,判不了")) continue stat["已判定项数"] += 1 stat["合格项数" if verdict == "合格" else "不合格项数"] += 1 bucket["合格" if verdict == "合格" else "不合格"] += 1 if verdict == "不合格": bad.append({"客户名称": clean_name(latest.get(COL_CLIENT)), "委托单号": latest.get(COL_ORDER), "样品编号": latest.get(COL_SAMPLE), "检测类别": latest.get(COL_CATEGORY), "检测项目": latest.get(COL_ITEM), "结果值": latest.get(COL_VALUE), "单位": latest.get(COL_UNIT), "判定标准": rule_text(rule), "报告日期": latest.get(COL_REPORT)}) return {"stat": stat, "detail": detail, "bad": bad, "pending": pending}三档要分清楚:未完成(报告还没出)与待确认(标准没登记、结果读不出)都不进合格率分母,前者只单独计数,后者还要挂出来给人补。分组合格率的分母是这一组的已判定项数——示例里同一委托单下 3 项、1 项待确认、2 项合格,算出来是 100.0% 而不是 66.7%。
步骤四:按客户口径出页
模板只留一套版式,页签按客户复制;填数按标签名反查行号。
出处:label_row()、section_row()、write_client_sheet()(文件后半)
def label_row(ws, text): """按 A 列的标签名反查行号——模板挪一行也不会填错地方。""" for r in range(1, ws.max_row + 1): if norm(ws.cell(r, 1).value) == norm(text): return r return None def section_row(ws, prefix): """找「一、」「二、」「三、」这种小节标题所在行。""" for r in range(1, ws.max_row + 1): if norm(ws.cell(r, 1).value).startswith(prefix): return r return None def write_client_sheet(ws, client, spec, summary): """把一家客户的汇总结果填进按模板复制出来的这一页。""" ws.cell(2, 2).value = client ws.cell(3, 2).value = PERIOD stat = summary["stat"] for name in INDICATORS: row = label_row(ws, name) if name == "合格率": ws.cell(row, 2).value = rate_text(stat["合格项数"], stat["已判定项数"]) else: ws.cell(row, 2).value = stat[name] head = section_row(ws, "二、") + 1 ws.cell(head, 1).value = spec["表头组名"] for i, (gkey, cell) in enumerate(summary["detail"].items()): r = head + 1 + i ws.cell(r, 1).value = gkey ws.cell(r, 2).value = cell["检测项数"] ws.cell(r, 3).value = cell["合格"] ws.cell(r, 4).value = cell["不合格"] ws.cell(r, 5).value = rate_text(cell["合格"], cell["合格"] + cell["不合格"]) bad_head = section_row(ws, "三、") + 1 if spec["含不合格明细"] == "是": for i, item in enumerate(summary["bad"]): r = bad_head + 1 + i ws.cell(r, 1).value = item["样品编号"] ws.cell(r, 2).value = item["检测项目"] ws.cell(r, 3).value = f"{item['结果值']} {item['单位']}".strip() ws.cell(r, 4).value = item["判定标准"] ws.cell(r, 5).value = item["报告日期"] else: ws.cell(bad_head + 1, 1).value = "按客户口径,本报表只给统计数,不列不合格明细。"按标签反查行号,是因为模板几乎一定会被改版:加一行表头、删一行标题,写死的坐标就全漂,而且漂了不报错。分组列的表头也不写死,直接取口径表里的「表头组名」。
出处:main()(文件末尾的读表与循环,RATE_TEXT之后)
results = read_table(RESULT_FILE, COL_ORDER) specs = {norm(r["客户名称"]): r for r in read_table(SPEC_FILE, "客户名称")} rules = {(norm(r["检测类别"]), norm(r["检测项目"])): r for r in read_table(RULE_FILE, "检测项目")} OUT_DIR.mkdir(parents=True, exist_ok=True) wb = load_workbook(TPL_FILE) tpl = wb["报表"] summary_rows, bad_all, pending_clients = [], [], [] for key, rows in group_by_client(results).items(): display = clean_name(rows[0].get(COL_CLIENT)) spec = specs.get(key) if spec is None: # 客户没进口径表:不猜口径、不出这一页,只在一览表里挂起 pending_clients.append(display) summary_rows.append([display, len(sample_set(rows)), "—", "—", "—", "—", "未进客户报表口径表,先补口径再出报表"]) continue summary = summarize_client(rows, spec, rules) sheet = wb.copy_worksheet(tpl) sheet.title = display[:31] write_client_sheet(sheet, display, spec, summary) stat = summary["stat"] summary_rows.append([display, stat["送检样品数"], stat["检测项数"], stat["合格项数"], stat["不合格项数"], rate_text(stat["合格项数"], stat["已判定项数"]), "正常"]) bad_all.extend(summary["bad"]) del wb["报表"] fill_summary(wb["汇总"], summary_rows) wb.save(REPORT_FILE) write_bad_csv(bad_all)出不了报表的客户不上报表面。它没进口径表,就在「汇总」页留一行备注,不猜默认口径硬塞。补上口径再跑一次,页签自然就出来了。另外del wb["报表"]不能漏——模板页留着,客户打开文件先看到的是一张空白表。(fill_summary()把一览写进首页、write_bad_csv()把不合格项落成 CSV,都在run.py里紧跟其后。)
三、几个一上手就会踩的坑
- 判空列写死。换一张表就得重新问「靠哪一列判空」;那列要是新表里根本没有,整张表读空还不报错。
- 模板明细空行必须画边框。不画的话 openpyxl 读回来这些行根本不存在(
max_row只到表头)。 - 客户名去空白再当键。尾巴粘个全角空格就是两家客户。
- 作废件不参与「取最新」。作废行的日期往往更新,按日期取最新会把不合格洗成合格。
- 分组合格率的分母是已判定项数。拿整组项数当分母,待确认的项会拉低合格率。
- 算不出来给「—」,不给 0%。一条都没判出来就印 0%,会被读成「全不合格」。
总结
多客户汇总报表的难点不在循环,在口径:把每家客户的要求挪进一张表,脚本读表决定分组维度和表头叫法;把判定方式挪进另一张表,判定逻辑就不再写死。剩下的三件事——复检去重、作废排除、未出报告单独计数——最容易出错,也最不容易被发现。示例可复跑,回读断言 43 项全绿。
完整源码
本文配套的可运行示例已开源,带上自己的三张 CSV 和模板就能复跑:
huang_jianhua0101/examples - Gitee.com
关于我
在实验室一线待了 13 年(9 年制药 + 4 年第三方检测),做的一直是实验室信息化。做过 STARLIMS 的甲方 PM——一期、二期两轮上线都由我主导(招标到 3Q 验证到验收全流程);也在系统上自己做过二次开发——把纸质的账号申请流程搬到线上跑;在 STARLIMS 之前还有 6 年多 CS 架构 LIMS 的使用与运维经验(其中一段经 Citrix 远程接入)。现在专做实验室里那些重复劳动:报表自动生成、仪器数据对接、合规文档批量处理。
本科物理化学、硕士计算机化学,既听得懂 QA 说的变更控制,也看得懂仪器导出的原始数据长什么样。SOP、偏差、OOS、样本流转这些词,不用你解释。
现在主要做这几类:
- 检验报告与台账批量生成:模板不动,数据自动填,格式一步不错
- 仪器数据对接:色谱、光谱、酶标仪导出的原始文件,解析、清洗、入库、转成报表
- 合规文档自动化:SOP、验证方案、批记录这类重复文档的批量生成与核对
- 数据完整性核查:按 ALCOA+ 逐条核对原始数据与记录是否对得上
手里有这类活儿卡着,或者只是想问问能不能自动化,都欢迎评论区聊,先把问题说清楚再谈怎么做。