ds_quickbooks/ai_rules.py

from dataclasses import dataclass
from typing import Any, Dict, List

from models import Receipt, Transaction
from 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)