from dataclasses import dataclass from typing import Dict, Any, List import config from models import Receipt, Transaction @dataclass class AIRule: name: str condition: str action: str source: str status: str = "active" class AIRulesEngine: def __init__(self): self.rules: List[AIRule] = [] self._load_default_rules() def _load_default_rules(self): self.rules = [ AIRule("exact_amount_match", "amount_diff <= 0.01", "auto_approve", "system"), AIRule("same_vendor_same_date", "vendor_match and date_diff <= 1", "high_confidence", "system"), AIRule("gas_station_pattern", "vendor_contains_gas_or_fuel", "categorize_transport", "system") ] def apply_rules(self, receipt: Receipt, transaction: Transaction) -> Dict[str, Any]: results = {"auto_approve": False, "confidence_boost": 0, "category": None} for rule in self.rules: if rule.status != "active": continue if self._evaluate_condition(rule.condition, receipt, transaction): self._execute_action(rule.action, results, receipt, transaction) return results def _evaluate_condition(self, condition: str, receipt: Receipt, transaction: Transaction) -> bool: """Safely evaluate rule conditions without using eval()""" amount_diff = abs(receipt.amount - abs(transaction.amount)) date_diff = abs((receipt.receipt_date - transaction.transaction_date).days) vendor_match = receipt.vendor.lower() in transaction.vendor.lower() or transaction.vendor.lower() in receipt.vendor.lower() vendor_lower = receipt.vendor.lower() vendor_contains_gas_or_fuel = 'gas' in vendor_lower or 'fuel' in vendor_lower # Handle specific condition types safely if condition == "amount_diff <= 0.01": return amount_diff <= 0.01 elif condition == "vendor_match and date_diff <= 1": return vendor_match and date_diff <= 1 elif condition == "vendor_contains_gas_or_fuel": return vendor_contains_gas_or_fuel else: # For any other conditions, try to evaluate them safely try: # Only allow safe operations safe_globals = { "amount_diff": amount_diff, "date_diff": date_diff, "vendor_match": vendor_match, "vendor_contains_gas_or_fuel": vendor_contains_gas_or_fuel, "receipt": receipt, "transaction": transaction, "abs": abs, "len": len, "min": min, "max": max, "sum": sum, "round": round } return eval(condition, safe_globals, {}) except (SyntaxError, NameError, TypeError) as e: print(f"Warning: Invalid condition '{condition}': {e}") return False def _execute_action(self, action: str, results: Dict[str, Any], receipt: Receipt, transaction: Transaction): if action == "auto_approve": results["auto_approve"] = True elif action == "high_confidence": results["confidence_boost"] += 0.2 elif action == "categorize_transport": results["category"] = "Transportation" def add_rule(self, rule: AIRule): self.rules.append(rule) def remove_rule(self, rule_name: str): self.rules = [r for r in self.rules if r.name != rule_name]