基于Python标准库实现CSV数据动态条件查询

核心方案 #

要实现灵活的动态条件筛选，核心是把用户输入的条件表达式转换成可执行的判断逻辑，同时利用Python标准库的csv模块处理数据。这里提供两种实用方案，兼顾安全性和灵活性：

方案1：安全的比较器映射（无eval） #

这种方式通过预定义操作符对应的判断函数，手动解析条件字符串，完全避免eval()的安全风险，适合对安全性要求高的场景。

实现步骤

• 预定义支持的比较操作符与对应逻辑的映射字典
• 解析用户输入的条件字符串，拆分为单个可验证的条件单元
• 遍历CSV每行数据，逐个验证条件单元，全部满足则保留该行

代码示例

import csv
from typing import List, Dict

# 预定义支持的比较操作符
OPERATORS = {
    '==': lambda a, b: a == b,
    '!=': lambda a, b: a != b,
    '>': lambda a, b: a > b,
    '>=': lambda a, b: a >= b,
    '<': lambda a, b: a < b,
    '<=': lambda a, b: a <= b,
    'is': lambda a, b: a is b,
    'is not': lambda a, b: a is not b
}

def parse_condition(condition_str: str) -> List[Dict]:
    """解析条件字符串为可执行的条件单元列表"""
    conditions = []
    # 拆分AND连接的条件（默认条件用and分隔，需带空格）
    for cond in condition_str.split('and'):
        cond = cond.strip()
        # 优先匹配长操作符（如'is not'）
        for op in sorted(OPERATORS.keys(), key=len, reverse=True):
            if op in cond:
                left, right = cond.split(op, 1)
                left = left.strip()
                right = right.strip()
                # 解析左侧索引，比如element[1] -> 1
                if left.startswith('element[') and left.endswith(']'):
                    idx = int(left[8:-1])
                else:
                    raise ValueError(f"不支持的左侧表达式: {left}")
                # 解析右侧值，处理字符串、数字、None
                if right == 'None':
                    value = None
                elif right.startswith('"') and right.endswith('"'):
                    value = right[1:-1]
                elif right.replace('.', '', 1).isdigit():
                    value = float(right) if '.' in right else int(right)
                else:
                    raise ValueError(f"不支持的值类型: {right}")
                conditions.append({'idx': idx, 'op': op, 'value': value})
                break
        else:
            raise ValueError(f"不支持的操作符: {cond}")
    return conditions

def filter_csv(csv_path: str, condition_str: str) -> List[List[str]]:
    """根据条件筛选CSV行"""
    conditions = parse_condition(condition_str)
    filtered_rows = []
    with open(csv_path, 'r', newline='', encoding='utf-8') as f:
        reader = csv.reader(f)
        for row in reader:
            match = True
            for cond in conditions:
                idx = cond['idx']
                # 处理索引超出行长度的情况
                if idx >= len(row):
                    match = False
                    break
                cell_value = row[idx]
                # 转换单元格类型以匹配目标值类型
                target_type = type(cond['value'])
                if target_type is int:
                    cell_value = int(cell_value)
                elif target_type is float:
                    cell_value = float(cell_value)
                elif target_type is type(None):
                    cell_value = None if cell_value == '' else cell_value
                # 执行比较判断
                if not OPERATORS[cond['op']](cell_value, cond['value']):
                    match = False
                    break
            if match:
                filtered_rows.append(row)
    return filtered_rows

# 使用示例
if __name__ == '__main__':
    result = filter_csv('data.csv', 'element[1]==2 and element[3]!=5 and element[6]=="Bob"')
    for row in result:
        print(row)

方案2：受控的eval（更灵活） #

如果需要支持更复杂的表达式（比如逻辑或、简单函数调用），可以用eval()但严格限制命名空间，只允许访问行数据和安全的内置函数，避免恶意代码执行。

代码示例

import csv

def filter_csv_with_eval(csv_path: str, condition_str: str) -> List[List[str]]:
    filtered_rows = []
    # 定义安全命名空间，仅开放必要的内置类型和行数据
    safe_namespace = {
        '__builtins__': {
            'int': int,
            'float': float,
            'str': str,
            'None': None
        }
    }
    with open(csv_path, 'r', newline='', encoding='utf-8') as f:
        reader = csv.reader(f)
        for row in reader:
            safe_namespace['element'] = row
            try:
                # 执行条件判断
                if eval(condition_str, safe_namespace):
                    filtered_rows.append(row)
            except Exception as e:
                print(f"条件执行错误: {e}")
                continue
    return filtered_rows

# 使用示例
if __name__ == '__main__':
    result = filter_csv_with_eval('data.csv', 'element[1]==2 and element[3]!=5 and element[6]=="Bob"')
    for row in result:
        print(row)

关键注意事项 #

• 类型转换：CSV读取的单元格默认是字符串，必须根据条件中的目标值类型做转换，否则会出现字符串与数字比较的逻辑错误
• 安全防护：使用eval()时，必须严格限制命名空间，禁止访问exec、os等危险内置对象
• 表头适配：如果CSV包含表头，建议改用csv.DictReader，条件可写成row['age']>=18 and row['name']!='Bob'，更直观易读
• 错误处理：添加异常捕获，处理条件解析错误、索引越界、类型转换失败等场景，避免程序直接崩溃

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