from dataclasses import dataclass from typing import Any, Dict, List from schemas import Receipt, Transaction from services.tax_rules_engine import TaxRulesEngine @dataclass class AIRule: name: str condition: str action: str source: str status: str = "active" class AIRulesEngine: def __init__(self): self.rules: List[AIRule] = [] self.tax_rules_engine = TaxRulesEngine() 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", ), # Tax-related rules AIRule( "fx_currency_mismatch", "currency_mismatch", "flag_fx_review", "tax_system", ), AIRule( "meals_entertainment", "is_meals_entertainment", "apply_me_tax_rule", "tax_system", ), AIRule( "provincial_tax_calculation", "has_address_info", "calculate_provincial_tax", "tax_system", ), ] def apply_rules(self, receipt: Receipt, transaction: Transaction) -> Dict[str, Any]: results = { "auto_approve": False, "confidence_boost": 0, "category": None, "tax_analysis": {}, } 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 # Tax-related conditions currency_mismatch = receipt.currency != transaction.currency is_meals_entertainment = receipt.is_meals_entertainment has_address_info = ( receipt.billing_address is not None or receipt.shipping_address is not None ) # 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 elif condition == "currency_mismatch": return currency_mismatch elif condition == "is_meals_entertainment": return is_meals_entertainment elif condition == "has_address_info": return has_address_info 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, "currency_mismatch": currency_mismatch, "is_meals_entertainment": is_meals_entertainment, "has_address_info": has_address_info, "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" elif action == "flag_fx_review": # Apply FX rule and flag for review fx_result = self.tax_rules_engine.apply_fx_rule(receipt, transaction) results["tax_analysis"]["fx"] = fx_result if fx_result.get("requires_manual_review", False): results["confidence_boost"] -= 0.1 # Reduce confidence for FX issues elif action == "apply_me_tax_rule": # Apply meals & entertainment rule me_result = self.tax_rules_engine.apply_meals_entertainment_rule(receipt) results["tax_analysis"]["meals_entertainment"] = me_result elif action == "calculate_provincial_tax": # Calculate provincial tax tax_result = self.tax_rules_engine.apply_sales_tax_rule(receipt) results["tax_analysis"]["sales_tax"] = tax_result 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] def apply_tax_rules( self, receipt: Receipt, transaction: Transaction = None ) -> Dict[str, Any]: """Apply all tax rules to a receipt/transaction pair""" return self.tax_rules_engine.apply_all_tax_rules(receipt, transaction)