bolade
126 lines
5.8 KiB
Python
126 lines
5.8 KiB
Python
from dataclasses import dataclass
|
|
from typing import Dict, Any, List
|
|
import config
|
|
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) |