219 lines
6.3 KiB
Python
219 lines
6.3 KiB
Python
#!/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()
|