收集表里的空行、重复记录和首尾空格看似好处理,规则含糊却可能误删有效数据。本文提供一套 Python 脚本,按工作表清洗 .xlsx 文件,并另存原始备份、清洗结果和校验报告:默认仅删除整行完全相同的记录,保留部分缺失值,不自动补值;只有确认业务唯一标识后,才按指定字段去重,缺少去重键的记录也会保留。如何在整理数据的同时,让每一步处理都可核查?
— 此摘要由AI分析文章内容生成,仅供参考。
如果你要整理收集表、客户表或运营数据,可以用 Python 把 Excel 中的全空行、重复记录和首尾空格按明确规则处理,并另存清洗结果与校验报告。下面的做法不会覆盖原文件;默认只删除整行完全相同的记录,不会擅自判断某个字段代表什么,也不会自动填补缺失值。
安装准备
这套脚本适用于以第一行为表头的 .xlsx 工作簿。先安装 pandas 和 Excel 读写引擎:
python -m pip install pandas openpyxl
将下面代码保存为 clean_excel.py,并修改代码开头的输入、输出文件名。脚本会为每个工作表分别清洗,并生成原始备份、清洗后的工作簿和校验报告。
配置规则并运行
默认的重复判断依据是“所有字段都相同”。如果你已经确认某个字段或字段组合可以作为业务上的唯一标识,可以把 DUPLICATE_KEYS 改成对应的列名,例如 ["手机号"]。使用业务字段去重前,先确认同一标识重复出现确实意味着重复记录;缺少去重字段的记录会保留,不会因为空键相同而被删除。
from pathlib import Path
import shutil
import pandas as pd
INPUT = Path("收集表.xlsx")
OUTPUT = Path("收集表_清洗后.xlsx")
REPORT = Path("收集表_校验报告.xlsx")
BACKUP = INPUT.with_name(f"{INPUT.stem}_原始备份{INPUT.suffix}")
# 默认整行完全相同才算重复。
# 如需按已确认的业务字段去重,可改为例如 ["手机号"]。
DUPLICATE_KEYS = None
def normalize_cell(value):
"""去除文本首尾空格,并将空字符串转为空值;不改写其他内容。"""
if isinstance(value, str):
value = value.strip()
return pd.NA if value == "" else value
return value
def clean_sheet(sheet_name, original):
# 假设 Excel 第一行为表头,所以第一条数据对应第 2 行。
excel_rows = pd.Series(
range(2, len(original) + 2),
index=original.index,
)
normalized = original.apply(
lambda column: column.map(normalize_cell)
)
# 只删除所有字段均为空的记录;部分字段为空的记录继续保留。
nonblank_mask = ~normalized.isna().all(axis=1)
work = normalized.loc[nonblank_mask].copy()
work_rows = excel_rows.loc[nonblank_mask]
# 缺失清单基于删除重复记录之前的数据,便于检查所有非空行。
missing_records = []
for index, row in work.iterrows():
for column in work.columns:
if pd.isna(row[column]):
missing_records.append({
"工作表": sheet_name,
"Excel行号": int(work_rows.loc[index]),
"字段": str(column),
"处理后值": "空值",
})
if DUPLICATE_KEYS is None:
duplicate_mask = work.duplicated(keep="first")
key_columns = list(work.columns)
else:
missing_keys = [key for key in DUPLICATE_KEYS if key not in work.columns]
if missing_keys:
raise ValueError(
f"工作表“{sheet_name}”找不到去重字段:{missing_keys}"
)
# 去重键不完整的记录不参与自动去重,留待人工核对。
eligible = work[DUPLICATE_KEYS].notna().all(axis=1)
duplicate_mask = pd.Series(False, index=work.index)
duplicate_mask.loc[eligible] = work.loc[eligible].duplicated(
subset=DUPLICATE_KEYS,
keep="first",
)
key_columns = DUPLICATE_KEYS
duplicate_records = []
for index in work.index[duplicate_mask]:
record = {
"工作表": sheet_name,
"删除记录Excel行号": int(work_rows.loc[index]),
}
for column in key_columns:
record[str(column)] = work.loc[index, column]
duplicate_records.append(record)
cleaned = work.loc[~duplicate_mask].copy()
summary = {
"工作表": sheet_name,
"导入数据行数": len(original),
"删除全空行数": int((~nonblank_mask).sum()),
"去重前非空行数": len(work),
"删除重复记录数": int(duplicate_mask.sum()),
"清洗后行数": len(cleaned),
"保留记录中的空值单元格数": len(missing_records),
"重复判断依据": (
"所有字段完全相同"
if DUPLICATE_KEYS is None
else "字段:" + "、".join(DUPLICATE_KEYS)
),
}
return cleaned, summary, duplicate_records, missing_records
def main():
if INPUT.suffix.lower() != ".xlsx":
raise ValueError("本示例仅处理 .xlsx 文件;请先确认文件格式。")
paths = [INPUT, OUTPUT, REPORT, BACKUP]
if len({path.resolve() for path in paths}) != len(paths):
raise ValueError("输入、输出、报告和备份文件路径不能相同。")
if not INPUT.exists():
raise FileNotFoundError(f"找不到输入文件:{INPUT}")
# 不覆盖已有文件,避免误替换先前结果或备份。
for path in (OUTPUT, REPORT, BACKUP):
if path.exists():
raise FileExistsError(
f"文件已存在:{path}。请更改文件名或先手动确认后处理。"
)
shutil.copy2(INPUT, BACKUP)
sheets = pd.read_excel(INPUT, sheet_name=None, engine="openpyxl")
cleaned_sheets = {}
summaries = []
all_duplicates = []
all_missing = []
for sheet_name, frame in sheets.items():
cleaned, summary, duplicates, missing = clean_sheet(sheet_name, frame)
cleaned_sheets[sheet_name] = cleaned
summaries.append(summary)
all_duplicates.extend(duplicates)
all_missing.extend(missing)
# 将清洗后的各工作表写入新文件。
with pd.ExcelWriter(OUTPUT, engine="openpyxl") as writer:
for sheet_name, frame in cleaned_sheets.items():
frame.to_excel(writer, sheet_name=sheet_name, index=False)
summary_df = pd.DataFrame(summaries)
duplicate_columns = ["工作表", "删除记录Excel行号"]
duplicate_columns += (
list(sheets[next(iter(sheets))].columns)
if DUPLICATE_KEYS is None and sheets
else (DUPLICATE_KEYS or [])
)
duplicates_df = pd.DataFrame(all_duplicates)
if duplicates_df.empty:
duplicates_df = pd.DataFrame(columns=duplicate_columns)
missing_df = pd.DataFrame(
all_missing,
columns=["工作表", "Excel行号", "字段", "处理后值"],
)
with pd.ExcelWriter(REPORT, engine="openpyxl") as writer:
summary_df.to_excel(writer, sheet_name="数量汇总", index=False)
duplicates_df.to_excel(writer, sheet_name="重复记录", index=False)
missing_df.to_excel(writer, sheet_name="空值清单", index=False)
print(f"原始备份:{BACKUP}")
print(f"清洗结果:{OUTPUT}")
print(f"校验报告:{REPORT}")
if __name__ == "__main__":
main()
运行:
python clean_excel.py
如果你修改了输出文件名,也要确保它与已有文件不重名。脚本检测到输出、报告或备份文件已存在时会停止,不会直接覆盖。
看懂校验报告
报告工作簿包含三个工作表:
- 数量汇总:按原工作表列出导入行数、删除全空行数、去重前非空行数、删除重复记录数和清洗后行数。可用“去重前非空行数 − 删除重复记录数 = 清洗后行数”核对处理结果。
- 重复记录:列出被删除记录在原工作表中的行号及重复判断字段。每组相同记录默认保留最先出现的一条。
- 空值清单:列出清理首尾空格后仍为空的单元格。脚本不会用默认值补齐这些字段,是否补值应由你根据业务规则决定。
“删除全空行”只指所有字段都为空的记录;只有部分单元格为空的记录仍会保留。“空值清单”在去重前生成,因此其中也可能包含后来被判定为重复并删除的记录。
使用时注意
- 先确认重复规则。 整行相同适合识别字段值完全一致的记录,不等于业务上所有重复都已找全。若按手机号、订单号等字段去重,必须先确认该字段或组合确实能识别同一条记录。
- 区分空白与缺失。 脚本会把仅含空格的文本按空值处理,但不会删除部分缺失的行,也不会替你推断空值含义。
- 核对备份与输出。 原文件会复制为带有“原始备份”后缀的文件;清洗结果与报告分别保存。检查数量汇总和异常清单后,再决定是否将清洗结果用于后续工作。
- 注意格式限制。 pandas 适合处理以单元格数据为主的表格,但重新写出工作簿时,不应假设原有格式、宏、公式、合并单元格或其他 Excel 特性都会完整保留。若文件依赖这些内容,先在副本上测试,或改用能保留所需工作簿结构的处理方式。

评论列表 (2条):
加载更多评论 Loading...