条件检查场景下替代eval函数的实现方案咨询（附JSON示例）

嘿，我完全理解你为什么想换掉eval——安全风险高，而且代码可读性和维护性都差。针对你这个场景，我有两个实用的替代方案，都能避免eval的问题：

方案1：操作符映射 + 轻量逻辑表达式解析（适合简单场景） #

这个方案的核心是把操作符（比如>、<）映射到内置的比较函数，然后手动解析outcome_selection_criteria里的逻辑规则，完全不用eval。

实现步骤： #

1. 定义操作符与比较函数的映射字典，用operator模块的内置函数来避免重复造轮子。
2. 逐个计算checking_condition里的每个条件结果，得到一个布尔值列表。
3. 解析outcome_selection_criteria（比如index0 and index1），把indexN替换成布尔列表对应的索引值，然后判断逻辑是否成立。

代码示例： #

import operator
import re

def evaluate_single_condition(condition, context):
    # 映射操作符到内置比较函数
    op_map = {
        ">": operator.gt,
        "<": operator.lt,
        "==": operator.eq,
        ">=": operator.ge,
        "<=": operator.le,
        "!=": operator.ne
    }
    
    # 获取数据库字段的实际值
    left_val = context[condition["db_key"]]
    # 处理右边的值：如果是上下文变量则取值，否则转换为对应类型
    right_val = context.get(condition["db_value"], condition["db_value"])
    
    # 类型转换（确保左右值类型一致，避免比较错误）
    if isinstance(left_val, int):
        right_val = int(right_val)
    elif isinstance(left_val, float):
        right_val = float(right_val)
    # 日期类型处理示例（根据你的实际需求调整）
    elif hasattr(left_val, "strftime"):
        from datetime import datetime
        right_val = datetime.strptime(right_val, "%Y-%m-%d")
    
    # 执行比较
    return op_map[condition["operator"]](left_val, right_val)

def evaluate_selection_criteria(criteria, condition_results):
    # 把indexN替换成布尔列表的索引访问
    expr = re.sub(r"index(\d+)", r"\1", criteria)
    # 拆分逻辑表达式（仅支持and/or的扁平逻辑，复杂嵌套请用方案2）
    parts = expr.split()
    result = True
    current_op = "and"
    
    for part in parts:
        if part in ["and", "or"]:
            current_op = part
            continue
        idx = int(part)
        cond_result = condition_results[idx]
        
        if current_op == "and":
            result = result and cond_result
        elif current_op == "or":
            result = result or cond_result
    
    return result

def get_selected_outcome(scenarios, context):
    for scenario in scenarios:
        # 计算所有检查条件的结果
        cond_results = [evaluate_single_condition(cond, context) for cond in scenario["checking_condition"]]
        # 判断是否符合选择标准
        if evaluate_selection_criteria(scenario["outcome_selection_criteria"], cond_results):
            return scenario["outcome_name"]
    return None

优缺点： #

• ✅ 优点：代码简单易懂，实现成本低，适合逻辑不复杂的场景（比如只有and/or的扁平表达式）。
• ❌ 缺点：不支持嵌套逻辑（比如(index0 or index1) and index2），如果需要复杂规则，维护成本会上升。

方案2：使用AST模块解析逻辑表达式（进阶安全版） #

如果你的outcome_selection_criteria可能包含嵌套逻辑，用Python的ast模块是更安全、更灵活的选择。AST可以解析表达式的抽象语法树，只允许我们指定的安全操作，完全杜绝恶意代码执行的可能。

代码示例： #

import operator
import ast
import re

def evaluate_single_condition(condition, context):
    # 和方案1的这个函数完全一致
    op_map = {
        ">": operator.gt,
        "<": operator.lt,
        "==": operator.eq,
        ">=": operator.ge,
        "<=": operator.le,
        "!=": operator.ne
    }
    
    left_val = context[condition["db_key"]]
    right_val = context.get(condition["db_value"], condition["db_value"])
    
    if isinstance(left_val, int):
        right_val = int(right_val)
    elif isinstance(left_val, float):
        right_val = float(right_val)
    elif hasattr(left_val, "strftime"):
        from datetime import datetime
        right_val = datetime.strptime(right_val, "%Y-%m-%d")
    
    return op_map[condition["operator"]](left_val, right_val)

def safe_evaluate_criteria(criteria, condition_results):
    # 替换indexN为results[N]的语法
    expr_str = re.sub(r"index(\d+)", r"results[\1]", criteria)
    # 解析为AST树
    tree = ast.parse(expr_str, mode="eval")
    
    # 自定义AST访问器，只允许安全的节点类型
    class SafeLogicEvaluator(ast.NodeVisitor):
        def visit_BoolOp(self, node):
            # 处理and/or逻辑
            values = [self.visit(v) for v in node.values]
            if isinstance(node.op, ast.And):
                return all(values)
            elif isinstance(node.op, ast.Or):
                return any(values)
            else:
                raise ValueError(f"不支持的逻辑操作符: {type(node.op).__name__}")
        
        def visit_Subscript(self, node):
            # 只允许访问condition_results列表
            if not (isinstance(node.value, ast.Name) and node.value.id == "results"):
                raise ValueError(f"只能访问results列表，禁止其他变量")
            # 处理索引
            index = self.visit(node.slice) if hasattr(node.slice, "value") else node.slice.value
            if not isinstance(index, int):
                raise ValueError(f"索引必须是整数")
            if index < 0 or index >= len(condition_results):
                raise ValueError(f"索引{index}超出结果列表范围")
            return condition_results[index]
        
        def visit_Constant(self, node):
            # 只允许布尔常量
            if isinstance(node.value, bool):
                return node.value
            raise ValueError(f"只允许布尔常量")
        
        def generic_visit(self, node):
            # 拦截所有未定义的节点类型
            raise ValueError(f"不支持的语法: {type(node).__name__}")
    
    evaluator = SafeLogicEvaluator()
    try:
        return evaluator.visit(tree.body)
    except Exception as e:
        raise ValueError(f"解析条件失败: {str(e)}")

def get_selected_outcome(scenarios, context):
    for scenario in scenarios:
        cond_results = [evaluate_single_condition(cond, context) for cond in scenario["checking_condition"]]
        try:
            if safe_evaluate_criteria(scenario["outcome_selection_criteria"], cond_results):
                return scenario["outcome_name"]
        except ValueError as e:
            print(f"处理场景{scenario['outcome_name']}时出错: {e}")
            continue
    return None

优缺点： #

• ✅ 优点：支持复杂嵌套逻辑（比如(index0 or index1) and (index2 or index3)），完全安全，能拦截任何恶意代码。
• ❌ 缺点：代码量稍大，需要对AST有基本理解，但维护性和扩展性更强。

测试示例 #

用你提供的scenario测试一下：

from datetime import datetime

scenarios = [
    {
        "outcome_name":"outcome1", 
        "checking_condition":[
            { "db_key":"purchase_date", "operator":">", "db_value":"requested_date" }, 
            { "db_key":"purchase_amount", "operator":"<", "db_value":"5000" }
        ], 
        "outcome_selection_criteria":"index0 and index1" 
    }
]

# 模拟数据库和请求的上下文数据
context = {
    "purchase_date": datetime(2024, 5, 1),
    "requested_date": datetime(2024, 4, 1),
    "purchase_amount": 3000
}

selected_outcome = get_selected_outcome(scenarios, context)
print(f"选中的结果: {selected_outcome}")  # 输出: outcome1

注意事项 #

1. 类型转换：一定要确保比较的左右值类型一致，否则会出现不符合预期的结果（比如字符串和数字比较）。
2. 错误处理：添加必要的异常捕获，避免单个场景的错误导致整个流程崩溃。
3. 上下文安全：确保context里的内容是可信的，虽然AST方案已经限制了访问范围，但仍要避免不可信数据的注入。

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