基于YAML条件过滤Dataframe异常：无法生成匹配标记列

1. YAML配置文件核心错误 #

• 缩进层级错误：condition_4被错误嵌套在condition_3下，导致无法被识别为独立条件。需将condition_4与其他condition_*保持同级缩进。
• range对象无法被YAML解析：YAML不支持直接序列化Python的range对象，yaml.safe_load会将range(0,8)解析为字符串而非可用的范围对象。需改为存储数值范围的上下限，示例：

  condition_1:
    'A': 1
    'B': 5
    'C_min': 0
    'C_max': 8
    'D': 21
  

2. 数据类型不匹配问题 #

csv.DictReader读取的所有列值均为字符串类型，但YAML配置中的数值是整数类型，直接用==比较会永远不成立（例如'1' == 1返回False）。必须先将每行的列值转换为对应数值类型再做比较。

3. 条件逻辑错误 #

• 原代码中判断C列的逻辑是row['C'] == config[...]['C']，即使range能被正确解析，也应该判断数值是否在范围内，而非直接比较对象相等。
• 部分条件的D值是单个整数（如condition_1的D:21），但代码中用了row['D'] in config[...]['D']，单个整数无法用in判断（会触发异常），需兼容单个值和列表值的判断逻辑。

4. 遍历与赋值的潜在风险 #

虽然遍历字典列表时修改row会同步到原列表，但如果遍历过程中出现异常（如键不存在、类型转换失败），会导致部分行未被赋值matched。需添加异常处理，确保所有行都能得到标记。

修复后的完整代码 #

修正后的YAML配置 #

general_1:
  condition_1:
    'A': 1
    'B': 5
    'C_min': 0
    'C_max': 8
    'D': 21

  condition_2:
    'A': 1
    'B': 4
    'C_min': 9
    'C_max': 200
    'D': 22

  condition_3:
    'A': 1
    'B': 3
    'C_min': 3
    'C_max': 200
    'D': 22

  condition_4:
    'A': 1
    'B': 6
    'C_min': 3
    'C_max': 200
    'D': [21, 101, 102, 241, 242, 341, 342, 343, 344, 345, 346, 347, 348, 349, 351, 352, 353, 354, 355, 356, 357, 551, 552, 553, 554, 555, 556, 665, 667, 767, 861, 862]

修复后的Python代码 #

import yaml
import csv

def filter_columns(data_list, yaml_file):
    with open(yaml_file) as f:
        config = yaml.safe_load(f)
    conditions = config['general_1'].values()
    
    for row in data_list:
        # 转换数据类型并处理异常
        try:
            a = int(row['A'])
            b = int(row['B'])
            c = int(row['C'])
            d = int(row['D'])
        except (ValueError, KeyError):
            row['matched'] = 0
            continue
        
        matched = 0
        # 遍历所有条件，无需硬编码
        for cond in conditions:
            if a != cond['A'] or b != cond['B']:
                continue
            if not (cond['C_min'] <= c <= cond['C_max']):
                continue
            # 兼容单个值与列表值的D列判断
            d_target = cond['D']
            if isinstance(d_target, list):
                if d in d_target:
                    matched = 1
                    break
            else:
                if d == d_target:
                    matched = 1
                    break
        row['matched'] = matched
    return data_list

# 读取输入CSV
input_file = "your_input.csv"
config_file = "config.yaml"

with open(input_file, 'r') as f:
    reader = csv.DictReader(f)
    data = [row for row in reader]

# 处理数据并生成标记列
filtered_data = filter_columns(data, config_file)

# 可选：将结果写入输出CSV
with open('output.csv', 'w', newline='') as f:
    writer = csv.DictWriter(f, fieldnames=filtered_data[0].keys())
    writer.writeheader()
    writer.writerows(filtered_data)

关键优化点 #

• 循环遍历所有条件，避免硬编码100+条件的冗余代码
• 兼容单个值和列表值的D列判断逻辑
• 添加类型转换异常处理，确保所有行都能得到标记
• 修正YAML配置结构，保证所有条件可被正常读取

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