如何从已验证CSV生成关联规则实现Pandas数据自动校验

问题背景 #

现有如下结构的Pandas DataFrame：

需要实现：

• 生成依赖其他列的校验规则（比如Fruit为Apple时Color可以是Red/Green/Yellow，但Banana的Color不能是Red；Eaten?为Yes时Date Eaten不能为空，为No时无要求）
• 从已手动验证的CSV文件中自动提取规则并保存
• 用规则校验新数据，输出包含所有无效行的DataFrame

此前手动设置规则因组合过多不可行，决策树方案不适合精确字符串校验，寻求可行方案。

可行解决方案 #

1. 从已验证数据提取合法规则 #

已验证的CSV是"黄金标准"，可从中提取两类核心规则：

• 列组合合法性规则：提取合法的Fruit-Color这类关联列组合
• 必填依赖规则：明确Eaten?状态对应的Date Eaten必填要求

代码示例：提取并保存规则

import pandas as pd
import json

# 读取已验证的合法数据集
valid_df = pd.read_csv("validated_data.csv")

# 提取合法的Fruit-Color组合
valid_fruit_color = valid_df.dropna(subset=["Fruit", "Color"]).groupby("Fruit")["Color"].unique().to_dict()
# 转换为可序列化的列表格式
valid_fruit_color = {k: v.tolist() for k, v in valid_fruit_color.items()}

# 提取Eaten?对应的Date Eaten必填规则
eaten_date_rule = {}
for eaten_val in valid_df["Eaten?"].dropna().unique():
    subset = valid_df[valid_df["Eaten?"] == eaten_val]
    # 若合法数据中该Eaten?值下所有Date Eaten都非空，则标记为必填
    requires_date = subset["Date Eaten"].notna().all()
    eaten_date_rule[eaten_val] = requires_date

# 将规则保存到JSON文件
rules = {
    "valid_fruit_color": valid_fruit_color,
    "eaten_date_rule": eaten_date_rule
}

with open("validation_rules.json", "w") as f:
    json.dump(rules, f, indent=4)

2. 使用规则校验新数据 #

加载保存的规则，对新数据逐行校验，标记并提取无效行：

代码示例：校验新数据

import pandas as pd
import json

# 加载预先生成的规则
with open("validation_rules.json", "r") as f:
    rules = json.load(f)

valid_fruit_color = rules["valid_fruit_color"]
eaten_date_rule = rules["eaten_date_rule"]

# 读取待校验的新数据
new_df = pd.read_csv("new_data.csv")

# 初始化无效标记和原因列
new_df["is_invalid"] = False
new_df["invalid_reason"] = ""

# 校验Fruit-Color组合合法性
def check_fruit_color(row):
    fruit = row["Fruit"]
    color = row["Color"]
    if pd.notna(fruit) and pd.notna(color):
        if fruit in valid_fruit_color and color not in valid_fruit_color[fruit]:
            return f"Invalid color '{color}' for fruit '{fruit}'"
    return ""

new_df["invalid_reason"] += new_df.apply(check_fruit_color, axis=1)

# 校验Date Eaten必填规则
def check_date_eaten(row):
    eaten_val = row["Eaten?"]
    date_eaten = row["Date Eaten"]
    if pd.notna(eaten_val) and eaten_val in eaten_date_rule:
        if eaten_date_rule[eaten_val] and pd.isna(date_eaten):
            return "; Date Eaten is required when 'Eaten?' is 'Yes'"
    return ""

new_df["invalid_reason"] += new_df.apply(check_date_eaten, axis=1)

# 更新无效行标记
new_df["is_invalid"] = new_df["invalid_reason"] != ""

# 提取所有无效行
invalid_rows = new_df[new_df["is_invalid"]].copy()

# 保存无效行到文件
invalid_rows.to_csv("invalid_data.csv", index=False)

print("Invalid rows found:")
print(invalid_rows)

方案优势 #

• 完全基于已验证数据自动生成规则，无需手动枚举海量组合
• 规则以JSON格式存储，易于维护和调整
• 校验逻辑模块化，可快速扩展其他列的依赖校验规则

内容的提问来源于stack exchange，提问作者Him