"""Shared rule evaluation for posts and comments.""" from typing import Any, Dict def get_nested_value(obj: Dict[str, Any], path: str) -> Any: """Get a nested dict value using dot notation.""" value = obj for key in path.split("."): if isinstance(value, dict) and key in value: value = value[key] else: return None return value def evaluate_rule(value: Any, operator: str, target: Any) -> bool: """Evaluate one rule operator.""" if value is None: return False if operator == "equals": return value == target if operator == "not_equals": return value != target if operator == "in": return value in target if operator == "not_in": return value not in target if operator == "min": return value >= target if operator == "max": return value <= target if operator == "after": return value > target if operator == "before": return value < target if operator == "contains": return target in value if operator == "excludes": if isinstance(value, list): return not any(item in target for item in value) return value not in target if operator == "includes": if isinstance(value, list): return target in value return False if operator == "includes_any": if isinstance(value, list) and isinstance(target, list): for item in value: if isinstance(item, dict): for rule in target: if ( isinstance(rule, dict) and item.get("topic") == rule.get("topic") and item.get("confidence", 0) >= rule.get("confidence_min", 0) ): return True elif item in target: return True return False if operator == "min_length": return len(str(value)) >= target if operator == "max_length": return len(str(value)) <= target return False def apply_rules(item: Dict[str, Any], rules: Dict[str, Dict[str, Any]]) -> bool: """Return True when all field rules pass.""" if not rules: return True for field_path, rule_def in rules.items(): value = get_nested_value(item, field_path) for operator, target in rule_def.items(): if not evaluate_rule(value, operator, target): return False return True