验证Pandas DataFrame中LOV列是否符合映射DataFrame规则

1. 准备示例数据 #

先将提供的示例数据转换成Pandas DataFrame：

import pandas as pd
import numpy as np

# 员工信息表
employee_df = pd.DataFrame({
    'Country': ['CZ', 'CZ', 'CZ', 'CZ', 'CZ'],
    'Employee Name': ['Jonathan', 'Peter', 'Maro', 'Marisa', 'Petra'],
    'Contract Type': ['permanent', np.nan, 'Fixed', 'Contractor', 'Permanent'],
    'Grade': ['1', '6', '.', '01-01-2020', '5'],
    'Education': ['Male', 'male', 'N/A', 'Female', 'Female']
})

# LOV映射表
lov_mapping_df = pd.DataFrame({
    'Country': ['CZ', 'CZ', 'US', 'US', 'US', 'AE', 'CZ', 'CZ', 'CZ', 'CZ', 'CZ', 'US', 'US', 'CZ', 'CZ', 'SK', 'SK', 'SK', 'AE', 'AE'],
    'LOV Column': ['Contract Type', 'Contract Type', 'Contract Type', 'Contract Type', 'Contract Type', 'Contract Type', 'Grade', 'Grade', 'Grade', 'Grade', 'Grade', 'Contract Type', 'Contract Type', 'Education', 'Education', 'Education', 'Education', 'Education', 'Gender', 'Gender'],
    'Values': ['Permanent', 'Fixed', 'Permanent', 'Fixed', 'Contractor', 'Permanent', '1', '2', '3', '4', '5', 'Manager', 'Non Manager', '1st Degree', '2nd Degree', 'A', 'B', 'C', 'Male', 'Female']
})

2. 核心实现逻辑 #

针对指定国家（比如示例中的CZ），按以下步骤执行合规性检查：

• 从映射表中提取该国家所有LOV列的允许值，整理成列名: 允许值集合的字典（严格区分大小写）
• 过滤出员工表中目标国家的数据
• 逐列检查：排除空值（包括np.nan和N/A这类文本空值），找出不在允许值集合中的数据，记录违规值和对应的行号

3. 完整代码实现 #

def check_lov_compliance(employee_df, lov_mapping_df, target_country):
    # 提取目标国家的LOV规则，按列分组整理成字典
    country_lov = lov_mapping_df[lov_mapping_df['Country'] == target_country]
    lov_rules = country_lov.groupby('LOV Column')['Values'].apply(set).to_dict()
    
    # 过滤目标国家的员工数据，重置行号方便后续定位
    target_employees = employee_df[employee_df['Country'] == target_country].reset_index(drop=True)
    
    # 逐列检查合规性
    for col_name, allowed_values in lov_rules.items():
        print(f"\n{col_name}")
        # 统一处理空值：将文本型空值转为np.nan
        col_data = target_employees[col_name].replace('N/A', np.nan)
        # 筛选非空且不在允许值范围内的行
        non_compliant_mask = ~col_data.isna() & ~col_data.isin(allowed_values)
        # 提取违规值和对应行号
        non_compliant_values = col_data[non_compliant_mask].tolist()
        error_rows = non_compliant_mask[non_compliant_mask].index.tolist()
        
        print(f"Values non compliant in the column: {non_compliant_values}")
        print(f"Rows error occured: {error_rows}")

# 执行检查，目标国家设为CZ
check_lov_compliance(employee_df, lov_mapping_df, 'CZ')

4. 运行结果 #

运行上述代码后，输出结果与示例完全一致：

Contract Type
Values non compliant in the column: ['permanent', 'Contractor']
Rows error occured: [0, 3]

Education
Values non compliant in the column: ['Male', 'male', 'Female', 'Female']
Rows error occured: [0, 1, 3, 4]

Grade
Values non compliant in the column: ['6', '.', '01-01-2020']
Rows error occured: [1, 2, 3]

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