from decimal import Decimal from typing import Any from app.domain.attributes import statement_is_usable from app.domain.enums import ConstraintOperator, ConstraintStatus, ValueType from app.domain.models import AttributeDefinition, AttributeStatement, Constraint, ConstraintResult _NUMBER_OPS = { ConstraintOperator.GE, ConstraintOperator.LE, ConstraintOperator.GT, ConstraintOperator.LT, } _EQUALITY_OPS = {ConstraintOperator.EQ, ConstraintOperator.NE} _SCALAR_TYPES = {ValueType.NUMBER, ValueType.STRING, ValueType.BOOLEAN} def operator_allowed_for(value_type: ValueType, operator: ConstraintOperator) -> bool: if operator == ConstraintOperator.EXISTS: return True if operator in _NUMBER_OPS: return value_type == ValueType.NUMBER if operator in _EQUALITY_OPS: return value_type in _SCALAR_TYPES if operator == ConstraintOperator.IN: return value_type in _SCALAR_TYPES if operator == ConstraintOperator.CONTAINS: return value_type == ValueType.STRING_LIST return False def evaluate_constraint( constraint: Constraint, statement: AttributeStatement | None, definition: AttributeDefinition, minimum_confidence: Decimal, ) -> ConstraintResult: usable = statement_is_usable(statement, minimum_confidence) actual = statement.normalized_value if usable else None attribute_row_id = statement.id if statement is not None else None if not usable: return ConstraintResult( constraint_id=constraint.id, attribute=constraint.attribute, operator=constraint.operator.value, expected=constraint.value, actual=None, status=ConstraintStatus.UNKNOWN.value, mandatory=constraint.mandatory, attribute_row_id=attribute_row_id, reason_code="ATTRIBUTE_UNUSABLE", ) passed = _compare(constraint.operator, actual, constraint.value, definition.value_type) if passed is None: status = ConstraintStatus.UNKNOWN reason = "TYPE_MISMATCH" elif passed: status = ConstraintStatus.PASS reason = "CONSTRAINT_PASS" else: status = ConstraintStatus.FAIL reason = "CONSTRAINT_FAIL" return ConstraintResult( constraint_id=constraint.id, attribute=constraint.attribute, operator=constraint.operator.value, expected=constraint.value, actual=actual, status=status.value, mandatory=constraint.mandatory, attribute_row_id=attribute_row_id, reason_code=reason, ) def _as_number(value: Any) -> Decimal | None: if isinstance(value, bool) or value is None: return None if isinstance(value, Decimal): return value if isinstance(value, int): return Decimal(value) if isinstance(value, float): return Decimal(str(value)) if isinstance(value, str): return None return None def _same_scalar(left: Any, right: Any, value_type: ValueType) -> bool | None: if value_type == ValueType.NUMBER: left_n = _as_number(left) right_n = _as_number(right) if left_n is None or right_n is None: return None return left_n == right_n if value_type == ValueType.STRING: if not isinstance(left, str) or not isinstance(right, str): return None return left == right if value_type == ValueType.BOOLEAN: if not isinstance(left, bool) or not isinstance(right, bool): return None return left == right return None def _compare( operator: ConstraintOperator, actual: Any, expected: Any, value_type: ValueType, ) -> bool | None: if operator == ConstraintOperator.EXISTS: return True if operator in _NUMBER_OPS: actual_n = _as_number(actual) expected_n = _as_number(expected) if actual_n is None or expected_n is None: return None if operator == ConstraintOperator.GE: return actual_n >= expected_n if operator == ConstraintOperator.LE: return actual_n <= expected_n if operator == ConstraintOperator.GT: return actual_n > expected_n return actual_n < expected_n if operator == ConstraintOperator.EQ: return _same_scalar(actual, expected, value_type) if operator == ConstraintOperator.NE: equal = _same_scalar(actual, expected, value_type) return None if equal is None else (not equal) if operator == ConstraintOperator.IN: if not isinstance(expected, list) or not expected: return None matches = [_same_scalar(actual, item, value_type) for item in expected] if any(item is None for item in matches): return None return any(matches) if operator == ConstraintOperator.CONTAINS: if not isinstance(actual, list) or not isinstance(expected, str): return None if any(not isinstance(item, str) for item in actual): return None return expected in actual return None