#!/usr/bin/env python # -*- coding: utf-8 -*- """ 条件表达式求值器 负责解析和执行条件分支的表达式 """ import logging import operator from decimal import Decimal from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) class ConditionEvaluator: """ 条件求值器 支持的操作符: - eq: 等于 - ne: 不等于 - gt: 大于 - gte: 大于等于 - lt: 小于 - lte: 小于等于 - contains: 包含 - not_contains: 不包含 - in: 在...中 - not_in: 不在...中 - empty: 为空 - not_empty: 不为空 """ # 操作符映射 OPERATORS = { 'eq': operator.eq, 'ne': operator.ne, 'gt': operator.gt, 'gte': operator.ge, 'lt': operator.lt, 'lte': operator.le, } def evaluate_groups(self, groups: List[Dict], form_data: Dict) -> bool: """ 求值条件组列表(组之间是 OR 关系) Args: groups: 条件组列表,每个组包含 conditions 数组 form_data: 表单数据 Returns: bool: 任一组满足则返回 True """ if not groups: return True for group in groups: if self.evaluate_group(group, form_data): return True return False def evaluate_group(self, group: Dict, form_data: Dict) -> bool: """ 求值单个条件组(组内条件是 AND 关系) Args: group: 条件组,包含 conditions 数组 form_data: 表单数据 Returns: bool: 所有条件都满足才返回 True """ conditions = group.get('conditions', []) if not conditions: return True for condition in conditions: if not self.evaluate_condition(condition, form_data): return False return True def evaluate_condition(self, condition: Dict, form_data: Dict) -> bool: """ 求值单个条件 Args: condition: 条件定义 {field, operator, value} form_data: 表单数据 Returns: bool: 条件是否满足 """ field = condition.get('field', '') op = condition.get('operator', 'eq') expected_value = condition.get('value') actual_value = self._get_field_value(form_data, field) logger.info( f"条件求值: field={field}, operator={op}, " f"expected={expected_value!r}(type={type(expected_value).__name__}), " f"actual={actual_value!r}(type={type(actual_value).__name__})" ) try: result = self._compare(actual_value, op, expected_value) logger.info(f"条件求值结果: {result}") return result except Exception as e: logger.warning(f"条件求值失败: {condition}, 错误: {e}") return False def _get_field_value(self, data: Dict, field: str) -> Any: """ 获取字段值,支持嵌套路径(如 user.dept.name) """ if not field: return None parts = field.split('.') value = data for part in parts: if isinstance(value, dict): value = value.get(part) else: return None return value def _compare(self, actual: Any, op: str, expected: Any) -> bool: """ 执行比较操作 """ # 空值检查 if op == 'empty': return self._is_empty(actual) if op == 'not_empty': return not self._is_empty(actual) # 包含检查 if op == 'contains': return self._contains(actual, expected) if op == 'not_contains': return not self._contains(actual, expected) # 集合检查 if op == 'in': return self._in_list(actual, expected) if op == 'not_in': return not self._in_list(actual, expected) # 数值比较(需要类型转换) if op in ('gt', 'gte', 'lt', 'lte'): actual = self._to_number(actual) expected = self._to_number(expected) if actual is None or expected is None: return False # 等于/不等于:先尝试原始比较,如果类型不同则统一转字符串再比较 if op in ('eq', 'ne'): if type(actual) != type(expected) and actual is not None and expected is not None: result = self.OPERATORS[op](str(actual).strip(), str(expected).strip()) logger.debug(f"类型不一致,转字符串比较: {str(actual)!r} {op} {str(expected)!r} = {result}") return result return self.OPERATORS[op](actual, expected) # 其他标准比较 if op in self.OPERATORS: return self.OPERATORS[op](actual, expected) return actual == expected def _is_empty(self, value: Any) -> bool: """检查值是否为空""" if value is None: return True if isinstance(value, str) and value.strip() == '': return True if isinstance(value, (list, dict)) and len(value) == 0: return True return False def _contains(self, actual: Any, expected: Any) -> bool: """检查是否包含""" if actual is None: return False if isinstance(actual, str): return str(expected) in actual if isinstance(actual, (list, tuple)): return expected in actual return False def _in_list(self, actual: Any, expected: Any) -> bool: """检查是否在列表中""" if not isinstance(expected, (list, tuple)): expected = [expected] return actual in expected def _to_number(self, value: Any) -> Optional[float]: """转换为数字""" if value is None: return None if isinstance(value, (int, float, Decimal)): return float(value) if isinstance(value, str): try: return float(value) except ValueError: return None return None # 全局实例 condition_evaluator = ConditionEvaluator()