Python脚本实现Excel数据质量自动检测：重复、缺失与逻辑错误标记

脚本专家

在日常数据处理工作中，从Excel中人工检查几千行数据的重复项、缺失值和逻辑错误不仅耗时，而且容易遗漏。借助Python的pandas和openpyxl库，我们可以编写一个可复用的数据质量检测脚本，自动标记问题行并输出质检报告。

## 检测原理

数据质量检测主要基于三类检查：
- 重复检测：通过 pandas 的 duplicated() 方法识别完全重复或关键字段重复的行。
- 缺失检测：利用 isna() 或 isnull() 找出空值、空字符串等缺失情况。
- 逻辑检测：根据业务规则（如金额≥0、年龄18-65等）使用布尔条件过滤异常值。

整个流程：读取原始数据 → 重复检测 → 缺失检测 → 逻辑检测 → 生成报告（包含问题数据表和高亮标记）。

## 环境准备

运行以下命令安装依赖：

2. pip install pandas openpyxl

复制代码

## 完整代码实现

以下是一个自定义函数 data_quality_check，它接收输入文件、必填字段列表和自定义规则字典，输出检测报告和问题数据文件。

2. import pandas as pd
3. import numpy as np
4. from openpyxl import load_workbook
5. from openpyxl.styles import PatternFill
7. def data_quality_check(input_file, required_columns=None, rules=None, output_file="数据质检报告.xlsx"):
8.     """
9.     对 Excel 数据进行质量检查，输出检测报告
10.     参数:
11.         input_file: 待检查的 Excel 文件
12.         required_columns: 必填字段列表
13.         rules: 自定义规则字典，如 {'金额': '>=0', '年龄': '18-65'}
14.         output_file: 输出报告文件名
15.     """
16.     df = pd.read_excel(input_file, engine='openpyxl')
17.     print(f"总数据: {len(df)} 行 × {len(df.columns)} 列\n")
19.     # ==================== 1. 重复项检测 ====================
20.     print("=" * 50)
21.     print("【1. 重复项检测】")
22.     duplicate_mask = df.duplicated(keep='first')
23.     duplicate_count = duplicate_mask.sum()
24.     print(f"完全重复行: {duplicate_count} 行")
25.     if duplicate_count > 0:
26.         df['是否重复'] = duplicate_mask
28.     # ==================== 2. 缺失值检测 ====================
29.     print("\n【2. 缺失值检测】")
30.     missing_report = df.isnull().sum()
31.     missing_report = missing_report[missing_report > 0]
32.     if len(missing_report) > 0:
33.         for col, count in missing_report.items():
34.             pct = count / len(df) * 100
35.             print(f"  {col}: {count} 个缺失 ({pct:.1f}%)")
36.     else:
37.         print("  无缺失值 √")
39.     if required_columns:
40.         for col in required_columns:
41.             if col in df.columns:
42.                 missing = df[col].isna().sum()
43.                 blank = (df[col].astype(str).str.strip() == '').sum()
44.                 total_issues = missing + blank
45.                 if total_issues > 0:
46.                     print(f"  ⚠ 必填字段 '{col}' 有 {total_issues} 个空值！")
47.                     df[f'{col}_必填缺失'] = df[col].isna() | (df[col].astype(str).str.strip() == '')
49.     # ==================== 3. 逻辑错误检测 ====================
50.     print("\n【3. 逻辑错误检测】")
51.     if rules:
52.         for col, rule in rules.items():
53.             if col not in df.columns:
54.                 print(f"  跳过: 列 '{col}' 不存在")
55.                 continue
56.             try:
57.                 if rule.startswith('>='):
58.                     threshold = float(rule[2:])
59.                     mask = df[col] < threshold
60.                     df[f'{col}_逻辑异常'] = mask
61.                     count = mask.sum()
62.                     print(f"  {col} < {threshold}: {count} 行异常")
63.                 elif rule.startswith('<='):
64.                     threshold = float(rule[2:])
65.                     mask = df[col] > threshold
66.                     df[f'{col}_逻辑异常'] = mask
67.                     count = mask.sum()
68.                     print(f"  {col} > {threshold}: {count} 行异常")
69.                 elif '-' in rule:
70.                     min_val, max_val = rule.split('-')
71.                     min_val, max_val = float(min_val), float(max_val)
72.                     mask = (df[col] < min_val) | (df[col] > max_val)
73.                     df[f'{col}_逻辑异常'] = mask
74.                     count = mask.sum()
75.                     print(f"  {col} 不在 [{min_val}-{max_val}]: {count} 行异常")
76.                 elif rule == '>0':
77.                     mask = df[col] <= 0
78.                     df[f'{col}_逻辑异常'] = mask
79.                     count = mask.sum()
80.                     print(f"  {col} <= 0: {count} 行异常")
81.             except Exception as e:
82.                 print(f"  规则执行失败 {col}: {e}")
84.     # ==================== 4. 生成报告 ====================
85.     print("\n" + "=" * 50)
86.     print("【4. 生成质检报告】")
87.     flag_columns = [col for col in df.columns if col.endswith(('_重复', '_缺失', '_必填缺失', '_逻辑异常'))]
88.     if flag_columns:
89.         df['是否有问题'] = df[flag_columns].any(axis=1)
90.         issue_count = df['是否有问题'].sum()
91.         print(f"问题行总数: {issue_count} / {len(df)} ({issue_count/len(df)*100:.1f}%)")
93.         issue_df = df[df['是否有问题'] == True]
94.         clean_df = df[df['是否有问题'] == False]
96.         issue_df.drop(columns=['是否有问题'] + flag_columns, inplace=True, errors='ignore')
97.         issue_df.to_excel(output_file.replace('.xlsx', '_问题数据.xlsx'), index=False, engine='openpyxl')
98.         print(f"问题数据已保存: {output_file.replace('.xlsx', '_问题数据.xlsx')}")
100.         df.to_excel(output_file, index=False, engine='openpyxl')
101.         print(f"完整报告已保存: {output_file}")
103.     return df
105. # ==================== 使用示例 ====================
106. if __name__ == "__main__":
107.     data_quality_check(
108.         input_file="销售数据表.xlsx",
109.         required_columns=["工号", "姓名", "部门", "金额"],
110.         rules={
111.             "金额": ">=0",
112.             "数量": ">0",
113.             "年龄": "18-65",
114.             "折扣率": "0-1",
115.         },
116.         output_file="数据质检报告.xlsx"
117.     )

复制代码

## 进阶技巧

### 技巧1：在Excel中用颜色高亮问题行

利用 openpyxl 给问题行添加红色背景，方便直接查看。

2. def highlight_issues(excel_file):
3.     """用红色背景高亮有问题的行"""
4.     wb = load_workbook(excel_file)
5.     ws = wb.active
6.     red_fill = PatternFill(start_color='FFC7CE', fill_type='solid')
7.     issue_col = 'N'  # 假设'是否有问题'列在第N列
8.     for row in range(2, ws.max_row + 1):
9.         if ws[f'{issue_col}{row}'].value == True:
10.             for cell in ws[row]:
11.                 cell.fill = red_fill
12.     wb.save(excel_file)

复制代码

### 技巧2：自动计算数据质量评分

通过缺失比例和重复比例计算一个0-100的质量分数：

2. total_cells = len(df) * len(df.columns)
3. missing_cells = df.isnull().sum().sum()
4. duplicate_rows = df.duplicated().sum()
5. quality_score = 100 - (missing_cells / total_cells * 50) - (duplicate_rows / len(df) * 50)
6. print(f"数据质量评分: {quality_score:.1f} / 100")

复制代码

### 技巧3：基于3σ原则检测异常值

对于数值列，可以使用统计学方法自动识别离群点：

2. def detect_outliers(df, column):
3.     """检测数值列的异常值（3σ 原则）"""
4.     mean = df[column].mean()
5.     std = df[column].std()
6.     lower = mean - 3 * std
7.     upper = mean + 3 * std
8.     outliers = df[(df[column] < lower) | (df[column] > upper)]
9.     print(f"{column} 异常值: {len(outliers)} 行 (范围: {lower:.2f} ~ {upper:.2f})")
10.     return outliers

复制代码

## 常见问题与处理

### Q1：日期列缺失检测不到？
如果Excel中日期实际存储为空字符串或0，pandas不会自动视为NaN。解决方法是提前统一替换：

2. df['日期'] = df['日期'].replace('', pd.NaT)
3. missing = df['日期'].isna().sum()

复制代码

### Q2：数值列中混入了文本？
使用 pd.to_numeric 将列强制转换为数值，无法转换的会变成NaN，从而可以统计异常：

2. df['金额'] = pd.to_numeric(df['金额'], errors='coerce')
3. print(f"非数值单元格: {df['金额'].isna().sum()}")

复制代码

### Q3：如何处理大小写不一致？
统一转为大写并去除空格，比如“北京”与“beijing”视为不同，但可统一格式：

2. df['城市'] = df['城市'].str.upper().str.strip()

复制代码

## 总结

本文实现了一个通用的Excel数据质量自动检查脚本，覆盖了重复、缺失、逻辑三类核心检测，并提供了高亮行、质量评分、异常值检测等增强功能。这段代码可以直接集成到数据处理流水线中，作为数据清洗的前置步骤，显著提升质检效率与准确性。

热心网友7

感谢楼主分享这么实用的脚本！手动检查Excel数据确实是苦力活，这个自动化方案能节省大量时间。我特别关注逻辑检测部分，因为工作中经常遇到类似年龄范围、金额区间这类条件。想请教一下：如果规则像“年龄18-65”这种区间判断，代码示例里只处理了 `>=` 和 `