bolade
1f530da7c4
- Reorganized imports in main.py for better readability and structure. - Enhanced logging configuration and added more detailed log messages throughout the application. - Improved error handling and response formatting in transaction import endpoints. - Streamlined transaction processing logic for CSV and image uploads. - Updated matching engine to enhance match results with rules and improved logging. - Refactored tax rules engine for better clarity and maintainability. - Cleaned up requirements.txt by removing specific versioning for easier dependency management.
277 lines
9.8 KiB
Python
277 lines
9.8 KiB
Python
import logging
|
|
from typing import Any, Dict, Optional
|
|
|
|
from models import Address, Asset, Receipt, Transaction
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TaxRulesEngine:
|
|
"""Engine to handle tax calculations based on the four tax rules"""
|
|
|
|
# Provincial tax rates (simplified - in production, use a tax rate API)
|
|
PROVINCIAL_TAX_RATES = {
|
|
"ON": 0.13, # Ontario HST
|
|
"QC": 0.14975, # Quebec QST
|
|
"BC": 0.12, # British Columbia
|
|
"AB": 0.05, # Alberta
|
|
"SK": 0.11, # Saskatchewan
|
|
"MB": 0.12, # Manitoba
|
|
"NS": 0.15, # Nova Scotia
|
|
"NB": 0.15, # New Brunswick
|
|
"NL": 0.15, # Newfoundland and Labrador
|
|
"PE": 0.15, # Prince Edward Island
|
|
"NT": 0.05, # Northwest Territories
|
|
"NU": 0.05, # Nunavut
|
|
"YT": 0.05, # Yukon
|
|
}
|
|
|
|
def __init__(self):
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
def apply_sales_tax_rule(self, receipt: Receipt) -> Dict[str, Any]:
|
|
"""
|
|
Sales Tax Rule: Apply correct sales tax based on billing vs shipping addresses
|
|
"""
|
|
try:
|
|
# Determine which address to use for tax calculation
|
|
tax_address = self._get_tax_address(receipt)
|
|
|
|
if not tax_address:
|
|
return {
|
|
"success": False,
|
|
"error": "No valid address found for tax calculation",
|
|
"calculated_tax": 0.0,
|
|
"tax_rate": 0.0,
|
|
}
|
|
|
|
# Get tax rate for the province
|
|
tax_rate = self.PROVINCIAL_TAX_RATES.get(tax_address.province, 0.0)
|
|
|
|
# Calculate tax amount
|
|
calculated_tax = receipt.amount * tax_rate
|
|
|
|
return {
|
|
"success": True,
|
|
"calculated_tax": calculated_tax,
|
|
"tax_rate": tax_rate,
|
|
"tax_address": tax_address.province,
|
|
"rule_applied": "Sales Tax Rule",
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error applying sales tax rule: {str(e)}")
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"calculated_tax": 0.0,
|
|
"tax_rate": 0.0,
|
|
}
|
|
|
|
def _get_tax_address(self, receipt: Receipt) -> Optional[Address]:
|
|
"""Determine which address to use for tax calculation"""
|
|
# Rule: Use shipping address if different from billing, otherwise use billing
|
|
if receipt.shipping_address and receipt.billing_address:
|
|
if self._addresses_different(
|
|
receipt.billing_address, receipt.shipping_address
|
|
):
|
|
return receipt.shipping_address
|
|
else:
|
|
return receipt.billing_address
|
|
elif receipt.shipping_address:
|
|
return receipt.shipping_address
|
|
elif receipt.billing_address:
|
|
return receipt.billing_address
|
|
else:
|
|
return None
|
|
|
|
def _addresses_different(self, billing: Address, shipping: Address) -> bool:
|
|
"""Check if billing and shipping addresses are different"""
|
|
return (
|
|
billing.province != shipping.province
|
|
or billing.city != shipping.city
|
|
or billing.postal_code != shipping.postal_code
|
|
)
|
|
|
|
def apply_fx_rule(
|
|
self, receipt: Receipt, transaction: Transaction
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Foreign Exchange Rule: Handle currency mismatches
|
|
"""
|
|
try:
|
|
# Check for currency mismatch
|
|
if receipt.currency != transaction.currency:
|
|
fx_discrepancy = abs(receipt.amount - abs(transaction.amount))
|
|
|
|
return {
|
|
"success": True,
|
|
"fx_discrepancy": fx_discrepancy,
|
|
"receipt_currency": receipt.currency,
|
|
"transaction_currency": transaction.currency,
|
|
"receipt_amount": receipt.amount,
|
|
"transaction_amount": abs(transaction.amount),
|
|
"requires_manual_review": True,
|
|
"rule_applied": "Foreign Exchange Rule",
|
|
}
|
|
else:
|
|
return {
|
|
"success": True,
|
|
"fx_discrepancy": 0.0,
|
|
"requires_manual_review": False,
|
|
"rule_applied": "No FX Rule (same currency)",
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error applying FX rule: {str(e)}")
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"fx_discrepancy": 0.0,
|
|
"requires_manual_review": False,
|
|
}
|
|
|
|
def calculate_straight_line_depreciation(
|
|
self, asset: Asset, year: int
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Straight-Line Depreciation for accounting purposes
|
|
"""
|
|
try:
|
|
if year > asset.useful_life_years:
|
|
return {
|
|
"success": False,
|
|
"error": f"Year {year} exceeds useful life of {asset.useful_life_years} years",
|
|
"depreciation": 0.0,
|
|
}
|
|
|
|
# Straight-line formula: (Cost - Residual Value) / Useful Life
|
|
annual_depreciation = (
|
|
asset.purchase_amount - asset.residual_value
|
|
) / asset.useful_life_years
|
|
|
|
return {
|
|
"success": True,
|
|
"depreciation": annual_depreciation,
|
|
"book_value": asset.purchase_amount - (annual_depreciation * year),
|
|
"method": "Straight-Line",
|
|
"rule_applied": "Depreciation Rule (Accounting)",
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error calculating straight-line depreciation: {str(e)}")
|
|
return {"success": False, "error": str(e), "depreciation": 0.0}
|
|
|
|
def calculate_cca_depreciation(self, asset: Asset, year: int) -> Dict[str, Any]:
|
|
"""
|
|
CCA (Capital Cost Allowance) Depreciation for tax purposes
|
|
"""
|
|
try:
|
|
if year < 1:
|
|
return {
|
|
"success": False,
|
|
"error": "Year must be at least 1",
|
|
"depreciation": 0.0,
|
|
}
|
|
|
|
# CCA uses declining balance method
|
|
book_value = asset.purchase_amount
|
|
total_depreciation = 0.0
|
|
|
|
for current_year in range(1, year + 1):
|
|
# CCA is calculated on the declining balance
|
|
cca_amount = book_value * asset.cca_rate
|
|
book_value -= cca_amount
|
|
total_depreciation += cca_amount
|
|
|
|
# Stop if book value reaches residual value
|
|
if book_value <= asset.residual_value:
|
|
break
|
|
|
|
return {
|
|
"success": True,
|
|
"depreciation": cca_amount, # Current year depreciation
|
|
"total_depreciation": total_depreciation,
|
|
"book_value": max(book_value, asset.residual_value),
|
|
"method": "CCA Declining Balance",
|
|
"rule_applied": "Depreciation Rule (Tax)",
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error calculating CCA depreciation: {str(e)}")
|
|
return {"success": False, "error": str(e), "depreciation": 0.0}
|
|
|
|
def apply_meals_entertainment_rule(self, receipt: Receipt) -> Dict[str, Any]:
|
|
"""
|
|
Meals & Entertainment Tax Deduction Rule
|
|
"""
|
|
try:
|
|
if not receipt.is_meals_entertainment:
|
|
return {
|
|
"success": True,
|
|
"tax_deduction": receipt.amount,
|
|
"accounting_deduction": receipt.amount,
|
|
"rule_applied": "No M&E Rule (not meals/entertainment)",
|
|
}
|
|
|
|
# For tax purposes: 50% deductible
|
|
tax_deduction = receipt.amount * 0.5
|
|
|
|
# For accounting purposes: 100% deductible
|
|
accounting_deduction = receipt.amount
|
|
|
|
# Sales tax is fully deductible for accounting
|
|
tax_on_meal = receipt.tax
|
|
|
|
return {
|
|
"success": True,
|
|
"tax_deduction": tax_deduction,
|
|
"accounting_deduction": accounting_deduction,
|
|
"tax_on_meal": tax_on_meal,
|
|
"rule_applied": "Meals & Entertainment Rule",
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error applying meals & entertainment rule: {str(e)}")
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"tax_deduction": 0.0,
|
|
"accounting_deduction": 0.0,
|
|
}
|
|
|
|
def apply_all_tax_rules(
|
|
self, receipt: Receipt, transaction: Transaction = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Apply all tax rules to a receipt
|
|
"""
|
|
results = {
|
|
"receipt_id": receipt.id,
|
|
"rules_applied": [],
|
|
"sales_tax": {},
|
|
"fx_analysis": {},
|
|
"meals_entertainment": {},
|
|
}
|
|
|
|
# Apply Sales Tax Rule
|
|
sales_tax_result = self.apply_sales_tax_rule(receipt)
|
|
results["sales_tax"] = sales_tax_result
|
|
if sales_tax_result["success"]:
|
|
results["rules_applied"].append("Sales Tax Rule")
|
|
|
|
# Apply FX Rule (if transaction provided)
|
|
if transaction:
|
|
fx_result = self.apply_fx_rule(receipt, transaction)
|
|
results["fx_analysis"] = fx_result
|
|
if fx_result["success"]:
|
|
results["rules_applied"].append("Foreign Exchange Rule")
|
|
|
|
# Apply Meals & Entertainment Rule
|
|
me_result = self.apply_meals_entertainment_rule(receipt)
|
|
results["meals_entertainment"] = me_result
|
|
if me_result["success"]:
|
|
results["rules_applied"].append("Meals & Entertainment Rule")
|
|
|
|
return results
|