Spaces:
Sleeping
Sleeping
| # backend/app/api/process.py | |
| from fastapi import APIRouter, HTTPException, Depends | |
| from sqlalchemy.orm import Session | |
| from typing import List | |
| from app.models.schemas import Transaction, ProcessResponse, ProcessRequest | |
| from app.ai.orchestrator import process_transactions | |
| from app.db.database import get_db | |
| from app.db.models import JournalEntry, LineItem, Account | |
| from app.api.auth import get_current_user | |
| from app.db.models import User | |
| router = APIRouter() | |
| import hashlib | |
| from app.api.auth import get_current_user_id | |
| from app.api.rate_limiter import process_rate_limiter | |
| async def trigger_ai_processing( | |
| req: ProcessRequest, | |
| db: Session = Depends(get_db), # <-- 1. This opens the connection to Supabase | |
| user: User = Depends(get_current_user) # <-- 2. Get the authenticated user | |
| ): | |
| """Takes parsed transactions, runs them through the AI, and saves them to the Vault.""" | |
| user_id = user.user_id # Extract the ID string | |
| try: | |
| # 1. Filter out duplicates BEFORE running expensive AI | |
| unique_transactions = [] | |
| skipped_count = 0 | |
| seen_hashes = set() # Track hashes we've seen in THIS file | |
| for txn in req.transactions: | |
| # Create a stable fingerprint locked to the user | |
| raw_string = f"{user_id}|{txn.date}|{txn.description}|{txn.amount}" | |
| tx_hash = hashlib.sha256(raw_string.encode()).hexdigest() | |
| # Check if we already processed this exact row in THIS file | |
| if tx_hash in seen_hashes: | |
| skipped_count += 1 | |
| continue | |
| # Check if hash already exists in DB from a previous upload | |
| exists = db.query(JournalEntry).filter(JournalEntry.tx_hash == tx_hash).first() | |
| if exists: | |
| skipped_count += 1 | |
| else: | |
| seen_hashes.add(tx_hash) | |
| txn.tx_hash = tx_hash | |
| unique_transactions.append(txn) | |
| print(f"🔄 Deduplication: Skipped {skipped_count} existing transactions. Processing {len(unique_transactions)} new ones.") | |
| if not unique_transactions: | |
| return { | |
| "message": f"Ignored {skipped_count} duplicates. No new transactions to process.", | |
| "total_processed": 0, | |
| "results": [] | |
| } | |
| # 2. Run the 5-Tier Categorization Engine | |
| categorized_results = process_transactions(unique_transactions, user_id, db) | |
| # 3. The Immutable Audit Trail (Double-Entry) | |
| print("Saving AI decisions to Supabase Vault...") | |
| # Grab the user's source account | |
| source_account = db.query(Account).filter( | |
| Account.id == req.source_account_id, | |
| Account.user_id == user_id | |
| ).first() | |
| if not source_account: | |
| raise HTTPException(status_code=400, detail="Invalid source account selected.") | |
| # Grab fallback accounts | |
| default_expense_account = db.query(Account).filter(Account.user_id == user_id, Account.account_type == "EXPENSE").first() | |
| default_revenue_account = db.query(Account).filter(Account.user_id == user_id, Account.account_type == "REVENUE").first() | |
| # Find credit card account for transfers | |
| credit_card_account = db.query(Account).filter(Account.user_id == user_id, Account.name == "Credit Card").first() | |
| checking_account = db.query(Account).filter(Account.user_id == user_id, Account.name == "Checking Account").first() | |
| for result in categorized_results: | |
| raw = result.get("raw_transaction", {}) | |
| desc_lower = raw.get("description", "").lower() | |
| abs_amount = abs(float(raw.get("amount", 0.0))) | |
| # --- TRANSFER DETECTION LOGIC --- | |
| is_money_leaving = float(raw.get("amount", 0.0)) < 0 | |
| is_transfer = any(keyword in desc_lower for keyword in ["credit card payment", "autopay", "brex pay", "amex pay", "chase cc"]) | |
| if is_transfer and credit_card_account and checking_account: | |
| print(f"🔄 Transfer Detected: {raw.get('description')} - Bypassing AI Categorizer") | |
| # A payment is a transfer from Checking to Credit Card | |
| # DR Credit Card (Liability decreases) | |
| # CR Checking Account (Asset decreases) | |
| journal_entry = JournalEntry( | |
| user_id=user_id, | |
| tx_hash=raw.get("tx_hash"), | |
| date=raw.get("date"), | |
| description=raw.get("description"), | |
| status="categorized" | |
| ) | |
| db.add(journal_entry) | |
| db.flush() | |
| debit_leg = LineItem( | |
| journal_entry_id=journal_entry.id, | |
| account_id=credit_card_account.id, | |
| amount=abs_amount, | |
| entry_type="DEBIT" | |
| ) | |
| credit_leg = LineItem( | |
| journal_entry_id=journal_entry.id, | |
| account_id=checking_account.id, | |
| amount=abs_amount, | |
| entry_type="CREDIT" | |
| ) | |
| db.add(debit_leg) | |
| db.add(credit_leg) | |
| continue # Skip normal expense/revenue processing | |
| # --- NORMAL EXPENSE/REVENUE LOGIC --- | |
| # Pre-AI guard: amount sign is deterministic truth — don't let AI override it | |
| if not is_money_leaving and source_account.account_type == "ASSET": | |
| # Positive amount into a checking account = Revenue, bypass AI entirely | |
| print(f"💰 Revenue Detected: {raw.get('description')} — routing to SaaS Subscriptions") | |
| target_account = default_revenue_account | |
| else: | |
| predicted_account_name = result.get("account_name", "Software Subscriptions") | |
| target_account = db.query(Account).filter( | |
| Account.user_id == user_id, | |
| Account.name == predicted_account_name | |
| ).first() | |
| # Smart fallback: if AI returned an unknown account name | |
| if not target_account: | |
| target_account = default_expense_account if is_money_leaving else default_revenue_account | |
| print(f"⚠️ Account '{predicted_account_name}' not found — falling back to {'EXPENSE' if is_money_leaving else 'REVENUE'} default") | |
| # 1. Create the Receipt Header | |
| journal_entry = JournalEntry( | |
| user_id=user_id, | |
| tx_hash=raw.get("tx_hash"), | |
| date=raw.get("date"), | |
| description=raw.get("description"), | |
| status=result.get("status", "categorized"), | |
| ambiguity_reason=result.get("ambiguity_reason") | |
| ) | |
| db.add(journal_entry) | |
| db.flush() # Force PostgreSQL to generate the ID so we can attach the legs! | |
| # 2. Create the Legs (Debits and Credits) | |
| # The direction of the flow is strictly dictated by whether money | |
| # is entering (> 0) or leaving (< 0) the source account. | |
| debit_leg = LineItem( | |
| journal_entry_id=journal_entry.id, | |
| account_id=target_account.id if is_money_leaving else source_account.id, | |
| amount=abs_amount, | |
| entry_type="DEBIT" | |
| ) | |
| credit_leg = LineItem( | |
| journal_entry_id=journal_entry.id, | |
| account_id=source_account.id if is_money_leaving else target_account.id, | |
| amount=abs_amount, | |
| entry_type="CREDIT" | |
| ) | |
| db.add(debit_leg) | |
| db.add(credit_leg) | |
| # Push all records to Supabase securely | |
| db.commit() | |
| print("✅ Successfully saved to Supabase!") | |
| return { | |
| "message": f"Processing complete. Ignored {skipped_count} duplicates.", | |
| "total_processed": len(categorized_results), | |
| "results": categorized_results | |
| } | |
| except Exception as e: | |
| import traceback | |
| import logging | |
| # Log internally only — never expose stack trace to client | |
| logging.error("AI Processing Error:\n" + traceback.format_exc()) | |
| db.rollback() | |
| raise HTTPException(status_code=500, detail="An internal error occurred during processing. Please try again.") |