使用Python自动标记Excel中的重复项、缺失值与逻辑错误

原始数据
  ↓
[重复检测] → 标记重复行
  ↓
[缺失检测] → 标记缺失字段
  ↓
[逻辑检测] → 标记异常值
  ↓
生成检测报告 + 高亮问题行

pip install pandas openpyxl

import pandas as pd
import numpy as np
from openpyxl import load_workbook
from openpyxl.styles import PatternFill

def data_quality_check(input_file, required_columns=None, rules=None, output_file="数据质检报告.xlsx"):
    """
    对 Excel 数据进行质量检查，输出检测报告

    参数:
        input_file: 待检查的 Excel 文件
        required_columns: 必填字段列表
        rules: 自定义规则字典，如 {'金额': '>=0', '年龄': '18-65'}
        output_file: 输出报告文件名
    """
    df = pd.read_excel(input_file, engine='openpyxl')
    print(f"总数据: {len(df)} 行 × {len(df.columns)} 列\n")

    # ==================== 1. 重复项检测 ====================
    print("=" * 50)
    print("【1. 重复项检测】")

    # 完全重复的行
    duplicate_mask = df.duplicated(keep='first')
    duplicate_count = duplicate_mask.sum()
    print(f"完全重复行: {duplicate_count} 行")

    if duplicate_count > 0:
        # 记录重复行
        df['是否重复'] = duplicate_mask

    # ==================== 2. 缺失值检测 ====================
    print("\n【2. 缺失值检测】")

    missing_report = df.isnull().sum()
    missing_report = missing_report[missing_report > 0]

    if len(missing_report) > 0:
        for col, count in missing_report.items():
            pct = count / len(df) * 100
            print(f"  {col}: {count} 个缺失 ({pct:.1f}%)")
    else:
        print("  无缺失值 ✓")

    # 必填字段检查
    if required_columns:
        for col in required_columns:
            if col in df.columns:
                missing = df[col].isna().sum()
                blank = (df[col].astype(str).str.strip() == '').sum()
                total_issues = missing + blank
                if total_issues > 0:
                    print(f"  ⚠ 必填字段 '{col}' 有 {total_issues} 个空值！")
                    df[f'{col}_必填缺失'] = df[col].isna() | (df[col].astype(str).str.strip() == '')

    # ==================== 3. 逻辑错误检测 ====================
    print("\n【3. 逻辑错误检测】")

    if rules:
        for col, rule in rules.items():
            if col not in df.columns:
                print(f"  跳过: 列 '{col}' 不存在")
                continue

            try:
                if rule.startswith('>='):
                    threshold = float(rule[2:])
                    mask = df[col] < threshold
                    df[f'{col}_逻辑异常'] = mask
                    count = mask.sum()
                    print(f"  {col} < {threshold}: {count} 行异常")

                elif rule.startswith('<='):
                    threshold = float(rule[2:])
                    mask = df[col] > threshold
                    df[f'{col}_逻辑异常'] = mask
                    count = mask.sum()
                    print(f"  {col} > {threshold}: {count} 行异常")

                elif '-' in rule:
                    min_val, max_val = rule.split('-')
                    min_val, max_val = float(min_val), float(max_val)
                    mask = (df[col] < min_val) | (df[col] > max_val)
                    df[f'{col}_逻辑异常'] = mask
                    count = mask.sum()
                    print(f"  {col} 不在 [{min_val}-{max_val}]: {count} 行异常")

                elif rule == '>0':
                    mask = df[col] <= 0
                    df[f'{col}_逻辑异常'] = mask
                    count = mask.sum()
                    print(f"  {col} <= 0: {count} 行异常")

            except Exception as e:
                print(f"  规则执行失败 {col}: {e}")

    # ==================== 4. 生成报告 ====================
    print("\n" + "=" * 50)
    print("【4. 生成质检报告】")

    # 找出所有标记列（异常标记）
    flag_columns = [col for col in df.columns if col.endswith(('_重复', '_缺失', '_必填缺失', '_逻辑异常'))]

    if flag_columns:
        # 只要有任何一个标记为 True，就认为是问题行
        df['是否有问题'] = df[flag_columns].any(axis=1)
        issue_count = df['是否有问题'].sum()
        print(f"问题行总数: {issue_count} / {len(df)} ({issue_count/len(df)*100:.1f}%)")

        # 分离正常数据和问题数据
        issue_df = df[df['是否有问题'] == True]
        clean_df = df[df['是否有问题'] == False]

        # 保存问题数据
        issue_df.drop(columns=['是否有问题'] + flag_columns, inplace=True, errors='ignore')
        issue_df.to_excel(output_file.replace('.xlsx', '_问题数据.xlsx'), index=False, engine='openpyxl')
        print(f"问题数据已保存: {output_file.replace('.xlsx', '_问题数据.xlsx')}")

    # 保存完整报告
    df.to_excel(output_file, index=False, engine='openpyxl')
    print(f"完整报告已保存: {output_file}")

    return df


# ==================== 使用示例 ====================
if __name__ == "__main__":
    # 执行数据质检
    data_quality_check(
        input_file="销售数据表.xlsx",
        required_columns=["工号", "姓名", "部门", "金额"],  # 必填字段
        rules={
            "金额": ">=0",           # 金额不能为负数
            "数量": ">0",            # 数量必须大于 0
            "年龄": "18-65",         # 年龄范围 18-65
            "折扣率": "0-1",         # 折扣率 0-1 之间
        },
        output_file="数据质检报告.xlsx"
    )

def highlight_issues(excel_file):
    """用红色背景高亮有问题的行"""
    wb = load_workbook(excel_file)
    ws = wb.active
    red_fill = PatternFill(start_color='FFC7CE', fill_type='solid')

    # 假设"是否有问题"列在第 N 列
    issue_col = 'N'

    for row in range(2, ws.max_row + 1):
        if ws[f'{issue_col}{row}'].value == True:
            for cell in ws[row]:
                cell.fill = red_fill

    wb.save(excel_file)

# 计算数据质量分数（100 分制）
total_cells = len(df) * len(df.columns)
missing_cells = df.isnull().sum().sum()
duplicate_rows = df.duplicated().sum()

quality_score = 100 - (missing_cells / total_cells * 50) - (duplicate_rows / len(df) * 50)
print(f"数据质量评分: {quality_score:.1f} / 100")

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

# 空字符串也视为缺失
df['日期'] = df['日期'].replace('', pd.NaT)
missing = df['日期'].isna().sum()

# 尝试将列转为数值，无法转换的会变成 NaN
df['金额'] = pd.to_numeric(df['金额'], errors='coerce')
# 然后检查新产生的 NaN
print(f"非数值单元格: {df['金额'].isna().sum()}")

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